strong-tie/inbound-calls
0
1'use strict'2 3/* eslint-disable no-var */4 5var test = require('tape')6var buildQueue = require('../')7 8test('concurrency', function (t) {9 t.plan(6)10 t.throws(buildQueue.bind(null, worker, 0))11 t.throws(buildQueue.bind(null, worker, NaN))12 t.doesNotThrow(buildQueue.bind(null, worker, 1))13 14 var queue = buildQueue(worker, 1)15 t.throws(function () {16 queue.concurrency = 017 })18 t.throws(function () {19 queue.concurrency = NaN20 })21 t.doesNotThrow(function () {22 queue.concurrency = 223 })24 25 function worker (arg, cb) {26 cb(null, true)27 }28})29 30test('worker execution', function (t) {31 t.plan(3)32 33 var queue = buildQueue(worker, 1)34 35 queue.push(42, function (err, result) {36 t.error(err, 'no error')37 t.equal(result, true, 'result matches')38 })39 40 function worker (arg, cb) {41 t.equal(arg, 42)42 cb(null, true)43 }44})45 46test('limit', function (t) {47 t.plan(4)48 49 var expected = [10, 0]50 var queue = buildQueue(worker, 1)51 52 queue.push(10, result)53 queue.push(0, result)54 55 function result (err, arg) {56 t.error(err, 'no error')57 t.equal(arg, expected.shift(), 'the result matches')58 }59 60 function worker (arg, cb) {61 setTimeout(cb, arg, null, arg)62 }63})64 65test('multiple executions', function (t) {66 t.plan(15)67 68 var queue = buildQueue(worker, 1)69 var toExec = [1, 2, 3, 4, 5]70 var count = 071 72 toExec.forEach(function (task) {73 queue.push(task, done)74 })75 76 function done (err, result) {77 t.error(err, 'no error')78 t.equal(result, toExec[count - 1], 'the result matches')79 }80 81 function worker (arg, cb) {82 t.equal(arg, toExec[count], 'arg matches')83 count++84 setImmediate(cb, null, arg)85 }86})87 88test('multiple executions, one after another', function (t) {89 t.plan(15)90 91 var queue = buildQueue(worker, 1)92 var toExec = [1, 2, 3, 4, 5]93 var count = 094 95 queue.push(toExec[0], done)96 97 function done (err, result) {98 t.error(err, 'no error')99 t.equal(result, toExec[count - 1], 'the result matches')100 if (count < toExec.length) {101 queue.push(toExec[count], done)102 }103 }104 105 function worker (arg, cb) {106 t.equal(arg, toExec[count], 'arg matches')107 count++108 setImmediate(cb, null, arg)109 }110})111 112test('set this', function (t) {113 t.plan(3)114 115 var that = {}116 var queue = buildQueue(that, worker, 1)117 118 queue.push(42, function (err, result) {119 t.error(err, 'no error')120 t.equal(this, that, 'this matches')121 })122 123 function worker (arg, cb) {124 t.equal(this, that, 'this matches')125 cb(null, true)126 }127})128 129test('drain', function (t) {130 t.plan(4)131 132 var queue = buildQueue(worker, 1)133 var worked = false134 135 queue.push(42, function (err, result) {136 t.error(err, 'no error')137 t.equal(result, true, 'result matches')138 })139 140 queue.drain = function () {141 t.equal(true, worked, 'drained')142 }143 144 function worker (arg, cb) {145 t.equal(arg, 42)146 worked = true147 setImmediate(cb, null, true)148 }149})150 151test('pause && resume', function (t) {152 t.plan(13)153 154 var queue = buildQueue(worker, 1)155 var worked = false156 var expected = [42, 24]157 158 t.notOk(queue.paused, 'it should not be paused')159 160 queue.pause()161 162 queue.push(42, function (err, result) {163 t.error(err, 'no error')164 t.equal(result, true, 'result matches')165 })166 167 queue.push(24, function (err, result) {168 t.error(err, 'no error')169 t.equal(result, true, 'result matches')170 })171 172 t.notOk(worked, 'it should be paused')173 t.ok(queue.paused, 'it should be paused')174 175 queue.resume()176 queue.pause()177 queue.resume()178 queue.resume() // second resume is a no-op179 180 function worker (arg, cb) {181 t.notOk(queue.paused, 'it should not be paused')182 t.ok(queue.running() <= queue.concurrency, 'should respect the concurrency')183 t.equal(arg, expected.shift())184 worked = true185 process.nextTick(function () { cb(null, true) })186 }187})188 189test('pause in flight && resume', function (t) {190 t.plan(16)191 192 var queue = buildQueue(worker, 1)193 var expected = [42, 24, 12]194 195 t.notOk(queue.paused, 'it should not be paused')196 197 queue.push(42, function (err, result) {198 t.error(err, 'no error')199 t.equal(result, true, 'result matches')200 t.ok(queue.paused, 'it should be paused')201 process.nextTick(function () {202 queue.resume()203 queue.pause()204 queue.resume()205 })206 })207 208 queue.push(24, function (err, result) {209 t.error(err, 'no error')210 t.equal(result, true, 'result matches')211 t.notOk(queue.paused, 'it should not be paused')212 })213 214 queue.push(12, function (err, result) {215 t.error(err, 'no error')216 t.equal(result, true, 'result matches')217 t.notOk(queue.paused, 'it should not be paused')218 })219 220 queue.pause()221 222 function worker (arg, cb) {223 t.ok(queue.running() <= queue.concurrency, 'should respect the concurrency')224 t.equal(arg, expected.shift())225 process.nextTick(function () { cb(null, true) })226 }227})228 229test('altering concurrency', function (t) {230 t.plan(24)231 232 var queue = buildQueue(worker, 1)233 234 queue.push(24, workDone)235 queue.push(24, workDone)236 queue.push(24, workDone)237 238 queue.pause()239 240 queue.concurrency = 3 // concurrency changes are ignored while paused241 queue.concurrency = 2242 243 queue.resume()244 245 t.equal(queue.running(), 2, '2 jobs running')246 247 queue.concurrency = 3248 249 t.equal(queue.running(), 3, '3 jobs running')250 251 queue.concurrency = 1252 253 t.equal(queue.running(), 3, '3 jobs running') // running jobs can't be killed254 255 queue.push(24, workDone)256 queue.push(24, workDone)257 queue.push(24, workDone)258 queue.push(24, workDone)259 260 function workDone (err, result) {261 t.error(err, 'no error')262 t.equal(result, true, 'result matches')263 }264 265 function worker (arg, cb) {266 t.ok(queue.running() <= queue.concurrency, 'should respect the concurrency')267 setImmediate(function () {268 cb(null, true)269 })270 }271})272 273test('idle()', function (t) {274 t.plan(12)275 276 var queue = buildQueue(worker, 1)277 278 t.ok(queue.idle(), 'queue is idle')279 280 queue.push(42, function (err, result) {281 t.error(err, 'no error')282 t.equal(result, true, 'result matches')283 t.notOk(queue.idle(), 'queue is not idle')284 })285 286 queue.push(42, function (err, result) {287 t.error(err, 'no error')288 t.equal(result, true, 'result matches')289 // it will go idle after executing this function290 setImmediate(function () {291 t.ok(queue.idle(), 'queue is now idle')292 })293 })294 295 t.notOk(queue.idle(), 'queue is not idle')296 297 function worker (arg, cb) {298 t.notOk(queue.idle(), 'queue is not idle')299 t.equal(arg, 42)300 setImmediate(cb, null, true)301 }302})303 304test('saturated', function (t) {305 t.plan(9)306 307 var queue = buildQueue(worker, 1)308 var preworked = 0309 var worked = 0310 311 queue.saturated = function () {312 t.pass('saturated')313 t.equal(preworked, 1, 'started 1 task')314 t.equal(worked, 0, 'worked zero task')315 }316 317 queue.push(42, done)318 queue.push(42, done)319 320 function done (err, result) {321 t.error(err, 'no error')322 t.equal(result, true, 'result matches')323 }324 325 function worker (arg, cb) {326 t.equal(arg, 42)327 preworked++328 setImmediate(function () {329 worked++330 cb(null, true)331 })332 }333})334 335test('length', function (t) {336 t.plan(7)337 338 var queue = buildQueue(worker, 1)339 340 t.equal(queue.length(), 0, 'nothing waiting')341 queue.push(42, done)342 t.equal(queue.length(), 0, 'nothing waiting')343 queue.push(42, done)344 t.equal(queue.length(), 1, 'one task waiting')345 queue.push(42, done)346 t.equal(queue.length(), 2, 'two tasks waiting')347 348 function done (err, result) {349 t.error(err, 'no error')350 }351 352 function worker (arg, cb) {353 setImmediate(function () {354 cb(null, true)355 })356 }357})358 359test('getQueue', function (t) {360 t.plan(10)361 362 var queue = buildQueue(worker, 1)363 364 t.equal(queue.getQueue().length, 0, 'nothing waiting')365 queue.push(42, done)366 t.equal(queue.getQueue().length, 0, 'nothing waiting')367 queue.push(42, done)368 t.equal(queue.getQueue().length, 1, 'one task waiting')369 t.equal(queue.getQueue()[0], 42, 'should be equal')370 queue.push(43, done)371 t.equal(queue.getQueue().length, 2, 'two tasks waiting')372 t.equal(queue.getQueue()[0], 42, 'should be equal')373 t.equal(queue.getQueue()[1], 43, 'should be equal')374 375 function done (err, result) {376 t.error(err, 'no error')377 }378 379 function worker (arg, cb) {380 setImmediate(function () {381 cb(null, true)382 })383 }384})385 386test('unshift', function (t) {387 t.plan(8)388 389 var queue = buildQueue(worker, 1)390 var expected = [1, 2, 3, 4]391 392 queue.push(1, done)393 queue.push(4, done)394 queue.unshift(3, done)395 queue.unshift(2, done)396 397 function done (err, result) {398 t.error(err, 'no error')399 }400 401 function worker (arg, cb) {402 t.equal(expected.shift(), arg, 'tasks come in order')403 setImmediate(function () {404 cb(null, true)405 })406 }407})408 409test('unshift && empty', function (t) {410 t.plan(2)411 412 var queue = buildQueue(worker, 1)413 var completed = false414 415 queue.pause()416 417 queue.empty = function () {418 t.notOk(completed, 'the task has not completed yet')419 }420 421 queue.unshift(1, done)422 423 queue.resume()424 425 function done (err, result) {426 completed = true427 t.error(err, 'no error')428 }429 430 function worker (arg, cb) {431 setImmediate(function () {432 cb(null, true)433 })434 }435})436 437test('push && empty', function (t) {438 t.plan(2)439 440 var queue = buildQueue(worker, 1)441 var completed = false442 443 queue.pause()444 445 queue.empty = function () {446 t.notOk(completed, 'the task has not completed yet')447 }448 449 queue.push(1, done)450 451 queue.resume()452 453 function done (err, result) {454 completed = true455 t.error(err, 'no error')456 }457 458 function worker (arg, cb) {459 setImmediate(function () {460 cb(null, true)461 })462 }463})464 465test('kill', function (t) {466 t.plan(5)467 468 var queue = buildQueue(worker, 1)469 var expected = [1]470 471 var predrain = queue.drain472 473 queue.drain = function drain () {474 t.fail('drain should never be called')475 }476 477 queue.push(1, done)478 queue.push(4, done)479 queue.unshift(3, done)480 queue.unshift(2, done)481 queue.kill()482 483 function done (err, result) {484 t.error(err, 'no error')485 setImmediate(function () {486 t.equal(queue.length(), 0, 'no queued tasks')487 t.equal(queue.running(), 0, 'no running tasks')488 t.equal(queue.drain, predrain, 'drain is back to default')489 })490 }491 492 function worker (arg, cb) {493 t.equal(expected.shift(), arg, 'tasks come in order')494 setImmediate(function () {495 cb(null, true)496 })497 }498})499 500test('killAndDrain', function (t) {501 t.plan(6)502 503 var queue = buildQueue(worker, 1)504 var expected = [1]505 506 var predrain = queue.drain507 508 queue.drain = function drain () {509 t.pass('drain has been called')510 }511 512 queue.push(1, done)513 queue.push(4, done)514 queue.unshift(3, done)515 queue.unshift(2, done)516 queue.killAndDrain()517 518 function done (err, result) {519 t.error(err, 'no error')520 setImmediate(function () {521 t.equal(queue.length(), 0, 'no queued tasks')522 t.equal(queue.running(), 0, 'no running tasks')523 t.equal(queue.drain, predrain, 'drain is back to default')524 })525 }526 527 function worker (arg, cb) {528 t.equal(expected.shift(), arg, 'tasks come in order')529 setImmediate(function () {530 cb(null, true)531 })532 }533})534 535test('pause && idle', function (t) {536 t.plan(11)537 538 var queue = buildQueue(worker, 1)539 var worked = false540 541 t.notOk(queue.paused, 'it should not be paused')542 t.ok(queue.idle(), 'should be idle')543 544 queue.pause()545 546 queue.push(42, function (err, result) {547 t.error(err, 'no error')548 t.equal(result, true, 'result matches')549 })550 551 t.notOk(worked, 'it should be paused')552 t.ok(queue.paused, 'it should be paused')553 t.notOk(queue.idle(), 'should not be idle')554 555 queue.resume()556 557 t.notOk(queue.paused, 'it should not be paused')558 t.notOk(queue.idle(), 'it should not be idle')559 560 function worker (arg, cb) {561 t.equal(arg, 42)562 worked = true563 process.nextTick(cb.bind(null, null, true))564 process.nextTick(function () {565 t.ok(queue.idle(), 'is should be idle')566 })567 }568})569 570test('push without cb', function (t) {571 t.plan(1)572 573 var queue = buildQueue(worker, 1)574 575 queue.push(42)576 577 function worker (arg, cb) {578 t.equal(arg, 42)579 cb()580 }581})582 583test('unshift without cb', function (t) {584 t.plan(1)585 586 var queue = buildQueue(worker, 1)587 588 queue.unshift(42)589 590 function worker (arg, cb) {591 t.equal(arg, 42)592 cb()593 }594})595 596test('push with worker throwing error', function (t) {597 t.plan(5)598 var q = buildQueue(function (task, cb) {599 cb(new Error('test error'), null)600 }, 1)601 q.error(function (err, task) {602 t.ok(err instanceof Error, 'global error handler should catch the error')603 t.match(err.message, /test error/, 'error message should be "test error"')604 t.equal(task, 42, 'The task executed should be passed')605 })606 q.push(42, function (err) {607 t.ok(err instanceof Error, 'push callback should catch the error')608 t.match(err.message, /test error/, 'error message should be "test error"')609 })610})611 612test('unshift with worker throwing error', function (t) {613 t.plan(5)614 var q = buildQueue(function (task, cb) {615 cb(new Error('test error'), null)616 }, 1)617 q.error(function (err, task) {618 t.ok(err instanceof Error, 'global error handler should catch the error')619 t.match(err.message, /test error/, 'error message should be "test error"')620 t.equal(task, 42, 'The task executed should be passed')621 })622 q.unshift(42, function (err) {623 t.ok(err instanceof Error, 'unshift callback should catch the error')624 t.match(err.message, /test error/, 'error message should be "test error"')625 })626})627 628test('pause/resume should trigger drain event', function (t) {629 t.plan(1)630 631 var queue = buildQueue(worker, 1)632 queue.pause()633 queue.drain = function () {634 t.pass('drain should be called')635 }636 637 function worker (arg, cb) {638 cb(null, true)639 }640 641 queue.resume()642})643 644test('paused flag', function (t) {645 t.plan(2)646 647 var queue = buildQueue(function (arg, cb) {648 cb(null)649 }, 1)650 t.equal(queue.paused, false)651 queue.pause()652 t.equal(queue.paused, true)653})654