fastq - npm Package Compare versions

Comparing version 1.12.0 to 1.13.0

index.d.ts

@@ -30,2 +30,3 @@ declare function fastq<C, T = any, R = any>(context: C, worker: fastq.worker<C, T, R>, concurrency: number): fastq.queue<T, R>
     unshift(task: T): Promise<R>
+    drained(): Promise<void>
   }
@@ -32,0 +33,0 @@

package.json

 {
   "name": "fastq",
-  "version": "1.12.0",
+  "version": "1.13.0",
   "description": "Fast, in memory work queue",
@@ -5,0 +5,0 @@ "main": "queue.js",

queue.js

@@ -226,2 +226,3 @@ 'use strict'
   queue.unshift = unshift
+  queue.drained = drained
 
@@ -267,2 +268,15 @@ return queue
   }
+
+  function drained () {
+    var previousDrain = queue.drain
+
+    var p = new Promise(function (resolve) {
+      queue.drain = function () {
+        previousDrain()
+        resolve()
+      }
+    })
+
+    return p
+  }
 }
@@ -269,0 +283,0 @@

README.md

@@ -96,3 +96,3 @@ # fastq
 type Task = {
-  id: number  
+  id: number
 }
@@ -128,3 +128,3 @@
   // No need for a try-catch block, fastq handles errors automatically
-  console.log(arg.id)  
+  console.log(arg.id)
 }
@@ -296,2 +296,9 @@ ```
 
+<a name="drained"></a>
+#### queue.drained() => Promise
+
+Wait for the queue to be drained. The returned `Promise` will be resolved when all tasks in the queue have been processed by a worker.
+
+This promise could be ignored as it will not lead to a `'unhandledRejection'`.
+
 ## License
@@ -298,0 +305,0 @@

test/promise.js

@@ -62,2 +62,72 @@ 'use strict'
 
+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) {
@@ -64,0 +134,0 @@ t.plan(1)

No alert changes

Improved metrics

Total package byte prevSize

38219

increased by4.72%

Lines of code

1008

increased by6.44%

Number of lines in readme file

310

increased by2.31%

No dependency changes