Method queue

src/master/pool.ts:319–370

(taskFunction: TaskRunFunction<ThreadType, any>)

  }

  public queue(taskFunction: TaskRunFunction<ThreadType, any>) {
    const { maxQueuedJobs = Infinity } = this.options

    if (this.isClosing) {
      throw Error(`Cannot schedule pool tasks after terminate() has been called.`)
    }
    if (this.initErrors.length > 0) {
      throw this.initErrors[0]
    }

    const taskID = this.nextTaskID++
    const taskCompletion = this.taskCompletion(taskID)

    taskCompletion.catch((error) => {
      // Prevent unhandled rejections here as we assume the user will use
      // `pool.completed()`, `pool.settled()` or `task.catch()` to handle errors
      this.debug(`Task #${taskID} errored:`, error)
    })

    const task: QueuedTask<ThreadType, any> = {
      id: taskID,
      run: taskFunction,
      cancel: () => {
        if (this.taskQueue.indexOf(task) === -1) return
        this.taskQueue = this.taskQueue.filter(someTask => someTask !== task)
        this.eventSubject.next({
          type: PoolEventType.taskCanceled,
          taskID: task.id
        })
      },
      then: taskCompletion.then.bind(taskCompletion)
    }

    if (this.taskQueue.length >= maxQueuedJobs) {
      throw Error(
        "Maximum number of pool tasks queued. Refusing to queue another one.\n" +
        "This usually happens for one of two reasons: We are either at peak " +
        "workload right now or some tasks just won't finish, thus blocking the pool."
      )
    }

    this.debug(`Queueing task #${task.id}...`)
    this.taskQueue.push(task)

    this.eventSubject.next({
      type: PoolEventType.taskQueued,
      taskID: task.id
    })

    this.scheduleWork()
    return task
  }

  public async terminate(force?: boolean) {
    this.isClosing = true