Function createAsyncPipeline

lib/internal/streams/iter/pull.js:772–862

* Create an async pipeline from source through transforms. * @yields {Uint8Array[]}

(source, transforms, signal)

 * @yields {Uint8Array[]}
 */
async function* createAsyncPipeline(source, transforms, signal) {
  // Check for abort
  signal?.throwIfAborted();

  // Fast path: no transforms, just yield normalized source directly
  if (transforms.length === 0) {
    yield* yieldAbortable(source, signal);
    return;
  }

  const normalized = yieldAbortable(source, signal);

  // Create internal controller for transform cancellation.
  // Note: if signal was already aborted, we threw above - no need to check here.
  const controller = new AbortController();
  let abortHandler;
  if (signal) {
    abortHandler = () => {
      controller.abort(signal.reason ??
        lazyDOMException('Aborted', 'AbortError'));
    };
    signal.addEventListener('abort', abortHandler, { __proto__: null, once: true });
  }

  // Apply transforms - fuse consecutive stateless transforms into a single
  // generator layer to avoid unnecessary async generator ticks.
  //
  // INVARIANT: Each transform invocation MUST receive its own fresh options
  // object ({ __proto__: null, signal }). Transforms may mutate the options
  // object, so sharing a single object across invocations would allow one
  // transform to corrupt the options seen by another. The signal is shared
  // across calls (mutations to it are acceptable), but the containing options
  // object must be unique per call. This is enforced inside
  // applyFusedStatelessAsyncTransforms and applyStatefulAsyncTransform, which
  // accept the signal directly and create the options object per invocation.
  // DO NOT pass a pre-built options object.
  let current = normalized;
  const transformSignal = controller.signal;
  let statelessRun = [];

  for (let i = 0; i < transforms.length; i++) {
    const transform = transforms[i];
    if (isTransformObject(transform)) {
      // Flush any accumulated stateless run before the stateful transform
      if (statelessRun.length > 0) {
        current = applyFusedStatelessAsyncTransforms(current, statelessRun,
                                                     transformSignal);
        statelessRun = [];
      }
      const opts = { __proto__: null, signal: transformSignal };
      if (transform[kValidatedTransform]) {
        current = applyValidatedStatefulAsyncTransform(
          current, transform.transform, opts);
      } else {
        current = applyStatefulAsyncTransform(
          current, transform.transform, opts);
      }
    } else {