Method take

packages/node-core/src/utils/queues/autoQueue.ts:137–193

()

  }

  private async take(): Promise<void> {
    if (this.pendingPromise) return;
    if (this._abort) {
      return;
    }

    while (!this._abort) {
      const action = this.queue.take();

      // No more actions to start, take will be called again when new items are pushed
      if (!action) break;

      this.pendingPromise = true;

      const p = timeout(
        Promise.resolve(action.task()),
        this.taskTimeoutSec,
        `${this.name} Queue process task timeout in ${this.taskTimeoutSec} seconds. Please increase --timeout`
      )
        .then((result) => {
          // Queue was flushed while task was running, we ned to discard now
          if (this.nextTask > action.index) {
            action.reject(new TaskFlushedError(this.name));
            return;
          }
          this.outOfOrderTasks[action.index] = {action, result};
        })
        .catch((error) => {
          // Queue was flushed while task was running, we ned to discard now
          if (this.nextTask > action.index) {
            action.reject(new TaskFlushedError(this.name));
            return;
          }
          this.outOfOrderTasks[action.index] = {action, error};
        })
        .finally(() => {
          const index = this.runningTasks.indexOf(p);
          // If the index is -1 then the queue will have been flushed
          if (index >= 0) {
            this.processOutOfOrderTasks();
            this.runningTasks.splice(index, 1);
          }
        });

      this.runningTasks.push(p);

      if (this.runningTasks.length >= this.concurrency) {
        // Load up more when any task completes
        await Promise.any(this.runningTasks);
      }
    }

    // Processed all items in the queue
    this._resolveIdle?.();
    this._resolveIdle = undefined;
    this.pendingPromise = false;
  }