process-cursor-concurrently/index.test.js at master · dflupu/process-cursor-concurrently · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
const test = require('ava')

const {MongoClient} = require('mongodb')
const {MongoMemoryServer} = require('mongodb-memory-server')

const processCursorConcurrently = require('.')

let mongoCollection

test.before(async () => {
    const mongod = new MongoMemoryServer()
    const mongoUri = await mongod.getUri()

    const mongoClient = new MongoClient(mongoUri, {useUnifiedTopology: true})
    await mongoClient.connect()

    mongoCollection = mongoClient.db().collection('test')
})

test.beforeEach(async () => {
    await mongoCollection.deleteMany({})
})

test.serial('evenly divisible number of items', async t => {
    await testWithParameters(t, {itemCount: 100, concurrency: 10})
})

test.serial('non evenly divisible number of items', async t => {
    await testWithParameters(t, {itemCount: 101, concurrency: 10})
})

test.serial('zero items', async t => {
    await testWithParameters(t, {itemCount: 0, concurrency: 10})
})

test.serial('less items than concurrency', async t => {
    await testWithParameters(t, {itemCount: 1, concurrency: 5})
})

test.serial('concurrency set to 1', async t => {
    await testWithParameters(t, {itemCount: 10, concurrency: 1})
})

test.serial('sync function that throws', async t => {
    const itemCount = 100
    const cursor = await insertAndGetCursor(itemCount)

    let calls = 0

    await t.throwsAsync(processCursorConcurrently(cursor, _ => {
        calls += 1
        throw new Error('test')
    }, {concurrency: 10}))

    t.is(calls, 10)
})

test.serial('async function that throws', async t => {
    const itemCount = 100
    const cursor = await insertAndGetCursor(itemCount)

    let calls = 0

    await t.throwsAsync(processCursorConcurrently(cursor, async _ => {
        calls += 1

        return new Promise((resolve, reject) => {
            setTimeout(reject(new Error('test')), 1)
        })
    }, {concurrency: 10}))

    t.is(calls, 10)
})

test.serial('onProgress function is called', async t => {
    const progressCalls = []

    await testWithParameters(t, {
        itemCount: 101,
        concurrency: 10,
        onProgress: countProcessed => progressCalls.push(countProcessed)
    })

    t.is(progressCalls.length, 101)

    for (let i = 1; i <= 101; i++) {
        progressCalls.includes(i)
    }
})

async function testWithParameters(t, {itemCount, concurrency, onProgress}) {
    const cursor = await insertAndGetCursor(itemCount)
    const results = []

    await processCursorConcurrently(cursor, item => results.push(item), {concurrency, onProgress})
    t.is(results.length, itemCount)
}

async function insertAndGetCursor(itemCount) {
    for (let i = 0; i < itemCount; i++) {
        await mongoCollection.insertOne({_id: i})
    }

    return mongoCollection.find()
}