promise.js 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221
  1. 'use strict'
  2. const test = require('tape')
  3. const buildQueue = require('../').promise
  4. const { promisify } = require('util')
  5. const sleep = promisify(setTimeout)
  6. const immediate = promisify(setImmediate)
  7. test('concurrency', function (t) {
  8. t.plan(2)
  9. t.throws(buildQueue.bind(null, worker, 0))
  10. t.doesNotThrow(buildQueue.bind(null, worker, 1))
  11. async function worker (arg) {
  12. return true
  13. }
  14. })
  15. test('worker execution', async function (t) {
  16. const queue = buildQueue(worker, 1)
  17. const result = await queue.push(42)
  18. t.equal(result, true, 'result matches')
  19. async function worker (arg) {
  20. t.equal(arg, 42)
  21. return true
  22. }
  23. })
  24. test('limit', async function (t) {
  25. const queue = buildQueue(worker, 1)
  26. const [res1, res2] = await Promise.all([queue.push(10), queue.push(0)])
  27. t.equal(res1, 10, 'the result matches')
  28. t.equal(res2, 0, 'the result matches')
  29. async function worker (arg) {
  30. await sleep(arg)
  31. return arg
  32. }
  33. })
  34. test('multiple executions', async function (t) {
  35. const queue = buildQueue(worker, 1)
  36. const toExec = [1, 2, 3, 4, 5]
  37. const expected = ['a', 'b', 'c', 'd', 'e']
  38. let count = 0
  39. await Promise.all(toExec.map(async function (task, i) {
  40. const result = await queue.push(task)
  41. t.equal(result, expected[i], 'the result matches')
  42. }))
  43. async function worker (arg) {
  44. t.equal(arg, toExec[count], 'arg matches')
  45. return expected[count++]
  46. }
  47. })
  48. test('drained', async function (t) {
  49. const queue = buildQueue(worker, 2)
  50. const toExec = new Array(10).fill(10)
  51. let count = 0
  52. async function worker (arg) {
  53. await sleep(arg)
  54. count++
  55. }
  56. toExec.forEach(function (i) {
  57. queue.push(i)
  58. })
  59. await queue.drained()
  60. t.equal(count, toExec.length)
  61. toExec.forEach(function (i) {
  62. queue.push(i)
  63. })
  64. await queue.drained()
  65. t.equal(count, toExec.length * 2)
  66. })
  67. test('drained with exception should not throw', async function (t) {
  68. const queue = buildQueue(worker, 2)
  69. const toExec = new Array(10).fill(10)
  70. async function worker () {
  71. throw new Error('foo')
  72. }
  73. toExec.forEach(function (i) {
  74. queue.push(i)
  75. })
  76. await queue.drained()
  77. })
  78. test('drained with drain function', async function (t) {
  79. let drainCalled = false
  80. const queue = buildQueue(worker, 2)
  81. queue.drain = function () {
  82. drainCalled = true
  83. }
  84. const toExec = new Array(10).fill(10)
  85. let count = 0
  86. async function worker (arg) {
  87. await sleep(arg)
  88. count++
  89. }
  90. toExec.forEach(function () {
  91. queue.push()
  92. })
  93. await queue.drained()
  94. t.equal(count, toExec.length)
  95. t.equal(drainCalled, true)
  96. })
  97. test('set this', async function (t) {
  98. t.plan(1)
  99. const that = {}
  100. const queue = buildQueue(that, worker, 1)
  101. await queue.push(42)
  102. async function worker (arg) {
  103. t.equal(this, that, 'this matches')
  104. }
  105. })
  106. test('unshift', async function (t) {
  107. const queue = buildQueue(worker, 1)
  108. const expected = [1, 2, 3, 4]
  109. await Promise.all([
  110. queue.push(1),
  111. queue.push(4),
  112. queue.unshift(3),
  113. queue.unshift(2)
  114. ])
  115. t.is(expected.length, 0)
  116. async function worker (arg) {
  117. t.equal(expected.shift(), arg, 'tasks come in order')
  118. }
  119. })
  120. test('push with worker throwing error', async function (t) {
  121. t.plan(5)
  122. const q = buildQueue(async function (task, cb) {
  123. throw new Error('test error')
  124. }, 1)
  125. q.error(function (err, task) {
  126. t.ok(err instanceof Error, 'global error handler should catch the error')
  127. t.match(err.message, /test error/, 'error message should be "test error"')
  128. t.equal(task, 42, 'The task executed should be passed')
  129. })
  130. try {
  131. await q.push(42)
  132. } catch (err) {
  133. t.ok(err instanceof Error, 'push callback should catch the error')
  134. t.match(err.message, /test error/, 'error message should be "test error"')
  135. }
  136. })
  137. test('unshift with worker throwing error', async function (t) {
  138. t.plan(2)
  139. const q = buildQueue(async function (task, cb) {
  140. throw new Error('test error')
  141. }, 1)
  142. try {
  143. await q.unshift(42)
  144. } catch (err) {
  145. t.ok(err instanceof Error, 'push callback should catch the error')
  146. t.match(err.message, /test error/, 'error message should be "test error"')
  147. }
  148. })
  149. test('no unhandledRejection (push)', async function (t) {
  150. function handleRejection () {
  151. t.fail('unhandledRejection')
  152. }
  153. process.once('unhandledRejection', handleRejection)
  154. const q = buildQueue(async function (task, cb) {
  155. throw new Error('test error')
  156. }, 1)
  157. q.push(42)
  158. await immediate()
  159. process.removeListener('unhandledRejection', handleRejection)
  160. })
  161. test('no unhandledRejection (unshift)', async function (t) {
  162. function handleRejection () {
  163. t.fail('unhandledRejection')
  164. }
  165. process.once('unhandledRejection', handleRejection)
  166. const q = buildQueue(async function (task, cb) {
  167. throw new Error('test error')
  168. }, 1)
  169. q.unshift(42)
  170. await immediate()
  171. process.removeListener('unhandledRejection', handleRejection)
  172. })