Merge pull request #1205 from caolan/dll

Implment queues using DLLs

1 files changed, 16 insertions, 29 deletions

diff --git a/lib/priorityQueue.js b/lib/priorityQueue.js
index 9ea24bc..e895d8b 100644
--- a/lib/priorityQueue.js
+++ b/lib/priorityQueue.js
@@ -31,25 +31,11 @@ import queue from './queue';
  * * The `unshift` method was removed.
  */
 export default function(worker, concurrency) {
-    function _compareTasks(a, b) {
-        return a.priority - b.priority;
-    }
-
-    function _binarySearch(sequence, item, compare) {
-        var beg = -1,
-            end = sequence.length - 1;
-        while (beg < end) {
-            var mid = beg + ((end - beg + 1) >>> 1);
-            if (compare(item, sequence[mid]) >= 0) {
-                beg = mid;
-            } else {
-                end = mid - 1;
-            }
-        }
-        return beg;
-    }
+    // Start with a normal queue
+    var q = queue(worker, concurrency);
 
-    function _insert(q, data, priority, callback) {
+    // Override push to accept second parameter representing priority
+    q.push = function(data, priority, callback) {
         if (callback != null && typeof callback !== 'function') {
             throw new Error('task callback must be a function');
         }
@@ -63,6 +49,12 @@ export default function(worker, concurrency) {
                 q.drain();
             });
         }
+
+        var nextNode = q._tasks.head;
+        while (nextNode && priority >= nextNode.priority) {
+            nextNode = nextNode.next;
+        }
+
         arrayEach(data, function(task) {
             var item = {
                 data: task,
@@ -70,18 +62,13 @@ export default function(worker, concurrency) {
                 callback: typeof callback === 'function' ? callback : noop
             };
 
-            q.tasks.splice(_binarySearch(q.tasks, item, _compareTasks) + 1, 0, item);
-
-            setImmediate(q.process);
+            if (nextNode) {
+                q._tasks.insertBefore(nextNode, item);
+            } else {
+                q._tasks.push(item);
+            }
         });
-    }
-
-    // Start with a normal queue
-    var q = queue(worker, concurrency);
-
-    // Override push to accept second parameter representing priority
-    q.push = function(data, priority, callback) {
-        _insert(q, data, priority, callback);
+        setImmediate(q.process);
     };
 
     // Remove unshift function