API веб-потоков

Реализация стандарта WHATWG Streams.

Обзор

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

Существует три основных типа объектов:

• ReadableStream — представляет источник потоковых данных.
• WritableStream — представляет получателя потоковых данных.
• TransformStream — представляет алгоритм преобразования потоковых данных.

Пример ReadableStream

В этом примере создаётся простой ReadableStream, который бесконечно отправляет текущую временную метку performance.now() раз в секунду. Для чтения данных из потока используется асинхронный итерируемый объект.

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

Совместимость с потоками Node.js

Потоки Node.js можно преобразовывать в веб-потоки и наоборот с помощью методов toWeb и fromWeb, доступных в объектах stream.Readable, stream.Writable и stream.Duplex.

Дополнительные сведения см. в соответствующей документации:

• stream.Readable.toWeb
• stream.Readable.fromWeb
• stream.Writable.toWeb
• stream.Writable.fromWeb
• stream.Duplex.toWeb
• stream.Duplex.fromWeb

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

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

Объект <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());
mjscopy

Асинхронный итератор будет получать данные из <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]);
jscopy

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

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, выделенных из пула. При передаче Buffer, <TypedArray> или <DataView> в readableStreamBYOBReader.read() нижележащий 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');
mjscopy

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

Класс: 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(),
]);
mjscopy

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.
  • cancel <Function> Пользовательская функция, вызываемая при отмене читаемой стороны TransformStream или прерывании записываемой стороны.
    • reason <any>
    • Возвращает: промис, выполненный с 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]);
jscopy

Класс: 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' или 'brotli'.

compressionStream.readable

• Тип: <ReadableStream>

compressionStream.writable

• Тип: <WritableStream>

Класс: DecompressionStream

new DecompressionStream(format)

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

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

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

import { bytes } 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 bytes(readable);
console.log(`from readable: ${data.length}`);
// Prints: from readable: 27

const { bytes } = 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);
bytes(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
});