only run drain once when multiple empty tasks are pushed [#1791]

lib/internal/queue.js

@@ -45,6 +45,7 @@ export default function queue(worker, concurrency, payload) {
     }
 
     var processingScheduled = false;
+    var drainScheduled = false;
     function _insert(data, insertAtFront, rejectOnError, callback) {
         if (callback != null && typeof callback !== 'function') {
             throw new Error('task callback must be a function');
@@ -124,7 +125,13 @@ export default function queue(worker, concurrency, payload) {
     function _maybeDrain(data) {
         if (data.length === 0 && q.idle()) {
             // call drain immediately if there are no tasks
-            setImmediate(() => trigger('drain'));
+            if (!drainScheduled) {
+                drainScheduled = true
+                setImmediate(() => {
+                    trigger('drain')
+                    drainScheduled = false
+                })
+            }
             return true
         }
         return false
@@ -141,7 +148,6 @@ export default function queue(worker, concurrency, payload) {
         }
         off(name)
         on(name, handler)
-
     }
 
     var isProcessing = false;

test/queue.js

@@ -670,6 +670,46 @@ describe('queue', function(){
         done();
     });
 
+    it('should only call drain once when empty tasks are pushed', (done) => {
+        const q = async.queue(() => {
+            throw new Error('should not be called')
+        })
+
+        let numCalled = 0
+        q.drain(() => {
+            numCalled++
+        })
+        q.push([])
+        q.push([])
+        q.push([])
+
+        setTimeout(() => {
+            expect(numCalled).to.equal(1)
+            done()
+        }, 50);
+    });
+
+    it('should not schedule another drain call if one is running', (done) => {
+        const q = async.queue(() => {
+            throw new Error('should not be called')
+        })
+
+        let numCalled = 0
+        q.drain(() => {
+            if (numCalled > 0) {
+                throw new Error('drain should not be called more than one')
+            }
+            numCalled++
+            q.push([])
+        })
+        q.push([])
+
+        setTimeout(() => {
+            expect(numCalled).to.equal(1)
+            done()
+        }, 50);
+    });
+
     context('q.saturated(): ', () => {
         it('should call the saturated callback if tasks length is concurrency', (done) => {
             var calls = [];