Итерируемые потоки

Модуль node:stream/iter предоставляет API потоковой передачи, построенный на итерируемых объектах, а не на событийной иерархии классов Readable/Writable/Transform или интерфейсах потоков Web Streams ReadableStream/WritableStream/TransformStream.

Этот модуль доступен только при включённом флаге CLI --experimental-stream-iter.

Потоки представлены как AsyncIterable<Uint8Array[]> (асинхронные) или Iterable<Uint8Array[]> (синхронные). Нет базовых классов для расширения — участвовать может любой объект, реализующий протокол итерируемых объектов. Преобразования представляют собой обычные функции или объекты с методом transform.

Данные передаются пакетами (Uint8Array[] за итерацию), что позволяет распределить затраты на асинхронные операции.

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

Основные понятия

Байтовые потоки

Все данные в этом API представлены в виде байтов Uint8Array. Строки автоматически кодируются в UTF-8 при передаче в from(), push() или pipeTo(). Это устраняет неоднозначность кодировок и обеспечивает передачу без копирования между потоками и нативным кодом.

Пакетная обработка

Каждая итерация возвращает пакет — массив фрагментов Uint8Array (Uint8Array[]). Пакетная обработка позволяет распределить затраты на await и создание Promise между несколькими фрагментами. Потребитель, обрабатывающий по одному фрагменту за раз, может просто перебирать внутренний массив:

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);
    }
  }
}

Преобразования

Преобразования бывают двух видов:

• Без сохранения состояния — функция (chunks, options) => result, вызываемая один раз для каждого пакета. Получает Uint8Array[] (или null в качестве сигнала завершения) и объект options. Возвращает Uint8Array[], null или итерируемый объект с фрагментами.

• С сохранением состояния — объект { transform(source, options) }, в котором transform является генератором (синхронным или асинхронным), получающим весь входной итерируемый объект и объект options и выдающим результат. Этот вариант используется для сжатия, шифрования и любых преобразований, которым необходимо буферизовать данные между пакетами.

Оба варианта получают параметр options со следующим свойством:

• options.signal <AbortSignal> Объект AbortSignal, срабатывающий при отмене конвейера, возникновении ошибки или прекращении чтения потребителем. Преобразования могут проверять signal.aborted или прослушивать событие 'abort' для раннего освобождения ресурсов.

Сигнал завершения (null) отправляется после завершения источника, предоставляя преобразованиям возможность выдать завершающие данные (например, окончания потока сжатия).

// 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`)];
        }
      }
    }
  },
};
jscopy

Чтение по запросу и отправка данных

API поддерживает две модели:

• Чтение по запросу — данные передаются по запросу. pull() и pullSync() создают отложенные конвейеры, которые читают данные из источника только при переборе потребителем.

• Отправка данных — данные записываются явно. push() создаёт пару «писатель/читаемый поток» с обратным давлением. Писатель отправляет данные, а читаемый поток потребляется как асинхронный итерируемый объект.

Обратное давление

В потоках с чтением по запросу обратное давление возникает естественным образом: темп задаёт потребитель, поэтому источник никогда не читается быстрее, чем потребитель способен обрабатывать данные. В потоках с отправкой данных необходимо управлять обратным давлением явно, поскольку производитель и потребитель работают независимо. Параметры highWaterMark и backpressure в push(), broadcast() и share() управляют этим поведением.

Модель с двумя буферами

В потоках с отправкой данных используется система буферизации из двух частей. Представьте себе ведро (слоты), которое наполняется через шланг (ожидающие записи), с поплавковым клапаном, закрывающимся, когда ведро заполнено:

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

• Слоты (ведро) — данные, готовые для потребителя; их количество ограничено значением highWaterMark. Когда потребитель считывает данные, все слоты разом переносятся в один пакет.

• Ожидающие записи (шланг) — записи, ожидающие свободного места в слотах. После того как потребитель извлечёт данные, ожидающие записи перемещаются в освободившиеся слоты, а соответствующие промисы завершаются.

Использование этих буферов в каждой политике:

Строгий режим (по умолчанию)

Строгий режим обнаруживает шаблоны «запустил и забыл», когда производитель вызывает write(), не дожидаясь завершения, что может привести к неограниченному росту потребления памяти. Он ограничивает буфер слотов и очередь ожидающих записей значением highWaterMark.

Если правильно ожидать завершения каждой записи, одновременно может существовать только одна ожидающая запись (ваша), поэтому лимит ожидающих записей не будет достигнут. Не ожидающие завершения записи накапливаются в очереди и вызывают ошибку при переполнении:

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

const { writer, readable } = push({ highWaterMark: 16 });

// 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({ highWaterMark: 16 });

  // 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, со временем будет выброшена ошибка:

// 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"
jscopy

Блокировка

В режиме блокировки количество слотов ограничено значением highWaterMark, но очередь ожидающих записей не ограничена. Записи, завершения которых ожидают, блокируются, пока потребитель не освободит место, как и в строгом режиме. Отличие в том, что записи, завершения которых не ожидают, бесконечно накапливаются в очереди, не вызывая ошибки, — это может привести к утечке памяти, если производитель забудет вызвать await.

Этот режим используется по умолчанию в классических потоках Node.js и потоках Web Streams. Используйте его, если вы контролируете производителя и уверены, что он правильно ожидает завершения операций, или при переносе кода с этих API.

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

const { writer, readable } = push({
  highWaterMark: 16,
  backpressure: 'block',
});

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({
    highWaterMark: 16,
    backpressure: 'block',
  });

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

Удаление самых старых данных

Записи не ожидают завершения. Когда буфер слотов заполнен, самый старый буферизованный фрагмент удаляется, чтобы освободить место для новой записи. Потребитель всегда получает самые свежие данные. Этот режим полезен для прямых трансляций, телеметрии и любых сценариев, в которых актуальные данные ценнее устаревших.

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

// Keep only the 5 most recent readings
const { writer, readable } = push({
  highWaterMark: 5,
  backpressure: 'drop-oldest',
});

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

// Keep only the 5 most recent readings
const { writer, readable } = push({
  highWaterMark: 5,
  backpressure: 'drop-oldest',
});

Удаление самых новых данных

Записи не ожидают завершения. Когда буфер слотов заполнен, новая запись молча отбрасывается. Потребитель обрабатывает уже буферизованные данные, не перегружаясь новыми. Этот режим полезен для ограничения частоты или снижения нагрузки при высокой нагрузке.

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

// Accept up to 10 buffered items; discard anything beyond that
const { writer, readable } = push({
  highWaterMark: 10,
  backpressure: 'drop-newest',
});

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

// Accept up to 10 buffered items; discard anything beyond that
const { writer, readable } = push({
  highWaterMark: 10,
  backpressure: 'drop-newest',
});

Интерфейс Writer

Писателем может быть любой объект, соответствующий интерфейсу Writer. Обязателен только метод write(); все остальные методы необязательны.

У каждого асинхронного метода есть синхронный аналог *Sync, предназначенный для шаблона с резервным вариантом: сначала пробуется быстрый синхронный путь, а к асинхронной версии переходят только в том случае, если синхронный вызов сообщает, что не может завершиться:

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
mjscopy

writer.desiredSize

• <number> | <null>

Количество доступных слотов буфера до достижения верхнего порога. Возвращает null, если писатель закрыт или потребитель отключился.

Значение всегда неотрицательное.

writer.end([options])

• options <Object>
  • signal <AbortSignal> Отменить только эту операцию. Сигнал отменяет только ожидающий вызов end(); он не переводит сам писатель в состояние ошибки.
• Возвращает: <Promise> Выполняется с общим количеством записанных байтов.

Сообщить, что больше данные записываться не будут.

writer.endSync()

• Возвращает: <number> Общее количество записанных байтов или -1, если писатель не открыт.

Синхронный вариант writer.end(). Возвращает -1, если писатель уже закрыт или находится в состоянии ошибки. Может использоваться в шаблоне с резервным вариантом:

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

writer.fail(reason)

• reason <any>

Перевести писатель в конечное состояние ошибки. Если писатель уже закрыт или находится в состоянии ошибки, ничего не происходит. В отличие от write() и end(), fail() всегда выполняется синхронно, поскольку перевод писателя в состояние ошибки — это простое изменение состояния, не требующее асинхронной работы.

writer.write(chunk[, options])

• chunk <Uint8Array> | <string>
• options <Object>
  • signal <AbortSignal> Отменить только эту операцию записи. Сигнал отменяет только ожидающий вызов write(); он не переводит сам писатель в состояние ошибки.
• Возвращает: <Promise> Выполняется со значением undefined, когда в буфере появляется свободное место.

Записать фрагмент.

writer.writeSync(chunk)

• chunk <Uint8Array> | <string>
• Возвращает: <boolean> true, если запись принята, false, если буфер заполнен.

Синхронная запись. Не блокирует выполнение; возвращает false, если действует обратное давление.

writer.writev(chunks[, options])

• chunks <Uint8Array[]> | <string[]>
• options <Object>
  • signal <AbortSignal> Отменить только эту операцию записи. Сигнал отменяет только ожидающий вызов writev(); он не переводит сам писатель в состояние ошибки.
• Возвращает: <Promise>

Записать несколько фрагментов одним пакетом.

writer.writevSync(chunks)

• chunks <Uint8Array[]> | <string[]>
• Возвращает: <boolean> true, если запись принята, false, если буфер заполнен.

Синхронная пакетная запись.

Модуль stream/iter

Все функции доступны как именованные экспорты и как свойства объекта пространства имён Stream:

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

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

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

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

Префикс node: в спецификаторе модуля можно не указывать.

Источники

from(input)

• input <string> | <ArrayBuffer> | <AsyncIterable> | <Object> Не должно быть null или undefined.
• Возвращает: {AsyncIterable<Uint8Array[]>}

Создать асинхронный байтовый поток из заданных входных данных. Строки кодируются в UTF-8. Значения ArrayBuffer и ArrayBufferView оборачиваются в Uint8Array. Массивы и итерируемые объекты рекурсивно уплощаются и нормализуются.

Объекты, реализующие Symbol.for('Stream.toAsyncStreamable') или Symbol.for('Stream.toStreamable'), преобразуются с помощью соответствующих протоколов. Протокол toAsyncStreamable имеет приоритет над toStreamable, который, в свою очередь, имеет приоритет над протоколами итерации (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> | <Object> Не должно быть null или undefined.
• Возвращает: {Iterable<Uint8Array[]>}

Синхронная версия from(). Возвращает синхронный итерируемый объект. Не принимает асинхронные итерируемые объекты и промисы. Объекты, реализующие Symbol.for('Stream.toStreamable'), преобразуются с помощью этого протокола (он имеет приоритет над Symbol.iterator). Протокол toAsyncStreamable полностью игнорируется.

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'

Конвейеры

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

• source <AsyncIterable> Источник данных.
• ...transforms <Function> | <Object> Ноль или более преобразований для применения.
• writer <Object> Приёмник с методом write(chunk).
• options <Object>
  • signal <AbortSignal> Прервать конвейер.
  • preventClose <boolean> Если true, не вызывать writer.end() при завершении источника. По умолчанию: false.
  • preventFail <boolean> Если true, не вызывать writer.fail() при ошибке. По умолчанию: false.
• Возвращает: <Promise> Выполняется с общим количеством записанных байтов.

Передать источник через преобразования в писатель. Если у писателя есть метод writev(chunks), целые пакеты передаются одним вызовом (что позволяет использовать ввод-вывод по схеме scatter/gather).

Если писатель реализует необязательные методы *Sync (writeSync, writevSync, endSync), pipeTo() попытается сначала использовать синхронные методы как быстрый путь и перейдёт к асинхронным версиям только в том случае, если синхронные методы сообщат, что не могут завершиться (например, из-за обратного давления или необходимости дождаться следующего такта). fail() всегда вызывается синхронно.

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 должны быть синхронными. Не принимает асинхронные итерируемые объекты и промисы.

Для работы у writer должны быть методы *Sync (writeSync, writevSync, endSync) и fail().

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

• source <AsyncIterable> Источник данных.
• ...transforms <Function> | <Object> Ноль или более преобразований для применения.
• options <Object>
  • signal <AbortSignal> Прервать конвейер.
• Возвращает: {AsyncIterable<Uint8Array[]>}

Создать отложенный асинхронный конвейер. Данные не считываются из 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);

Использование 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} Синхронный источник данных.
• ...transforms <Function> | <Object> Ноль или более синхронных преобразований.
• Возвращает: {Iterable<Uint8Array[]>}

Синхронная версия pull(). Все преобразования должны быть синхронными.

Потоки с отправкой данных

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

• ...transforms <Function> | <Object> Необязательные преобразования, применяемые к читаемой стороне.
• options <Object>
  • highWaterMark <number> Максимальное количество буферизованных слотов до применения обратного давления. Должно быть >= 1; значения меньше 1 ограничиваются значением 1. По умолчанию: 4.
  • backpressure <string> Политика обратного давления: 'strict', 'block', 'drop-oldest' или 'drop-newest'. По умолчанию: 'strict'.
  • signal <AbortSignal> Прервать поток.
• Возвращает: <Object>
  • writer {PushWriter} Сторона записи.
  • readable {AsyncIterable<Uint8Array[]>} Читаемая сторона.

Создать поток с отправкой данных и обратным давлением. Писатель отправляет данные, а читаемая сторона потребляется как асинхронный итерируемый объект.

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][].

Дуплексные каналы

duplex([options])

• options <Object>
  • highWaterMark <number> Размер буфера для обоих направлений. По умолчанию: 4.
  • backpressure <string> Политика для обоих направлений. По умолчанию: 'strict'.
  • signal <AbortSignal> Сигнал отмены для обоих каналов.
  • a <Object> Параметры для направления от A к B. Переопределяют общие параметры.
    • highWaterMark <number>
    • backpressure <string>
  • b <Object> Параметры для направления от B к A. Переопределяют общие параметры.
    • highWaterMark <number>
    • backpressure <string>
• Возвращает: <Array> Пару [channelA, channelB] дуплексных каналов.

Создать пару соединённых дуплексных каналов для двунаправленной связи, аналогично socketpair(). Данные, записанные в писатель одного канала, появляются в читаемом потоке другого канала.

Каждый канал содержит:

• writer — объект [интерфейса Writer][] для отправки данных другому участнику.
• readable — AsyncIterable<Uint8Array[]> для чтения данных от другого участника.
• 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 client.writer.write('hello');
await client.writer.end();

console.log(await text(server.readable)); // handled by echo
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 client.writer.write('hello');
  await client.writer.end();

  console.log(await text(server.readable)); // handled by echo
  await serving;
}

run().catch(console.error);

Потребители

array(source[, options])

• source {AsyncIterable<Uint8Array[]>|Iterable<Uint8Array[]>}
• options <Object>
  • signal <AbortSignal>
  • limit <number> Максимальное количество байтов для чтения. Если общее количество собранных байтов превысит лимит, будет выброшена ошибка ERR_OUT_OF_RANGE
• Возвращает: <Promise> Выполняется с массивом объектов Uint8Array.

Собрать все фрагменты в массив значений Uint8Array (без объединения).

arrayBuffer(source[, options])

• source {AsyncIterable<Uint8Array[]>|Iterable<Uint8Array[]>}
• options <Object>
  • signal <AbortSignal>
  • limit <number> Максимальное количество байтов для чтения. Если общее количество собранных байтов превышает лимит, возникает ошибка ERR_OUT_OF_RANGE
• Возвращает: <Promise> Выполняется с объектом ArrayBuffer.

Собирает все байты в ArrayBuffer.

arrayBufferSync(source[, options])

• source {Iterable<Uint8Array[]>}
• options <Object>
  • limit <number> Максимальное количество байтов для чтения. Если общее количество собранных байтов превышает лимит, возникает ошибка ERR_OUT_OF_RANGE
• Возвращает: <ArrayBuffer>

Синхронная версия arrayBuffer().

arraySync(source[, options])

• source {Iterable<Uint8Array[]>}
• options <Object>
  • limit <number> Максимальное количество байтов для чтения. Если общее количество собранных байтов превышает лимит, возникает ошибка ERR_OUT_OF_RANGE
• Возвращает: <Uint8Array[]>

Синхронная версия array().

bytes(source[, options])

• source {AsyncIterable<Uint8Array[]>|Iterable<Uint8Array[]>}
• options <Object>
  • signal <AbortSignal>
  • limit <number> Максимальное количество байтов для чтения. Если общее количество собранных байтов превышает лимит, возникает ошибка ERR_OUT_OF_RANGE
• Возвращает: <Promise> Выполняется с объектом Uint8Array.

Собирает все байты из потока в один объект 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().

text(source[, options])

• source {AsyncIterable<Uint8Array[]>|Iterable<Uint8Array[]>}
• options <Object>
  • encoding <string> Кодировка текста. По умолчанию: 'utf-8'.
  • signal <AbortSignal>
  • limit <number> Максимальное количество байтов для чтения. Если общее количество собранных байтов превышает лимит, возникает ошибка ERR_OUT_OF_RANGE
• Возвращает: <Promise> Выполняется со значением string.

Собирает все байты и декодирует их как текст.

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

Вспомогательные функции

ondrain(drainable)

• drainable <Object> Объект, реализующий протокол drainable.
• Возвращает: <Promise> | <null>

Ожидает снятия ограничения обратного давления у записи в drainable-объект. Возвращает null, если объект не реализует протокол drainable, или промис, который выполняется со значением true, когда запись может принимать больше данных.

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

const { writer, readable } = push({ highWaterMark: 2 });
writer.writeSync('a');
writer.writeSync('b');

// 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({ highWaterMark: 2 });
  writer.writeSync('a');
  writer.writeSync('b');

  // 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<Uint8Array[]>|Iterable<Uint8Array[]>} Два или более итерируемых объекта.
• options <Object>
  • signal <AbortSignal>
• Возвращает: {AsyncIterable<Uint8Array[]>}

Объединяет несколько асинхронных итерируемых объектов, выдавая пакеты в хронологическом порядке (сначала поступают данные из источника, который выдал их раньше). Данные из всех источников считываются одновременно.

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) => void Вызывается для каждого пакета.
• Возвращает: <Function> Преобразование без состояния.

Создает сквозное преобразование, которое отслеживает пакеты, не изменяя их. Полезно для ведения журналов, сбора метрик и отладки.

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

const result = pull(
  from('hello'),
  tap((chunks) => 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) => console.log('Batch size:', chunks.length)),
  );
  console.log(await text(result));
}

run().catch(console.error);

tap() намеренно не препятствует изменению фрагментов на месте функцией обратного вызова наблюдателя; возвращаемые значения игнорируются.

tapSync(callback)

• callback <Function>
• Возвращает: <Function>

Синхронная версия tap().

Для нескольких потребителей

broadcast([options])

• options <Object>
  • highWaterMark <number> Размер буфера в слотах. Должен быть >= 1; значения меньше 1 ограничиваются значением 1. По умолчанию: 16.
  • backpressure <string> 'strict', 'block', 'drop-oldest' или 'drop-newest'. По умолчанию: 'strict'.
  • signal <AbortSignal>
• Возвращает: <Object>
  • writer {BroadcastWriter}
  • broadcast {Broadcast}

Создает канал широковещательной передачи данных для нескольких потребителей с моделью push. Один записывающий объект передает данные нескольким потребителям. У каждого потребителя есть независимый курсор в общем буфере.

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

• <number>

Количество фрагментов, находящихся в буфере в данный момент.

broadcast.cancel([reason])

• reason <Error>

Отменяет широковещательную передачу. Все потребители получают ошибку.

broadcast.consumerCount

• <number>

Количество активных потребителей.

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

• ...transforms <Function> | <Object>
• options <Object>
  • signal <AbortSignal>
• Возвращает: {AsyncIterable<Uint8Array[]>}

Создает нового потребителя. Каждый потребитель получает все данные, записанные в широковещательный канал с момента подписки. Необязательные преобразования применяются к представлению данных этого потребителя.

broadcast[Symbol.dispose]()

Псевдоним для broadcast.cancel().

Broadcast.from(input[, options])

• input <AsyncIterable>
• options <Object> То же, что и broadcast().
• Возвращает: <Object> { writer, broadcast }

Создает {Broadcast} из существующего источника. Источник считывается автоматически, а данные передаются всем подписчикам.

share(source[, options])

• source <AsyncIterable> Источник для совместного использования.
• options <Object>
  • highWaterMark <number> Размер буфера. Должен быть >= 1; значения меньше 1 ограничиваются значением 1. По умолчанию: 16.
  • backpressure <string> 'strict', 'block', 'drop-oldest' или 'drop-newest'. По умолчанию: 'strict'.
• Возвращает: {Share}

Создает общий поток для нескольких потребителей с моделью pull. В отличие от broadcast(), источник считывается только по запросу потребителя. Несколько потребителей используют общий буфер.

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

• <number>

Количество фрагментов, находящихся в буфере в данный момент.

share.cancel([reason])

• reason <Error>

Отменяет совместное использование. Все потребители получают ошибку.

share.consumerCount

• <number>

Количество активных потребителей.

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

• ...transforms <Function> | <Object>
• options <Object>
  • signal <AbortSignal>
• Возвращает: {AsyncIterable<Uint8Array[]>}

Создает нового потребителя общего источника.

share[Symbol.dispose]()

Псевдоним для share.cancel().

Share.from(input[, options])

• input <AsyncIterable>
• options <Object> То же, что и share().
• Возвращает: {Share}

Создает {Share} из существующего источника.

shareSync(source[, options])

• source {Iterable} Синхронный источник для совместного использования.
• options <Object>
  • highWaterMark <number> Должен быть >= 1; значения меньше 1 ограничиваются значением 1. По умолчанию: 16.
  • backpressure <string> По умолчанию: 'strict'.
• Возвращает: {SyncShare}

Синхронная версия share().

SyncShare.fromSync(input[, options])

• input {Iterable|SyncShareable}
• options <Object>
• Возвращает: {SyncShare}

Преобразования для сжатия и распаковки

Преобразования для сжатия и распаковки, используемые с pull(), pullSync(), pipeTo() и pipeToSync(), доступны в модуле node:zlib/iter. Подробности см. в документации node:zlib/iter.

Взаимодействие с классическими потоками

Эти вспомогательные функции обеспечивают взаимодействие классических потоков stream.Readable/stream.Writable с API stream/iter.

Функции fromReadable() и fromWritable() принимают объекты, совместимые по структуре (duck-typed), — входным данным не требуется напрямую наследоваться от stream.Readable или stream.Writable. Минимальные требования к каждому объекту описаны ниже для каждой функции.

fromReadable(readable)

• readable <stream.Readable> | <Object> Классический поток Readable или любой объект с методами read() и on().
• Возвращает: {AsyncIterable<Uint8Array[]>} Асинхронный итерируемый источник потока/итератора.

Преобразует классический поток Readable (или совместимый по структуре объект) в асинхронный итерируемый источник потока/итератора, который можно передать в from(), pull(), text() и т. д.

Если объект реализует протокол toAsyncStreamable (как это делает stream.Readable), используется этот протокол. В противном случае функция проверяет наличие методов read() и on() (EventEmitter) и оборачивает поток в пакетный асинхронный итератор.

Результат кэшируется для каждого экземпляра: два вызова fromReadable() с одним и тем же потоком возвращают один и тот же итерируемый объект.

Для потоков Readable в объектном режиме или с заданной кодировкой фрагменты автоматически нормализуются до 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> Классический поток Writable или любой объект с методами write() и on().
• options <Object>
  • backpressure <string> Политика обратного давления. По умолчанию: 'strict'.
    • 'strict' — запись отклоняется, когда буфер заполнен. Позволяет выявлять вызывающие стороны, игнорирующие обратное давление.
    • 'block' — при заполнении буфера запись ожидает события drain. Рекомендуется использовать с pipeTo().
    • 'drop-newest' — при заполнении буфера записываемые данные отбрасываются без предупреждения.
    • 'drop-oldest' — не поддерживается. Вызывает ERR_INVALID_ARG_VALUE.
• Возвращает: <Object> Адаптер Writer для потока/итератора.

Создает адаптер Writer для потока/итератора из классического потока Writable (или совместимого по структуре объекта). Адаптер можно передать в pipeTo() в качестве назначения.

Поскольку все операции записи в классический Writable по своей сути асинхронны, синхронные методы Writer (writeSync, writevSync, endSync) всегда возвращают false или -1, передавая выполнение асинхронному пути. Параметр options.signal для отдельной записи из интерфейса Writer также игнорируется.

Результат кэшируется для каждого экземпляра и политики обратного давления: два вызова fromWritable() с одним и тем же потоком и параметром backpressure возвращают один и тот же Writer.

Для потоков, совместимых по структуре, которые не предоставляют свойства writableHighWaterMark, writableLength или аналогичные, используются подходящие значения по умолчанию. Записываемые потоки в объектном режиме (если это можно определить) отклоняются, поскольку интерфейс Writer работает только с байтами.

import { Writable } from 'node:stream';
import { from, fromWritable, pipeTo } from 'node:stream/iter';

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

await pipeTo(from('hello world'),
             fromWritable(writable, { backpressure: 'block' }));

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: 'block' }));
}
run();

toReadable(source[, options])

• source <AsyncIterable> Источник AsyncIterable<Uint8Array[]>, например значение, возвращаемое pull() или from().
• options <Object>
  • highWaterMark <number> Размер внутреннего буфера в байтах, при достижении которого применяется обратное давление. По умолчанию: 65536 (64 КБ).
  • signal <AbortSignal> Необязательный сигнал для отмены чтения.
• Возвращает: <stream.Readable>

Создает stream.Readable в режиме байтов из AsyncIterable<Uint8Array[]> (встроенного формата пакетов, используемого API потока/итератора). Каждый Uint8Array в выданном пакете передается в Readable как отдельный фрагмент.

import { createWriteStream } from 'node:fs';
import { from, pull, toReadable } from 'node:stream/iter';
import { compressGzip } from 'node:zlib/iter';

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

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

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} Источник Iterable<Uint8Array[]>, например значение, возвращаемое pullSync() или fromSync().
• options <Object>
  • highWaterMark <number> Размер внутреннего буфера в байтах, при достижении которого применяется обратное давление. По умолчанию: 65536 (64 КБ).
• Возвращает: <stream.Readable>

Создает stream.Readable в режиме байтов из синхронного Iterable<Uint8Array[]>. Метод _read() синхронно получает данные из итератора, поэтому данные сразу доступны через 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> Writer для потока/итератора. Обязателен только метод write(); методы end(), fail(), writeSync(), writevSync(), endSync() и writev() необязательны.
• Возвращает: <stream.Writable>

Создает классический stream.Writable, работающий на основе Writer для потока/итератора.

При каждом вызове _write() / _writev() сначала выполняется синхронный метод Writer (writeSync / writevSync); если синхронный путь возвращает false или вызывает исключение, используется асинхронный метод. Аналогично, _final() сначала вызывает endSync(), а затем end(). Если синхронный путь завершается успешно, обратный вызов откладывается с помощью queueMicrotask для сохранения контракта асинхронного разрешения.

Свойству highWaterMark Writable присваивается значение Number.MAX_SAFE_INTEGER, чтобы фактически отключить внутреннюю буферизацию и позволить базовому Writer напрямую управлять обратным давлением.

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();

Символы протокола

Эти общеизвестные символы позволяют сторонним объектам участвовать в протоколе потоковой передачи без прямого импорта из node:stream/iter.

Stream.broadcastProtocol

• Значение: Symbol.for('Stream.broadcastProtocol')

Значение должно быть функцией. При вызове из Broadcast.from() она получает параметры, переданные в Broadcast.from(), и должна возвращать объект, соответствующий интерфейсу {Broadcast}. Реализация полностью настраиваемая — она может управлять потребителями, буферизацией и обратным давлением любым удобным способом.

import { 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 } = Broadcast();
    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, 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 } = Broadcast();
    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, если обратного давления нет, или промис, который выполнится со значением true-подобного типа, когда обратное давление прекратится.

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 backpressure

const { 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}. Реализация полностью настраиваемая — она может управлять общим источником, потребителями, буферизацией и обратным давлением любым удобным способом.

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}. Реализация полностью настраиваемая — она может управлять общим источником, потребителями и буферизацией любым удобным способом.

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 или другой потоковый объект.

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.

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'