Function applyFusedStatelessSyncTransforms

lib/internal/streams/iter/pull.js:456–498

* Apply a fused run of stateless sync transforms. * @param {Iterable } source * @param {Array } run - Array of stateless transform functions * @yields {Uint8Array[]}

(source, run)

 * @yields {Uint8Array[]}
 */
function* applyFusedStatelessSyncTransforms(source, run) {
  for (const chunks of source) {
    let current = chunks;
    for (let i = 0; i < run.length; i++) {
      const result = run[i](current);
      if (result === null) {
        current = null;
        break;
      }
      current = result;
    }
    if (current === null) continue;
    // Inline normalization with Uint8Array[] batch as the fast path,
    // matching the async pipeline's check order.
    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* processTransformResultSync(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++) {
      appendTransformResultSync(next, run[i](pending[j]));
    }
    appendTransformResultSync(next, run[i](null));
    pending = next;
  }
  for (let i = 0; i < pending.length; i++) {
    yield pending[i];
  }
}

/**
 * Apply a single stateful sync transform to a source.