Node.js module

node:stream/iter

namespace Broadcast
- function from(
  input: ByteReadableStream | SyncByteReadableStream | Broadcastable,
  options?: BroadcastOptions
  ): BroadcastResult;
  Create a Broadcast from an existing source. The source is consumed automatically and pushed to all subscribers.
  @param options
  Same as broadcast().
namespace Share
- function from(
  input: ByteReadableStream | SyncByteReadableStream | Shareable,
  options?: ShareOptions
  ): Share;
  Create a Share from an existing source.
  @param options
  Same as share().
namespace Stream
namespace SyncShare
- function from(
  input: SyncByteReadableStream | SyncShareable,
  options?: ShareSyncOptions
  ): SyncShare;
const broadcastProtocol: unique symbol
const drainableProtocol: unique symbol
const shareProtocol: unique symbol
const shareSyncProtocol: unique symbol
const toAsyncStreamable: unique symbol
const toStreamable: unique symbol
function array(
source: Source,
options?: ConsumeOptions
): Promise<Uint8Array<ArrayBufferLike>[]>;
function arrayBuffer(
source: Source,
options?: ConsumeOptions
): Promise<ArrayBuffer>;
function arrayBufferSync(
source: SyncSource,
options?: ConsumeSyncOptions
): ArrayBuffer;
function arraySync(
source: SyncSource,
options?: ConsumeSyncOptions
): Uint8Array<ArrayBufferLike>[];
function broadcast(
options?: BroadcastOptions
): BroadcastResult;
function bytes(
source: Source,
options?: ConsumeOptions
): Promise<Uint8Array<ArrayBufferLike>>;
function bytesSync(
source: SyncSource,
options?: ConsumeSyncOptions
): Uint8Array;
function duplex(
options?: DuplexOptions
): [DuplexChannel, DuplexChannel];
function from(
input: Source
): ByteReadableStream;
function fromReadable(
readable: ReadableStream
): ByteReadableStream;
Converts a classic Readable stream (or duck-typed equivalent) into a stream/iter async iterable source that can be passed to from(), pull(), text(), etc.
If the object implements the toAsyncStreamable protocol (as stream.Readable does), that protocol is used. Otherwise, the function duck-types on read() and on() (EventEmitter) and wraps the stream with a batched async iterator.
The result is cached per instance -- calling fromReadable() twice with the same stream returns the same iterable.
For object-mode or encoded Readable streams, chunks are automatically normalized to Uint8Array.
import { Readable } from 'node:stream'; import { fromReadable, text } from 'node:stream/iter'; const readable = new Readable({ read() { this.push('hello world'); this.push(null); }, }); const result = await text(fromReadable(readable)); console.log(result); // 'hello world'
@param readable
A classic Readable stream or any object with read() and on() methods.
@returns
A stream/iter async iterable source.
function fromSync(
input: SyncSource
): SyncByteReadableStream;
function fromWritable(
writable: WritableStream,
options?: FromWritableOptions
): Writer;
Creates a stream/iter Writer adapter from a classic Writable stream (or duck-typed equivalent). The adapter can be passed to pipeTo() as a destination.
Since all writes on a classic Writable are fundamentally asynchronous, the synchronous Writer methods (writeSync, writevSync, endSync) always return false or -1, deferring to the async path. The per-write options.signal parameter from the Writer interface is also ignored.
The result is cached per instance -- calling fromWritable() twice with the same stream returns the same Writer.
For duck-typed streams that do not expose writableHighWaterMark, writableLength, or similar properties, sensible defaults are used. Object-mode writables (if detectable) are rejected since the Writer interface is bytes-only.
import { Writable } from 'node:stream'; import { from, fromWritable, pipeTo } from 'node:stream/iter'; const writable = new Writable({ write(chunk, encoding, cb) { console.log(chunk.toString()); cb(); }, }); await pipeTo(from('hello world'), fromWritable(writable, { backpressure: 'block' }));
@param writable
A classic Writable stream or any object with write() and on() methods.
@returns
A stream/iter Writer adapter.
function merge(
...sources: Source[]
): ByteReadableStream;
function merge(
...args: [...sources: Source[], options: MergeOptions]
): ByteReadableStream;
function ondrain(
drainable: any
): null | Promise<boolean>;
function pipeTo(
source: Source,
writer: PartialWriter,
options?: PipeToOptions
): Promise<number>;
function pipeTo(
source: Source,
...args: [...transforms: Transform[], writer: PartialWriter]
): Promise<number>;
function pipeTo(
source: Source,
...args: [...transforms: Transform[], writer: PartialWriter, options: PipeToOptions]
): Promise<number>;
function pipeToSync(
source: SyncSource,
writer: PartialSyncWriter,
options?: PipeToSyncOptions
): number;
function pipeToSync(
source: SyncSource,
...args: [...transforms: SyncTransform[], writer: PartialSyncWriter]
): number;
function pipeToSync(
source: SyncSource,
...args: [...transforms: SyncTransform[], writer: PartialSyncWriter, options: PipeToSyncOptions]
): number;
function pull(
source: Source,
...transforms: Transform[]
): ByteReadableStream;
function pull(
source: Source,
...args: [...transforms: Transform[], options: PullOptions]
): ByteReadableStream;
function pullSync(
source: SyncSource,
...transforms: SyncTransform[]
): SyncByteReadableStream;
function push(
...transforms: Transform[]
): PushStreamResult;
function push(
...args: [...transforms: Transform[], options: PushStreamOptions]
): PushStreamResult;
function share(
source: Source,
options?: ShareOptions
): Share;
function shareSync(
source: SyncSource,
options?: ShareSyncOptions
): SyncShare;
function tap(
callback: StatelessTransformFn
): StatelessTransformFn;
function tapSync(
callback: SyncStatelessTransformFn
): SyncStatelessTransformFn;
function text(
source: Source,
options?: TextConsumeOptions
): Promise<string>;
function textSync(
source: SyncSource,
options?: TextConsumeSyncOptions
): string;
function toReadable(
source: Source,
options?: ToReadableOptions
): Readable;
Creates a byte-mode stream.Readable from an AsyncIterable<Uint8Array[]> (the native batch format used by the stream/iter API). Each Uint8Array in a yielded batch is pushed as a separate chunk into the Readable.
import { createWriteStream } from 'node:fs'; import { from, pull, toReadable } from 'node:stream/iter'; import { compressGzip } from 'node:zlib/iter'; const source = pull(from('hello world'), compressGzip()); const readable = toReadable(source); readable.pipe(createWriteStream('output.gz'));
@param source
An AsyncIterable<Uint8Array[]> source, such as the return value of pull() or from().
function toReadableSync(
source: SyncSource,
options?: ToReadableSyncOptions
): Readable;
Creates a byte-mode stream.Readable from a synchronous Iterable<Uint8Array[]>. The _read() method pulls from the iterator synchronously, so data is available immediately via readable.read().
import { fromSync, toReadableSync } from 'node:stream/iter'; const source = fromSync('hello world'); const readable = toReadableSync(source); console.log(readable.read().toString()); // 'hello world'
@param source
An Iterable<Uint8Array[]> source, such as the return value of pullSync() or fromSync().
function toWritable(
writer: PartialWriter
): Writable;
Creates a classic stream.Writable backed by a stream/iter Writer.
Each _write() / _writev() call attempts the Writer's synchronous method first (writeSync / writevSync), falling back to the async method if the sync path returns false or throws. Similarly, _final() tries endSync() before end(). When the sync path succeeds, the callback is deferred via queueMicrotask to preserve the async resolution contract.
The Writable's highWaterMark is set to Number.MAX_SAFE_INTEGER to effectively disable its internal buffering, allowing the underlying Writer to manage backpressure directly.
import { push, toWritable } from 'node:stream/iter'; const { writer, readable } = push(); const writable = toWritable(writer); writable.write('hello'); writable.end();
@param writer
A stream/iter Writer. Only the write() method is required; end(), fail(), writeSync(), writevSync(), endSync(), and writev() are optional.

Type definitions

interface AsyncStreamable
- [toAsyncStreamable](): Source;
interface Broadcast
- readonly bufferSize: number
- readonly consumerCount: number
- [Symbol.dispose](): void;
- cancel(
  reason?: any
  ): void;
- push(
  ...args: any[]
  ): ByteReadableStream;
interface Broadcastable
- [broadcastProtocol](
  options: BroadcastOptions
  ): Broadcast;
interface BroadcastOptions
- backpressure?: BackpressurePolicy
- highWaterMark?: number
- signal?: AbortSignal
interface BroadcastResult
- broadcast: Broadcast
- writer: BroadcastWriter
interface BroadcastWriter
- readonly desiredSize: null | number
- [Symbol.asyncDispose](): PromiseLike<void>;
- [Symbol.dispose](): void;
- [drainableProtocol](): null | Promise<boolean>;
- end(
  options?: WriteOptions
  ): Promise<number>;
- endSync(): number;
- fail(
  reason?: any
  ): void;
- write(
  chunk: string | Uint8Array<ArrayBufferLike>,
  options?: WriteOptions
  ): Promise<void>;
- writeSync(
  chunk: string | Uint8Array<ArrayBufferLike>
  ): boolean;
- writev(
  chunks: string | Uint8Array<ArrayBufferLike>[],
  options?: WriteOptions
  ): Promise<void>;
- writevSync(
  chunks: string | Uint8Array<ArrayBufferLike>[]
  ): boolean;
interface ConsumeOptions
- limit?: number
- signal?: AbortSignal
interface ConsumeSyncOptions
- limit?: number
interface Drainable
- [drainableProtocol](): null | Promise<boolean>;
interface DuplexChannel
- readonly readable: ByteReadableStream
- readonly writer: Writer
- [Symbol.asyncDispose](): PromiseLike<void>;
- close(): Promise<void>;
interface DuplexDirectionOptions
- backpressure?: BackpressurePolicy
- highWaterMark?: number
interface DuplexOptions
- a?: DuplexDirectionOptions
- b?: DuplexDirectionOptions
- backpressure?: BackpressurePolicy
- highWaterMark?: number
- signal?: AbortSignal
interface FromWritableOptions
- backpressure?: BackpressurePolicy
interface MergeOptions
- signal?: AbortSignal
interface PartialSyncWriter
- [Symbol.dispose]?: () => void
- readonly desiredSize?: null | number
- endSync?: () => number
- fail?: (reason?: any) => void
- writevSync?: (chunks: string | Uint8Array<ArrayBufferLike>[]) => number
- writeSync(
  chunk: string | Uint8Array<ArrayBufferLike>
  ): number;
interface PartialWriter
- [Symbol.asyncDispose]?: () => PromiseLike<void>
- [Symbol.dispose]?: () => void
- readonly desiredSize?: null | number
- end?: (options?: WriteOptions) => Promise<number>
- endSync?: () => number
- fail?: (reason?: any) => void
- writeSync?: (chunk: string | Uint8Array<ArrayBufferLike>) => boolean
- writev?: (chunks: string | Uint8Array<ArrayBufferLike>[], options?: WriteOptions) => Promise<void>
- writevSync?: (chunks: string | Uint8Array<ArrayBufferLike>[]) => boolean
- write(
  chunk: string | Uint8Array<ArrayBufferLike>,
  options?: WriteOptions
  ): Promise<void>;
interface PipeToOptions
- preventClose?: boolean
- preventFail?: boolean
- signal?: AbortSignal
interface PipeToSyncOptions
- preventClose?: boolean
- preventFail?: boolean
interface PullOptions
- signal?: AbortSignal
interface PushStreamOptions
- backpressure?: BackpressurePolicy
- highWaterMark?: number
- signal?: AbortSignal
interface PushStreamResult
- readable: ByteReadableStream
- writer: PushWriter
interface PushWriter
- readonly desiredSize: null | number
- [Symbol.asyncDispose](): PromiseLike<void>;
- [Symbol.dispose](): void;
- [drainableProtocol](): null | Promise<boolean>;
- end(
  options?: WriteOptions
  ): Promise<number>;
- endSync(): number;
- fail(
  reason?: any
  ): void;
- write(
  chunk: string | Uint8Array<ArrayBufferLike>,
  options?: WriteOptions
  ): Promise<void>;
- writeSync(
  chunk: string | Uint8Array<ArrayBufferLike>
  ): boolean;
- writev(
  chunks: string | Uint8Array<ArrayBufferLike>[],
  options?: WriteOptions
  ): Promise<void>;
- writevSync(
  chunks: string | Uint8Array<ArrayBufferLike>[]
  ): boolean;
interface Share
- readonly bufferSize: number
- readonly consumerCount: number
- [Symbol.dispose](): void;
- cancel(
  reason?: any
  ): void;
- pull(
  ...args: any[]
  ): ByteReadableStream;
interface Shareable
- [shareProtocol](
  options: ShareOptions
  ): Share;
interface ShareOptions
- backpressure?: BackpressurePolicy
- highWaterMark?: number
- signal?: AbortSignal
interface ShareSyncOptions
- backpressure?: BackpressurePolicy
- highWaterMark?: number
interface StatefulTransform
- transform(
  source: AsyncIterable<null | Uint8Array<ArrayBufferLike>[]>,
  options: TransformCallbackOptions
  ): AsyncIterable<TransformResult>;
interface StatelessTransformFn
interface Streamable
- [toStreamable](): SyncSource;
interface SyncShare
- readonly bufferSize: number
- readonly consumerCount: number
- [Symbol.dispose](): void;
- cancel(
  reason?: any
  ): void;
- pull(
  ...args: any
  ): SyncByteReadableStream;
interface SyncShareable
- [shareSyncProtocol](
  options: ShareSyncOptions
  ): SyncShare;
interface SyncStatefulTransform
- transform(
  source: Iterable<null | Uint8Array<ArrayBufferLike>[]>
  ): Iterable<SyncTransformResult>;
interface SyncStatelessTransformFn
interface SyncWriter
- readonly desiredSize: null | number
- [Symbol.dispose](): void;
- endSync(): number;
- fail(
  reason?: any
  ): void;
- writeSync(
  chunk: string | Uint8Array<ArrayBufferLike>
  ): number;
- writevSync(
  chunks: string | Uint8Array<ArrayBufferLike>[]
  ): number;
interface TextConsumeOptions
- encoding?: string
- limit?: number
- signal?: AbortSignal
interface TextConsumeSyncOptions
- encoding?: string
- limit?: number
interface ToReadableOptions
- highWaterMark?: number
- signal?: AbortSignal
interface ToReadableSyncOptions
- highWaterMark?: number
interface TransformCallbackOptions
- signal: AbortSignal
interface WriteOptions
- signal?: AbortSignal
interface Writer
- readonly desiredSize: null | number
- [Symbol.asyncDispose](): PromiseLike<void>;
- [Symbol.dispose](): void;
- end(
  options?: WriteOptions
  ): Promise<number>;
- endSync(): number;
- fail(
  reason?: any
  ): void;
- write(
  chunk: string | Uint8Array<ArrayBufferLike>,
  options?: WriteOptions
  ): Promise<void>;
- writeSync(
  chunk: string | Uint8Array<ArrayBufferLike>
  ): boolean;
- writev(
  chunks: string | Uint8Array<ArrayBufferLike>[],
  options?: WriteOptions
  ): Promise<void>;
- writevSync(
  chunks: string | Uint8Array<ArrayBufferLike>[]
  ): boolean;
type BackpressurePolicy = 'strict' | 'block' | 'drop-oldest' | 'drop-newest'
type ByteReadableStream = AsyncIterable<Uint8Array[]>
type Source = string | ArrayBufferLike | ArrayBufferView | Iterable<SyncSource> | AsyncIterable<Source> | Streamable | AsyncStreamable
type SyncByteReadableStream = Iterable<Uint8Array[]>
type SyncSource = string | ArrayBufferLike | ArrayBufferView | Iterable<SyncSource> | Streamable
type SyncTransform = SyncStatelessTransformFn | SyncStatefulTransform
type SyncTransformResult = string | ArrayBufferLike | ArrayBufferView | Iterable<SyncTransformResult>
type Transform = StatelessTransformFn | StatefulTransform
type TransformResult = string | ArrayBufferLike | ArrayBufferView | Iterable<SyncTransformResult> | AsyncIterable<TransformResult>

node:stream/iter

namespace Broadcast

namespace Share

namespace Stream

namespace SyncShare

Type definitions

interface AsyncStreamable

interface Broadcast

interface Broadcastable

interface BroadcastOptions

interface BroadcastResult

interface BroadcastWriter

interface ConsumeOptions

interface ConsumeSyncOptions

interface Drainable

interface DuplexChannel

interface DuplexDirectionOptions

interface DuplexOptions

interface FromWritableOptions

interface MergeOptions

interface PartialSyncWriter

interface PartialWriter

interface PipeToOptions

interface PipeToSyncOptions

interface PullOptions

interface PushStreamOptions

interface PushStreamResult

interface PushWriter

interface Share

interface Shareable

interface ShareOptions

interface ShareSyncOptions

interface StatefulTransform

interface StatelessTransformFn

interface Streamable

interface SyncShare

interface SyncShareable

interface SyncStatefulTransform

interface SyncStatelessTransformFn

interface SyncWriter

interface TextConsumeOptions

interface TextConsumeSyncOptions

interface ToReadableOptions

interface ToReadableSyncOptions

interface TransformCallbackOptions

interface WriteOptions

interface Writer