Обзор

Стандарт потоков WHATWG (или «веб-потоки») определяет API для обработки потоковых данных. Он похож на API потоков Node.js, но появился позже и стал «стандартным» API для потоковой передачи данных в различных средах JavaScript.

import {
  ReadableStream,
} from 'node:stream/web';

import {
  setInterval as every,
} from 'node:timers/promises';

import {
  performance,
} from 'node:perf_hooks';

const SECOND = 1000;

const stream = new ReadableStream({
  async start(controller) {
    for await (const _ of every(SECOND))
      controller.enqueue(performance.now());
  },
});

for await (const value of stream)
  console.log(value);

const {
  ReadableStream,
} = require('node:stream/web');

const {
  setInterval: every,
} = require('node:timers/promises');

const {
  performance,
} = require('node:perf_hooks');

const SECOND = 1000;

const stream = new ReadableStream({
  async start(controller) {
    for await (const _ of every(SECOND))
      controller.enqueue(performance.now());
  },
});

(async () => {
  for await (const value of stream)
    console.log(value);
})();

API

Класс: ReadableStream

new ReadableStream([underlyingSource [, strategy]])

• underlyingSource <Object>
  • start <Function> Пользовательская функция, которая вызывается немедленно при создании ReadableStream.
    • controller <ReadableStreamDefaultController> | <ReadableByteStreamController>
    • Возвращает: undefined или промис, выполненный с undefined.
  • pull <Function> Пользовательская функция, которая многократно вызывается, пока внутренняя очередь ReadableStream не заполнена. Операция может быть синхронной или асинхронной. В асинхронном случае функция не будет вызвана снова, пока не будет выполнен ранее возвращенный промис.
    • controller <ReadableStreamDefaultController> | <ReadableByteStreamController>
    • Возвращает: промис, выполненный с undefined.
  • cancel <Function> Пользовательская функция, которая вызывается при отмене ReadableStream.
    • reason <any>
    • Возвращает: промис, выполненный с undefined.
  • type <string> Должно быть 'bytes' или undefined.
  • autoAllocateChunkSize <number> Используется только, если type равно 'bytes'. Если задано ненулевое значение, буфер представления выделяется автоматически для ReadableByteStreamController.byobRequest. Если значение не задано, для передачи данных необходимо использовать внутренние очереди потока через ReadableStreamDefaultReader средства чтения по умолчанию.
• strategy <Object>
  • highWaterMark <number> Максимальный размер внутренней очереди до применения обратного давления.
  • size <Function> Пользовательская функция, используемая для определения размера каждого фрагмента данных.
    • chunk <any>
    • Возвращает: <number>

readableStream.locked

• Тип: <boolean> Устанавливается в true, если для этого <ReadableStream> имеется активное средство чтения.

По умолчанию свойство readableStream.locked имеет значение false. Пока активное средство чтения получает данные из потока, оно переключается на true.

readableStream.cancel([reason])

• reason <any>
• Возвращает: промис, выполненный с undefined после завершения отмены.

readableStream.getReader([options])

• options <Object>
  • mode <string> 'byob' или undefined
• Возвращает: <ReadableStreamDefaultReader> | <ReadableStreamBYOBReader>

import { ReadableStream } from 'node:stream/web';

const stream = new ReadableStream();

const reader = stream.getReader();

console.log(await reader.read());

const { ReadableStream } = require('node:stream/web');

const stream = new ReadableStream();

const reader = stream.getReader();

reader.read().then(console.log);

Приводит к тому, что readableStream.locked становится true.

readableStream.pipeThrough(transform[, options])

• transform <Object>
  • readable <ReadableStream> Поток ReadableStream, в который transform.writable будет передавать потенциально измененные данные, полученные из этого ReadableStream.
  • writable <WritableStream> Поток WritableStream, в который будут записываться данные этого ReadableStream.
• options <Object>
  • preventAbort <boolean> Если значение равно true, ошибки в этом ReadableStream не приведут к прерыванию transform.writable.
  • preventCancel <boolean> Если значение равно true, ошибки в целевом transform.writable не приведут к отмене этого ReadableStream.
  • preventClose <boolean> Если значение равно true, закрытие этого ReadableStream не приведет к закрытию transform.writable.
  • signal <AbortSignal> Позволяет отменить передачу данных с помощью <AbortController>.
• Возвращает: <ReadableStream> Из transform.readable.

Соединяет этот <ReadableStream> с парой <ReadableStream> и <WritableStream>, переданной в аргументе transform, так что данные из этого <ReadableStream> записываются в transform.writable, при необходимости преобразуются, а затем передаются в transform.readable. После настройки конвейера возвращается transform.readable.

Пока операция передачи данных активна, readableStream.locked становится true.

import {
  ReadableStream,
  TransformStream,
} from 'node:stream/web';

const stream = new ReadableStream({
  start(controller) {
    controller.enqueue('a');
  },
});

const transform = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});

const transformedStream = stream.pipeThrough(transform);

for await (const chunk of transformedStream)
  console.log(chunk);
  // Prints: A

const {
  ReadableStream,
  TransformStream,
} = require('node:stream/web');

const stream = new ReadableStream({
  start(controller) {
    controller.enqueue('a');
  },
});

const transform = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});

const transformedStream = stream.pipeThrough(transform);

(async () => {
  for await (const chunk of transformedStream)
    console.log(chunk);
    // Prints: A
})();

readableStream.pipeTo(destination[, options])

• destination <WritableStream> Поток <WritableStream>, в который будут записываться данные этого ReadableStream.
• options <Object>
  • preventAbort <boolean> Если значение равно true, ошибки в этом ReadableStream не приведут к прерыванию destination.
  • preventCancel <boolean> Если значение равно true, ошибки в destination не приведут к отмене этого ReadableStream.
  • preventClose <boolean> Если значение равно true, закрытие этого ReadableStream не приведет к закрытию destination.
  • signal <AbortSignal> Позволяет отменить передачу данных с помощью <AbortController>.
• Возвращает: промис, выполненный с undefined

Пока операция передачи данных активна, readableStream.locked становится true.

readableStream.tee()

• Возвращает: <ReadableStream[]>

Возвращает пару новых экземпляров <ReadableStream>, которым будут передаваться данные этого ReadableStream. Каждый из них получит одинаковые данные.

Приводит к тому, что readableStream.locked становится true.

readableStream.values([options])

• options <Object>
  • preventCancel <boolean> Если значение равно true, предотвращает закрытие <ReadableStream> при неожиданном завершении асинхронного итератора. По умолчанию: false.

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

Пока асинхронный итератор активен, readableStream.locked становится true.

import { Buffer } from 'node:buffer';

const stream = new ReadableStream(getSomeSource());

for await (const chunk of stream.values({ preventCancel: true }))
  console.log(Buffer.from(chunk).toString()); copy

Асинхронная итерация

Объект <ReadableStream> поддерживает протокол асинхронного итератора с использованием синтаксиса for await.

import { Buffer } from 'node:buffer';

const stream = new ReadableStream(getSomeSource());

for await (const chunk of stream)
  console.log(Buffer.from(chunk).toString()); copy

Асинхронный итератор будет получать данные из <ReadableStream>, пока поток не завершится.

По умолчанию, если асинхронный итератор завершится раньше времени (из-за break, return или throw), <ReadableStream> будет закрыт. Чтобы предотвратить автоматическое закрытие <ReadableStream>, используйте метод readableStream.values() для получения асинхронного итератора и установите для параметра preventCancel значение true.

<ReadableStream> не должен быть заблокирован (то есть у него не должно быть существующего активного средства чтения). Во время асинхронной итерации <ReadableStream> будет заблокирован.

Передача с помощью postMessage()

Экземпляр <ReadableStream> можно передать с помощью <MessagePort>.

const stream = new ReadableStream(getReadableSourceSomehow());

const { port1, port2 } = new MessageChannel();

port1.onmessage = ({ data }) => {
  data.getReader().read().then((chunk) => {
    console.log(chunk);
  });
};

port2.postMessage(stream, [stream]); copy

ReadableStream.from(iterable)

• iterable <Iterable> Объект, реализующий протокол итерируемого объекта Symbol.asyncIterator или Symbol.iterator.

Вспомогательный метод, создающий новый <ReadableStream> из итерируемого объекта.

import { ReadableStream } from 'node:stream/web';

async function* asyncIterableGenerator() {
  yield 'a';
  yield 'b';
  yield 'c';
}

const stream = ReadableStream.from(asyncIterableGenerator());

for await (const chunk of stream)
  console.log(chunk); // Prints: 'a', 'b', 'c'

const { ReadableStream } = require('node:stream/web');

async function* asyncIterableGenerator() {
  yield 'a';
  yield 'b';
  yield 'c';
}

(async () => {
  const stream = ReadableStream.from(asyncIterableGenerator());

  for await (const chunk of stream)
    console.log(chunk); // Prints: 'a', 'b', 'c'
})();

Чтобы передать результирующий <ReadableStream> в <WritableStream>, <Iterable> должен выдавать последовательность объектов <Buffer>, <TypedArray> или <DataView>.

import { ReadableStream } from 'node:stream/web';
import { Buffer } from 'node:buffer';

async function* asyncIterableGenerator() {
  yield Buffer.from('a');
  yield Buffer.from('b');
  yield Buffer.from('c');
}

const stream = ReadableStream.from(asyncIterableGenerator());

await stream.pipeTo(createWritableStreamSomehow());

const { ReadableStream } = require('node:stream/web');
const { Buffer } = require('node:buffer');

async function* asyncIterableGenerator() {
  yield Buffer.from('a');
  yield Buffer.from('b');
  yield Buffer.from('c');
}

const stream = ReadableStream.from(asyncIterableGenerator());

(async () => {
  await stream.pipeTo(createWritableStreamSomehow());
})();

Класс: ReadableStreamDefaultReader

По умолчанию вызов readableStream.getReader() без аргументов возвращает экземпляр ReadableStreamDefaultReader. Средство чтения по умолчанию рассматривает фрагменты данных, передаваемые через поток, как непрозрачные значения, благодаря чему <ReadableStream> может работать практически с любыми значениями JavaScript.

new ReadableStreamDefaultReader(stream)

• stream <ReadableStream>

Создает новый <ReadableStreamDefaultReader>, заблокированный для указанного <ReadableStream>.

readableStreamDefaultReader.cancel([reason])

• reason <any>
• Возвращает: промис, выполненный с undefined.

Отменяет <ReadableStream> и возвращает промис, который выполняется после отмены нижележащего потока.

readableStreamDefaultReader.closed

• Тип: <Promise> Выполняется с undefined, когда связанный <ReadableStream> закрыт, или отклоняется, если в потоке возникает ошибка либо блокировка средства чтения снимается до завершения закрытия потока.

readableStreamDefaultReader.read()

• Возвращает: промис, выполненный с объектом:
  • value <any>
  • done <boolean>

Запрашивает следующий фрагмент данных из нижележащего <ReadableStream> и возвращает промис, который выполняется с этими данными, когда они становятся доступными.

readableStreamDefaultReader.releaseLock()

Снимает блокировку этого средства чтения с нижележащего <ReadableStream>.

Класс: ReadableStreamBYOBReader

ReadableStreamBYOBReader — это альтернативное средство чтения для потоков <ReadableStream>, работающих с байтами (то есть потоков, созданных с параметром underlyingSource.type, равным 'bytes' при создании ReadableStream).

Название BYOB — сокращение от «bring your own buffer» («используйте собственный буфер»). Этот шаблон позволяет эффективнее считывать байтовые данные, избегая лишнего копирования.

import {
  open,
} from 'node:fs/promises';

import {
  ReadableStream,
} from 'node:stream/web';

import { Buffer } from 'node:buffer';

class Source {
  type = 'bytes';
  autoAllocateChunkSize = 1024;

  async start(controller) {
    this.file = await open(new URL(import.meta.url));
    this.controller = controller;
  }

  async pull(controller) {
    const view = controller.byobRequest?.view;
    const {
      bytesRead,
    } = await this.file.read({
      buffer: view,
      offset: view.byteOffset,
      length: view.byteLength,
    });

    if (bytesRead === 0) {
      await this.file.close();
      this.controller.close();
    }
    controller.byobRequest.respond(bytesRead);
  }
}

const stream = new ReadableStream(new Source());

async function read(stream) {
  const reader = stream.getReader({ mode: 'byob' });

  const chunks = [];
  let result;
  do {
    result = await reader.read(Buffer.alloc(100));
    if (result.value !== undefined)
      chunks.push(Buffer.from(result.value));
  } while (!result.done);

  return Buffer.concat(chunks);
}

const data = await read(stream);
console.log(Buffer.from(data).toString()); copy

new ReadableStreamBYOBReader(stream)

• stream <ReadableStream>

Создает новый ReadableStreamBYOBReader, заблокированный для указанного <ReadableStream>.

readableStreamBYOBReader.cancel([reason])

• reason <any>
• Возвращает: промис, выполненный с undefined.

Отменяет <ReadableStream> и возвращает промис, который выполняется после отмены нижележащего потока.

readableStreamBYOBReader.closed

• Тип: <Promise> Выполняется с undefined, когда связанный <ReadableStream> закрыт, или отклоняется, если в потоке возникает ошибка либо блокировка средства чтения снимается до завершения закрытия потока.

readableStreamBYOBReader.read(view[, options])

• view <Buffer> | <TypedArray> | <DataView>
• options <Object>
  • min <number> Если задано, возвращенный промис выполнится только тогда, когда станет доступно указанное количество элементов min. Если значение не задано, промис выполняется, когда доступен хотя бы один элемент.
• Возвращает: промис, выполненный с объектом:
  • value <TypedArray> | <DataView>
  • done <boolean>

Запрашивает следующий фрагмент данных из нижележащего <ReadableStream> и возвращает промис, который выполняется с этими данными, когда они становятся доступными.

Не передавайте в этот метод экземпляр объекта <Buffer> из пула. Буферизованные объекты Buffer создаются с помощью Buffer.allocUnsafe() или Buffer.from() либо часто возвращаются различными функциями обратного вызова модуля node:fs. В таких объектах Buffer используется общий нижележащий объект <ArrayBuffer>, содержащий данные всех буферизованных экземпляров Buffer. Когда в readableStreamBYOBReader.read() передается объект Buffer, <TypedArray> или <DataView>, нижележащий ArrayBuffer представления отсоединяется, делая недействительными все существующие представления, которые могут ссылаться на этот ArrayBuffer. Это может привести к серьезным последствиям для приложения.

readableStreamBYOBReader.releaseLock()

Снимает блокировку этого средства чтения с нижележащего <ReadableStream>.

Класс: ReadableStreamDefaultController

У каждого <ReadableStream> есть контроллер, отвечающий за внутреннее состояние и управление очередью потока. ReadableStreamDefaultController — это реализация контроллера по умолчанию для ReadableStream, не работающих с байтами.

readableStreamDefaultController.close()

Закрывает <ReadableStream>, с которым связан этот контроллер.

readableStreamDefaultController.desiredSize

• Тип: <number>

Возвращает объем данных, необходимый для заполнения очереди <ReadableStream>.

readableStreamDefaultController.enqueue([chunk])

• chunk <any>

Добавляет новый фрагмент данных в очередь <ReadableStream>.

readableStreamDefaultController.error([error])

• error <any>

Сообщает об ошибке, из-за которой в <ReadableStream> возникает ошибка и поток закрывается.

Класс: ReadableByteStreamController

У каждого <ReadableStream> есть контроллер, отвечающий за внутреннее состояние и управление очередью потока. ReadableByteStreamController предназначен для ReadableStream, работающих с байтами.

readableByteStreamController.byobRequest

• Тип: <ReadableStreamBYOBRequest>

readableByteStreamController.close()

Закрывает <ReadableStream>, с которым связан этот контроллер.

readableByteStreamController.desiredSize

• Тип: <number>

Возвращает объем данных, необходимый для заполнения очереди <ReadableStream>.

readableByteStreamController.enqueue(chunk)

• chunk: <Buffer> | <TypedArray> | <DataView>

Добавляет новый фрагмент данных в очередь <ReadableStream>.

readableByteStreamController.error([error])

• error <any>

Сообщает об ошибке, из-за которой в <ReadableStream> возникает ошибка и поток закрывается.

Класс: ReadableStreamBYOBRequest

При использовании ReadableByteStreamController в потоках, работающих с байтами, и ReadableStreamBYOBReader свойство readableByteStreamController.byobRequest предоставляет доступ к экземпляру ReadableStreamBYOBRequest, представляющему текущий запрос на чтение. Этот объект используется для доступа к ArrayBuffer/TypedArray, предоставленному для заполнения в рамках запроса на чтение, и содержит методы, позволяющие сообщить о готовности данных.

readableStreamBYOBRequest.respond(bytesWritten)

• bytesWritten <number>

Сигнализирует о том, что в readableStreamBYOBRequest.view записано указанное число байтов: bytesWritten.

readableStreamBYOBRequest.respondWithNewView(view)

• view <Buffer> | <TypedArray> | <DataView>

Сигнализирует, что запрос выполнен: байты записаны в новое значение Buffer, TypedArray или DataView.

readableStreamBYOBRequest.view

• Тип: <Buffer> | <TypedArray> | <DataView>

Класс: WritableStream

WritableStream — это место назначения, куда передаются данные потока.

import {
  WritableStream,
} from 'node:stream/web';

const stream = new WritableStream({
  write(chunk) {
    console.log(chunk);
  },
});

await stream.getWriter().write('Hello World'); copy

new WritableStream([underlyingSink[, strategy]])

• underlyingSink <Object>
  • start <Function> Функция, определяемая пользователем и вызываемая сразу после создания WritableStream.
    • controller <WritableStreamDefaultController>
    • Возвращает: undefined или промис, выполненный с undefined.
  • write <Function> Функция, определяемая пользователем и вызываемая после записи фрагмента данных в WritableStream.
    • chunk <any>
    • controller <WritableStreamDefaultController>
    • Возвращает: промис, выполненный с undefined.
  • close <Function> Функция, определяемая пользователем и вызываемая при закрытии WritableStream.
    • Возвращает: промис, выполненный с undefined.
  • abort <Function> Функция, определяемая пользователем и вызываемая для принудительного закрытия WritableStream.
    • reason <any>
    • Возвращает: промис, выполненный с undefined.
  • type <any> Параметр type зарезервирован для будущего использования и должен иметь значение undefined.
• strategy <Object>
  • highWaterMark <number> Максимальный размер внутренней очереди до применения обратного давления.
  • size <Function> Функция, определяемая пользователем и используемая для определения размера каждого фрагмента данных.
    • chunk <any>
    • Возвращает: <number>

writableStream.abort([reason])

• reason <any>
• Возвращает: промис, выполненный с undefined.

Принудительно завершает WritableStream. Все записи в очереди будут отменены, а связанные с ними промисы — отклонены.

writableStream.close()

• Возвращает: промис, выполненный с undefined.

Закрывает WritableStream, когда больше не ожидается дополнительных записей.

writableStream.getWriter()

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

Создает и возвращает новый экземпляр средства записи, который можно использовать для записи данных в WritableStream.

writableStream.locked

• Тип: <boolean>

По умолчанию свойство writableStream.locked имеет значение false; оно переключается на true, пока к этому WritableStream подключено активное средство записи.

Передача с помощью postMessage()

Экземпляр <WritableStream> можно передать с помощью <MessagePort>.

const stream = new WritableStream(getWritableSinkSomehow());

const { port1, port2 } = new MessageChannel();

port1.onmessage = ({ data }) => {
  data.getWriter().write('hello');
};

port2.postMessage(stream, [stream]); copy

Класс: WritableStreamDefaultWriter

new WritableStreamDefaultWriter(stream)

• stream <WritableStream>

Создает новый WritableStreamDefaultWriter, заблокированный для указанного WritableStream.

writableStreamDefaultWriter.abort([reason])

• reason <any>
• Возвращает: промис, выполненный с undefined.

Принудительно завершает WritableStream. Все записи в очереди будут отменены, а связанные с ними промисы — отклонены.

writableStreamDefaultWriter.close()

• Возвращает: промис, выполненный с undefined.

Закрывает WritableStream, когда больше не ожидается дополнительных записей.

writableStreamDefaultWriter.closed

• Тип: <Promise> Выполняется с undefined, когда связанный <WritableStream> закрыт, или отклоняется, если в потоке возникает ошибка либо блокировка средства записи снимается до завершения закрытия потока.

writableStreamDefaultWriter.desiredSize

• Тип: <number>

Объем данных, необходимый для заполнения очереди <WritableStream>.

writableStreamDefaultWriter.ready

• Тип: <Promise> Выполняется с undefined, когда средство записи готово к использованию.

writableStreamDefaultWriter.releaseLock()

Снимает блокировку этого средства записи с базового <ReadableStream>.

writableStreamDefaultWriter.write([chunk])

• chunk: <any>
• Возвращает: промис, выполненный с undefined.

Добавляет новый фрагмент данных в очередь <WritableStream>.

Класс: WritableStreamDefaultController

WritableStreamDefaultController управляет внутренним состоянием <WritableStream>.

writableStreamDefaultController.error([error])

• error <any>

Вызывается пользовательским кодом, чтобы сообщить об ошибке при обработке данных WritableStream. При вызове <WritableStream> будет прерван, а ожидающие записи отменены.

writableStreamDefaultController.signal

• Тип: <AbortSignal> AbortSignal, который можно использовать для отмены ожидающих операций записи или закрытия при прерывании <WritableStream>.

Класс: TransformStream

TransformStream состоит из <ReadableStream> и <WritableStream>, соединенных таким образом, что данные, записанные в WritableStream, принимаются и, возможно, преобразуются перед помещением в очередь ReadableStream.

import {
  TransformStream,
} from 'node:stream/web';

const transform = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});

await Promise.all([
  transform.writable.getWriter().write('A'),
  transform.readable.getReader().read(),
]); copy

new TransformStream([transformer[, writableStrategy[, readableStrategy]]])

• transformer <Object>
  • start <Function> Функция, определяемая пользователем и вызываемая сразу после создания TransformStream.
    • controller <TransformStreamDefaultController>
    • Возвращает: undefined или промис, выполненный с undefined
  • transform <Function> Функция, определяемая пользователем, которая получает и, возможно, изменяет фрагмент данных, записанный в transformStream.writable, прежде чем передать его в transformStream.readable.
    • chunk <any>
    • controller <TransformStreamDefaultController>
    • Возвращает: промис, выполненный с undefined.
  • flush <Function> Функция, определяемая пользователем и вызываемая непосредственно перед закрытием записываемой стороны TransformStream, что сигнализирует о завершении процесса преобразования.
    • controller <TransformStreamDefaultController>
    • Возвращает: промис, выполненный с undefined.
  • readableType <any> параметр readableType зарезервирован для будущего использования и должен иметь значение undefined.
  • writableType <any> параметр writableType зарезервирован для будущего использования и должен иметь значение undefined.
• writableStrategy <Object>
  • highWaterMark <number> Максимальный размер внутренней очереди до применения обратного давления.
  • size <Function> Функция, определяемая пользователем и используемая для определения размера каждого фрагмента данных.
    • chunk <any>
    • Возвращает: <number>
• readableStrategy <Object>
  • highWaterMark <number> Максимальный размер внутренней очереди до применения обратного давления.
  • size <Function> Функция, определяемая пользователем и используемая для определения размера каждого фрагмента данных.
    • chunk <any>
    • Возвращает: <number>

transformStream.readable

• Тип: <ReadableStream>

transformStream.writable

• Тип: <WritableStream>

Передача с помощью postMessage()

Экземпляр <TransformStream> можно передать с помощью <MessagePort>.

const stream = new TransformStream();

const { port1, port2 } = new MessageChannel();

port1.onmessage = ({ data }) => {
  const { writable, readable } = data;
  // ...
};

port2.postMessage(stream, [stream]); copy

Класс: TransformStreamDefaultController

TransformStreamDefaultController управляет внутренним состоянием TransformStream.

transformStreamDefaultController.desiredSize

• Тип: <number>

Объем данных, необходимый для заполнения очереди читаемой стороны.

transformStreamDefaultController.enqueue([chunk])

• chunk <any>

Добавляет фрагмент данных в очередь читаемой стороны.

transformStreamDefaultController.error([reason])

• reason <any>

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

transformStreamDefaultController.terminate()

Закрывает читаемую сторону потока преобразования и вызывает принудительное закрытие записываемой стороны с ошибкой.

Класс: ByteLengthQueuingStrategy

new ByteLengthQueuingStrategy(init)

• init <Object>
  • highWaterMark <number>

byteLengthQueuingStrategy.highWaterMark

• Тип: <number>

byteLengthQueuingStrategy.size

• Тип: <Function>
  • chunk <any>
  • Возвращает: <number>

Класс: CountQueuingStrategy

new CountQueuingStrategy(init)

• init <Object>
  • highWaterMark <number>

countQueuingStrategy.highWaterMark

• Тип: <number>

countQueuingStrategy.size

• Тип: <Function>
  • chunk <any>
  • Возвращает: <number>

Класс: TextEncoderStream

new TextEncoderStream()

Создает новый экземпляр TextEncoderStream.

textEncoderStream.encoding

• Тип: <string>

Кодировка, поддерживаемая экземпляром TextEncoderStream.

textEncoderStream.readable

• Тип: <ReadableStream>

textEncoderStream.writable

• Тип: <WritableStream>

Класс: TextDecoderStream

new TextDecoderStream([encoding[, options]])

• encoding <string> Определяет encoding, поддерживаемую этим экземпляром TextDecoder. По умолчанию: 'utf-8'.
• options <Object>
  • fatal <boolean> true, если ошибки декодирования являются фатальными.
  • ignoreBOM <boolean> Если задано значение true, TextDecoderStream включит метку порядка байтов в декодированный результат. Если задано значение false, метка порядка байтов будет удалена из результата. Этот параметр используется только тогда, когда encoding имеет значение 'utf-8', 'utf-16be' или 'utf-16le'. По умолчанию: false.

Создает новый экземпляр TextDecoderStream.

textDecoderStream.encoding

• Тип: <string>

Кодировка, поддерживаемая экземпляром TextDecoderStream.

textDecoderStream.fatal

• Тип: <boolean>

Значение будет true, если ошибки декодирования приводят к выбрасыванию TypeError.

textDecoderStream.ignoreBOM

• Тип: <boolean>

Значение будет true, если результат декодирования будет включать метку порядка байтов.

textDecoderStream.readable

• Тип: <ReadableStream>

textDecoderStream.writable

• Тип: <WritableStream>

Класс: CompressionStream

new CompressionStream(format)

• format <string> Одно из значений: 'deflate', 'deflate-raw' или 'gzip'.

compressionStream.readable

• Тип: <ReadableStream>

compressionStream.writable

• Тип: <WritableStream>

Класс: DecompressionStream

new DecompressionStream(format)

• format <string> Одно из значений: 'deflate', 'deflate-raw' или 'gzip'.

decompressionStream.readable

• Тип: <ReadableStream>

decompressionStream.writable

• Тип: <WritableStream>

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

Вспомогательные функции для чтения потоков предоставляют стандартные способы чтения данных из потоков.

Они доступны через:

import {
  arrayBuffer,
  blob,
  buffer,
  json,
  text,
} from 'node:stream/consumers';

const {
  arrayBuffer,
  blob,
  buffer,
  json,
  text,
} = require('node:stream/consumers');

streamConsumers.arrayBuffer(stream)

• stream <ReadableStream> | <stream.Readable> | <AsyncIterator>
• Возвращает: <Promise> Выполняется с ArrayBuffer, содержащим всё содержимое потока.

import { arrayBuffer } from 'node:stream/consumers';
import { Readable } from 'node:stream';
import { TextEncoder } from 'node:util';

const encoder = new TextEncoder();
const dataArray = encoder.encode('hello world from consumers!');

const readable = Readable.from(dataArray);
const data = await arrayBuffer(readable);
console.log(`from readable: ${data.byteLength}`);
// Prints: from readable: 76

const { arrayBuffer } = require('node:stream/consumers');
const { Readable } = require('node:stream');
const { TextEncoder } = require('node:util');

const encoder = new TextEncoder();
const dataArray = encoder.encode('hello world from consumers!');
const readable = Readable.from(dataArray);
arrayBuffer(readable).then((data) => {
  console.log(`from readable: ${data.byteLength}`);
  // Prints: from readable: 76
});

streamConsumers.blob(stream)

• stream <ReadableStream> | <stream.Readable> | <AsyncIterator>
• Возвращает: <Promise> Выполняется с объектом <Blob>, содержащим всё содержимое потока.

import { blob } from 'node:stream/consumers';

const dataBlob = new Blob(['hello world from consumers!']);

const readable = dataBlob.stream();
const data = await blob(readable);
console.log(`from readable: ${data.size}`);
// Prints: from readable: 27

const { blob } = require('node:stream/consumers');

const dataBlob = new Blob(['hello world from consumers!']);

const readable = dataBlob.stream();
blob(readable).then((data) => {
  console.log(`from readable: ${data.size}`);
  // Prints: from readable: 27
});

streamConsumers.buffer(stream)

• stream <ReadableStream> | <stream.Readable> | <AsyncIterator>
• Возвращает: <Promise> Выполняется с объектом <Buffer>, содержащим всё содержимое потока.

import { buffer } from 'node:stream/consumers';
import { Readable } from 'node:stream';
import { Buffer } from 'node:buffer';

const dataBuffer = Buffer.from('hello world from consumers!');

const readable = Readable.from(dataBuffer);
const data = await buffer(readable);
console.log(`from readable: ${data.length}`);
// Prints: from readable: 27

const { buffer } = require('node:stream/consumers');
const { Readable } = require('node:stream');
const { Buffer } = require('node:buffer');

const dataBuffer = Buffer.from('hello world from consumers!');

const readable = Readable.from(dataBuffer);
buffer(readable).then((data) => {
  console.log(`from readable: ${data.length}`);
  // Prints: from readable: 27
});

streamConsumers.json(stream)

• stream <ReadableStream> | <stream.Readable> | <AsyncIterator>
• Возвращает: <Promise> Выполняется с содержимым потока, разобранным как строка в кодировке UTF-8 и переданным затем в JSON.parse().

import { json } from 'node:stream/consumers';
import { Readable } from 'node:stream';

const items = Array.from(
  {
    length: 100,
  },
  () => ({
    message: 'hello world from consumers!',
  }),
);

const readable = Readable.from(JSON.stringify(items));
const data = await json(readable);
console.log(`from readable: ${data.length}`);
// Prints: from readable: 100

const { json } = require('node:stream/consumers');
const { Readable } = require('node:stream');

const items = Array.from(
  {
    length: 100,
  },
  () => ({
    message: 'hello world from consumers!',
  }),
);

const readable = Readable.from(JSON.stringify(items));
json(readable).then((data) => {
  console.log(`from readable: ${data.length}`);
  // Prints: from readable: 100
});

streamConsumers.text(stream)

• stream <ReadableStream> | <stream.Readable> | <AsyncIterator>
• Возвращает: <Promise> Выполняется с содержимым потока, разобранным как строка в кодировке UTF-8.

import { text } from 'node:stream/consumers';
import { Readable } from 'node:stream';

const readable = Readable.from('Hello world from consumers!');
const data = await text(readable);
console.log(`from readable: ${data.length}`);
// Prints: from readable: 27

const { text } = require('node:stream/consumers');
const { Readable } = require('node:stream');

const readable = Readable.from('Hello world from consumers!');
text(readable).then((data) => {
  console.log(`from readable: ${data.length}`);
  // Prints: from readable: 27
});