BREAKING CHANGE: awaitable queues

8 files changed, 256 insertions, 188 deletions

diff --git a/lib/cargo.js b/lib/cargo.js
index 5060223..1afdc0e 100644
--- a/lib/cargo.js
+++ b/lib/cargo.js
@@ -1,36 +1,6 @@
 import queue from './internal/queue';
 
 /**
- * A cargo of tasks for the worker function to complete. Cargo inherits all of
- * the same methods and event callbacks as [`queue`]{@link module:ControlFlow.queue}.
- * @typedef {Object} CargoObject
- * @memberOf module:ControlFlow
- * @property {Function} length - A function returning the number of items
- * waiting to be processed. Invoke like `cargo.length()`.
- * @property {number} payload - An `integer` for determining how many tasks
- * should be process per round. This property can be changed after a `cargo` is
- * created to alter the payload on-the-fly.
- * @property {Function} push - Adds `task` to the `queue`. The callback is
- * called once the `worker` has finished processing the task. Instead of a
- * single task, an array of `tasks` can be submitted. The respective callback is
- * used for every task in the list. Invoke like `cargo.push(task, [callback])`.
- * @property {Function} saturated - A callback that is called when the
- * `queue.length()` hits the concurrency and further tasks will be queued.
- * @property {Function} empty - A callback that is called when the last item
- * from the `queue` is given to a `worker`.
- * @property {Function} drain - A callback that is called when the last item
- * from the `queue` has returned from the `worker`.
- * @property {Function} idle - a function returning false if there are items
- * waiting or being processed, or true if not. Invoke like `cargo.idle()`.
- * @property {Function} pause - a function that pauses the processing of tasks
- * until `resume()` is called. Invoke like `cargo.pause()`.
- * @property {Function} resume - a function that resumes the processing of
- * queued tasks when the queue is paused. Invoke like `cargo.resume()`.
- * @property {Function} kill - a function that removes the `drain` callback and
- * empties remaining tasks from the queue forcing it to go idle. Invoke like `cargo.kill()`.
- */
-
-/**
  * Creates a `cargo` object with the specified payload. Tasks added to the
  * cargo will be processed altogether (up to the `payload` limit). If the
  * `worker` is in progress, the task is queued until it becomes available. Once
@@ -53,7 +23,7 @@ import queue from './internal/queue';
  * @param {number} [payload=Infinity] - An optional `integer` for determining
  * how many tasks should be processed per round; if omitted, the default is
  * unlimited.
- * @returns {module:ControlFlow.CargoObject} A cargo object to manage the tasks. Callbacks can
+ * @returns {module:ControlFlow.QueueObject} A cargo object to manage the tasks. Callbacks can
  * attached as certain properties to listen for specific events during the
  * lifecycle of the cargo and inner queue.
  * @example
@@ -73,9 +43,8 @@ import queue from './internal/queue';
  * cargo.push({name: 'bar'}, function(err) {
  *     console.log('finished processing bar');
  * });
- * cargo.push({name: 'baz'}, function(err) {
- *     console.log('finished processing baz');
- * });
+ * await cargo.push({name: 'baz'});
+ * console.log('finished processing baz');
  */
 export default function cargo(worker, payload) {
     return queue(worker, 1, payload);
diff --git a/lib/internal/queue.js b/lib/internal/queue.js
index 751b562..fb23e81 100644
--- a/lib/internal/queue.js
+++ b/lib/internal/queue.js
@@ -3,8 +3,6 @@ import setImmediate from './setImmediate';
 import DLL from './DoublyLinkedList';
 import wrapAsync from './wrapAsync';
 
-const noop = () => {}
-
 export default function queue(worker, concurrency, payload) {
     if (concurrency == null) {
         concurrency = 1;
@@ -16,6 +14,35 @@ export default function queue(worker, concurrency, payload) {
     var _worker = wrapAsync(worker);
     var numRunning = 0;
     var workersList = [];
+    const events = {
+        error: [],
+        drain: [],
+        saturated: [],
+        unsaturated: [],
+        empty: []
+    }
+
+    function on (event, handler) {
+        events[event].push(handler)
+    }
+
+    function once (event, handler) {
+        const handleAndRemove = (...args) => {
+            off(event, handleAndRemove)
+            handler(...args)
+        }
+        events[event].push(handleAndRemove)
+    }
+
+    function off (event, handler) {
+        if (!event) return Object.keys(events).forEach(ev => events[ev] = [])
+        if (!handler) return events[event] = []
+        events[event] = events[event].filter(ev => ev !== handler)
+    }
+
+    function trigger (event, ...args) {
+        events[event].forEach(handler => handler(...args))
+    }
 
     var processingScheduled = false;
     function _insert(data, insertAtFront, callback) {
@@ -23,25 +50,32 @@ export default function queue(worker, concurrency, payload) {
             throw new Error('task callback must be a function');
         }
         q.started = true;
-        if (!Array.isArray(data)) {
-            data = [data];
-        }
-        if (data.length === 0 && q.idle()) {
-            // call drain immediately if there are no tasks
-            return setImmediate(() => q.drain());
+        if (Array.isArray(data)) {
+            if (data.length === 0 && q.idle()) {
+                // call drain immediately if there are no tasks
+                return setImmediate(() => trigger('drain'));
+            }
+
+            return data.map(datum => _insert(datum, insertAtFront, callback));
         }
 
-        for (var i = 0, l = data.length; i < l; i++) {
-            var item = {
-                data: data[i],
-                callback: callback || noop
-            };
+        var res;
 
-            if (insertAtFront) {
-                q._tasks.unshift(item);
-            } else {
-                q._tasks.push(item);
+        var item = {
+            data,
+            callback: callback || function (err, ...args) {
+                // we don't care about the error, let the global error handler
+                // deal with it
+                if (err) return
+                if (args.length <= 1) return res(args[0])
+                res(args)
             }
+        };
+
+        if (insertAtFront) {
+            q._tasks.unshift(item);
+        } else {
+            q._tasks.push(item);
         }
 
         if (!processingScheduled) {
@@ -51,9 +85,15 @@ export default function queue(worker, concurrency, payload) {
                 q.process();
             });
         }
+
+        if (!callback) {
+            return new Promise((resolve) => {
+                res = resolve
+            })
+        }
     }
 
-    function _next(tasks) {
+    function _createCB(tasks) {
         return function (err, ...args) {
             numRunning -= 1;
 
@@ -70,21 +110,35 @@ export default function queue(worker, concurrency, payload) {
                 task.callback(err, ...args);
 
                 if (err != null) {
-                    q.error(err, task.data);
+                    trigger('error', err, task.data);
                 }
             }
 
             if (numRunning <= (q.concurrency - q.buffer) ) {
-                q.unsaturated();
+                trigger('unsaturated')
             }
 
             if (q.idle()) {
-                q.drain();
+                trigger('drain')
             }
             q.process();
         };
     }
 
+    const eventMethod = (name) => (handler) => {
+        if (!handler) {
+            return new Promise((resolve, reject) => {
+                once(name, (err, data) => {
+                    if (err) return reject(err)
+                    resolve(data)
+                })
+            })
+        }
+        off(name)
+        on(name, handler)
+
+    }
+
     var isProcessing = false;
     var q = {
         _tasks: new DLL(),
@@ -93,23 +147,23 @@ export default function queue(worker, concurrency, payload) {
         },
         concurrency,
         payload,
-        saturated: noop,
-        unsaturated:noop,
+        saturated: eventMethod('saturated'),
+        unsaturated: eventMethod('unsaturated'),
         buffer: concurrency / 4,
-        empty: noop,
-        drain: noop,
-        error: noop,
+        empty: eventMethod('empty'),
+        drain: eventMethod('drain'),
+        error: eventMethod('error'),
         started: false,
         paused: false,
         push (data, callback) {
-            _insert(data, false, callback);
+            return _insert(data, false, callback);
         },
         kill () {
-            q.drain = noop;
+            off()
             q._tasks.empty();
         },
         unshift (data, callback) {
-            _insert(data, true, callback);
+            return _insert(data, true, callback);
         },
         remove (testFn) {
             q._tasks.remove(testFn);
@@ -135,14 +189,14 @@ export default function queue(worker, concurrency, payload) {
                 numRunning += 1;
 
                 if (q._tasks.length === 0) {
-                    q.empty();
+                    trigger('empty');
                 }
 
                 if (numRunning === q.concurrency) {
-                    q.saturated();
+                    trigger('saturated');
                 }
 
-                var cb = onlyOnce(_next(tasks));
+                var cb = onlyOnce(_createCB(tasks));
                 _worker(data, cb);
             }
             isProcessing = false;
diff --git a/lib/queue.js b/lib/queue.js
index 4462f80..4bee1bd 100644
--- a/lib/queue.js
+++ b/lib/queue.js
@@ -18,6 +18,9 @@ import wrapAsync from './internal/wrapAsync';
  * @property {number} concurrency - an integer for determining how many `worker`
  * functions should be run in parallel. This property can be changed after a
  * `queue` is created to alter the concurrency on-the-fly.
+ * @property {number} payload - an integer that specifies how many items are
+ * passed to the worker function at a time. only applies if this is a
+ * [cargo]{@link module:ControlFlow.cargo} object
  * @property {Function} push - add a new task to the `queue`. Calls `callback`
  * once the `worker` has finished processing the task. Instead of a single task,
  * a `tasks` array can be submitted. The respective callback is used for every
@@ -30,20 +33,25 @@ import wrapAsync from './internal/wrapAsync';
  * [priorityQueue]{@link module:ControlFlow.priorityQueue} object.
  * Invoked with `queue.remove(testFn)`, where `testFn` is of the form
  * `function ({data, priority}) {}` and returns a Boolean.
- * @property {Function} saturated - a callback that is called when the number of
- * running workers hits the `concurrency` limit, and further tasks will be
- * queued.
- * @property {Function} unsaturated - a callback that is called when the number
- * of running workers is less than the `concurrency` & `buffer` limits, and
- * further tasks will not be queued.
+ * @property {Function} saturated - a function that sets a callback that is
+ * called when the number of running workers hits the `concurrency` limit, and
+ * further tasks will be queued.  If the callback is omitted, `q.saturated()`
+ * returns a promise for the next occurrence.
+ * @property {Function} unsaturated - a function that sets a callback that is
+ * called when the number of running workers is less than the `concurrency` &
+ * `buffer` limits, and further tasks will not be queued. If the callback is
+ * omitted, `q.unsaturated()` returns a promise for the next occurrence.
  * @property {number} buffer - A minimum threshold buffer in order to say that
  * the `queue` is `unsaturated`.
- * @property {Function} empty - a callback that is called when the last item
- * from the `queue` is given to a `worker`.
- * @property {Function} drain - a callback that is called when the last item
- * from the `queue` has returned from the `worker`.
- * @property {Function} error - a callback that is called when a task errors.
- * Has the signature `function(error, task)`.
+ * @property {Function} empty - a function that sets a callback that is called
+ * when the last item from the `queue` is given to a `worker`. If the callback
+ * is omitted, `q.empty()` returns a promise for the next occurrence.
+ * @property {Function} drain - a function that sets a callback that is called
+ * when the last item from the `queue` has returned from the `worker`. If the
+ * callback is omitted, `q.drain()` returns a promise for the next occurrence.
+ * @property {Function} error - a function that sets a callback that is called
+ * when a task errors. Has the signature `function(error, task)`. If the
+ * callback is omitted, `error()` returns a promise for the next error.
  * @property {boolean} paused - a boolean for determining whether the queue is
  * in a paused state.
  * @property {Function} pause - a function that pauses the processing of tasks
@@ -65,6 +73,12 @@ import wrapAsync from './internal/wrapAsync';
  * for (let item of q) {
  *     console.log(item)
  * }
+ *
+ * q.drain(() => {
+ *     console.log('all done')
+ * })
+ * // or
+ * await q.drain()
  */
 
 /**
@@ -96,14 +110,16 @@ import wrapAsync from './internal/wrapAsync';
  * }, 2);
  *
  * // assign a callback
- * q.drain = function() {
+ * q.drain(function() {
  *     console.log('all items have been processed');
- * };
+ * });
+ * // or await the end
+ * await q.drain()
  *
  * // assign an error callback
- * q.error = function(err, task) {
+ * q.error(function(err, task) {
  *     console.error('task experienced an error');
- * };
+ * });
  *
  * // add some items to the queue
  * q.push({name: 'foo'}, function(err) {
diff --git a/test/cargo.js b/test/cargo.js
index 2191eb8..59e28f5 100644
--- a/test/cargo.js
+++ b/test/cargo.js
@@ -59,7 +59,7 @@ describe('cargo', () => {
         }, 30);
 
 
-        c.drain = function () {
+        c.drain(() => {
             expect(call_order).to.eql([
                 'process 1 2', 'callback 1', 'callback 2',
                 'process 3 4', 'callback 3', 'callback 4',
@@ -67,7 +67,7 @@ describe('cargo', () => {
             ]);
             expect(c.length()).to.equal(0);
             done();
-        };
+        });
     });
 
     it('without callback', (done) => {
@@ -95,7 +95,7 @@ describe('cargo', () => {
             c.push(5);
         }, 80);
 
-        c.drain = function() {
+        c.drain(() => {
             expect(call_order).to.eql([
                 'process 1',
                 'process 2',
@@ -103,7 +103,7 @@ describe('cargo', () => {
                 'process 5'
             ]);
             done();
-        }
+        })
     });
 
     it('bulk task', (done) => {
@@ -145,9 +145,9 @@ describe('cargo', () => {
         }, 3);
 
         var drainCounter = 0;
-        c.drain = function () {
+        c.drain(() => {
             drainCounter++;
-        };
+        });
 
         for(var i = 0; i < 10; i++){
             c.push(i);
@@ -172,7 +172,7 @@ describe('cargo', () => {
         }
 
         var drainCounter = 0;
-        c.drain = function () {
+        c.drain(() => {
             drainCounter++;
 
             if (drainCounter === 1) {
@@ -181,7 +181,7 @@ describe('cargo', () => {
                 expect(drainCounter).to.equal(2);
                 done();
             }
-        };
+        });
 
         loadCargo();
     });
@@ -195,15 +195,15 @@ describe('cargo', () => {
         }, 1);
         q.concurrency = 3;
 
-        q.saturated = function() {
+        q.saturated(() => {
             assert(q.running() == 3, 'cargo should be saturated now');
             calls.push('saturated');
-        };
-        q.empty = function() {
+        });
+        q.empty(() => {
             assert(q.length() === 0, 'cargo should be empty now');
             calls.push('empty');
-        };
-        q.drain = function() {
+        });
+        q.drain(() => {
             assert(
                 q.length() === 0 && q.running() === 0,
                 'cargo should be empty now and no more workers should be running'
@@ -227,7 +227,7 @@ describe('cargo', () => {
                 'drain'
             ]);
             done();
-        };
+        });
         q.push('foo', () => {calls.push('foo cb');});
         q.push('bar', () => {calls.push('bar cb');});
         q.push('zoo', () => {calls.push('zoo cb');});
@@ -249,9 +249,7 @@ describe('cargo', () => {
             setTimeout(cb, 25);
         }, 1);
 
-        cargo.drain = function () {
-            done();
-        };
+        cargo.drain(done);
 
         expect(cargo.payload).to.equal(1);
 
@@ -286,11 +284,11 @@ describe('cargo', () => {
             });
         }, 2);
 
-        cargo.drain = function() {
+        cargo.drain(() => {
             expect(cargo.workersList()).to.eql([]);
             expect(cargo.running()).to.equal(0);
             done();
-        };
+        });
 
         cargo.push('foo');
         cargo.push('bar');
@@ -306,10 +304,10 @@ describe('cargo', () => {
             });
         }, 2);
 
-        cargo.drain = function() {
+        cargo.drain(() => {
             expect(cargo.running()).to.equal(0);
             done();
-        };
+        });
 
         cargo.push('foo');
         cargo.push('bar');
diff --git a/test/cargoQueue.js b/test/cargoQueue.js
index 517422c..46801c7 100644
--- a/test/cargoQueue.js
+++ b/test/cargoQueue.js
@@ -61,7 +61,7 @@ describe('cargoQueue', () => {
 
         expect(c.length()).to.equal(2);
 
-        c.drain = function () {
+        c.drain(() => {
             expect(call_order).to.eql([
                 'process 1 2', 'callback 1', 'callback 2',
                 'process 3', 'callback 3',
@@ -69,7 +69,7 @@ describe('cargoQueue', () => {
             ]);
             expect(c.length()).to.equal(0);
             done();
-        };
+        });
     });
 
     it('without callback', (done) => {
@@ -83,7 +83,7 @@ describe('cargoQueue', () => {
                 c.push(4);
                 setImmediate(() => {
                     c.push(5);
-                    c.drain = function complete () {
+                    c.drain(() => {
                         expect(call_order).to.eql([
                             'process 1',
                             'process 2',
@@ -91,7 +91,7 @@ describe('cargoQueue', () => {
                             'process 5'
                         ]);
                         done();
-                    }
+                    })
                 })
             })
         })
@@ -135,9 +135,9 @@ describe('cargoQueue', () => {
         }, 3, 2);
 
         var drainCounter = 0;
-        c.drain = function () {
+        c.drain(() => {
             drainCounter++;
-        };
+        });
 
         for(var i = 0; i < 10; i++){
             c.push(i);
@@ -162,7 +162,7 @@ describe('cargoQueue', () => {
         }
 
         var drainCounter = 0;
-        c.drain = function () {
+        c.drain(() => {
             drainCounter++;
 
             if (drainCounter === 1) {
@@ -171,7 +171,7 @@ describe('cargoQueue', () => {
                 expect(drainCounter).to.equal(2);
                 done();
             }
-        };
+        });
 
         loadCargo();
     });
@@ -184,15 +184,15 @@ describe('cargoQueue', () => {
             async.setImmediate(cb);
         }, 3, 1);
 
-        q.saturated = function() {
+        q.saturated(() => {
             assert(q.running() == 3, 'cargoQueue should be saturated now');
             calls.push('saturated');
-        };
-        q.empty = function() {
+        });
+        q.empty(() => {
             assert(q.length() === 0, 'cargoQueue should be empty now');
             calls.push('empty');
-        };
-        q.drain = function() {
+        });
+        q.drain(() => {
             assert(
                 q.length() === 0 && q.running() === 0,
                 'cargoQueue should be empty now and no more workers should be running'
@@ -216,7 +216,7 @@ describe('cargoQueue', () => {
                 'drain'
             ]);
             done();
-        };
+        });
         q.push('foo', () => {calls.push('foo cb');});
         q.push('bar', () => {calls.push('bar cb');});
         q.push('zoo', () => {calls.push('zoo cb');});
@@ -238,9 +238,7 @@ describe('cargoQueue', () => {
             setTimeout(cb, 25);
         }, 1, 1);
 
-        cargo.drain = function () {
-            done();
-        };
+        cargo.drain(done);
 
         expect(cargo.payload).to.equal(1);
 
@@ -264,9 +262,7 @@ describe('cargoQueue', () => {
             setTimeout(cb, 25);
         }, 1, 1);
 
-        cargo.drain = function () {
-            done();
-        };
+        cargo.drain(done);
 
         expect(cargo.concurrency).to.equal(1);
 
@@ -301,11 +297,11 @@ describe('cargoQueue', () => {
             });
         }, 1, 2);
 
-        cargo.drain = function() {
+        cargo.drain(() => {
             expect(cargo.workersList()).to.eql([]);
             expect(cargo.running()).to.equal(0);
             done();
-        };
+        });
 
         cargo.push('foo');
         cargo.push('bar');
@@ -321,10 +317,10 @@ describe('cargoQueue', () => {
             });
         }, 1, 1);
 
-        cargo.drain = function() {
+        cargo.drain(() => {
             expect(cargo.running()).to.equal(0);
             done();
-        };
+        });
 
         cargo.push(['foo', 'bar', 'baz', 'boo']);
     })
diff --git a/test/es2017/asyncFunctions.js b/test/es2017/asyncFunctions.js
index efb6555..68a89e2 100644
--- a/test/es2017/asyncFunctions.js
+++ b/test/es2017/asyncFunctions.js
@@ -350,10 +350,10 @@ module.exports = function () {
             result.push(await Promise.resolve(val));
         }, 2)
 
-        q.drain = () => {
+        q.drain(() => {
             expect(result).to.eql([[1, 2], [3]]);
             done();
-        };
+        });
 
         q.push(1);
         q.push(2);
@@ -366,10 +366,10 @@ module.exports = function () {
             result.push(await Promise.resolve(val));
         }, 2)
 
-        q.drain = () => {
+        q.drain(() => {
             expect(result).to.eql([1, 2, 3]);
             done();
-        };
+        });
 
         q.push(1);
         q.push(2);
diff --git a/test/es2017/awaitableFunctions.js b/test/es2017/awaitableFunctions.js
index b45b56c..ccf9baf 100644
--- a/test/es2017/awaitableFunctions.js
+++ b/test/es2017/awaitableFunctions.js
@@ -571,6 +571,45 @@ module.exports = function () {
         expect(calls).to.eql([1, 2, 3, 4])
     });
 
+    it('should work with queues', async () => {
+        const q = async.queue(async (data) => {
+            if (data === 6) throw new Error('oh noes')
+            await new Promise(resolve => setTimeout(resolve, 10))
+            return data
+        }, 5)
+
+        const calls = []
+
+        q.error().catch(d => calls.push('error ' + d))
+        q.saturated().then(() => calls.push('saturated'))
+        q.unsaturated().then(() => calls.push('unsaturated'))
+        q.empty().then(() => calls.push('empty'))
+
+        q.push(1).then(d => calls.push('push cb ' + d))
+        q.push(2).then(d => calls.push('push cb ' + d))
+        q.push([3, 4, 5, 6]).map(p => p.then(d => calls.push('push cb ' + d)))
+        q.push(7).then(d => calls.push('push cb ' + d))
+        q.push(8).then(d => calls.push('push cb ' + d))
+
+        const multiP = Promise.all(q.push([9, 10]))
+
+        await q.drain()
+        await multiP
+        expect(calls).to.eql([
+            'saturated',
+            'push cb 1',
+            'push cb 2',
+            'push cb 3',
+            'push cb 4',
+            'push cb 5',
+            'empty',
+            'error Error: oh noes',
+            'push cb 7',
+            'unsaturated',
+            'push cb 8'
+        ])
+    })
+
     /*
      * Util
      */
diff --git a/test/queue.js b/test/queue.js
index f103aaa..58302b4 100644
--- a/test/queue.js
+++ b/test/queue.js
@@ -50,7 +50,7 @@ describe('queue', function(){
         expect(q.length()).to.equal(4);
         expect(q.concurrency).to.equal(2);
 
-        q.drain = function () {
+        q.drain(() => {
             expect(call_order).to.eql([
                 'process 2', 'callback 2',
                 'process 1', 'callback 1',
@@ -60,7 +60,7 @@ describe('queue', function(){
             expect(q.concurrency).to.equal(2);
             expect(q.length()).to.equal(0);
             done();
-        };
+        });
     });
 
     it('default concurrency', (done) => {
@@ -103,7 +103,7 @@ describe('queue', function(){
         expect(q.length()).to.equal(4);
         expect(q.concurrency).to.equal(1);
 
-        q.drain = function () {
+        q.drain(() => {
             expect(call_order).to.eql([
                 'process 1', 'callback 1',
                 'process 2', 'callback 2',
@@ -113,7 +113,7 @@ describe('queue', function(){
             expect(q.concurrency).to.equal(1);
             expect(q.length()).to.equal(0);
             done();
-        };
+        });
     });
 
     it('zero concurrency', (done) => {
@@ -132,10 +132,10 @@ describe('queue', function(){
             callback(task.name === 'foo' ? new Error('fooError') : null);
         }, 2);
 
-        q.drain = function() {
+        q.drain(() => {
             expect(results).to.eql(['bar', 'fooError']);
             done();
-        };
+        });
 
         q.push({name: 'bar'}, (err) => {
             if(err) {
@@ -163,17 +163,17 @@ describe('queue', function(){
             callback(task.name === 'foo' ? new Error('fooError') : null);
         }, 2);
 
-        q.error = function(error, task) {
+        q.error((error, task) => {
             expect(error).to.exist;
             expect(error.message).to.equal('fooError');
             expect(task.name).to.equal('foo');
             results.push('fooError');
-        };
+        });
 
-        q.drain = function() {
+        q.drain (() => {
             expect(results).to.eql(['fooError', 'bar']);
             done();
-        };
+        });
 
         q.push({name: 'foo'});
 
@@ -201,9 +201,7 @@ describe('queue', function(){
             q.push('');
         }
 
-        q.drain = function(){
-            done();
-        };
+        q.drain(done);
 
         setTimeout(() => {
             expect(q.concurrency).to.equal(1);
@@ -245,7 +243,7 @@ describe('queue', function(){
         q.push(3);
         q.push(4);
 
-        q.drain = function () {
+        q.drain(() => {
             expect(running).to.eql(0);
             expect(concurrencyList).to.eql([1, 2, 2, 2]);
             expect(call_order).to.eql([
@@ -255,7 +253,7 @@ describe('queue', function(){
                 'process 3'
             ]);
             done();
-        };
+        });
     });
 
     it('push with non-function', (done) => {
@@ -320,7 +318,7 @@ describe('queue', function(){
         expect(q.length()).to.equal(4);
         expect(q.concurrency).to.equal(2);
 
-        q.drain = function () {
+        q.drain(() => {
             expect(call_order).to.eql([
                 'process 2', 'callback 2',
                 'process 1', 'callback 1',
@@ -330,7 +328,7 @@ describe('queue', function(){
             expect(q.concurrency).to.equal(2);
             expect(q.length()).to.equal(0);
             done();
-        };
+        });
     });
 
     it('idle', (done) => {
@@ -351,11 +349,11 @@ describe('queue', function(){
         // Queue is busy when tasks added
         expect(q.idle()).to.equal(false);
 
-        q.drain = function() {
+        q.drain(() => {
             // Queue is idle after drain
             expect(q.idle()).to.equal(true);
             done();
-        };
+        });
     });
 
     it('pause', (done) => {
@@ -397,7 +395,7 @@ describe('queue', function(){
             q.resume();
             q.push(5);
             q.push(6);
-            q.drain = drain;
+            q.drain(drain);
         }
         function drain () {
             expect(concurrencyList).to.eql([1, 2, 2, 1, 2, 2]);
@@ -436,10 +434,10 @@ describe('queue', function(){
         q.push({ id: 4 });
         q.push({ id: 5 });
 
-        q.drain = function () {
+        q.drain(() => {
             expect(call_order).to.eql([1, 2, 3, 4, 5]);
             done();
-        };
+        });
     });
 
     it('start paused', (done) => {
@@ -462,9 +460,7 @@ describe('queue', function(){
             q.resume();
         }, 5);
 
-        q.drain = function () {
-            done();
-        };
+        q.drain(done);
     });
 
     it('kill', (done) => {
@@ -496,15 +492,15 @@ describe('queue', function(){
         }, 3);
         q.concurrency = 3;
 
-        q.saturated = function() {
+        q.saturated(() => {
             assert(q.running() == 3, 'queue should be saturated now');
             calls.push('saturated');
-        };
-        q.empty = function() {
+        });
+        q.empty(() => {
             assert(q.length() === 0, 'queue should be empty now');
             calls.push('empty');
-        };
-        q.drain = function() {
+        });
+        q.drain(() => {
             assert(
                 q.length() === 0 && q.running() === 0,
                 'queue should be empty now and no more workers should be running'
@@ -528,7 +524,7 @@ describe('queue', function(){
                 'drain'
             ]);
             done();
-        };
+        });
         q.push('foo', () => {calls.push('foo cb');});
         q.push('bar', () => {calls.push('bar cb');});
         q.push('zoo', () => {calls.push('zoo cb');});
@@ -544,7 +540,7 @@ describe('queue', function(){
             async.setImmediate(cb);
         }, 3);
 
-        q.drain = function() {
+        q.drain(() => {
             assert(
                 q.length() === 0 && q.running() === 0,
                 'queue should be empty now and no more workers should be running'
@@ -554,7 +550,7 @@ describe('queue', function(){
                 'drain'
             ]);
             done();
-        };
+        });
         q.push([]);
     });
 
@@ -568,14 +564,14 @@ describe('queue', function(){
             async.setImmediate(cb);
         }, 1);
 
-        q.empty = function () {
+        q.empty(() => {
             calls.push('empty');
             assert(q.idle() === false,
                 'tasks should be running when empty is called')
             expect(q.running()).to.equal(1);
-        }
+        })
 
-        q.drain = function() {
+        q.drain(() => {
             calls.push('drain');
             expect(calls).to.eql([
                 'empty',
@@ -583,7 +579,7 @@ describe('queue', function(){
                 'drain'
             ]);
             done();
-        };
+        });
         q.push(1);
     });
 
@@ -593,13 +589,13 @@ describe('queue', function(){
             async.setImmediate(cb);
         }, 2);
 
-        q.saturated = function () {
+        q.saturated(() => {
             saturatedCalled = true;
-        };
-        q.drain = function () {
+        })
+        q.drain(() => {
             assert(saturatedCalled, "saturated not called");
             done();
-        };
+        })
 
         q.push(['foo', 'bar', 'baz', 'moo']);
     });
@@ -623,10 +619,10 @@ describe('queue', function(){
                 calls.push('process ' + task);
                 async.setImmediate(cb);
             }, 4);
-            q.saturated = function() {
+            q.saturated(() => {
                 calls.push('saturated');
-            };
-            q.empty = function() {
+            });
+            q.empty(() => {
                 expect(calls.indexOf('saturated')).to.be.above(-1);
                 setTimeout(() => {
                     expect(calls).eql([
@@ -645,7 +641,7 @@ describe('queue', function(){
                     ]);
                     done();
                 }, 50);
-            };
+            });
             q.push('foo0', () => {calls.push('foo0 cb');});
             q.push('foo1', () => {calls.push('foo1 cb');});
             q.push('foo2', () => {calls.push('foo2 cb');});
@@ -683,10 +679,10 @@ describe('queue', function(){
                 calls.push('process ' + task);
                 async.setImmediate(cb);
             }, 4);
-            q.unsaturated = function() {
+            q.unsaturated(() => {
                 calls.push('unsaturated');
-            };
-            q.empty = function() {
+            });
+            q.empty(() => {
                 expect(calls.indexOf('unsaturated')).to.be.above(-1);
                 setTimeout(() => {
                     expect(calls).eql([
@@ -708,7 +704,7 @@ describe('queue', function(){
                     ]);
                     done();
                 }, 50);
-            };
+            });
             q.push('foo0', () => {calls.push('foo0 cb');});
             q.push('foo1', () => {calls.push('foo1 cb');});
             q.push('foo2', () => {calls.push('foo2 cb');});
@@ -726,11 +722,11 @@ describe('queue', function(){
                 });
             }, 2);
 
-            q.drain = function() {
+            q.drain(() => {
                 expect(q.workersList().length).to.equal(0);
                 expect(q.running()).to.equal(0);
                 done();
-            };
+            });
 
             q.push('foo');
             q.push('bar');
@@ -767,11 +763,11 @@ describe('queue', function(){
                 });
             }, 2);
 
-            q.drain = function() {
+            q.drain(() => {
                 expect(q.workersList()).to.eql([]);
                 expect(q.workersList().length).to.equal(q.running());
                 done();
-            };
+            });
 
             q.push('foo');
             q.push('bar');
@@ -792,10 +788,10 @@ describe('queue', function(){
             return node.data === 3;
         });
 
-        q.drain = function () {
+        q.drain(() => {
             expect(result).to.eql([1, 2, 4, 5]);
             done();
-        }
+        })
     });
 
     it('should be iterable', (done) => {
@@ -811,9 +807,9 @@ describe('queue', function(){
 
         expect([...q]).to.eql([1, 2, 3, 4, 5]);
 
-        q.drain = function () {
+        q.drain(() => {
             expect([...q]).to.eql([]);
             done();
-        }
+        })
     })
 });