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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
'use strict'
const test = require('tape')
const buildQueue = require('../').promise
const { promisify } = require('util')
const sleep = promisify(setTimeout)
const immediate = promisify(setImmediate)
test('concurrency', function (t) {
t.plan(2)
t.throws(buildQueue.bind(null, worker, 0))
t.doesNotThrow(buildQueue.bind(null, worker, 1))
async function worker (arg) {
return true
}
})
test('worker execution', async function (t) {
const queue = buildQueue(worker, 1)
const result = await queue.push(42)
t.equal(result, true, 'result matches')
async function worker (arg) {
t.equal(arg, 42)
return true
}
})
test('limit', async function (t) {
const queue = buildQueue(worker, 1)
const [res1, res2] = await Promise.all([queue.push(10), queue.push(0)])
t.equal(res1, 10, 'the result matches')
t.equal(res2, 0, 'the result matches')
async function worker (arg) {
await sleep(arg)
return arg
}
})
test('multiple executions', async function (t) {
const queue = buildQueue(worker, 1)
const toExec = [1, 2, 3, 4, 5]
const expected = ['a', 'b', 'c', 'd', 'e']
let count = 0
await Promise.all(toExec.map(async function (task, i) {
const result = await queue.push(task)
t.equal(result, expected[i], 'the result matches')
}))
async function worker (arg) {
t.equal(arg, toExec[count], 'arg matches')
return expected[count++]
}
})
test('set this', async function (t) {
t.plan(1)
const that = {}
const queue = buildQueue(that, worker, 1)
await queue.push(42)
async function worker (arg) {
t.equal(this, that, 'this matches')
}
})
test('unshift', async function (t) {
const queue = buildQueue(worker, 1)
const expected = [1, 2, 3, 4]
await Promise.all([
queue.push(1),
queue.push(4),
queue.unshift(3),
queue.unshift(2)
])
t.is(expected.length, 0)
async function worker (arg) {
t.equal(expected.shift(), arg, 'tasks come in order')
}
})
test('push with worker throwing error', async function (t) {
t.plan(5)
const q = buildQueue(async function (task, cb) {
throw new Error('test error')
}, 1)
q.error(function (err, task) {
t.ok(err instanceof Error, 'global error handler should catch the error')
t.match(err.message, /test error/, 'error message should be "test error"')
t.equal(task, 42, 'The task executed should be passed')
})
try {
await q.push(42)
} catch (err) {
t.ok(err instanceof Error, 'push callback should catch the error')
t.match(err.message, /test error/, 'error message should be "test error"')
}
})
test('unshift with worker throwing error', async function (t) {
t.plan(2)
const q = buildQueue(async function (task, cb) {
throw new Error('test error')
}, 1)
try {
await q.unshift(42)
} catch (err) {
t.ok(err instanceof Error, 'push callback should catch the error')
t.match(err.message, /test error/, 'error message should be "test error"')
}
})
test('no unhandledRejection (push)', async function (t) {
function handleRejection () {
t.fail('unhandledRejection')
}
process.once('unhandledRejection', handleRejection)
const q = buildQueue(async function (task, cb) {
throw new Error('test error')
}, 1)
q.push(42)
await immediate()
process.removeListener('unhandledRejection', handleRejection)
})
test('no unhandledRejection (unshift)', async function (t) {
function handleRejection () {
t.fail('unhandledRejection')
}
process.once('unhandledRejection', handleRejection)
const q = buildQueue(async function (task, cb) {
throw new Error('test error')
}, 1)
q.unshift(42)
await immediate()
process.removeListener('unhandledRejection', handleRejection)
})
|
