Function applyFusedStatelessAsyncTransforms

lib/internal/streams/iter/pull.js:573–633

* Apply a fused run of stateless async transforms to a source. * All transforms in the run are applied in a tight synchronous loop per batch, * avoiding the overhead of N async generator ticks for N transforms. * * INVARIANT: This function accepts a signal, NOT a pre-built options object. * A f

(source, run, signal)

 * @yields {Uint8Array[]}
 */
async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
  for await (const chunks of source) {
    let current = chunks;
    for (let i = 0; i < run.length; i++) {
      const result = run[i](current, { __proto__: null, signal });
      if (result === null) {
        current = null;
        break;
      }
      if (isPromise(result)) {
        const resolved = await result;
        if (resolved === null) {
          current = null;
          break;
        }
        current = resolved;
      } else {
        current = result;
      }
    }
    if (current === null) continue;
    // Normalize the final output
    if (isUint8ArrayBatch(current)) {
      if (current.length > 0) yield current;
    } else if (isUint8Array(current)) {
      yield [current];
    } else if (typeof current === 'string') {
      yield [toUint8Array(current)];
    } else if (isAnyArrayBuffer(current)) {
      yield [new Uint8Array(current)];
    } else if (ArrayBufferIsView(current)) {
      yield [arrayBufferViewToUint8Array(current)];
    } else {
      yield* processTransformResultAsync(current);
    }
  }
  // Flush each transform after all upstream data, including data emitted by
  // earlier flushes, has been processed by that transform.
  let pending = [];
  for (let i = 0; i < run.length; i++) {
    const next = [];
    for (let j = 0; j < pending.length; j++) {
      const pendingResult = appendTransformResultAsync(
        next,
        run[i](pending[j], { __proto__: null, signal }));
      if (pendingResult !== undefined) {
        await pendingResult;
      }
    }
    const flushResult = appendTransformResultAsync(
      next,
      run[i](null, { __proto__: null, signal }));
    if (flushResult !== undefined) {
      await flushResult;
    }
    pending = next;
  }
  for (let i = 0; i < pending.length; i++) {