123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221 |
- '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('drained', async function (t) {
- const queue = buildQueue(worker, 2)
- const toExec = new Array(10).fill(10)
- let count = 0
- async function worker (arg) {
- await sleep(arg)
- count++
- }
- toExec.forEach(function (i) {
- queue.push(i)
- })
- await queue.drained()
- t.equal(count, toExec.length)
- toExec.forEach(function (i) {
- queue.push(i)
- })
- await queue.drained()
- t.equal(count, toExec.length * 2)
- })
- test('drained with exception should not throw', async function (t) {
- const queue = buildQueue(worker, 2)
- const toExec = new Array(10).fill(10)
- async function worker () {
- throw new Error('foo')
- }
- toExec.forEach(function (i) {
- queue.push(i)
- })
- await queue.drained()
- })
- test('drained with drain function', async function (t) {
- let drainCalled = false
- const queue = buildQueue(worker, 2)
- queue.drain = function () {
- drainCalled = true
- }
- const toExec = new Array(10).fill(10)
- let count = 0
- async function worker (arg) {
- await sleep(arg)
- count++
- }
- toExec.forEach(function () {
- queue.push()
- })
- await queue.drained()
- t.equal(count, toExec.length)
- t.equal(drainCalled, true)
- })
- 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)
- })
|