add timeout support to blocking queue operations

dist/index.d.ts

@@ -6,16 +6,18 @@ declare class WaitQueue<T> {
             done: boolean;
         };
     };
-    queue: LinkedList;
-    listeners: LinkedList;
+    queue: LinkedList<T>;
+    listeners: LinkedList<(err?: Error | undefined) => unknown>;
     get length(): number;
+    numWaiters(): number;
     empty(): void;
     clear(): void;
     clearListeners(): void;
-    unshift(...items: T[]): number;
-    push(...items: T[]): number;
-    shift(): Promise<T>;
-    pop(): Promise<T>;
+    unshift(item: T): number;
+    push(item: T): number;
+    private _remove;
+    shift(timeout?: number): Promise<T>;
+    pop(timeout?: number): Promise<T>;
     private _flush;
 }
 export = WaitQueue;

dist/index.js

@@ -10,26 +10,6 @@ var __values = (this && this.__values) || function(o) {
     };
     throw new TypeError(s ? "Object is not iterable." : "Symbol.iterator is not defined.");
 };
-var __read = (this && this.__read) || function (o, n) {
-    var m = typeof Symbol === "function" && o[Symbol.iterator];
-    if (!m) return o;
-    var i = m.call(o), r, ar = [], e;
-    try {
-        while ((n === void 0 || n-- > 0) && !(r = i.next()).done) ar.push(r.value);
-    }
-    catch (error) { e = { error: error }; }
-    finally {
-        try {
-            if (r && !r.done && (m = i["return"])) m.call(i);
-        }
-        finally { if (e) throw e.error; }
-    }
-    return ar;
-};
-var __spread = (this && this.__spread) || function () {
-    for (var ar = [], i = 0; i < arguments.length; i++) ar = ar.concat(__read(arguments[i]));
-    return ar;
-};
 var __importDefault = (this && this.__importDefault) || function (mod) {
     return (mod && mod.__esModule) ? mod : { "default": mod };
 };
@@ -38,13 +18,6 @@ var __importDefault = (this && this.__importDefault) || function (mod) {
  * https://github.com/flarestart/wait-queue
  */
 var LinkedList_1 = __importDefault(require("./libs/LinkedList"));
-var nextLoop = (function () {
-    if (typeof setImmediate === 'function') {
-        return setImmediate;
-    }
-    /* istanbul ignore next */
-    return function (fn) { return setTimeout(fn, 0); };
-})();
 var WaitQueue = /** @class */ (function () {
     function WaitQueue() {
         this.queue = new LinkedList_1.default();
@@ -54,9 +27,12 @@ var WaitQueue = /** @class */ (function () {
         get: function () {
             return this.queue.length;
         },
-        enumerable: true,
+        enumerable: false,
         configurable: true
     });
+    WaitQueue.prototype.numWaiters = function () {
+        return this.listeners.length;
+    };
     WaitQueue.prototype.empty = function () {
         this.queue = new LinkedList_1.default();
     };
@@ -80,83 +56,70 @@ var WaitQueue = /** @class */ (function () {
         }
         this.listeners = new LinkedList_1.default();
     };
-    WaitQueue.prototype.unshift = function () {
-        var _a;
-        var items = [];
-        for (var _i = 0; _i < arguments.length; _i++) {
-            items[_i] = arguments[_i];
-        }
-        (_a = this.queue).unshift.apply(_a, __spread(items));
+    WaitQueue.prototype.unshift = function (item) {
+        this.queue.unshift(item);
         this._flush();
         return this.length;
     };
-    WaitQueue.prototype.push = function () {
-        var _a;
-        var items = [];
-        for (var _i = 0; _i < arguments.length; _i++) {
-            items[_i] = arguments[_i];
-        }
-        (_a = this.queue).push.apply(_a, __spread(items));
+    WaitQueue.prototype.push = function (item) {
+        this.queue.push(item);
         this._flush();
         return this.length;
     };
-    WaitQueue.prototype.shift = function () {
+    WaitQueue.prototype._remove = function (type, timeout) {
         var _this = this;
+        var fn;
+        switch (type) {
+            case 'SHIFT':
+                fn = this.queue.shift.bind(this.queue);
+                break;
+            case 'POP':
+                fn = this.queue.pop.bind(this.queue);
+                break;
+        }
         return new Promise(function (resolve, reject) {
-            if (_this.queue.length > 0) {
-                return resolve(_this.queue.shift());
+            var self = _this;
+            if (self.queue.length > 0) {
+                return resolve(fn());
             }
             else {
-                _this.listeners.push(function (err) {
-                    if (err) {
-                        return reject(err);
+                var timerId_1;
+                var listener_1 = self.listeners.push(function (err) {
+                    if (timerId_1) {
+                        clearTimeout(timerId_1);
+                        timerId_1 = undefined;
                     }
-                    return resolve(_this.queue.shift());
-                });
-            }
-        });
-    };
-    WaitQueue.prototype.pop = function () {
-        var _this = this;
-        return new Promise(function (resolve, reject) {
-            if (_this.queue.length > 0) {
-                return resolve(_this.queue.pop());
-            }
-            else {
-                _this.listeners.push(function (err) {
                     if (err) {
                         return reject(err);
                     }
-                    return resolve(_this.queue.pop());
+                    return resolve(fn());
                 });
+                timerId_1 = (timeout && timeout > 0) ?
+                    setTimeout(function () {
+                        timerId_1 = undefined;
+                        self.listeners.remove(listener_1);
+                        reject(new Error("Timed Out"));
+                    }, timeout) : undefined;
             }
         });
     };
+    WaitQueue.prototype.shift = function (timeout) {
+        return this._remove('SHIFT', timeout);
+    };
+    WaitQueue.prototype.pop = function (timeout) {
+        return this._remove('POP', timeout);
+    };
     WaitQueue.prototype._flush = function () {
-        if (this.queue.length > 0 && this.listeners.length > 0) {
+        while (this.queue.length > 0 && this.listeners.length > 0) {
             var listener = this.listeners.shift();
             listener.call(this);
-            // delay next loop
-            nextLoop(this._flush.bind(this));
         }
     };
     return WaitQueue;
 }());
 if (typeof Symbol === 'function' && typeof Symbol.iterator === 'symbol') {
     WaitQueue.prototype[Symbol.iterator] = function () {
-        var node = this.queue._front;
-        return {
-            next: function () {
-                if (node === null) {
-                    return { value: null, done: true };
-                }
-                else {
-                    var r = { value: node.item, done: false };
-                    node = node._next;
-                    return r;
-                }
-            },
-        };
+        return this.queue[Symbol.iterator]();
     };
 }
 module.exports = WaitQueue;

dist/libs/LinkedList.d.ts

@@ -1,23 +1,25 @@
-interface Node {
-    _next: Node | null;
-    _prev: Node | null;
-    item: any;
+interface Node<T> {
+    _next: Node<T>;
+    _prev: Node<T>;
+    _removed?: boolean;
+    item: T;
 }
-declare class LinkedList {
+declare class LinkedList<T> {
     [Symbol.iterator]: () => {
         next: () => {
-            value: any;
+            value: T;
             done: boolean;
         };
     };
     _length: number;
-    _front: Node | null;
-    _end: Node | null;
-    get length(): number;
+    _head: Node<T>;
+    constructor();
     empty(): void;
-    push(...items: any[]): number;
-    shift(): any;
-    unshift(...items: any[]): number;
-    pop(): any;
+    get length(): number;
+    push(item: T): Node<T>;
+    unshift(item: T): Node<T>;
+    pop(): T;
+    shift(): T;
+    remove(node: Node<T>): void;
 }
 export default LinkedList;