feat: allow worker to be an async function

Author: fent

README.md

@@ -15,6 +15,12 @@ const worker = (arg1, arg2, callback) => {
   someAsyncFunction(calculation(arg1, arg2), callback);
 };
 
+// Worker can be an async function.
+const worker = async (arg1) => {
+  await anotherSyncFunction(arg1);
+  return await andAnotherOne();
+};
+
 // create a queue with max 5 concurrency every second
 let q = new TimeQueue(worker, { concurrency: 5, every: 1000 });
 

lib/index.js

@@ -14,6 +14,7 @@ module.exports = class TimeQueue extends EventEmitter {
     super();
 
     this.worker = worker;
+    this._workerAsync = worker.constructor.name == 'AsyncFunction';
     options = options || {};
     this.concurrency = options.concurrency || 1;
     this.every = options.every || 0;
@@ -45,12 +46,15 @@ module.exports = class TimeQueue extends EventEmitter {
    * @return {Promise?}
    */
   push(...args) {
-    // Returns a promise if no `callback` is given.
-    if (args.length < this.worker.length) {
+    // Returns a promise no `callback` is given.
+    if (this._workerAsync && args.length === this.worker.length ||
+      !this._workerAsync && args.length < this.worker.length) {
       return new Promise((resolve, reject) => {
         // Add any missing arguments.
-        while (args.length < this.worker.length - 1) {
-          args.push(undefined);
+        if (!this._workerAsync) {
+          while (args.length < this.worker.length - 1) {
+            args.push(undefined);
+          }
         }
         this.push(...args, (err, results) => {
           if (err) return reject(err);
@@ -92,7 +96,7 @@ module.exports = class TimeQueue extends EventEmitter {
    *
    * @param {Array.<Object>} args
    */
-  _process(args) {
+  async _process(args) {
     const callback = args.pop();
     let finished = false;
     let every = ~~this.every;
@@ -152,12 +156,19 @@ module.exports = class TimeQueue extends EventEmitter {
       }, timeout);
     }
 
-    // Add custom callback to args.
-    const args2 = args.slice();
-    args2.push(taskCallback);
-
     // Call the worker.
-    this.worker(...args2);
+    if (this._workerAsync) {
+      try {
+        taskCallback(null, await this.worker(...args));
+      } catch (err) {
+        taskCallback(err);
+      }
+    } else {
+      // Add custom callback to args.
+      const args2 = args.slice();
+      args2.push(taskCallback);
+      this.worker(...args2);
+    }
   }
 
 

test/main-test.js

@@ -53,6 +53,76 @@ describe('Create a queue and add to it', () => {
       assert.equal(q.timeout, 0);
     });
   });
+
+  describe('with an async worker', () => {
+    it('Still respects concurrency', (done) => {
+      const q = new TimeQueue(async (a, b) => {
+        return await new Promise((resolve) => {
+          process.nextTick(() => resolve(a + b));
+        });
+      }, { concurrency: 2 });
+      let result1, result2, result3;
+      q.push(1, 2, (err, result) => {
+        assert.ifError(err);
+        result1 = result;
+      });
+      q.push(3, 4, (err, result) => {
+        assert.ifError(err);
+        result2 = result;
+      });
+      q.push(5, 6, (err, result) => {
+        assert.ifError(err);
+        result3 = result;
+      });
+      assert.equal(q.active, 2);
+      q.on('drain', () => {
+        assert.equal(result1, 3);
+        assert.equal(result2, 7);
+        assert.equal(result3, 11);
+        done();
+      });
+    });
+    it('Pushing to worker can be awaited', async () => {
+      const q = new TimeQueue(async (a, b) => {
+        return new Promise((resolve) => {
+          process.nextTick(() => resolve(a + b));
+        });
+      }, { concurrency: 1 });
+      let result1 = await q.push(1, 2);
+      let result2 = await q.push(3, 4);
+      assert.equal(result1, 3);
+      assert.equal(result2, 7);
+    });
+    describe('that errors', () => {
+      it('Calls callback with error', (done) => {
+        const q = new TimeQueue(async (a) => {
+          return new Promise((resolve, reject) => {
+            process.nextTick(() => reject(Error('no: ' + a)));
+          });
+        });
+        q.push('one', (err) => {
+          assert.ok(err);
+          assert.equal(err.message, 'no: one');
+          done();
+        });
+      });
+      it('Throws when awaited', async () => {
+        const q = new TimeQueue(async (a) => {
+          return new Promise((resolve, reject) => {
+            process.nextTick(() => reject(Error('no: ' + a)));
+          });
+        });
+        try {
+          await q.push('okay');
+        } catch (err) {
+          assert.ok(err);
+          assert.equal(err.message, 'no: okay');
+          return;
+        }
+        throw Error('shoult not get here');
+      });
+    });
+  });
 });