Function pipeTo

lib/internal/streams/iter/pull.js:991–1134

* Write an async source through transforms to a writer. * @param {AsyncIterable |Iterable } source * @param {...(Function|object)} args - Transforms, writer, and optional options * @returns {Promise } Total bytes written

(source, ...args)

 * @returns {Promise<number>} Total bytes written
 */
async function pipeTo(source, ...args) {
  const { transforms, writer, options } = parsePipeToArgs(args, 'write');
  if (options?.signal !== undefined) {
    validateAbortSignal(options.signal, 'options.signal');
  }

  const signal = options?.signal;

  // Check for abort
  signal?.throwIfAborted();

  const hasWriteSync = typeof writer.writeSync === 'function';
  const useSyncIterableFastPath =
    hasWriteSync && canUseSyncIterablePipeToFastPath(source, transforms, signal);
  const normalized = useSyncIterableFastPath ? undefined : from(source);

  let totalBytes = 0;
  const hasWritev = typeof writer.writev === 'function';
  const hasWritevSync = typeof writer.writevSync === 'function';
  const hasEndSync = typeof writer.endSync === 'function';

  // Async fallback for writeBatch when sync write fails partway through.
  // Continues writing from batch[startIndex] using async write().
  async function writeBatchAsyncFallback(batch, startIndex) {
    for (let i = startIndex; i < batch.length; i++) {
      const chunk = batch[i];
      if (hasWriteSync && writer.writeSync(chunk)) {
        // Sync retry succeeded
      } else {
        const result = writer.write(
          chunk, signal ? { __proto__: null, signal } : undefined);
        if (result !== undefined) {
          await result;
        }
      }
      totalBytes += TypedArrayPrototypeGetByteLength(chunk);
    }
  }

  // Write a batch using try-fallback: sync first, async if needed.
  // Returns undefined on sync success, or a Promise when async fallback
  // is required. Callers must check: const p = writeBatch(b); if (p) await p;
  function writeBatch(batch) {
    if (hasWritev && batch.length > 1) {
      if (!hasWritevSync || !writer.writevSync(batch)) {
        const opts = signal ? { __proto__: null, signal } : undefined;
        const writevResult = writer.writev(batch, opts);
        if (writevResult === undefined) {
          for (let i = 0; i < batch.length; i++) {
            totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
          }
          return;
        }
        return PromisePrototypeThen(PromiseResolve(writevResult), () => {
          for (let i = 0; i < batch.length; i++) {
            totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
          }
        });