Function normalizeAsyncSource

lib/internal/streams/iter/from.js:337–419

* Normalize an async streamable source, yielding batches of Uint8Array. * @param {AsyncIterable|Iterable} source * @yields {Uint8Array[]}

(source)

 * @yields {Uint8Array[]}
 */
async function* normalizeAsyncSource(source) {
  // Prefer async iteration if available
  if (isAsyncIterable(source)) {
    for await (const value of source) {
      // Fast path 1: value is already a Uint8Array[] batch
      if (isUint8ArrayBatch(value)) {
        if (value.length > 0) {
          yield value;
        }
        continue;
      }
      // Fast path 2: value is a single Uint8Array (very common)
      if (isUint8Array(value)) {
        yield [value];
        continue;
      }
      // Slow path: normalize the value
      const batch = [];
      for await (const chunk of normalizeAsyncValue(value)) {
        ArrayPrototypePush(batch, chunk);
      }
      if (batch.length > 0) {
        yield batch;
      }
    }
    return;
  }

  // Fall back to sync iteration - batch sync values together with a bound.
  if (isSyncIterable(source)) {
    let batch = [];

    for (const value of source) {
      // Fast path 1: value is already a Uint8Array[] batch
      if (isUint8ArrayBatch(value)) {
        // Flush any accumulated batch first
        if (batch.length > 0) {
          yield batch;
          batch = [];
        }
        yield* yieldBoundedBatch(value);
        continue;
      }
      // Fast path 2: value is a single Uint8Array (very common)
      if (isUint8Array(value)) {
        ArrayPrototypePush(batch, value);
        if (batch.length === FROM_BATCH_SIZE) {
          yield batch;
          batch = [];
        }
        continue;
      }
      // Slow path: normalize the value - must flush and yield individually
      if (batch.length > 0) {
        yield batch;
        batch = [];
      }
      let asyncBatch = [];