Add timeouts

Author: tshemsedinov

JavaScript/2-timeouts.js

@@ -0,0 +1,122 @@
+'use strict';
+
+class Queue {
+  constructor(concurrency) {
+    this.concurrency = concurrency;
+    this.count = 0;
+    this.waiting = [];
+    this.waitingStart = [];
+    this.onProcess = null;
+    this.onDone = null;
+    this.onSuccess = null;
+    this.onFailure = null;
+    this.onDrain = null;
+    this.waitTimeout = Infinity;
+    this.processTimeout = Infinity;
+  }
+  static channels(concurrency) {
+    return new Queue(concurrency);
+  }
+  wait(msec) {
+    this.waitTimeout = msec;
+    return this;
+  }
+  timeout(msec) {
+    this.processTimeout = msec;
+    return this;
+  }
+  add(task) {
+    const hasChannel = this.count < this.concurrency;
+    if (hasChannel) {
+      this.next(task);
+      return;
+    }
+    this.waiting.push(task);
+    this.waitingStart.push(Date.now());
+  }
+  next(task) {
+    this.count++;
+    let timer = null;
+    let finished = false;
+    const { processTimeout, onProcess } = this;
+    const finish = (err, res) => {
+      if (finished) return;
+      finished = true;
+      if (timer) clearTimeout(timer);
+      this.count--;
+      this.finish(err, res);
+      if (this.waiting.length > 0) this.takeNext();
+    };
+    if (processTimeout !== Infinity) {
+      const err = new Error('Process timed out');
+      timer = setTimeout(finish, processTimeout, err, task);
+    }
+    onProcess(task, finish);
+  }
+  takeNext() {
+    const { waiting, waitingStart, waitTimeout } = this;
+    const task = waiting.shift();
+    const start = waitingStart.shift();
+    if (waitTimeout !== Infinity) {
+      const delay = Date.now() - start;
+      if (delay > waitTimeout) {
+        const err = new Error('Waiting timed out');
+        this.finish(err, task);
+        if (waiting.length > 0) this.takeNext();
+        return;
+      }
+    }
+    this.next(task);
+    return;
+  }
+  finish(err, res) {
+    const { onFailure, onSuccess, onDone, onDrain } = this;
+    if (err) {
+      if (onFailure) onFailure(err, res);
+    } else if (onSuccess) {
+      onSuccess(res);
+    }
+    if (onDone) onDone(err, res);
+    if (this.count === 0 && onDrain) onDrain();
+  }
+  process(listener) {
+    this.onProcess = listener;
+    return this;
+  }
+  done(listener) {
+    this.onDone = listener;
+    return this;
+  }
+  success(listener) {
+    this.onSuccess = listener;
+    return this;
+  }
+  failure(listener) {
+    this.onFailure = listener;
+    return this;
+  }
+  drain(listener) {
+    this.onDrain = listener;
+    return this;
+  }
+}
+
+// Usage
+
+const job = (task, next) => {
+  console.log(`Process: ${task.name}`);
+  setTimeout(next, task.interval, null, task).unref();
+};
+
+const queue = Queue.channels(3)
+  .wait(4000)
+  .timeout(5000)
+  .process(job)
+  .done((err, task) => console.log(`Done: ${task.name}`))
+  .success(task => console.log(`Success: ${task.name}`))
+  .failure((err, task) => console.log(`Failure: ${err} ${task.name}`))
+  .drain(() => console.log('Queue drain'));
+
+for (let i = 0; i < 10; i++) {
+  queue.add({ name: `Task${i}`, interval: i * 1000 });
+}