Function pipeToSync

src/pull.ts:610–658

(
  source: Iterable<Uint8Array[]>,
  ...args: (SyncTransform | SyncWriter | PipeToSyncOptions)[]
)

  // Yield results (filter out null from final output)
  let completed = false;
  try {
    for await (const batch of abortableAsyncIterable(current, controller.signal)) {
      if (batch !== null) {
        yield batch;
      }
    }
    completed = true;
  } catch (error) {
    if (!controller.signal.aborted) {
      try {
        controller.abort(error instanceof Error ? error : new Error(String(error)));
      } catch { /* transform signal listeners may throw — suppress */ }
    }
    throw error;
  } finally {
    if (!completed && !controller.signal.aborted) {
      try {
        controller.abort(new DOMException('Aborted', 'AbortError'));
      } catch { /* transform signal listeners may throw — suppress */ }
    }
    // Clean up user signal listener to prevent holding controller alive
    if (signal && abortHandler) {
      signal.removeEventListener('abort', abortHandler);
    }
  }
}

// =============================================================================
// Public API: pull() and pullSync()
// =============================================================================

/**
 * Create a sync pull-through pipeline with transforms.
 *
 * @param source - The sync streamable source
 * @param args - Variadic transforms
 * @returns A sync iterable yielding Uint8Array[] batches
 */
export function pullSync(
  source: SyncStreamable,
  ...transforms: SyncTransform[]
): SyncByteStreamReadable {
  return {
    *[Symbol.iterator]() {
      yield* createSyncPipeline(source, transforms);
    },
  };
}

/**
 * Create an async pull-through pipeline with transforms.
 *