taskqueue.js

taskqueue.js

const EventEmitter = require('events')

class TaskQueue {
  constructor(concurrent) {
    this.concurrent = concurrent || 1
    this.unprocessed = []
    this.processing = []
    this.processingCount = 0
    this.eventEmitter = new EventEmitter()
    this.eventEmitter.on('processed', () => {
      this.processing.push(this.unprocessed.shift())
      this.next()
    })
  }

  runTask(task) {
    this.unprocessed.push(task)
    if(this.processingCount < this.concurrent) {
      this.processing.push(this.unprocessed.shift())
      this.next()
    }
  }

  async next() {
    const func = this.processing.shift()
    if(func) {
      this.processingCount += 1
      await func()
      this.processingCount -= 1
      if(this.processingCount < this.concurrent) {
        this.eventEmitter.emit('processed')   
      }
    }
  }
}

function task() {
  return new Promise(resolve => {
    setTimeout(() => {
      resolve("stuff")
    }, 1000)
  }).then(console.log)
}

const queue = new TaskQueue(4)

queue.runTask(task)
queue.runTask(task)
queue.runTask(task)
queue.runTask(task)
queue.runTask(task)
queue.runTask(task)