Method _executePipeline

packages/client/lib/client/index.ts:1509–1560

* @internal

(
    commands: Array<RedisMultiQueuedCommand>,
    selectedDB?: number
  )

   * @internal
   */
  async _executePipeline(
    commands: Array<RedisMultiQueuedCommand>,
    selectedDB?: number
  ) {
    if (!this._self.#socket.isOpen) {
      return Promise.reject(new ClientClosedError());
    }

    const batchSize = commands.length;

    return trace(CHANNELS.TRACE_BATCH,
      async () => {
        const chainId = Symbol('Pipeline Chain');
        const promise = Promise.all(
          commands.map(({ args }) => {
            const traced = trace(CHANNELS.TRACE_COMMAND,
              () => this._self.#queue.addCommand(args, {
                chainId,
                typeMapping: this._commandOptions?.typeMapping
              }),
              () => ({
                ...this._self.#commandTraceContext(args),
                batchMode: 'PIPELINE' as const,
                batchSize
              })
            );
            // Prevent unhandled rejection from tracePromise wrapper; individual
            // rejections are collected by Promise.all, but the tracePromise wrapper
            // is a separate branch that nobody awaits.
            traced.catch(noop);
            return traced;
          })
        );
        this._self.#scheduleWrite();

        const result = await promise;

        if (selectedDB !== undefined) {
          this._self.#selectedDB = selectedDB;
        }

        return result;
      },
      () => ({
        batchMode: 'PIPELINE' as const,
        batchSize,
        database: this._self.#selectedDB,
        clientId: this._self._clientId,
        ...this._self.#socketTraceContext()
      })
    );
  }

  /**
   * @internal