Method process

lib/queue.js:662–754

(concurrency, handler)

  }

  process(concurrency, handler) {
    if (!this.settings.isWorker) {
      throw new Error('Cannot call Queue#process on a non-worker');
    }

    if (this.handler) {
      throw new Error('Cannot call Queue#process twice');
    }

    if (this.paused) {
      throw new Error('closed');
    }

    if (typeof concurrency === 'function') {
      handler = concurrency;
      concurrency = defaults['#process'].concurrency;
    }

    // If the handler throws a synchronous exception (only applicable to
    // non-`async` functions), catch it, and fail the job.
    const catchExceptions = true;
    this.handler = helpers.wrapAsync(handler, catchExceptions);
    this.running = 0;
    this.queued = 1;
    this.concurrency = concurrency;

    const jobTick = () => {
      if (this.paused) {
        this.queued -= 1;
        return;
      }

      // invariant: in this code path, this.running < this.concurrency, always
      // after spoolup, this.running + this.queued === this.concurrency
      finally_(
        this._getNextJob().then((job) => {
          // We're shutting down.
          if (this.paused) {
            // This job will get picked up later as a stalled job if we happen
            // to get here. We can't easily process this job because at this
            // point Queue#close has already captured the activeJobs set in a
            // Promise.all call, and we'd prefer to delay a job than
            // half-process it.
            this.queued -= 1;
            return;
          }

          this.running += 1;
          this.queued -= 1;
          if (this.running + this.queued < this.concurrency) {
            this.queued += 1;
            setImmediate(jobTick);
          }

          if (!job) {
            // Per comment in Queue#_waitForJob, this branch is possible when
            // the job is removed before processing can take place, but after
            // being initially acquired.