feat: let TimeQueue#push() return a promise when no callback is given

BREAKING CHANGE: `error` event no longer emitted when a callback isn't
given to `push()`. Instead, `push()` will return a promise that can
reject.

Author: fent

README.md

@@ -18,14 +18,17 @@ const worker = (arg1, arg2, callback) => {
 // create a queue with max 5 concurrency every second
 let q = new TimeQueue(worker, { concurrency: 5, every: 1000 });
 
-// push tasks onto the queue
+// Push tasks onto the queue.
 q.push(42, 24);
 q.push(2, 74);
 
-// optional callback when pushing tasks
-q.push(2, 2, (err) => {
+// Optional callback when pushing tasks.
+q.push(2, 2, (err, result) => {
   // task finished
 });
+
+// Can use promise/await syntax instead.
+let result = await q.push(3, 5);
 ```
 
 
@@ -40,17 +43,17 @@ Creates a new instance of a queue. Worker must be a function with a callback for
 
   // How much time in milliseconds to allow no more than
   // the max number of concurrent tasks to run.
-  // If the max amount of concurrent tasks are finished faster than the limit,
-  // they will be queued.
+  // If the max amount of concurrent tasks finish faster than this time limit,
+  // additional tasks will wait until enough time has passed before starting.
 , every: 0
 
   // Maximum number of tasks to keep in the queue.
   // While full, pushed tasks will be ignored.
 , maxQueued: Infinity
 
-  // If set, will emit an `error` event if a tasks takes too much time.
-  // if callback was given to that task,
-  // it will be called with the error instead.
+  // If set, tasks will error if they take too much time.
+  // if callback was given to that task, it will be called with the error,
+  // otherwise, the returned promise should be `caught`.
 , timeout: 0
 }
 ```
@@ -86,11 +89,6 @@ Empties queue and clears the timeouts TimeQueue sets to keep track of running ta
 
 # Events
 
-### Event: 'error'
-* `Error`
-
-Emitted when there is an error processing a task and a callback isn't given to the `push` method.
-
 ### Event: 'full'
 
 Queue is full.

lib/index.js

@@ -42,8 +42,23 @@ module.exports = class TimeQueue extends EventEmitter {
    *
    * @param {Object} ...args
    * @param {Function(!Error, ...)} callback
+   * @return {Promise?}
    */
   push(...args) {
+    // Returns a promise if no `callback` is given.
+    if (args.length < this.worker.length) {
+      return new Promise((resolve, reject) => {
+        // Add any missing arguments.
+        while (args.length < this.worker.length - 1) {
+          args.push(undefined);
+        }
+        this.push(...args, (err, results) => {
+          if (err) return reject(err);
+          resolve(results);
+        });
+      });
+    }
+
     if (this.isFull()) {
       return;
     }
@@ -78,7 +93,7 @@ module.exports = class TimeQueue extends EventEmitter {
    * @param {Array.<Object>} args
    */
   _process(args) {
-    const callback = args.splice(this.worker.length - 1, 1)[0];
+    const callback = args.pop();
     let finished = false;
     let every = ~~this.every;
     let timedOut;
@@ -104,7 +119,7 @@ module.exports = class TimeQueue extends EventEmitter {
     let timeout = ~~this.timeout;
     let tid;
 
-    const taskCallback = (err, ...args) => {
+    const taskCallback = (err, result) => {
       // If this task has timed out, and the callback is called again
       // from the worker, ignore it.
       if (!taskTimedOut) {
@@ -118,19 +133,9 @@ module.exports = class TimeQueue extends EventEmitter {
         throw Error('Callback from worker should only be called once');
       }
       callbackCalled = true;
-
       this.finished++;
       this.active--;
-
-      if (typeof callback === 'function') {
-        // If a callback was given with the task,
-        // call it when the task is finished.
-        callback(err, ...args);
-
-      } else if (err) {
-        // Otherwise emit an `error` event if there was an error with the task.
-        this.emit('error', err);
-      }
+      callback(err, result);
 
       finished = true;
       if (timedOut) {
@@ -147,11 +152,6 @@ module.exports = class TimeQueue extends EventEmitter {
       }, timeout);
     }
 
-    // Add missing arguments.
-    while (args.length < this.worker.length - 1) {
-      args.push(undefined);
-    }
-
     // Add custom callback to args.
     const args2 = args.slice();
     args2.push(taskCallback);

test/main-test.js

@@ -4,7 +4,7 @@ const assert = require('assert');
 
 describe('Create a queue and add to it', () => {
   let full = false;
-  const q = new TimeQueue(process.nextTick, { concurrency: 3 });
+  const q = new TimeQueue(callback => process.nextTick(callback), { concurrency: 3 });
 
   q.on('full', () => {
     full = true;
@@ -79,6 +79,16 @@ describe('Create a queue with variable number of arguments', () => {
     assert.equal(lastC, 'hello');
   });
 
+  it('Returns a promise that fulfills', async () => {
+    const q = new TimeQueue((a, b, c, callback) => {
+      process.nextTick(() => {
+        callback(null, a + b + c);
+      });
+    }, { concurrency: 10, every: 1000 });
+    let result = await q.push(1, 2, 3);
+    assert.equal(result, 6);
+  });
+
   describe('Push with callback', () => {
     it('Calls callback when task finishes', (done) => {
       q.push(3, 2, 1, done);
@@ -125,12 +135,14 @@ describe('Create a queue with a worker that always errors', () => {
     });
   }, { concurrency: 10 });
 
-  it('Emits an `error` event', (done) => {
-    q.push();
-    q.once('error', (err) => {
+  it('Trhows an error', async () => {
+    try {
+      await q.push();
+    } catch (err) {
       assert.equal(err.message, 'gotcha');
-      done();
-    });
+      return;
+    }
+    throw Error('should have thrown');
   });
 
   describe('Push task with callback', () => {

test/timeout-test.js

@@ -26,9 +26,9 @@ describe('Create a queue with a timeout', () => {
       setTimeout(callback, 100);
     }, { timeout: 50 });
 
-    it('Should throw an error', (done) => {
-      q.push(3, 4);
-      q.on('error', (err) => {
+    it('Should reject', (done) => {
+      let p = q.push(3, 4);
+      p.catch((err) => {
         assert(err);
         assert.equal(err.message, 'Task timed out');
         assert.equal(err.args[0], 3);