hard kill #1335 :volcano:queue-hard-kill

1 files changed, 12 insertions, 2 deletions

diff --git a/lib/internal/queue.js b/lib/internal/queue.js
index 7a438da..95ddd91 100644
--- a/lib/internal/queue.js
+++ b/lib/internal/queue.js
@@ -19,6 +19,9 @@ export default function queue(worker, concurrency, payload) {
         if (callback != null && typeof callback !== 'function') {
             throw new Error('task callback must be a function');
         }
+        if (q.killed) {
+            throw new Error('cannot push to killed queue');
+        }
         q.started = true;
         if (!isArray(data)) {
             data = [data];
@@ -47,8 +50,11 @@ export default function queue(worker, concurrency, payload) {
 
     function _next(tasks) {
         return rest(function(args){
-            workers -= 1;
+            if (q.killed) {
+                return;
+            }
 
+            workers -= 1;
             for (var i = 0, l = tasks.length; i < l; i++) {
                 var task = tasks[i];
                 var index = indexOf(workersList, task, 0);
@@ -89,11 +95,15 @@ export default function queue(worker, concurrency, payload) {
         error: noop,
         started: false,
         paused: false,
+        killed: false,
         push: function (data, callback) {
             _insert(data, false, callback);
         },
         kill: function () {
+            q.killed = true;
             q.drain = noop;
+            workersList = null;
+            workers = 0;
             q._tasks.empty();
         },
         unshift: function (data, callback) {
@@ -102,7 +112,7 @@ export default function queue(worker, concurrency, payload) {
         process: function () {
             // Avoid trying to start too many processing operations. This can occur
             // when callbacks resolve synchronously (#1267).
-            if (isProcessing) {
+            if (isProcessing || q.killed) {
                 return;
             }
             isProcessing = true;