Iterable Streams

Iterable Streams

Stability: 1 - Experimental – Enable this API with the [--experimental-stream-iter][] CLI flag.

The node:stream/iter module provides a streaming API built on iterables rather than the event-driven Readable/Writable/Transform class hierarchy, or the Web Streams ReadableStream/WritableStream/TransformStream interfaces.

Streams are represented as {AsyncIterable} (async) or {Iterable} (sync). There are no base classes to extend -- any object implementing the iterable protocol can participate. Transforms are plain functions or objects with a transform method.

Data flows in batches ({Uint8Array[]} per iteration) to amortize the cost of async operations.

The module implements the WinterTC [Iterable Streams API][] draft. The classic stream interop functions ([fromReadable()][], [fromWritable()][], [toReadable()][], [toReadableSync()][] and [toWritable()][]), [Broadcast.from()][], [Share.from()][], [SyncShare.fromSync()][] and the protocol symbols exported by Stream are Node.js extensions that are not part of the draft.

import { from, pull, text } from 'node:stream/iter';
import { compressGzip, decompressGzip } from 'node:zlib/iter';

// Compress and decompress a string
const compressed = pull(from('Hello, world!'), compressGzip());
const result = await text(pull(compressed, decompressGzip()));
console.log(result); // 'Hello, world!'

const { from, pull, text } = require('node:stream/iter');
const { compressGzip, decompressGzip } = require('node:zlib/iter');

async function run() {
  // Compress and decompress a string
  const compressed = pull(from('Hello, world!'), compressGzip());
  const result = await text(pull(compressed, decompressGzip()));
  console.log(result); // 'Hello, world!'
}

run().catch(console.error);

import { open } from 'node:fs/promises';
import { text, pipeTo } from 'node:stream/iter';
import { compressGzip, decompressGzip } from 'node:zlib/iter';

// Read a file, compress, write to another file
const src = await open('input.txt', 'r');
const dst = await open('output.gz', 'w');
await pipeTo(src.pull(), compressGzip(), dst.writer({ autoClose: true }));
await src.close();

// Read it back
const gz = await open('output.gz', 'r');
console.log(await text(gz.pull(decompressGzip(), { autoClose: true })));

const { open } = require('node:fs/promises');
const { text, pipeTo } = require('node:stream/iter');
const { compressGzip, decompressGzip } = require('node:zlib/iter');

async function run() {
  // Read a file, compress, write to another file
  const src = await open('input.txt', 'r');
  const dst = await open('output.gz', 'w');
  await pipeTo(src.pull(), compressGzip(), dst.writer({ autoClose: true }));
  await src.close();

  // Read it back
  const gz = await open('output.gz', 'r');
  console.log(await text(gz.pull(decompressGzip(), { autoClose: true })));
}

run().catch(console.error);

Concepts

Byte streams

All data in this API is represented as {Uint8Array} bytes. Strings are automatically UTF-8 encoded when passed to from(), push(), or pipeTo(). This removes ambiguity around encodings and enables zero-copy transfers between streams and native code.

Batching

Each iteration yields a batch -- an {Array} of {Uint8Array} chunks ({Uint8Array[]}). Batching amortizes the cost of await and {Promise} creation across multiple chunks. A consumer that processes one chunk at a time can simply iterate the inner array:

for await (const batch of source) {
  for (const chunk of batch) {
    handle(chunk);
  }
}

async function run() {
  for await (const batch of source) {
    for (const chunk of batch) {
      handle(chunk);
    }
  }
}

Transforms

Transforms come in two forms:

Stateless -- a function (chunks, options) => result called once per batch. Receives Uint8Array[] (or null as the flush signal) and an options object. Returns {Uint8Array[]|null|Iterable}.
Stateful -- an object { transform(source, options) } where transform is a generator (sync or async) that receives the entire upstream iterable and an options object, and yields output. This form is used for compression, encryption, and any transform that needs to buffer across batches.

Both forms receive an options parameter with the following property:

options.signal {AbortSignal} An AbortSignal that fires when the pipeline is cancelled, encounters an error, or the consumer stops reading. Transforms can check signal.aborted or listen for the 'abort' event to perform early cleanup.

The flush signal (null) is sent after the source ends, giving transforms a chance to emit trailing data (e.g., compression footers).

// Stateless: uppercase transform
const upper = (chunks) => {
  if (chunks === null) return null; // flush
  return chunks.map((c) => new TextEncoder().encode(
    new TextDecoder().decode(c).toUpperCase(),
  ));
};

// Stateful: line splitter
const lines = {
  transform: async function*(source) {
    let partial = '';
    for await (const chunks of source) {
      if (chunks === null) {
        if (partial) yield [new TextEncoder().encode(partial)];
        continue;
      }
      for (const chunk of chunks) {
        const str = partial + new TextDecoder().decode(chunk);
        const parts = str.split('\n');
        partial = parts.pop();
        for (const line of parts) {
          yield [new TextEncoder().encode(`${line}\n`)];
        }
      }
    }
  },
};

Pull vs. push

The API supports two models:

Pull -- data flows on demand. pull() and pullSync() create lazy pipelines that only read from the source when the consumer iterates.
Push -- data is written explicitly. push() creates a writer/readable pair with backpressure. The writer pushes data in; the readable is consumed as an async iterable.

Backpressure

Pull streams have natural backpressure -- the consumer drives the pace, so the source is never read faster than the consumer can process. Push streams need explicit backpressure because the producer and consumer run independently. The budget and backpressure options on push(), broadcast(), and share() control how this works.

The two-buffer model

Push streams use a two-part buffering system. Think of it like a bucket (buffer) being filled through a hose (pending writes), with a float valve that closes when the bucket is full:

                          budget (e.g., 16384)
                                 |
    Producer                     v
       |                    +---------+
       v                    |         |
  [ write() ] ----+    +--->| buffer  |---> Consumer pulls
  [ write() ]     |    |    | (bucket)|     for await (...)
  [ write() ]     v    |    +---------+
              +--------+         ^
              | pending|         |
              | writes |    float valve
              | (hose) |    (backpressure)
              +--------+
                   ^
                   |
          'strict' mode limits this too!

Buffer (the bucket) -- data ready for the consumer, capped at budget bytes. When the consumer pulls, it drains all buffered data at once into a single batch.
Pending writes (the hose) -- writes waiting for buffer space. After the consumer drains, pending writes are promoted into the now-empty buffer and their promises settle.

How each policy uses these buffers:

Policy	Buffer limit	Pending writes limit
`'strict'`	`budget`	1
`'unbounded'`	`budget`	Unbounded
`'drop-oldest'`	`budget`	N/A (never waits)
`'drop-newest'`	`budget`	N/A (never waits)

Strict (default)

Strict mode catches “fire-and-forget” patterns where the producer calls write() without awaiting, which would cause unbounded memory growth. It limits the buffer to budget bytes and the pending writes queue to a single entry.

If you properly await each write, you can only ever have one pending write at a time (yours), so you never hit the pending writes limit. Unawaited writes accumulate in the pending queue and throw once it overflows:

import { push, text } from 'node:stream/iter';

const { writer, readable } = push({ budget: 16384 });

// Consumer must run concurrently -- without it, the first write
// that fills the buffer blocks the producer forever.
const consuming = text(readable);

// GOOD: awaited writes. The producer waits for the consumer to
// make room when the buffer is full.
for (const item of dataset) {
  await writer.write(item);
}
await writer.end();
console.log(await consuming);

const { push, text } = require('node:stream/iter');

async function run() {
  const { writer, readable } = push({ budget: 16384 });

  // Consumer must run concurrently -- without it, the first write
  // that fills the buffer blocks the producer forever.
  const consuming = text(readable);

  // GOOD: awaited writes. The producer waits for the consumer to
  // make room when the buffer is full.
  for (const item of dataset) {
    await writer.write(item);
  }
  await writer.end();
  console.log(await consuming);
}

run().catch(console.error);

Forgetting to await will eventually throw:

// BAD: fire-and-forget. Strict mode throws once both buffers fill.
for (const item of dataset) {
  writer.write(item); // Not awaited -- queues without bound
}
// --> throws "Backpressure violation: too many pending writes"

Unbounded

Unbounded mode caps buffered bytes at budget but places no limit on the pending writes queue. Awaited writes block until the consumer makes room, just like strict mode. The difference is that unawaited writes silently queue forever instead of throwing -- a potential memory leak if the producer forgets to await.

This is the mode that existing Node.js classic streams and Web Streams default to. Use it when you control the producer and know it awaits properly, or when migrating code from those APIs.

import { push, text } from 'node:stream/iter';

const { writer, readable } = push({
  budget: 16384,
  backpressure: 'unbounded',
});

const consuming = text(readable);

// Safe -- awaited writes block until the consumer reads.
for (const item of dataset) {
  await writer.write(item);
}
await writer.end();
console.log(await consuming);

const { push, text } = require('node:stream/iter');

async function run() {
  const { writer, readable } = push({
    budget: 16384,
    backpressure: 'unbounded',
  });

  const consuming = text(readable);

  // Safe -- awaited writes block until the consumer reads.
  for (const item of dataset) {
    await writer.write(item);
  }
  await writer.end();
  console.log(await consuming);
}

run().catch(console.error);

Drop-oldest

Writes never wait. When the slots buffer is full, the oldest buffered chunk is evicted to make room for the incoming write. The consumer always sees the most recent data. Useful for live feeds, telemetry, or any scenario where stale data is less valuable than current data.

import { push } from 'node:stream/iter';

// Keep only the most recent ~16 KB of readings
const { writer, readable } = push({
  budget: 16384,
  backpressure: 'drop-oldest',
});

const { push } = require('node:stream/iter');

// Keep only the most recent ~16 KB of readings
const { writer, readable } = push({
  budget: 16384,
  backpressure: 'drop-oldest',
});

Drop-newest

Writes never wait. When the slots buffer is full, the incoming write is silently discarded. The consumer processes what is already buffered without being overwhelmed by new data. Useful for rate-limiting or shedding load under pressure.

import { push } from 'node:stream/iter';

// Accept up to 16 KB of buffered data; discard anything beyond that
const { writer, readable } = push({
  budget: 16384,
  backpressure: 'drop-newest',
});

const { push } = require('node:stream/iter');

// Accept up to 16 KB of buffered data; discard anything beyond that
const { writer, readable } = push({
  budget: 16384,
  backpressure: 'drop-newest',
});

Writer interface

A writer is any object conforming to the Writer interface. Only write() is required; all other methods are optional.

Writer arguments use Web IDL conversion semantics. A non-Uint8Array chunk is converted to a USVString and then UTF-8 encoded. writev() and writevSync() accept any iterable object whose values can be converted to chunks. Writer option dictionaries treat null as an empty dictionary and ignore unknown members.

Arguments are converted before the write itself starts. If the conversion runs user code (for example a toString() method, or the iterator of a writev() argument) that writes to the same writer, those writes are ordered before the write whose argument is being converted, and they count against the same backpressure limits.

After end() or endSync() has been called, and until all buffered data has been consumed, the writer is closing. While closing, canWrite is null, write() and writev() reject with a TypeError, writeSync() and writevSync() return false, endSync() returns -1, and calling end() again returns the same promise as the first call.

Each async method has a synchronous *Sync counterpart designed for a try-fallback pattern: attempt the fast synchronous path first, and fall back to the async version only when the synchronous call indicates it could not complete:

if (!writer.writeSync(chunk)) await writer.write(chunk);
if (!writer.writevSync(chunks)) await writer.writev(chunks);
if (writer.endSync() < 0) await writer.end();
writer.fail(err);  // Always synchronous, no fallback needed

`writer.canWrite`

{boolean|null}

Returns true if the slots buffer has physical capacity (buffered data is below the configured byte budget), false if the budget is exhausted, or null if the writer is closed or the consumer has disconnected.

This reports physical capacity independently of the backpressure policy. With 'drop-oldest' or 'drop-newest', writes still complete when this is false by evicting buffered data or discarding the incoming data, respectively.

This is a hint, not a guarantee: the state can change between the check and the write. Use [ondrain()][] to wait for capacity rather than polling.

`writer.end([options])`

options {Object}
signal {AbortSignal} Cancel just this operation. The signal cancels only the pending end() call; it does not fail the writer itself.
Returns: {Promise} Fulfills with the total number of bytes written.

Signals that no more data will be written. Writes already waiting for buffer space remain ordered before the end of the stream, while later writes fail. If data is outstanding, the returned promise fulfills after the consumer pulls done: true beyond the final batch. If no data is buffered or pending, the writer closes immediately.

`writer.endSync()`

Returns: {number} Total bytes written, or -1 if ending cannot complete synchronously.

Synchronous variant of writer.end(). A return value of -1 only indicates that the operation could not complete synchronously; no assumption can be made about whether closing has started or why it could not complete. Use the try-fallback pattern to await completion:

const result = writer.endSync();
if (result < 0) {
  writer.end();
}

`writer.fail([reason])`

reason {any}

Put the writer into a terminal error state. If the writer is already closed or errored, this is a no-op. Unlike write() and end(), fail() is unconditionally synchronous because failing a writer is a pure state transition with no async work to perform. The reason is stored and propagated without modification. If omitted, the reason is undefined.

`writer[Symbol.asyncDispose]()`

If the writer is open, calls writer.fail(). If the writer is closing after end() or endSync(), waits for buffered data to drain. If the writer is already closed or errored, resolves immediately.

`writer.write(chunk[, options])`

chunk {Uint8Array|string}
options {Object}
signal {AbortSignal} Cancel just this write operation. The signal cancels only the pending write() call; it does not fail the writer itself.
Returns: {Promise} Fulfills with undefined when buffer space is available.

Write a chunk.

`writer.writeSync(chunk)`

chunk {Uint8Array|string}
Returns: {boolean} true if the write was accepted, false if the buffer is full.

Synchronous write. Does not block; returns false if backpressure is active.

`writer.writev(chunks[, options])`

chunks {Iterable} of {Uint8Array|string} values
options {Object}
signal {AbortSignal} Cancel just this write operation. The signal cancels only the pending writev() call; it does not fail the writer itself.
Returns: {Promise}

Write multiple chunks as a single batch.

`writer.writevSync(chunks)`

chunks {Iterable} of {Uint8Array|string} values
Returns: {boolean} true if the write was accepted, false if the buffer is full.

Synchronous batch write.

The `stream/iter` module

Most functions are available both as named exports and as properties of the Stream namespace object. The classic stream adapters (fromReadable(), fromWritable(), toReadable(), toReadableSync(), and toWritable()) and the static helper objects (Broadcast, Share, and SyncShare) are named exports only.

// Named exports
import { from, pull, bytes, Stream } from 'node:stream/iter';

// Namespace access
Stream.from('hello');

Options dictionaries defined by the Iterable Streams API use Web IDL conversion semantics. null is treated as an empty dictionary, unknown members are ignored, and known members are converted to their declared types before the operation runs. Conversion failures use Node.js error codes such as ERR_INVALID_ARG_TYPE, ERR_INVALID_ARG_VALUE, and ERR_OUT_OF_RANGE.

// Named exports
const { from, pull, bytes, Stream } = require('node:stream/iter');

// Namespace access
Stream.from('hello');

Including the node: prefix on the module specifier is optional.

Sources

`from(input)`

input {string|ArrayBuffer|ArrayBufferView|Iterable|AsyncIterable|Object} Must not be null or undefined.
Returns: {AsyncIterable} whose chunks fulfill with {Uint8Array[]}.

Create an async byte stream from the given input. Strings are UTF-8 encoded. ArrayBuffer and ArrayBufferView values are wrapped as Uint8Array. Arrays and iterables in input are recursively flattened and normalized. Flattened values may be split across implementation-defined bounded batches.

Objects implementing Symbol.for('Stream.toAsyncStreamable') or Symbol.for('Stream.toStreamable') are converted via those protocols. The toAsyncStreamable protocol takes precedence over toStreamable, which takes precedence over the iteration protocols (Symbol.asyncIterator, Symbol.iterator).

import { Buffer } from 'node:buffer';
import { from, text } from 'node:stream/iter';

console.log(await text(from('hello')));       // 'hello'
console.log(await text(from(Buffer.from('hello')))); // 'hello'

const { Buffer } = require('node:buffer');
const { from, text } = require('node:stream/iter');

async function run() {
  console.log(await text(from('hello')));       // 'hello'
  console.log(await text(from(Buffer.from('hello')))); // 'hello'
}

run().catch(console.error);

`fromSync(input)`

input {string|ArrayBuffer|ArrayBufferView|Iterable|Object} Must not be null or undefined.
Returns: {Iterable} whose chunks return {Uint8Array[]}

Synchronous version of [from()][]. Returns a sync iterable. Cannot accept async iterables or promises. Objects implementing Symbol.for('Stream.toStreamable') are converted via that protocol (takes precedence over Symbol.iterator). The toAsyncStreamable protocol is ignored entirely.

import { fromSync, textSync } from 'node:stream/iter';

console.log(textSync(fromSync('hello'))); // 'hello'

const { fromSync, textSync } = require('node:stream/iter');

console.log(textSync(fromSync('hello'))); // 'hello'

Pipelines

`pipeTo(source[, ...transforms], writer[, options])`

source {AsyncIterable|Iterable} The data source.
...transforms {Function|Object} Zero or more transforms to apply.
writer {Object} Destination with write(chunk) method.
options {Object}
signal {AbortSignal} Abort the pipeline. Aborting fails the destination writer unless preventFail is true.
preventClose {boolean} If true, do not call writer.end() when the source ends. Default: false.
preventFail {boolean} If true, do not call writer.fail() on error. Default: false.
Returns: {Promise} Fulfills with the total number of bytes written.

Pipe a source through transforms into a writer. If the writer has a writev(chunks) method, entire batches are passed in a single call (enabling scatter/gather I/O).

If the writer implements the optional *Sync methods (writeSync, writevSync, endSync), pipeTo() will attempt to use the synchronous methods first as a fast path, and fall back to the async versions only when the sync methods indicate they cannot complete (e.g., backpressure or waiting for the next tick). fail() is always called synchronously.

import { from, pipeTo } from 'node:stream/iter';
import { compressGzip } from 'node:zlib/iter';
import { open } from 'node:fs/promises';

const fh = await open('output.gz', 'w');
const totalBytes = await pipeTo(
  from('Hello, world!'),
  compressGzip(),
  fh.writer({ autoClose: true }),
);

const { from, pipeTo } = require('node:stream/iter');
const { compressGzip } = require('node:zlib/iter');
const { open } = require('node:fs/promises');

async function run() {
  const fh = await open('output.gz', 'w');
  const totalBytes = await pipeTo(
    from('Hello, world!'),
    compressGzip(),
    fh.writer({ autoClose: true }),
  );
}

run().catch(console.error);

`pipeToSync(source[, ...transforms], writer[, options])`

source {Iterable} The sync data source.
...transforms {Function|Object} Zero or more sync transforms.
writer {Object} Destination with write(chunk) method.
options {Object}
failOnIncompleteClose {boolean} If true, call writer.fail() when writer.endSync() cannot close the writer synchronously. Ignored when preventFail is true. This option is a Node.js extension. Default: false.
preventClose {boolean} Default: false.
preventFail {boolean} Default: false.
Returns: {number} Total bytes written.

Synchronous version of [pipeTo()][]. The source, all transforms, and the writer must be synchronous. Cannot accept async iterables or promises.

The writer must have the *Sync methods (writeSync, writevSync, endSync) and fail() for this to work.

pipeToSync() never falls back to the asynchronous writer methods. If writer.endSync() returns -1 because the writer cannot close synchronously (for example, a push() writer whose consumer has not read all of the data yet), pipeToSync() throws ERR_INVALID_STATE. All of the data was accepted by then, so by default the writer is not failed: it can still be closed, for example with await writer.end(). If the writer cannot be closed any other way (for example, it has no end() method), or the caller will not close it, set failOnIncompleteClose to fail it with the thrown error instead.

`pull(source[, ...transforms][, options])`

source {AsyncIterable|Iterable} The data source.
...transforms {Function|Object} Zero or more transforms to apply.
options {Object}
signal {AbortSignal} Abort the pipeline.
Returns: {AsyncIterable} whose chunks fulfill with {Uint8Array[]}

Create a lazy async pipeline. Source conversion and streamable protocol dispatch occur when pull() is called, but data is not read from source until the returned iterable is consumed. Transforms are applied in order.

When signal aborts, the pending read (or the next one) rejects with signal.reason, and so does every later read. If signal is already aborted, pull() still returns an iterable; reading from it rejects with signal.reason without reading from source.

import { from, pull, text } from 'node:stream/iter';

const asciiUpper = (chunks) => {
  if (chunks === null) return null;
  return chunks.map((c) => {
    for (let i = 0; i < c.length; i++) {
      c[i] -= (c[i] >= 97 && c[i] <= 122) * 32;
    }
    return c;
  });
};

const result = pull(from('hello'), asciiUpper);
console.log(await text(result)); // 'HELLO'

const { from, pull, text } = require('node:stream/iter');

const asciiUpper = (chunks) => {
  if (chunks === null) return null;
  return chunks.map((c) => {
    for (let i = 0; i < c.length; i++) {
      c[i] -= (c[i] >= 97 && c[i] <= 122) * 32;
    }
    return c;
  });
};

async function run() {
  const result = pull(from('hello'), asciiUpper);
  console.log(await text(result)); // 'HELLO'
}

run().catch(console.error);

Using an AbortSignal:

import { pull } from 'node:stream/iter';

const ac = new AbortController();
const result = pull(source, transform, { signal: ac.signal });
ac.abort(); // Pipeline throws AbortError on next iteration

const { pull } = require('node:stream/iter');

const ac = new AbortController();
const result = pull(source, transform, { signal: ac.signal });
ac.abort(); // Pipeline throws AbortError on next iteration

`pullSync(source[, ...transforms])`

source {Iterable} The sync data source.
...transforms {Function|Object} Zero or more sync transforms.
Returns: {Iterable} whose chunks return {Uint8Array[]}

Synchronous version of [pull()][]. Source conversion and streamable protocol dispatch occur when pullSync() is called. All transforms must be synchronous.

Push streams

`push([...transforms][, options])`

...transforms {Function|Object} Optional transforms applied to the readable side.
options {Object}
budget {number} Maximum number of buffered bytes before backpressure is applied. Must be a positive integer. Default: 16384.
backpressure {string} Backpressure policy: 'strict', 'unbounded', 'drop-oldest', or 'drop-newest'. Default: 'strict'.
signal {AbortSignal} Abort the stream. The signal remains active while buffered data drains after writer.end(); aborting during that time fails the writer and rejects the pending end() promise.
Returns: {Object}
writer {Writable} The writer side.
readable {AsyncIterable} whose chunks fulfill with {Uint8Array[]}

Create a push stream with backpressure. The writer pushes data in; the readable side is consumed as an async iterable.

import { push, text } from 'node:stream/iter';

const { writer, readable } = push();

// Producer and consumer must run concurrently. With strict backpressure
// (the default), awaited writes block until the consumer reads.
const producing = (async () => {
  await writer.write('hello');
  await writer.write(' world');
  await writer.end();
})();

console.log(await text(readable)); // 'hello world'
await producing;

const { push, text } = require('node:stream/iter');

async function run() {
  const { writer, readable } = push();

  // Producer and consumer must run concurrently. With strict backpressure
  // (the default), awaited writes block until the consumer reads.
  const producing = (async () => {
    await writer.write('hello');
    await writer.write(' world');
    await writer.end();
  })();

  console.log(await text(readable)); // 'hello world'
  await producing;
}

run().catch(console.error);

The writer returned by push() conforms to the [Writer interface][].

Zero-length chunks are accepted without being buffered: they are not delivered to the consumer, and writeSync() and write() report success for them even when backpressure is active.

Duplex channels

`duplex([options])`

options {Object}
budget {number} Buffer size in bytes for both directions. Must be a positive integer. Default: 16384.
backpressure {string} Policy for both directions. Default: 'strict'.
signal {AbortSignal} Cancellation signal for both channels.
a {Object} Options specific to the A-to-B direction. Overrides shared options.
- budget {number}
- backpressure {string}
b {Object} Options specific to the B-to-A direction. Overrides shared options.
- budget {number}
- backpressure {string}
Returns: {Array} A pair [channelA, channelB] of duplex channels.

Create a pair of connected duplex channels for bidirectional communication, similar to socketpair(). Data written to one channel's writer appears in the other channel's readable.

Each channel has:

writer — a [Writer interface][] object for sending data to the peer.
readable — an {AsyncIterable} for reading data from the peer.
close() — close this end of the channel (idempotent).
[Symbol.asyncDispose]() — async dispose support for await using.

import { duplex, text } from 'node:stream/iter';

const [client, server] = duplex();

// Server echoes back
const serving = (async () => {
  for await (const chunks of server.readable) {
    await server.writer.writev(chunks);
  }
  await server.writer.end();
})();

await client.writer.write('hello');
await client.writer.end();

console.log(await text(client.readable)); // 'hello'
await serving;

const { duplex, text } = require('node:stream/iter');

async function run() {
  const [client, server] = duplex();

  // Server echoes back
  const serving = (async () => {
    for await (const chunks of server.readable) {
      await server.writer.writev(chunks);
    }
    await server.writer.end();
  })();

  await client.writer.write('hello');
  await client.writer.end();

  console.log(await text(client.readable)); // 'hello'
  await serving;
}

run().catch(console.error);

Consumers

`array(source[, options])`

source {AsyncIterable|Iterable} whose chunks must be {Uint8Array[]}
options {Object}
signal {AbortSignal}
limit {number} Maximum number of bytes to consume. If the total bytes collected exceeds limit, an ERR_OUT_OF_RANGE error is thrown
Returns: {Promise} Fulfills with an array of Uint8Array objects.

Collect all chunks as an array of Uint8Array values (without concatenating).

`arrayBuffer(source[, options])`

source {AsyncIterable|Iterable} whose chunks must be {Uint8Array[]}
options {Object}
signal {AbortSignal}
limit {number} Maximum number of bytes to consume. If the total bytes collected exceeds limit, an ERR_OUT_OF_RANGE error is thrown
Returns: {Promise} Fulfills with an ArrayBuffer object.

Collect all bytes into an ArrayBuffer.

`arrayBufferSync(source[, options])`

source {Iterable} whose chunks must be {Uint8Array[]}
options {Object}
limit {number} Maximum number of bytes to consume. If the total bytes collected exceeds limit, an ERR_OUT_OF_RANGE error is thrown
Returns: {ArrayBuffer}

Synchronous version of [arrayBuffer()][].

`arraySync(source[, options])`

source {Iterable} whose chunks must be {Uint8Array[]}
options {Object}
limit {number} Maximum number of bytes to consume. If the total bytes collected exceeds limit, an ERR_OUT_OF_RANGE error is thrown
Returns: {Uint8Array[]}

Synchronous version of [array()][].

`bytes(source[, options])`

source {AsyncIterable|Iterable} whose chunks must be {Uint8Array[]}
options {Object}
signal {AbortSignal}
limit {number} Maximum number of bytes to consume. If the total bytes collected exceeds limit, an ERR_OUT_OF_RANGE error is thrown
Returns: {Promise} Fulfills with an Uint8Array object.

Collect all bytes from a stream into a single Uint8Array.

import { from, bytes } from 'node:stream/iter';

const data = await bytes(from('hello'));
console.log(data); // Uint8Array(5) [ 104, 101, 108, 108, 111 ]

const { from, bytes } = require('node:stream/iter');

async function run() {
  const data = await bytes(from('hello'));
  console.log(data); // Uint8Array(5) [ 104, 101, 108, 108, 111 ]
}

run().catch(console.error);

`bytesSync(source[, options])`

source {Iterable} whose chunks must be {Uint8Array[]}
options {Object}
limit {number} Maximum number of bytes to consume. If the total bytes collected exceeds limit, an ERR_OUT_OF_RANGE error is thrown
Returns: {Uint8Array}

Synchronous version of [bytes()][].

`dump(source[, options])`

source {AsyncIterable|Iterable} whose chunks must be {Uint8Array[]}
options {Object}
signal {AbortSignal}
limit {number} Maximum number of bytes to consume. If the total bytes read exceeds limit, an ERR_OUT_OF_RANGE error is thrown
Returns: {Promise} Fulfills with undefined.

Read a source to completion, discarding every chunk. Unlike the other consumers, dump() retains nothing. Memory tops-out at a single batch no matter how much data the source produces.

Use this to consume a stream when the content doesn't matter. For example, A QUIC stream only returns flow-control credit to the peer as its data is consumed, so a receiver that does not want the payload must still read it to completion.

If the source errors part-way through, the returned promise rejects with that error.

There is no default limit. dump() reads until the source is exhausted unless a limit is specified. When a limit is configured and the source exceeds it, the promise rejects and the source is cancelled. A partial dump is never reported as success.

import { dump, from, pull, tap } from 'node:stream/iter';

// Count the bytes flowing through a stream without retaining any of them.
let bytesSeen = 0;
const counter = tap((chunks) => {
  for (const chunk of chunks) bytesSeen += chunk.byteLength;
});

await dump(pull(from('hello world'), counter));
console.log(bytesSeen); // 11

const { dump, from, pull, tap } = require('node:stream/iter');

async function run() {
  // Count the bytes flowing through a stream without retaining any of them.
  let bytesSeen = 0;
  const counter = tap((chunks) => {
    for (const chunk of chunks) bytesSeen += chunk.byteLength;
  });

  await dump(pull(from('hello world'), counter));
  console.log(bytesSeen); // 11
}

run().catch(console.error);

`dumpSync(source[, options])`

source {Iterable} whose chunks must be {Uint8Array[]}
options {Object}
limit {number} Maximum number of bytes to consume. If the total bytes read exceeds limit, an ERR_OUT_OF_RANGE error is thrown
Returns: {undefined}

Synchronous version of [dump()][]. Throws ERR_INVALID_ARG_TYPE if source is not synchronously iterable.

`text(source[, options])`

source {AsyncIterable|Iterable} whose chunks must be {Uint8Array[]}
options {Object}
encoding {string} Text encoding. Default: 'utf-8'.
signal {AbortSignal}
limit {number} Maximum number of bytes to consume. If the total bytes collected exceeds limit, an ERR_OUT_OF_RANGE error is thrown
Returns: {Promise} Fulfills with a string.

Collect all bytes and decode as text.

import { from, text } from 'node:stream/iter';

console.log(await text(from('hello'))); // 'hello'

const { from, text } = require('node:stream/iter');

async function run() {
  console.log(await text(from('hello'))); // 'hello'
}

run().catch(console.error);

`textSync(source[, options])`

source {Iterable} whose chunks must be {Uint8Array[]}
options {Object}
encoding {string} Default: 'utf-8'.
limit {number} Maximum number of bytes to consume. If the total bytes collected exceeds limit, an ERR_OUT_OF_RANGE error is thrown
Returns: {string}

Synchronous version of [text()][].

Utilities

`ondrain(drainable)`

drainable {Object} An object implementing the drainable protocol.
Returns: {Promise|null}

Wait for a drainable writer to regain physical buffer capacity. Returns null if the object does not implement the drainable protocol, or a promise that fulfills with true when buffered data falls below the byte budget.

For writers using 'drop-oldest' or 'drop-newest', this waits for physical capacity even though writes do not block. This allows producers to avoid data loss by waiting before writing.

import { push, ondrain, text } from 'node:stream/iter';

const { writer, readable } = push({ budget: 16384 });
const chunk = new Uint8Array(8192);  // 8 KB
writer.writeSync(chunk);
writer.writeSync(chunk);  // 16 KB total -- buffer full

// Start consuming so the buffer can actually drain
const consuming = text(readable);

// Buffer is full -- wait for drain
const canWrite = await ondrain(writer);
if (canWrite) {
  await writer.write('c');
}
await writer.end();
await consuming;

const { push, ondrain, text } = require('node:stream/iter');

async function run() {
  const { writer, readable } = push({ budget: 16384 });
  const chunk = new Uint8Array(8192);  // 8 KB
  writer.writeSync(chunk);
  writer.writeSync(chunk);  // 16 KB total -- buffer full

  // Start consuming so the buffer can actually drain
  const consuming = text(readable);

  // Buffer is full -- wait for drain
  const canWrite = await ondrain(writer);
  if (canWrite) {
    await writer.write('c');
  }
  await writer.end();
  await consuming;
}

run().catch(console.error);

`merge(...sources[, options])`

...sources {AsyncIterable|Iterable} whose chunks must be {Uint8Array[]}
options {Object}
signal {AbortSignal}
Returns: {AsyncIterable} whose chunks fulfill with {Uint8Array[]}

Merge multiple async iterables by yielding batches in temporal order (whichever source produces data first). All sources are consumed concurrently, with at most one pending next() call per source.

If a source fails, the returned iterable rejects with its error. merge() calls return() on the other sources but does not wait for it to settle: an async generator source that is suspended in an await only runs its cleanup once that await completes.

import { from, merge, text } from 'node:stream/iter';

const merged = merge(from('hello '), from('world'));
console.log(await text(merged)); // Order depends on timing

const { from, merge, text } = require('node:stream/iter');

async function run() {
  const merged = merge(from('hello '), from('world'));
  console.log(await text(merged)); // Order depends on timing
}

run().catch(console.error);

`tap(callback)`

callback {Function} (chunks, options) => void Called with each batch and with null when the source ends. options.signal is the pipeline's {AbortSignal}.
Returns: {Function} A stateless transform.

Create a pass-through transform that observes batches without modifying them. Useful for logging, metrics, or debugging.

import { from, pull, text, tap } from 'node:stream/iter';

const result = pull(
  from('hello'),
  tap((chunks) => {
    if (chunks !== null) console.log('Batch size:', chunks.length);
  }),
);
console.log(await text(result));

const { from, pull, text, tap } = require('node:stream/iter');

async function run() {
  const result = pull(
    from('hello'),
    tap((chunks) => {
      if (chunks !== null) console.log('Batch size:', chunks.length);
    }),
  );
  console.log(await text(result));
}

run().catch(console.error);

tap() intentionally does not prevent in-place modification of the chunks by the tapping callback; but return values are ignored.

`tapSync(callback)`

callback {Function}
Returns: {Function}

Synchronous version of [tap()][].

Multi-consumer

`broadcast([options])`

options {Object}
budget {number} Buffer size in bytes. Must be a positive integer. Default: 65536.
backpressure {string} 'strict', 'unbounded', 'drop-oldest', or 'drop-newest'. Default: 'strict'.
signal {AbortSignal}
Returns: {Object}
writer {Writable}
broadcast {BroadcastChannel}

Create a push-model multi-consumer broadcast channel. A single writer pushes data to multiple consumers. Each consumer has an independent cursor into a shared buffer.

import { broadcast, text } from 'node:stream/iter';

const { writer, broadcast: bc } = broadcast();

// Create consumers before writing
const c1 = bc.push();  // Consumer 1
const c2 = bc.push();  // Consumer 2

// Producer and consumers must run concurrently. Awaited writes
// block when the buffer fills until consumers read.
const producing = (async () => {
  await writer.write('hello');
  await writer.end();
})();

const [r1, r2] = await Promise.all([text(c1), text(c2)]);
console.log(r1); // 'hello'
console.log(r2); // 'hello'
await producing;

const { broadcast, text } = require('node:stream/iter');

async function run() {
  const { writer, broadcast: bc } = broadcast();

  // Create consumers before writing
  const c1 = bc.push();  // Consumer 1
  const c2 = bc.push();  // Consumer 2

  // Producer and consumers must run concurrently. Awaited writes
  // block when the buffer fills until consumers read.
  const producing = (async () => {
    await writer.write('hello');
    await writer.end();
  })();

  const [r1, r2] = await Promise.all([text(c1), text(c2)]);
  console.log(r1); // 'hello'
  console.log(r2); // 'hello'
  await producing;
}

run().catch(console.error);

`broadcast.cancel([reason])`

reason {any}

Cancel the broadcast. If reason is provided, all consumers reject with that exact reason. If it is omitted, consumers complete normally.

Cancelling also closes the paired writer: afterwards its canWrite is null and write() rejects with a TypeError. This lets a [Broadcast.from()][] pump stop pulling from its source.

`broadcast.consumerCount`

{number}

The number of active consumers.

`broadcast.push([...transforms][, options])`

...transforms {Function|Object}
options {Object}
signal {AbortSignal}
Returns: {AsyncIterable} whose chunks fulfill with {Uint8Array[]}

Create a new consumer. Optional transforms are applied to this consumer's view of the data.

`broadcast[Symbol.dispose]()`

Alias for broadcast.cancel().

`Broadcast.from(input[, options])`

input {AsyncIterable|Iterable|BroadcastChannel}
options {Object} Same as broadcast().
Returns: {BroadcastChannel|Object} A broadcastProtocol input returns its {BroadcastChannel} directly. Other inputs return { writer, broadcast }.

Create a {BroadcastChannel} from an existing source. The source is consumed automatically and pushed to all subscribers.

`share(source[, options])`

source {AsyncIterable} The source to share.
options {Object}
budget {number} Buffer size in bytes. Must be a positive integer. Default: 65536.
backpressure {string} 'strict', 'unbounded', 'drop-oldest', or 'drop-newest'. Default: 'strict'.
signal {AbortSignal}
Returns: {Share}

Create a pull-model multi-consumer shared stream. Unlike broadcast(), the source is only read when a consumer pulls. Multiple consumers share a single buffer.

A consumer created with share.pull() starts reading at the oldest entry still in the buffer. Entries are released once every consumer has read them. When every consumer has detached, the b

（本文超出站点单文档体积上限，已截断）

Iterable Streams

Iterable Streams

Concepts

Byte streams

Batching

Transforms

Pull vs. push

Backpressure

The two-buffer model

Strict (default)

Unbounded

Drop-oldest

Drop-newest

Writer interface

writer.canWrite

writer.end([options])

writer.endSync()

writer.fail([reason])

writer[Symbol.asyncDispose]()

writer.write(chunk[, options])

writer.writeSync(chunk)

writer.writev(chunks[, options])

writer.writevSync(chunks)

The stream/iter module

Sources

from(input)

fromSync(input)

Pipelines

pipeTo(source[, ...transforms], writer[, options])

pipeToSync(source[, ...transforms], writer[, options])

pull(source[, ...transforms][, options])

pullSync(source[, ...transforms])

Push streams

push([...transforms][, options])

Duplex channels

duplex([options])

Consumers

array(source[, options])

arrayBuffer(source[, options])

arrayBufferSync(source[, options])

arraySync(source[, options])

bytes(source[, options])

bytesSync(source[, options])

dump(source[, options])

dumpSync(source[, options])

text(source[, options])

textSync(source[, options])

Utilities

ondrain(drainable)

merge(...sources[, options])

tap(callback)

tapSync(callback)

Multi-consumer

broadcast([options])

broadcast.cancel([reason])

broadcast.consumerCount

broadcast.push([...transforms][, options])

broadcast[Symbol.dispose]()

Broadcast.from(input[, options])

share(source[, options])

`writer.canWrite`

`writer.end([options])`

`writer.endSync()`

`writer.fail([reason])`

`writer[Symbol.asyncDispose]()`

`writer.write(chunk[, options])`

`writer.writeSync(chunk)`

`writer.writev(chunks[, options])`

`writer.writevSync(chunks)`

The `stream/iter` module

`from(input)`

`fromSync(input)`

`pipeTo(source[, ...transforms], writer[, options])`

`pipeToSync(source[, ...transforms], writer[, options])`

`pull(source[, ...transforms][, options])`

`pullSync(source[, ...transforms])`

`push([...transforms][, options])`

`duplex([options])`

`array(source[, options])`

`arrayBuffer(source[, options])`

`arrayBufferSync(source[, options])`

`arraySync(source[, options])`

`bytes(source[, options])`

`bytesSync(source[, options])`

`dump(source[, options])`

`dumpSync(source[, options])`

`text(source[, options])`

`textSync(source[, options])`

`ondrain(drainable)`

`merge(...sources[, options])`

`tap(callback)`

`tapSync(callback)`

`broadcast([options])`

`broadcast.cancel([reason])`

`broadcast.consumerCount`

`broadcast.push([...transforms][, options])`

`broadcast[Symbol.dispose]()`

`Broadcast.from(input[, options])`

`share(source[, options])`