queue.js

/*
Auther: s25g5d4 @ https://github.com/s25g5d4
Usage:

用法很簡單，只要 var q = new Queue(limit, timeout, delay)
第一個參數是這個 queue 最大同時併發量
第二個參數是任務執行的 timeout
當某個任務執行超過 timeout 時間後會自動被 reject
第三個參數是任務完成後強制延後下一個任務執行的時間

建出 Queue 物件後透過 q.queue(executor, name) 把任務排進 queue 裡
executor 與 promise 物件的參數 executor 一樣
是一個 function 接受 resolve 與 reject 作為參數
如果 queue 裡沒有東西會直接執行 executor
否則檢查頂到 limit 了沒，沒有就一樣直接執行
頂到 limit 就等目前執行中的任何一個任務結束才可執行
有 delay 的話會延遲執行
name 是會記錄在 console 的任務名稱

其他成員函數都是內部函數，不開放讓使用者呼叫

範例：
// 最多 1 個工作同時執行，超過 3 秒後強制 reject，每個工作延遲 500 毫秒
var q = new Queue(1, 3000, 500);
var task1 = q.queue((resolve, reject) => {
  setTimeout(() => { resolve(1); }, 1000);
}); // 回傳一個 Promise 物件
task1.then(data => console.log(data));

var task2 = (resolve, reject) => {
  setTimeout(() => { resolve(2); }, 1000);
};
var task3 = (resolve, reject) => {
  setTimeout(() => { resolve(3); }, 1000);
};
var task4 = (resolve, reject) => {
  setTimeout(() => { resolve(4); }, 1000);
};

var promiseTask234 = Promise.all( [task2, task3, task4].map(e => q.queue(e)) );

promiseTask234.then(data => console.log(data));

如果是某些函數本身就回傳 Promise 物件
則需要包裝如下（以 fetch 為範例）：
q.queue( (resolve, reject) => {
  fetch(url).then(resolve).catch(reject);
});



*/


class Queue {
  constructor(limit, ...options) {
    let timeout, delay;
    if (options.length) {
      if (typeof options[0] === 'object') {
        ({timeout, delay} = options[0]);
      }
      else {
        [timeout, delay] = options;
      }
    }

    this.limit = limit;
    this.timeout = timeout || 0;
    this.delay = delay || 0;
    this.slot = [];
    this.q = [];
  }

  queue(executor, name) {
    console.log(`queue: job ${name} queued`);

    const job = new Promise( (resolve, reject) => {
      this.q.push({
        'name':      name || '',
        'run':       executor,
        'resolve':   resolve,
        'reject':    reject,
        'timeout':   false,
        'timeoutid': undefined
      });
    });
    this.dequeue();

    return job;
  }

  dequeue() {
    const {q, slot, limit, timeout, delay} = this;

    if (slot.length < limit && q.length >= 1) {
      const job = q.shift();
      slot.push(job);
      console.log(`queue: job ${job.name} started`);

      if (timeout) job.timeoutid = setTimeout(this.jobTimeout.bind(this, job), timeout);

      const onFulfilled = data => {
        if (job.timeout) {
          return;
        }

        this.removeJob(job);
        setTimeout(this.dequeue.bind(this), delay); // force dequeue() run after current dequeue()
        if (job.timeoutid) clearTimeout(job.timeoutid);
        console.log(`queue: job ${job.name} resolved`);

        job.resolve(data);
      };

      const onRejected = reason => {
        if (job.timeout) {
          return;
        }

        this.removeJob(job);
        setTimeout(this.dequeue.bind(this), delay);
        if (job.timeoutid) clearTimeout(job.timeoutid);
        console.log(`queue: job ${job.name} rejected`);

        job.reject(reason);
      };

      job.run(onFulfilled, onRejected);
    }
  }

  jobTimeout(job) {
    this.removeJob(job);
    console.log(`queue: job ${job.name} timeout`);
    job.reject(new Error(`queue: job ${job.name || ''} timeout`));
    job = null;
  }

  removeJob(job) {
    let index = this.slot.indexOf(job);
    if (index >= 0) {
      this.slot.splice(index, 1);
      return;
    }

    index = this.q.indexOf(job);
    if (index >= 0) this.q.splice(index, 1);
  }
}

export default Queue;