Implement pause/resume

Author: tshemsedinov

JavaScript/3-pause.js

@@ -0,0 +1,157 @@
+'use strict';
+
+class Queue {
+  constructor(concurrency) {
+    this.paused = false;
+    this.concurrency = concurrency;
+    this.count = 0;
+    this.waiting = [];
+    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) {
+    if (!this.paused) {
+      const hasChannel = this.count < this.concurrency;
+      if (hasChannel) {
+        this.next(task);
+        return;
+      }
+    }
+    this.waiting.push({ task, start: 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.paused && 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, waitTimeout } = this;
+    const { task, start } = waiting.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) {
+          setTimeout(() => {
+            if (!this.paused) this.takeNext();
+          }, 0);
+        }
+        return;
+      }
+    }
+    const hasChannel = this.count < this.concurrency;
+    if (hasChannel) 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;
+  }
+  pause() {
+    this.paused = true;
+    return this;
+  }
+  resume() {
+    this.paused = false;
+    if (this.waiting.length > 0) {
+      const hasChannel = this.count < this.concurrency;
+      if (hasChannel) this.takeNext();
+    }
+    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'))
+  .pause();
+
+for (let i = 0; i < 10; i++) {
+  queue.add({ name: `Task${i}`, interval: i * 1000 });
+}
+
+console.log('Start paused');
+
+setTimeout(() => {
+  console.log('Resume');
+  queue.resume();
+}, 3000);
+
+setTimeout(() => {
+  console.log('Pause');
+  queue.pause();
+}, 4000);
+
+setTimeout(() => {
+  console.log('Resume');
+  queue.resume();
+}, 5000);