Merge pull request caolan#474 from maxkueng/pause-and-resume

queue.pause() and queue.resume()

Author: Caolan McMahon

README.md

@@ -1088,6 +1088,9 @@ methods:
    and further tasks will be queued.
 * `empty` - a callback that is called when the last item from the `queue` is given to a `worker`.
 * `drain` - a callback that is called when the last item from the `queue` has returned from the `worker`.
+* `paused` - a boolean for determining whether the queue is in a paused state
+* `pause()` - a function that pauses the processing of tasks until `resume()` is called.
+* `resume()` - a function that resumes the processing of queued tasks when the queue is paused.
 
 __Example__
 

lib/async.js

@@ -725,14 +725,15 @@
             saturated: null,
             empty: null,
             drain: null,
+            paused: false,
             push: function (data, callback) {
               _insert(q, data, false, callback);
             },
             unshift: function (data, callback) {
               _insert(q, data, true, callback);
             },
             process: function () {
-                if (workers < q.concurrency && q.tasks.length) {
+                if (!q.paused && workers < q.concurrency && q.tasks.length) {
                     var task = q.tasks.shift();
                     if (q.empty && q.tasks.length === 0) {
                         q.empty();
@@ -760,6 +761,16 @@
             },
             idle: function() {
                 return q.tasks.length + workers === 0;
+            },
+            pause: function () {
+                if (q.paused === true) { return; }
+                q.paused = true;
+                q.process();
+            },
+            resume: function () {
+                if (q.paused === false) { return; }
+                q.paused = false;
+                q.process();
             }
         };
         return q;

test/test-async.js

@@ -2300,6 +2300,57 @@ exports['queue idle'] = function(test) {
     }
 }
 
+exports['queue pause'] = function(test) {
+    var call_order = [],
+        task_timeout = 100,
+        pause_timeout = 300,
+        resume_timeout = 500,
+        tasks = [ 1, 2, 3, 4, 5, 6 ],
+
+        elapsed = (function () {
+            var start = +Date.now();
+            return function () { return Math.floor((+Date.now() - start) / 100) * 100; };
+        })();
+
+    var q = async.queue(function (task, callback) {
+        call_order.push('process ' + task);
+        call_order.push('timeout ' + elapsed());
+        callback();
+    });
+
+    function pushTask () {
+        var task = tasks.shift();
+        if (!task) { return; }
+        setTimeout(function () {
+            q.push(task);
+            pushTask();
+        }, task_timeout);
+    }
+    pushTask();
+
+    setTimeout(function () {
+        q.pause();
+        test.equal(q.paused, true);
+    }, pause_timeout);
+
+    setTimeout(function () {
+        q.resume();
+        test.equal(q.paused, false);
+    }, resume_timeout);
+
+    setTimeout(function () {
+        test.same(call_order, [
+            'process 1', 'timeout 100',
+            'process 2', 'timeout 200',
+            'process 3', 'timeout 500',
+            'process 4', 'timeout 500',
+            'process 5', 'timeout 500',
+            'process 6', 'timeout 600'
+        ]);
+        test.done();
+    }, 800);
+}
+
 exports['cargo'] = function (test) {
     var call_order = [],
         delays = [160, 160, 80];