Node.js v26.11.1 文档

概念#>

🌐 Concepts

字节流#>

🌐 Byte streams

这个 API 中的所有数据都表示为 <Uint8Array> 字节。当字符串传递给 from()、push() 或 pipeTo() 时，会自动进行 UTF-8 编码。这消除了编码的歧义，并且支持在流和本地代码之间进行零拷贝传输。

🌐 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

每次迭代会产生一个 批次 —— 一个 <Array> 的 <Uint8Array> 块（<Uint8Array[]>）。批处理可以将 await 和 <Promise> 的创建成本在多个块之间分摊。一个一次处理一个块的消费者可以直接迭代内部数组：

🌐 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:

• 无状态 — 一个函数 (chunks, options) => result，每批次调用一次。接收 Uint8Array[]（或者作为刷新信号的 null）和一个 options 对象。返回 <Uint8Array[]> | <null> | <Iterable>。
• 有状态 -- 一个对象 { transform(source, options) }，其中 transform 是一个生成器（同步或异步），接收整个上游可迭代对象和一个 options 对象，并生成输出。这种形式用于压缩、加密以及任何需要跨批处理缓冲的转换。

这两种形式都接收一个具有以下属性的 options 参数：

🌐 Both forms receive an options parameter with the following property:

• options.signal <AbortSignal> 当管道被取消、遇到错误或消费者停止读取时，会触发的 AbortSignal。Transform 可以检查 signal.aborted 或监听 'abort' 事件以执行提前清理。

在源结束后发送刷新信号（null），给转换器一个机会来输出尾随数据（例如，压缩尾部）。

🌐 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

该 API 支持两个模型：

🌐 The API supports two models:

• 拉取——数据按需流动。pull() 和 pullSync() 创建惰性管道，只有在消费者迭代时才从源读取。
• 推送 -- 数据是明确写入的。push() 创建一个具有背压的写入器/可读对象对。写入器推送数据；可读对象作为异步可迭代对象被消费。

反压#>

🌐 Backpressure

拉流（pull streams）有自然的背压 —— 消费者控制节奏，所以源数据永远不会比消费者处理得快。推流（push streams）需要显式的背压，因为生产者和消费者是独立运行的。push()、broadcast() 和 share() 上的 budget 和 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! 拷贝

• 缓冲区（桶）——为消费者准备的数据，最多为 budget 字节。当消费者拉取时，它会一次性将所有缓冲数据排入一个批次。
• 待写入（管道） -- 等待缓冲区空间的写入。在消费者清空后，待写入的内容会被提升到现在空的缓冲区中，并完成它们的承诺。

每项政策如何使用这些缓冲区：

🌐 How each policy uses these buffers:

严格（默认）#>

🌐 Strict (default)

严格模式会捕捉“发了就忘”的模式，即生产者调用 write() 而不等待，这会导致内存无限增长。它把缓冲区限制为 budget 字节，并且将待写入队列限制为单个条目。

🌐 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);拷贝

忘记 await 最终会抛出：

🌐 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

无界模式会将缓冲字节限制在 budget，但对待处理的写入队列没有限制。等待的写入会阻塞，直到消费者腾出空间，就像严格模式一样。不同的是，非等待的写入会无声地无限排队，而不是抛出错误——如果生产者忘记 await，可能会导致内存泄漏。

🌐 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.

这是现有 Node.js 经典流和 Web 流默认使用的模式。当你控制生产者并且知道它会正确等待时，或者在从这些 API 迁移代码时使用它。

🌐 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

写入操作永不等待。当槽缓冲区已满时，传入的写入会被静 silently丢弃。消费者处理已经缓冲的数据，而不会被新的数据淹没。在限流或在压力下减轻负载时非常有用。

🌐 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

作家是任何符合 Writer 接口的对象。只有 write() 是必需的；所有其他方法都是可选的。

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

写入器参数使用 Web IDL 转换语义。非 Uint8Array 的块会被转换为 USVString，然后进行 UTF-8 编码。writev() 和 writevSync() 接受任何值可以转换为块的可迭代对象。写入器选项字典将 null 视为空字典，并忽略未知成员。

🌐 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.

每个异步方法都有一个同步的 *Sync 对应方法，设计用于尝试-回退模式：首先尝试快速的同步路径，只有当同步调用表明无法完成时才回退到异步版本：

🌐 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>

如果插槽缓冲区有物理容量（缓冲的数据低于配置的字节预算），返回 true；如果预算用尽，返回 false；如果写入器已关闭或消费者已断开连接，返回 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.

这会独立于背压策略报告物理容量。使用'drop-oldest'或'drop-newest'时，当设置为false时，写入仍然会完成，分别通过驱逐缓存数据或丢弃传入数据。

🌐 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.

这只是一个提示，不是保证：状态可能在检查和写入之间改变。使用 ondrain() 来等待容量，而不是轮询。

🌐 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> 仅取消此操作。该信号仅取消挂起的 end() 调用；它不会使写入器本身失败。
• 返回：<Promise> 返回写入的总字节数。

表示不会再写入更多数据。已经在等待缓冲区空间的写入会在流结束前按顺序完成，而之后的写入会失败。如果有未完成的数据，当消费者拉取 done: true 超过最后一批数据时，返回的 promise 会被解决。如果没有缓冲或待处理的数据，写入器会立即关闭。

🌐 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()#>

• 返回：<number> 总共写入的字节数，如果结束无法同步完成，则为 -1。

writer.end() 的同步版本。返回值 -1 仅表示操作无法同步完成；不能对关闭是否已开始或为什么无法完成做任何假设。使用尝试-回退模式来等待完成：

🌐 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>

把写入器置于终端错误状态。如果写入器已经关闭或出错，这个操作不会有任何效果。不同于 write() 和 end()，fail() 是无条件同步的，因为让写入器失败只是一个纯状态转换，没有异步操作需要执行。原因会被存储并原样传播。如果省略，原因默认为 undefined。

🌐 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]()#>

如果写入器是打开的，调用 writer.fail()。如果写入器在 end() 或 endSync() 之后关闭，等待缓冲数据排空。如果写入器已经关闭或出错，则立即完成。

🌐 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> 仅取消此写操作。该信号只取消待处理的 write() 调用；它不会导致写入器本身失败。
• 返回：<Promise> 当缓冲区有可用空间时，用 undefined 填充。

写一块。

🌐 Write a chunk.

writer.writeSync(chunk)#>

• chunk <Uint8Array> | <string>
• 返回：<boolean> true 如果写入被接受，false 如果缓冲区已满。

同步写入。不阻塞；如果回压处于活动状态，则返回 false。

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

writer.writev(chunks[, options])#>

• chunks <Iterable> <Uint8Array> | <string> 值的
• options <Object>
  • signal <AbortSignal> 仅取消此写操作。该信号只取消待处理的 writev() 调用；它不会导致写入器本身失败。
• 返回：<Promise>

将多个块作为一个批处理写入。

🌐 Write multiple chunks as a single batch.

writer.writevSync(chunks)#>

• chunks <Iterable> <Uint8Array> | <string> 值的
• 返回：<boolean> true 如果写入被接受，false 如果缓冲区已满。

同步批量写入。

🌐 Synchronous batch write.

stream/iter 模块#>

🌐 The stream/iter module

大多数函数既可以作为命名导出使用，也可以作为 Stream 命名空间对象的属性使用。经典的流适配器（fromReadable()、fromWritable()、toReadable()、toReadableSync() 和 toWritable()）以及静态辅助对象（Broadcast、Share 和 SyncShare）只能作为命名导出使用。

🌐 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'); 拷贝

由 Iterable Streams API 定义的选项字典使用 Web IDL 转换语义。null 会被当作空字典处理，未知的成员会被忽略，已知成员会在操作运行前转换为它们声明的类型。转换失败会使用 Node.js 错误码，例如 ERR_INVALID_ARG_TYPE、ERR_INVALID_ARG_VALUE 和 ERR_OUT_OF_RANGE。

🌐 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'); 拷贝

在模块说明符中包含 node: 前缀是可选的。

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

来源#>

🌐 Sources

from(input)#>

• input <string> | <ArrayBuffer> | <ArrayBufferView> | <Iterable> | <AsyncIterable> | <Object> 不能是 null 或 undefined。
• 返回：<AsyncIterable> 其块用 <Uint8Array[]> 填充。

从给定输入创建一个异步字节流。字符串使用 UTF-8 编码。ArrayBuffer 和 ArrayBufferView 的值会被封装成 Uint8Array。input 中的数组和可迭代对象会递归展开并归一化。展开后的值可能会被分割成实现定义的有界批次。

🌐 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.

实现 Symbol.for('Stream.toAsyncStreamable') 或 Symbol.for('Stream.toStreamable') 的对象通过这些协议进行转换。toAsyncStreamable 协议优先于 toStreamable，而 toStreamable 又优先于迭代协议（Symbol.asyncIterator、Symbol.iterator）。

🌐 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> 不能是 null 或 undefined。
• 返回：<Iterable> 其块返回 <Uint8Array[]>

from() 的同步版本。返回一个同步可迭代对象。不能接受异步可迭代对象或 Promise。实现 Symbol.for('Stream.toStreamable') 的对象会通过该协议转换（优先于 Symbol.iterator）。toAsyncStreamable 协议将被完全忽略。

🌐 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> 数据源。
• ...transforms <Function> | <Object> 要应用的零个或多个转换。
• writer <Object> 使用 write(chunk) 方法的目的地。
• options <Object>
  • signal <AbortSignal> 中止管道。中止会导致目标写入器失败，除非 preventFail 是 true。
  • preventClose <boolean> 如果 true，当源结束时不要调用 writer.end()。默认值： false。
  • preventFail <boolean> 如果 true，在出现错误时不要调用 writer.fail()。默认： false。
• 返回：<Promise> 返回写入的总字节数。

将一个源通过转换传输到一个写入器。如果写入器有一个 writev(chunks) 方法，整个批次会在一次调用中传递（支持分散/聚集 I/O）。

🌐 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).

如果编写者实现了可选的 *Sync 方法（writeSync、writevSync、endSync），pipeTo() 将首先尝试使用同步方法作为快速路径，并且仅在同步方法表明无法完成时（例如，背压或等待下一个周期）才回退到异步版本。fail() 总是以同步方式调用。

🌐 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> 同步数据源。
• ...transforms <Function> | <Object> 零个或多个同步变换。
• writer <Object> 使用 write(chunk) 方法的目的地。
• options <Object>
  • preventClose <boolean> 默认: false。
  • preventFail <boolean> 默认: false。
• 返回：<number> 写入的总字节数。

pipeTo() 的同步版本。source、所有转换以及 writer 必须是同步的。不能接受异步可迭代对象或 Promise。

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

writer 必须具有 *Sync 方法（writeSync、writevSync、endSync）和 fail() 才能工作。

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

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

• source <AsyncIterable> | <Iterable> 数据源。
• ...transforms <Function> | <Object> 要应用的零个或多个转换。
• options <Object>
  • signal <AbortSignal> 中止管道。
• 返回：<AsyncIterable> 它的块用 <Uint8Array[]> 填充

创建一个懒惰的异步管道。源转换和可流式协议的分发会在调用 pull() 时发生，但数据直到返回的可迭代对象被消费时才会从 source 读取。已经中止的信号会在源转换后同步抛出。转换按顺序应用。

🌐 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. A signal that is already aborted is thrown synchronously after source conversion. Transforms are applied in order.

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);拷贝

使用 AbortSignal：

🌐 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 iterationconst { 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> 同步数据源。
• ...transforms <Function> | <Object> 零个或多个同步变换。
• 返回：<Iterable> 其块返回 <Uint8Array[]>

pull() 的同步版本。当调用 pullSync() 时，会进行源转换和可流式协议分发。所有转换都必须是同步的。

🌐 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> 可选转换应用于可读侧。
• options <Object>
  • budget <number> 在施加反压之前，缓冲的最大字节数。必须 >= 16384。默认值： 16384。
  • backpressure <string> 背压策略：'strict'、'unbounded'、'drop-oldest' 或 'drop-newest'。默认值：'strict'。
  • signal <AbortSignal> 中止流。信号在缓冲数据从 writer.end() 排出时仍然保持活跃，在此期间中止会导致写入失败，并拒绝待处理的 end() promise。
• 返回：<Object>
  • writer <Writable> 作家的方面。
  • readable <AsyncIterable> 它的块用 <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);拷贝

写入器返回 push() 符合 [Writer 接口][]。

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

双工通道#>

🌐 Duplex channels

duplex([options])#>

• options <Object>
  • budget <number> 双向的缓冲区大小（字节）。必须 >= 16384。默认值： 16384。
  • backpressure <string> 双向策略。 默认： 'strict'。
  • signal <AbortSignal> 两个通道的取消信号。
  • a <Object> 特定于 A 到 B 方向的选项。覆盖共享选项。
    • budget <number>
    • backpressure <string>
  • b <Object> 特定于 B 到 A 方向的选项。覆盖共享选项。
    • budget <number>
    • backpressure <string>
• 返回：<Array> 一对 [channelA, channelB] 双工通道。

创建一对连接的双工通道以实现双向通信，类似于 socketpair()。写入一个通道的写入端的数据会出现在另一个通道的可读端。

🌐 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 — 一个用于向节点发送数据的 [Writer 接口][] 对象。
• readable —— 一个用于从对等方读取数据的 <AsyncIterable>。
• close() — 关闭该端的通道（幂等）。
• [Symbol.asyncDispose]() — 为 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> 它的块必须是 <Uint8Array[]>
• options <Object>
  • signal <AbortSignal>
  • limit <number> 要消耗的最大字节数。如果收集的总字节数超过限制，将抛出 ERR_OUT_OF_RANGE 错误
• 返回：<Promise> 使用一组 Uint8Array 对象进行填充。

将所有块收集为 Uint8Array 值的数组（不进行连接）。

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

arrayBuffer(source[, options])#>

• source <AsyncIterable> | <Iterable> 它的块必须是 <Uint8Array[]>
• options <Object>
  • signal <AbortSignal>
  • limit <number> 要消耗的最大字节数。如果收集的总字节数超过限制，将抛出 ERR_OUT_OF_RANGE 错误
• 返回：<Promise> 使用 ArrayBuffer 对象完成。

将所有字节收集到 ArrayBuffer 中。

🌐 Collect all bytes into an ArrayBuffer.

arrayBufferSync(source[, options])#>

• source <Iterable> 它的块必须是 <Uint8Array[]>
• options <Object>
  • limit <number> 要消耗的最大字节数。如果收集的总字节数超过限制，将抛出 ERR_OUT_OF_RANGE 错误
• 返回：<ArrayBuffer>

arrayBuffer() 的同步版本。

🌐 Synchronous version of arrayBuffer().

arraySync(source[, options])#>

• source <Iterable> 它的块必须是 <Uint8Array[]>
• options <Object>
  • limit <number> 要消耗的最大字节数。如果收集的总字节数超过限制，将抛出 ERR_OUT_OF_RANGE 错误
• 返回：<Uint8Array[]>

array() 的同步版本。

🌐 Synchronous version of array().

bytes(source[, options])#>

• source <AsyncIterable> | <Iterable> 它的块必须是 <Uint8Array[]>
• options <Object>
  • signal <AbortSignal>
  • limit <number> 要消耗的最大字节数。如果收集的总字节数超过限制，将抛出 ERR_OUT_OF_RANGE 错误
• 返回：<Promise> 使用 Uint8Array 对象完成。

将流中的所有字节收集到一个 Uint8Array 中。

🌐 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> 它的块必须是 <Uint8Array[]>
• options <Object>
  • limit <number> 要消耗的最大字节数。如果收集的总字节数超过限制，将抛出 ERR_OUT_OF_RANGE 错误
• 返回：<Uint8Array>

bytes() 的同步版本。

🌐 Synchronous version of bytes().

text(source[, options])#>

• source <AsyncIterable> | <Iterable> 它的块必须是 <Uint8Array[]>
• options <Object>
  • encoding <string> 文本编码。默认： 'utf-8'。
  • signal <AbortSignal>
  • limit <number> 要消耗的最大字节数。如果收集的总字节数超过限制，将抛出 ERR_OUT_OF_RANGE 错误
• 返回：<Promise> 用 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> 它的块必须是 <Uint8Array[]>
• options <Object>
  • encoding <string> 默认: 'utf-8'。
  • limit <number> 要消耗的最大字节数。如果收集的总字节数超过限制，将抛出 ERR_OUT_OF_RANGE 错误
• 返回：<string>

text() 的同步版本。

🌐 Synchronous version of text().

实用工具#>

🌐 Utilities

ondrain(drainable)#>

• drainable <Object> 实现了可排水协议的对象。
• 返回：<Promise> | <null>

等待可排空的写入器恢复物理缓冲区容量。如果对象未实现可排空协议，则返回 null；或者返回一个承诺，当缓冲数据低于字节预算时，承诺会以 true 完成。

🌐 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.

对于使用 'drop-oldest' 或 'drop-newest' 的写入者来说，这会等待物理容量，即使写入不会阻塞。这允许生产者在写入前等待，从而避免数据丢失。

🌐 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> 它的块必须是 <Uint8Array[]>
• options <Object>
  • signal <AbortSignal>
• 返回：<AsyncIterable> 它的块用 <Uint8Array[]> 填充

通过按时间顺序产出批次来合并多个异步可迭代对象（无论哪个源先产生数据）。所有源都同时被消费。

🌐 Merge multiple async iterables by yielding batches in temporal order (whichever source produces data first). All sources are consumed concurrently.

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

const merged = merge(from('hello '), from('world'));
console.log(await text(merged)); // Order depends on timingconst { 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) => void 会在每个批次调用，并且当源结束时会调用 null。
• 返回：<Function> 无状态变换。

创建一个能够观察批处理而不修改它们的通道转换。适用于日志记录、指标或调试。

🌐 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() 故意不阻止 tapping 回调对块的就地修改；但返回值会被忽略。

tapSync(callback)#>

• callback <Function>
• 返回：<Function>

tap() 的同步版本。

🌐 Synchronous version of tap().

多消费者#>

🌐 Multi-consumer

broadcast([options])#>

• options <Object>
  • budget <number> 缓冲区大小（字节）。必须 >= 16384。 默认值: 65536。
  • backpressure <string> 'strict'、'unbounded'、'drop-oldest' 或 'drop-newest'。默认值： 'strict'。
  • signal <AbortSignal>
• 返回：<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>

取消广播。如果提供了 reason，所有消费者都会以这个确切原因拒绝。如果没有提供，消费者会正常完成。

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

broadcast.consumerCount#>

• <number>

活跃消费者的数量。

🌐 The number of active consumers.

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

• ...transforms <Function> | <Object>
• options <Object>
  • signal <AbortSignal>
• 返回：<AsyncIterable> 它的块用 <Uint8Array[]> 填充

创建一个新的消费者。可以选择对该消费者的数据视图应用可选的转换。

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

broadcast[Symbol.dispose]()#>

broadcast.cancel() 的别名。

🌐 Alias for broadcast.cancel().

Broadcast.from(input[, options])#>

• input <AsyncIterable> | <Iterable> | <BroadcastChannel>
• options <Object> 与 broadcast() 相同。
• 返回：<BroadcastChannel> | <Object> 一个 broadcastProtocol 输入会直接返回它的 <BroadcastChannel>。其他输入会返回 { writer, broadcast }。

从现有来源创建一个 <BroadcastChannel>。来源会被自动使用，并推送给所有订阅者。

🌐 Create a <BroadcastChannel> from an existing source. The source is consumed automatically and pushed to all subscribers.

share(source[, options])#>

• source <AsyncIterable> 共享的来源。
• options <Object>
  • budget <number> 缓冲区大小（字节）。必须 >= 16384。 默认值: 65536。
  • backpressure <string> 'strict'、'unbounded'、'drop-oldest' 或 'drop-newest'。默认值： 'strict'。
  • signal <AbortSignal>
• 返回：Share

创建一个拉取模式的多消费者共享流。与 broadcast() 不同，源只有在消费者拉取时才会被读取。多个消费者共享一个缓冲区。

🌐 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.

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

const shared = share(from('hello'));

const c1 = shared.pull();
const c2 = shared.pull();

// Consume concurrently to avoid deadlock with small buffers.
const [r1, r2] = await Promise.all([text(c1), text(c2)]);
console.log(r1); // 'hello'
console.log(r2); // 'hello'const { from, share, text } = require('node:stream/iter');

async function run() {
  const shared = share(from('hello'));

  const c1 = shared.pull();
  const c2 = shared.pull();

  // Consume concurrently to avoid deadlock with small buffers.
  const [r1, r2] = await Promise.all([text(c1), text(c2)]);
  console.log(r1); // 'hello'
  console.log(r2); // 'hello'
}

run().catch(console.error);拷贝

类：Share#>

🌐 Class: Share

静态方法：Share.from(input[, options])#>

🌐 Static method: Share.from(input[, options])

• input <AsyncIterable>
• options <Object> 与 share() 相同。
• 返回：Share

从现有来源创建一个 Share。

🌐 Create a Share from an existing source.

share.cancel([reason])#>

• reason <any>

取消共享。如果提供了 reason，所有消费者都会以这个确切原因拒绝。如果没有提供，消费者会正常完成。

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

share.consumerCount#>

• <number>

活跃消费者的数量。

🌐 The number of active consumers.

share.pull([...transforms][, options])#>

• ...transforms <Function> | <Object>
• options <Object>
  • signal <AbortSignal>
• 返回：<AsyncIterable> 它的块用 <Uint8Array[]> 填充

创建共享源的新消费者。

🌐 Create a new consumer of the shared source.

share[Symbol.dispose]()#>

share.cancel() 的别名。

🌐 Alias for share.cancel().

接口：Shareable#>

🌐 Interface: Shareable

sharable[Symbol.for('Stream.shareProtocol')]#>

• <Function> 返回一个 Share。

接口：SyncShareable#>

🌐 Interface: SyncShareable

sharable[Symbol.for('Stream.shareSyncProtocol')]#>

• <Function> 返回一个 SyncShare。

shareSync(source[, options])#>

• source <Iterable> 要共享的同步源。
• options <Object>
  • budget <number> 必须 >= 16384。 默认值： 65536。
  • backpressure <string> 'strict'、'drop-oldest' 或 'drop-newest'。 默认值: 'strict'。
• 返回：SyncShare

share() 的同步版本。

🌐 Synchronous version of share().

因为在同步上下文中无法等待，'unbounded' 不被支持，会抛出 ERR_INVALID_ARG_VALUE。使用 'drop-newest' 时，当消费者在预算耗尽时到达缓冲区末尾，会丢弃源中的一条条目，然后返回没有值的 { done: true }；消费者不会被分离，因此一旦最慢的消费者前进并释放预算，它可以继续运行。

🌐 Because there is no way to wait in a synchronous context, 'unbounded' is not supported and throws ERR_INVALID_ARG_VALUE. With 'drop-newest', a consumer that reaches the end of the buffer while the budget is exhausted discards a single entry from the source and then returns { done: true } without a value; the consumer is not detached, so it can resume once the slowest consumer advances and releases budget.

类：SyncShare#>

🌐 Class: SyncShare

静态方法：SyncShare.fromSync(input[, options])#>

🌐 Static method: SyncShare.fromSync(input[, options])

• input <Iterable>
• options <Object>
• 返回：SyncShare

share.cancel([reason])#>

• reason <any>

取消共享。如果提供了 reason，所有消费者都会抛出那个确切的原因。如果省略，则消费者正常完成。

🌐 Cancel the share. If reason is provided, all consumers throw that exact reason. If it is omitted, consumers complete normally.

share.consumerCount#>

• <number>

活跃消费者的数量。

🌐 The number of active consumers.

share.pull([...transforms])#>

• ...transforms <Function> | <Object>
• 返回：<Iterable> 其块返回 <Uint8Array[]>

创建共享源的新消费者。

🌐 Create a new consumer of the shared source.

share[Symbol.dispose]()#>

share.cancel() 的别名。

🌐 Alias for share.cancel().

压缩和解压变换#>

🌐 Compression and decompression transforms

pull()、pullSync()、pipeTo() 和 pipeToSync() 使用的压缩和解压缩转换可通过 node:zlib/iter 模块使用。详情请参见 node:zlib/iter 文档。

🌐 Compression and decompression transforms for use with pull(), pullSync(), pipeTo(), and pipeToSync() are available via the node:zlib/iter module. See the node:zlib/iter documentation for details.

经典流互操作#>

🌐 Classic stream interop

这些实用函数在经典的 stream.Readable/stream.Writable 流与 stream/iter API 之间搭建桥梁。

🌐 These utility functions bridge between classic stream.Readable/stream.Writable streams and the stream/iter API.

fromReadable() 和 fromWritable() 都接受动态类型对象——它们不要求输入直接继承 stream.Readable 或 stream.Writable。每个函数的最小契约如下所述。

🌐 Both fromReadable() and fromWritable() accept duck-typed objects -- they do not require the input to extend stream.Readable or stream.Writable directly. The minimum contract is described below for each function.

fromReadable(readable)#>

• readable <stream.Readable> | <Object> 一个经典的可读流或者一个拥有 read()、pipe()、destroy()、on() 和 removeListener() 方法的兼容对象。
• 返回：<AsyncIterable> 它的块用 <Uint8Array[]> 填充

将经典的可读流（或动态类型等效流）转换为可作为流/迭代异步可迭代源传递给 from()、pull()、text() 等的对象。

🌐 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.

如果对象实现了 toAsyncStreamable 协议（就像 stream.Readable 那样），将使用该协议。否则，该函数会对 read()、pipe()、destroy()、on() 和 removeListener()（EventEmitter）进行鸭子类型检查，并使用批量异步迭代器封装流。

🌐 If the object implements the toAsyncStreamable protocol (as stream.Readable does), that protocol is used. Otherwise, the function duck-types on read(), pipe(), destroy(), on(), and removeListener() (EventEmitter) and wraps the stream with a batched async iterator.

结果会被每个实例缓存——使用相同的流调用 fromReadable() 两次会返回相同的可迭代对象。

🌐 The result is cached per instance -- calling fromReadable() twice with the same stream returns the same iterable.

对于对象模式或编码的可读流，数据块会自动归一化为 Uint8Array。

🌐 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'const { Readable } = require('node:stream');
const { fromReadable, text } = require('node:stream/iter');

const readable = new Readable({
  read() { this.push('hello world'); this.push(null); },
});

async function run() {
  const result = await text(fromReadable(readable));
  console.log(result); // 'hello world'
}
run();拷贝

fromWritable(writable[, options])#>

• writable <stream.Writable> | <Object> 一个经典的可写流，或者一个拥有 write()、end()、destroy()、on() 和 removeListener() 方法的兼容对象。
• options <Object>
  • backpressure <string> 背压策略。默认值： 'strict'。
    • 'strict' —— 当缓冲区已满时，一次写入可能需要等待。之后的写入将被拒绝，直到它被接受或取消。
    • 'unbounded' —— 当缓冲区已满时，写入会排队。推荐与 pipeTo() 一起使用。
    • 'drop-newest' -- 当缓冲区已满时，写入会被悄无声息地丢弃。
    • 'drop-oldest' -- 不支持。会抛出 ERR_INVALID_ARG_VALUE。
• 返回：<Object> 一个流/迭代器写入器适配器。

从经典的 Writable 流（或动态类型的等效对象）创建一个流/迭代 Writer 适配器。该适配器可以作为目标传递给 pipeTo()。

🌐 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.

由于在经典可写对象上的所有写操作本质上都是异步的， 同步的 Writer 方法（writeSync、writevSync、endSync）总是返回 false 或 -1， 并依赖异步路径。 排队的 write() 或 writev() 可以在到达经典可写对象之前用它的 options.signal 取消。

🌐 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. A queued write() or writev() can be canceled with its options.signal before it reaches the classic Writable.

如果 writer.fail(reason) 收到一个非错误原因，经典 Writable 会被 ERR_FALSY_VALUE_REJECTION 或 ERR_OPERATION_FAILED 销毁。它的 reason 属性包含原始值，该值仍然是 Writer 存储的失败原因。

🌐 If writer.fail(reason) receives a non-Error reason, the classic Writable is destroyed with an ERR_FALSY_VALUE_REJECTION or ERR_OPERATION_FAILED error. Its reason property contains the original value, which remains the Writer's stored failure reason.

结果会根据每个实例和背压策略进行缓存——对相同的流和 backpressure 选项调用 fromWritable() 两次会返回相同的 Writer。

🌐 The result is cached per instance and backpressure policy -- calling fromWritable() twice with the same stream and backpressure option returns the same Writer.

对于不暴露 writableHighWaterMark、writableLength 或类似属性的动态类型流，将使用合理的默认值。可检测的对象模式可写流将被拒绝，因为 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: 'unbounded' }));const { Writable } = require('node:stream');
const { from, fromWritable, pipeTo } = require('node:stream/iter');

async function run() {
  const writable = new Writable({
    write(chunk, encoding, cb) { console.log(chunk.toString()); cb(); },
  });

  await pipeTo(from('hello world'),
               fromWritable(writable, { backpressure: 'unbounded' }));
}
run();拷贝

toReadable(source[, options])#>

• source <AsyncIterable> 它的块必须与 pull() 或 from() 的返回值 <Uint8Array[]> 相符。
• options <Object>
  • highWaterMark <number> 在应用反压之前的内部缓冲区大小（以字节为单位）。默认值： 65536（64 KB）。
  • signal <AbortSignal> 可选信号以中止可读流。
• 返回：<stream.Readable>

从 source（stream/iter API 使用的本地批处理格式）创建一个字节模式的 stream.Readable。每个批处理中产生的 Uint8Array 都会作为单独的块推入可读流。

🌐 Creates a byte-mode stream.Readable from the source (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.

经典流不能将任意值表示为发出的错误。非 Error 类型的原因会被封装在 ERR_FALSY_VALUE_REJECTION 或 ERR_OPERATION_FAILED 错误中，其 reason 属性包含原始值。

🌐 Classic streams cannot represent arbitrary values as emitted errors. A non-Error reason is wrapped in an ERR_FALSY_VALUE_REJECTION or ERR_OPERATION_FAILED error whose reason property contains the original value.

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'));const { createWriteStream } = require('node:fs');
const { from, pull, toReadable } = require('node:stream/iter');
const { compressGzip } = require('node:zlib/iter');

const source = pull(from('hello world'), compressGzip());
const readable = toReadable(source);

readable.pipe(createWriteStream('output.gz'));拷贝

toReadableSync(source[, options])#>

• source <Iterable> 其块必须返回 <Uint8Array[]>，例如 pullSync() 或 fromSync() 的返回值。
• options <Object>
  • highWaterMark <number> 在应用反压之前的内部缓冲区大小（以字节为单位）。默认值： 65536（64 KB）。
• 返回：<stream.Readable>

从 source 创建一个字节模式的 stream.Readable。_read() 方法同步地从迭代器中获取数据，因此可以通过 readable.read() 立即使用数据。

🌐 Creates a byte-mode stream.Readable from the source. 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'const { fromSync, toReadableSync } = require('node:stream/iter');

const source = fromSync('hello world');
const readable = toReadableSync(source);

console.log(readable.read().toString()); // 'hello world'拷贝

toWritable(writer)#>

• writer <Object> 一个流/迭代写入器。仅需要 write() 方法；end()、fail()、writeSync()、writevSync()、endSync() 和 writev() 是可选的。
• 返回：<stream.Writable>

创建一个由流/迭代 Writer 支持的经典 stream.Writable。

🌐 Creates a classic stream.Writable backed by a stream/iter Writer.

每个 _write() / _writev() 调用都会先尝试 Writer 的同步方法（writeSync / writevSync），如果同步路径返回 false 就会回退到异步方法。类似地，_final() 会先尝试 endSync() 再到 end()。当同步路径成功时，回调会通过 queueMicrotask 延迟处理，以保持异步解析的约定。

🌐 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. Similarly, _final() tries endSync() before end(). When the sync path succeeds, the callback is deferred via queueMicrotask to preserve the async resolution contract.

经典的流回调不能将任意值表示为错误。一个非 Error 类型的原因会在传给回调之前被封装在 ERR_FALSY_VALUE_REJECTION 或 ERR_OPERATION_FAILED 错误中。错误的 reason 属性包含原始值。

🌐 Classic stream callbacks cannot represent arbitrary values as errors. A non-Error reason is wrapped in an ERR_FALSY_VALUE_REJECTION or ERR_OPERATION_FAILED error before it is passed to the callback. The error's reason property contains the original value.

在成功完成之前销毁可写对象会调用 writer.fail()。如果 fail() 不可用，则在 Writer 实现时使用 Symbol.dispose 或 Symbol.asyncDispose。

🌐 Destroying the Writable before successful completion calls writer.fail(). If fail() is unavailable, Symbol.dispose or Symbol.asyncDispose is used when implemented by the Writer.

Writable 使用默认的经典流 highWaterMark。经典流的背压会限制等待写入到底层 Writer 的数据，而 Writer 控制活动的 _write() 或 _writev() 操作的完成。

🌐 The Writable uses the default classic stream highWaterMark. Classic stream backpressure bounds writes waiting to reach the underlying Writer, while the Writer controls completion of the active _write() or _writev() operation.

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

const { writer, readable } = push();
const writable = toWritable(writer);

writable.write('hello');
writable.end();const { push, toWritable } = require('node:stream/iter');

const { writer, readable } = push();
const writable = toWritable(writer);

writable.write('hello');
writable.end();拷贝

协议符号#>

🌐 Protocol symbols

这些众所周知的符号允许第三方对象参与流协议，而无需直接从 node:stream/iter 导入。

🌐 These well-known symbols allow third-party objects to participate in the streaming protocol without importing from node:stream/iter directly.

Stream.broadcastProtocol#>

• 值：Symbol.for('Stream.broadcastProtocol')

这个值必须是一个函数。当被 Broadcast.from() 调用时，它会接收传给 Broadcast.from() 的选项，并且必须返回一个符合 <BroadcastChannel> 接口的对象。实现方式完全自定义——它可以随意管理消费者、缓冲和背压。

🌐 The value must be a function. When called by Broadcast.from(), it receives the options passed to Broadcast.from() and must return an object conforming to the <BroadcastChannel> interface. The implementation is fully custom -- it can manage consumers, buffering, and backpressure however it wants.

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

// This example defers to the built-in Broadcast, but a custom
// implementation could use any mechanism.
class MessageBus {
  #broadcast;
  #writer;

  constructor() {
    const { writer, broadcast } = createBroadcast();
    this.#writer = writer;
    this.#broadcast = broadcast;
  }

  [Symbol.for('Stream.broadcastProtocol')](options) {
    return this.#broadcast;
  }

  send(data) {
    this.#writer.write(new TextEncoder().encode(data));
  }

  close() {
    this.#writer.end();
  }
}

const bus = new MessageBus();
const broadcast = Broadcast.from(bus);
const consumer = broadcast.push();
bus.send('hello');
bus.close();
console.log(await text(consumer)); // 'hello'const {
  broadcast: createBroadcast,
  Broadcast,
  text,
} = require('node:stream/iter');

// This example defers to the built-in Broadcast, but a custom
// implementation could use any mechanism.
class MessageBus {
  #broadcast;
  #writer;

  constructor() {
    const { writer, broadcast } = createBroadcast();
    this.#writer = writer;
    this.#broadcast = broadcast;
  }

  [Symbol.for('Stream.broadcastProtocol')](options) {
    return this.#broadcast;
  }

  send(data) {
    this.#writer.write(new TextEncoder().encode(data));
  }

  close() {
    this.#writer.end();
  }
}

const bus = new MessageBus();
const broadcast = Broadcast.from(bus);
const consumer = broadcast.push();
bus.send('hello');
bus.close();
text(consumer).then(console.log); // 'hello'拷贝

Stream.drainableProtocol#>

• 值：Symbol.for('Stream.drainableProtocol')

实现以使写入器与 ondrain() 兼容。如果没有背压，该方法应返回 null；如果存在背压，则返回一个在背压消除时兑现为真值的 Promise。

🌐 Implement to make a writer compatible with ondrain(). The method should return null if no backpressure, or a promise that fulfills with a truthy value when backpressure clears.

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

class CustomWriter {
  #queue = [];
  #drain = null;
  #closed = false;
  [Symbol.for('Stream.drainableProtocol')]() {
    if (this.#closed) return null;
    if (this.#queue.length < 3) return Promise.resolve(true);
    this.#drain ??= Promise.withResolvers();
    return this.#drain.promise;
  }
  write(chunk) {
    this.#queue.push(chunk);
  }
  flush() {
    this.#queue.length = 0;
    this.#drain?.resolve(true);
    this.#drain = null;
  }
  close() {
    this.#closed = true;
  }
}
const writer = new CustomWriter();
const ready = ondrain(writer);
console.log(ready); // Promise { true } -- no backpressureconst { ondrain } = require('node:stream/iter');

class CustomWriter {
  #queue = [];
  #drain = null;
  #closed = false;

  [Symbol.for('Stream.drainableProtocol')]() {
    if (this.#closed) return null;
    if (this.#queue.length < 3) return Promise.resolve(true);
    this.#drain ??= Promise.withResolvers();
    return this.#drain.promise;
  }

  write(chunk) {
    this.#queue.push(chunk);
  }

  flush() {
    this.#queue.length = 0;
    this.#drain?.resolve(true);
    this.#drain = null;
  }

  close() {
    this.#closed = true;
  }
}

const writer = new CustomWriter();
const ready = ondrain(writer);
console.log(ready); // Promise { true } -- no backpressure拷贝

Stream.shareProtocol#>

• 值：Symbol.for('Stream.shareProtocol')

该值必须是一个函数。当被 Share.from() 调用时，它会接收传递给 Share.from() 的选项，并且必须返回一个符合 Share 接口的对象。实现完全自定义——它可以以任何方式管理共享源、消费者、缓冲和背压。

🌐 The value must be a function. When called by Share.from(), it receives the options passed to Share.from() and must return an object conforming to the Share interface. The implementation is fully custom -- it can manage the shared source, consumers, buffering, and backpressure however it wants.

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

// This example defers to the built-in share(), but a custom
// implementation could use any mechanism.
class DataPool {
  #share;

  constructor(source) {
    this.#share = share(source);
  }

  [Symbol.for('Stream.shareProtocol')](options) {
    return this.#share;
  }
}

const pool = new DataPool(
  (async function* () {
    yield 'hello';
  })(),
);

const shared = Share.from(pool);
const consumer = shared.pull();
console.log(await text(consumer)); // 'hello'const { share, Share, text } = require('node:stream/iter');

// This example defers to the built-in share(), but a custom
// implementation could use any mechanism.
class DataPool {
  #share;

  constructor(source) {
    this.#share = share(source);
  }

  [Symbol.for('Stream.shareProtocol')](options) {
    return this.#share;
  }
}

const pool = new DataPool(
  (async function* () {
    yield 'hello';
  })(),
);

const shared = Share.from(pool);
const consumer = shared.pull();
text(consumer).then(console.log); // 'hello'拷贝

Stream.shareSyncProtocol#>

• 值：Symbol.for('Stream.shareSyncProtocol')

该值必须是一个函数。当被 SyncShare.fromSync() 调用时，它会接收到传递给 SyncShare.fromSync() 的选项，并且必须返回一个符合 SyncShare 接口的对象。实现完全自定义——它可以按照自己的方式管理共享源、消费者和缓冲。

🌐 The value must be a function. When called by SyncShare.fromSync(), it receives the options passed to SyncShare.fromSync() and must return an object conforming to the SyncShare interface. The implementation is fully custom -- it can manage the shared source, consumers, and buffering however it wants.

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

// This example defers to the built-in shareSync(), but a custom
// implementation could use any mechanism.
class SyncDataPool {
  #share;

  constructor(source) {
    this.#share = shareSync(source);
  }

  [Symbol.for('Stream.shareSyncProtocol')](options) {
    return this.#share;
  }
}

const encoder = new TextEncoder();
const pool = new SyncDataPool(
  function* () {
    yield [encoder.encode('hello')];
  }(),
);

const shared = SyncShare.fromSync(pool);
const consumer = shared.pull();
console.log(textSync(consumer)); // 'hello'const { shareSync, SyncShare, textSync } = require('node:stream/iter');

// This example defers to the built-in shareSync(), but a custom
// implementation could use any mechanism.
class SyncDataPool {
  #share;

  constructor(source) {
    this.#share = shareSync(source);
  }

  [Symbol.for('Stream.shareSyncProtocol')](options) {
    return this.#share;
  }
}

const encoder = new TextEncoder();
const pool = new SyncDataPool(
  function* () {
    yield [encoder.encode('hello')];
  }(),
);

const shared = SyncShare.fromSync(pool);
const consumer = shared.pull();
console.log(textSync(consumer)); // 'hello'拷贝

Stream.toAsyncStreamable#>

• 值：Symbol.for('Stream.toAsyncStreamable')

该值必须是一个将对象转换为可流式传输值的函数。当对象传递给 from() 时，会调用此方法来生成实际数据。它可以返回任何可解析为字符串、Uint8Array、AsyncIterable、Iterable 或其他可流式传输对象的值。

🌐 The value must be a function that converts the object into a streamable value. When the object is passed to from(), this method is called to produce the actual data. It may return any value that resolves to a string, Uint8Array, AsyncIterable, Iterable, or another streamable object.

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

class Greeting {
  #name;

  constructor(name) {
    this.#name = name;
  }

  [Symbol.for('Stream.toAsyncStreamable')]() {
    return `hello ${this.#name}`;
  }
}

const stream = from(new Greeting('world'));
console.log(await text(stream)); // 'hello world'const { from, text } = require('node:stream/iter');

class Greeting {
  #name;

  constructor(name) {
    this.#name = name;
  }

  [Symbol.for('Stream.toAsyncStreamable')]() {
    return `hello ${this.#name}`;
  }
}

const stream = from(new Greeting('world'));
text(stream).then(console.log); // 'hello world'拷贝

Stream.toStreamable#>

• 值：Symbol.for('Stream.toStreamable')

这个值必须是一个函数，能同步地将对象转换成可流式处理的值。当对象被传给 fromSync() 时，会调用这个方法来生成实际的数据。它必须同步返回一个可流式处理的值：一个字符串、Uint8Array 或 Iterable。

🌐 The value must be a function that synchronously converts the object into a streamable value. When the object is passed to fromSync(), this method is called to produce the actual data. It must synchronously return a streamable value: a string, Uint8Array, or Iterable.

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

class Greeting {
  #name;

  constructor(name) {
    this.#name = name;
  }

  [Symbol.for('Stream.toStreamable')]() {
    return `hello ${this.#name}`;
  }
}

const stream = fromSync(new Greeting('world'));
console.log(textSync(stream)); // 'hello world'const { fromSync, textSync } = require('node:stream/iter');

class Greeting {
  #name;

  constructor(name) {
    this.#name = name;
  }

  [Symbol.for('Stream.toStreamable')]() {
    return `hello ${this.#name}`;
  }
}

const stream = fromSync(new Greeting('world'));
console.log(textSync(stream)); // 'hello world'拷贝