Очередь задач с ограничением их числа выполнения.

concurrent-promise-queue.md

Videos

• https://www.youtube.com/watch?v=Lg46AH8wFvg&t=2s
• https://github.com/HowProgrammingWorks/ConcurrentQueue
• https://github.com/sindresorhus/p-queue

Edafe Emunotor. Building a Concurrent Promise Queue with JavaScript

https://www.codementor.io/@edafeadjekeemunotor/building-a-concurrent-promise-queue-with-javascript-1ano2eof0v

concurrent.js

function concurrent(tasks, iteratee, onDone) {
  let completed = 0;

  function onFinish() {
    if (++completed === tasks.length) {
      onDone();
    }
  }
  
  tasks.forEach((task, index) => {
    iteratee(task, index, tasks, onFinish);
  });
}

raf-queue.js

class TaskQueue {
  constructor() {
    this.queue = [];
    this.requestId = null;
  }

  addTask(task) {
    this.queue.unshift(task);
  }

  run() {
    const self = this;

    let tasksNumber = self.queue.length;
    if (!tasksNumber) return;

    const tasks = self.queue.splice(0, tasksNumber);

    this.requestId = requestAnimationFrame(function(time) {
      while(tasksNumber) {
        const task = tasks.pop();
        task.tick(time);
        tasksNumber--;
      };
      if (self.queue.length) {
        self.run();
      }
    });
  }

  destroy() {
    cancelAnimationFrame(this.requestId);
  }
}

serial.js

function iterate(index) {
  if(index === tasks.length) {
    return finish();
  }
  
  const task = tasks[index];
  task(function() {
    iterate(index + 1);
  });
}

function finish() {
  //обход завершен
}

iterate(0);

// =======
function serial(collection, iteratorCallback, onDone) {
  function iterate(index) {
    if (index === collection.length) {
      return onDone();
    }

    const item = collection[index];
    
    iteratorCallback(item, index, collection, function() {
      iterate(index + 1);
    });
  }

  iterate(0);
}

task-queue-promised.js

const { EventEmitter } = require('events');

class TaskQueue extends EventEmitter {
  constructor(concurrency) {
    super();
    this.concurrency = concurrency;
    this.running = 0;
    this.queue = [];
  }

  pushTask(task) {
    this.queue.unshift(task);
    this.scheduleNext();
    return this;
  }
  
  scheduleNext() {
    process.nextTick(() => this.next());
  }

  next() {
    if (this.running === 0 && this.queue.length === 0) {
      return this.emit('empty');
    }

    while (this.running < this.concurrency && this.queue.length) {
      const task = this.queue.pop();
      task().finally(() => {
        this.running--;
        this.next();      
      });
      this.running++;
    }
  }
  
  runTask(task) {
    return new Promise((resolve, reject) => {
      this.queue.push(() => {
        return task().then(resolve, reject)
      })
      this.scheduleNext();
    })
  }
};

task-queue.js

const { EventEmitter } = require('events');

class TaskQueue extends EventEmitter {
  constructor(concurrency) {
    super();
    this.concurrency = concurrency;
    this.running = 0;
    this.queue = [];
  }

  pushTask(task) {
    this.queue.unshift(task);
    this.scheduleNext();
    return this;
  }
  
  scheduleNext() {
    process.nextTick(() => this.next());
  }

  next() {
    if (this.running === 0 && this.queue.length === 0) {
      return this.emit('empty');
    }

    while (this.running < this.concurrency && this.queue.length) {
      const task = this.queue.pop();
      task((error) => {
        if (error) {
          this.emit('error', error);
        }
        this.running--;
        this.scheduleNext();
      });
      this.running++;
    }
  }
};