Web Streams API

[История]

Версия	Изменения
v21.0.0	Больше не является экспериментальной.
v18.0.0	Использование этого API больше не вызывает предупреждение во время выполнения.
v16.5.0	Добавлено в: v16.5.0

[Стабильно: 2 - Стабильно]

Стабильно: 2 Стабильность: 2 - Стабильно

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

Обзор

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

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

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

Пример `ReadableStream`

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

ESMCJS

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

Class: `ReadableStream`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь доступен в глобальном объекте.
v16.5.0	Добавлено в: v16.5.0

`new ReadableStream([underlyingSource [, strategy]])`

Добавлено в: v16.5.0

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

`readableStream.locked`

Добавлено в версии: v16.5.0

Тип: <boolean> Устанавливается в true, если для данного <ReadableStream> есть активный читатель.

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

`readableStream.cancel([reason])`

Добавлено в версии: v16.5.0

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

`readableStream.getReader([options])`

Добавлено в версии: v16.5.0

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

ESMCJS

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])`

Добавлено в версии: v16.5.0

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 во время активной операции конвейера.

ESMCJS

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])`

Добавлено в версии: v16.5.0

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

[История]

Версия	Изменения
v18.10.0, v16.18.0	Поддержка разветвления читаемого байтового потока.
v16.5.0	Добавлено в версии: v16.5.0

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

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

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

`readableStream.values([options])`

Добавлено в версии: v16.5.0

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

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

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

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

`ReadableStream.from(iterable)`

Добавлено в: v20.6.0

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

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

ESMCJS

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

Class: `ReadableStreamDefaultReader`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь представлен в глобальном объекте.
v16.5.0	Добавлено в: v16.5.0

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

`new ReadableStreamDefaultReader(stream)`

Добавлено в: v16.5.0

stream <ReadableStream>

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

`readableStreamDefaultReader.cancel([reason])`

Добавлено в: v16.5.0

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

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

`readableStreamDefaultReader.closed`

Добавлено в: v16.5.0

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

`readableStreamDefaultReader.read()`

Добавлено в: v16.5.0

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

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

`readableStreamDefaultReader.releaseLock()`

Добавлено в версии: v16.5.0

Освобождает блокировку этого считывателя для базового <ReadableStream>.

Класс: `ReadableStreamBYOBReader`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь доступен в глобальном объекте.
v16.5.0	Добавлено в версии: v16.5.0

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

`new ReadableStreamBYOBReader(stream)`

Добавлено в версии: v16.5.0

stream <ReadableStream>

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

`readableStreamBYOBReader.cancel([reason])`

Добавлено в: v16.5.0

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

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

`readableStreamBYOBReader.closed`

Добавлено в: v16.5.0

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

`readableStreamBYOBReader.read(view[, options])`

[История]

Версия	Изменения
v21.7.0, v20.17.0	Добавлен параметр `min`.
v16.5.0	Добавлено в: v16.5.0

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

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

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

`readableStreamBYOBReader.releaseLock()`

Добавлено в: v16.5.0

Освобождает блокировку этого читателя для базового <ReadableStream>.

Класс: `ReadableStreamDefaultController`

Добавлено в: v16.5.0

Каждый <ReadableStream> имеет контроллер, который отвечает за внутреннее состояние и управление очередью потока. ReadableStreamDefaultController является реализацией контроллера по умолчанию для ReadableStream, которые не являются байт-ориентированными.

`readableStreamDefaultController.close()`

Добавлено в: v16.5.0

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

`readableStreamDefaultController.desiredSize`

Добавлено в: v16.5.0

Тип: <number>

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

`readableStreamDefaultController.enqueue([chunk])`

Добавлено в: v16.5.0

chunk <any>

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

`readableStreamDefaultController.error([error])`

Добавлено в: v16.5.0

error <any>

Сигнализирует об ошибке, которая приводит к тому, что <ReadableStream> выдает ошибку и закрывается.

Класс: `ReadableByteStreamController`

[История]

Версия	Изменения
v18.10.0	Поддержка обработки BYOB pull запроса от выпущенного reader.
v16.5.0	Добавлено в: v16.5.0

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

`readableByteStreamController.byobRequest`

Добавлено в: v16.5.0

Тип: <ReadableStreamBYOBRequest>

`readableByteStreamController.close()`

Добавлено в: v16.5.0

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

`readableByteStreamController.desiredSize`

Добавлено в: v16.5.0

Тип: <number>

Возвращает количество данных, оставшихся для заполнения очереди <ReadableStream>.

`readableByteStreamController.enqueue(chunk)`

Добавлено в: v16.5.0

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

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

`readableByteStreamController.error([error])`

Добавлено в: v16.5.0

error <any>

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

Класс: `ReadableStreamBYOBRequest`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь представлен в глобальном объекте.
v16.5.0	Добавлено в: v16.5.0

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

`readableStreamBYOBRequest.respond(bytesWritten)`

Добавлено в: v16.5.0

bytesWritten <number>

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

`readableStreamBYOBRequest.respondWithNewView(view)`

Добавлено в: v16.5.0

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

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

`readableStreamBYOBRequest.view`

Добавлено в: v16.5.0

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

Класс: `WritableStream`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь доступен в глобальном объекте.
v16.5.0	Добавлено в: v16.5.0

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

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

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

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

`new WritableStream([underlyingSink[, strategy]])`

Добавлено в: v16.5.0

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])`

Добавлено в: v16.5.0

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

Резко завершает WritableStream. Все поставленные в очередь операции записи будут отменены с отклонением соответствующих им Promise.

`writableStream.close()`

Добавлено в: v16.5.0

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

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

`writableStream.getWriter()`

Добавлено в: v16.5.0

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

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

`writableStream.locked`

Добавлено в: v16.5.0

Тип: <boolean>

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

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

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

const stream = new WritableStream(getWritableSinkSomehow());

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

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

port2.postMessage(stream, [stream]);

Класс: `WritableStreamDefaultWriter`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь представлен в глобальном объекте.
v16.5.0	Добавлено в: v16.5.0

`new WritableStreamDefaultWriter(stream)`

Добавлено в: v16.5.0

stream <WritableStream>

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

`writableStreamDefaultWriter.abort([reason])`

Добавлено в: v16.5.0

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

`writableStreamDefaultWriter.close()`

Добавлено в версии: v16.5.0

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

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

`writableStreamDefaultWriter.closed`

Добавлено в версии: v16.5.0

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

`writableStreamDefaultWriter.desiredSize`

Добавлено в версии: v16.5.0

Тип: <number>

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

`writableStreamDefaultWriter.ready`

Добавлено в версии: v16.5.0

Тип: <Promise> Исполняется со значением undefined, когда писатель готов к использованию.

`writableStreamDefaultWriter.releaseLock()`

Добавлено в версии: v16.5.0

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

`writableStreamDefaultWriter.write([chunk])`

Добавлено в версии: v16.5.0

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

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

Class: `WritableStreamDefaultController`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь доступен в глобальном объекте.
v16.5.0	Добавлено в версии: v16.5.0

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

`writableStreamDefaultController.error([error])`

Добавлено в версии: v16.5.0

error <any>

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

`writableStreamDefaultController.signal`

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

Класс: `TransformStream`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь представлен в глобальном объекте.
v16.5.0	Добавлено в: v16.5.0

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

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

Добавлено в: v16.5.0

transformer <Object>
- start <Function> Определяемая пользователем функция, которая вызывается сразу при создании TransformStream.
- controller <TransformStreamDefaultController>
- Возвращает: undefined или promise, выполненный с undefined
- transform <Function> Определяемая пользователем функция, которая принимает и, возможно, изменяет фрагмент данных, записанный в transformStream.writable, прежде чем перенаправить его в transformStream.readable.
- chunk <any>
- controller <TransformStreamDefaultController>
- Возвращает: Promise, выполненный с undefined.
- flush <Function> Определяемая пользователем функция, которая вызывается непосредственно перед закрытием записываемой стороны TransformStream, сигнализируя об окончании процесса преобразования.
- controller <TransformStreamDefaultController>
- Возвращает: Promise, выполненный с 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`

Добавлено в: v16.5.0

Тип: <ReadableStream>

`transformStream.writable`

Добавлено в: v16.5.0

Тип: <WritableStream>

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

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

const stream = new TransformStream();

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

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

port2.postMessage(stream, [stream]);

Класс: `TransformStreamDefaultController`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь представлен в глобальном объекте.
v16.5.0	Добавлено в: v16.5.0

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

`transformStreamDefaultController.desiredSize`

Добавлено в: v16.5.0

Тип: <number>

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

`transformStreamDefaultController.enqueue([chunk])`

Добавлено в: v16.5.0

chunk <any>

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

`transformStreamDefaultController.error([reason])`

Добавлено в: v16.5.0

reason <any>

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

`transformStreamDefaultController.terminate()`

Добавлено в: v16.5.0

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

Класс: `ByteLengthQueuingStrategy`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь представлен в глобальном объекте.
v16.5.0	Добавлено в: v16.5.0

`new ByteLengthQueuingStrategy(init)`

Добавлено в: v16.5.0

init <Object>
- highWaterMark <number>

`byteLengthQueuingStrategy.highWaterMark`

Добавлено в: v16.5.0

Тип: <number>

`byteLengthQueuingStrategy.size`

Добавлено в: v16.5.0

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

Класс: `CountQueuingStrategy`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь доступен в глобальном объекте.
v16.5.0	Добавлено в: v16.5.0

`new CountQueuingStrategy(init)`

Добавлено в: v16.5.0

init <Object>
- highWaterMark <number>

`countQueuingStrategy.highWaterMark`

Добавлено в: v16.5.0

Тип: <number>

`countQueuingStrategy.size`

Добавлено в: v16.5.0

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

Класс: `TextEncoderStream`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь доступен в глобальном объекте.
v16.6.0	Добавлено в: v16.6.0

`new TextEncoderStream()`

Added in: v16.6.0

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

`textEncoderStream.encoding`

Added in: v16.6.0

Тип: <string>

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

`textEncoderStream.readable`

Added in: v16.6.0

Тип: <ReadableStream>

`textEncoderStream.writable`

Added in: v16.6.0

Тип: <WritableStream>

Class: `TextDecoderStream`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь представлен в глобальном объекте.
v16.6.0	Added in: v16.6.0

`new TextDecoderStream([encoding[, options]])`

Added in: v16.6.0

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`

Added in: v16.6.0

Тип: <string>

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

`textDecoderStream.fatal`

Added in: v16.6.0

Тип: <boolean>

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

`textDecoderStream.ignoreBOM`

Добавлено в: v16.6.0

Тип: <boolean>

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

`textDecoderStream.readable`

Добавлено в: v16.6.0

Тип: <ReadableStream>

`textDecoderStream.writable`

Добавлено в: v16.6.0

Тип: <WritableStream>

Класс: `CompressionStream`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь представлен в глобальном объекте.
v17.0.0	Добавлено в: v17.0.0

`new CompressionStream(format)`

[История]

Версия	Изменения
v21.2.0, v20.12.0	format теперь принимает значение `deflate-raw`.
v17.0.0	Добавлено в: v17.0.0

format <string> Одно из 'deflate', 'deflate-raw' или 'gzip'.

`compressionStream.readable`

Добавлено в: v17.0.0

Тип: <ReadableStream>

`compressionStream.writable`

Добавлено в: v17.0.0

Тип: <WritableStream>

Класс: `DecompressionStream`

[История]

Версия	Изменения
v18.0.0	Этот класс теперь представлен в глобальном объекте.
v17.0.0	Добавлено в: v17.0.0

`new DecompressionStream(format)`

[История]

Версия	Изменения
v21.2.0, v20.12.0	format теперь принимает значение `deflate-raw`.
v17.0.0	Добавлено в: v17.0.0

format <string> Одно из 'deflate', 'deflate-raw' или 'gzip'.

`decompressionStream.readable`

Добавлено в: v17.0.0

Тип: <ReadableStream>

`decompressionStream.writable`

Добавлено в: v17.0.0

Тип: <WritableStream>

Утилиты-потребители

Добавлено в: v16.7.0

Функции утилит-потребителей предоставляют общие опции для использования потоков.

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

ESMCJS

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

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

`streamConsumers.arrayBuffer(stream)`

Добавлено в: v16.7.0

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

ESMCJS

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

Добавлено в: v16.7.0

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

ESMCJS

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

Добавлено в: v16.7.0

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

ESMCJS

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

Добавлено в: v16.7.0

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

ESMCJS

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

Добавлено в: v16.7.0

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

ESMCJS

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}`);
// Выводит: 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}`);
  // Выводит: from readable: 27
});

Web Streams API ​

Обзор ​

Пример ReadableStream ​

API ​

Class: ReadableStream ​

new ReadableStream([underlyingSource [, strategy]]) ​

readableStream.locked ​

readableStream.cancel([reason]) ​

readableStream.getReader([options]) ​

readableStream.pipeThrough(transform[, options]) ​

readableStream.pipeTo(destination[, options]) ​

readableStream.tee() ​

readableStream.values([options]) ​

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

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

ReadableStream.from(iterable) ​

Class: ReadableStreamDefaultReader ​

new ReadableStreamDefaultReader(stream) ​

readableStreamDefaultReader.cancel([reason]) ​

readableStreamDefaultReader.closed ​

readableStreamDefaultReader.read() ​

readableStreamDefaultReader.releaseLock() ​

Класс: ReadableStreamBYOBReader ​

new ReadableStreamBYOBReader(stream) ​

readableStreamBYOBReader.cancel([reason]) ​

readableStreamBYOBReader.closed ​

readableStreamBYOBReader.read(view[, options]) ​

readableStreamBYOBReader.releaseLock() ​

Класс: ReadableStreamDefaultController ​

readableStreamDefaultController.close() ​

readableStreamDefaultController.desiredSize ​

readableStreamDefaultController.enqueue([chunk]) ​

readableStreamDefaultController.error([error]) ​

Класс: ReadableByteStreamController ​

readableByteStreamController.byobRequest ​

readableByteStreamController.close() ​

readableByteStreamController.desiredSize ​

readableByteStreamController.enqueue(chunk) ​

readableByteStreamController.error([error]) ​

Класс: ReadableStreamBYOBRequest ​

readableStreamBYOBRequest.respond(bytesWritten) ​

readableStreamBYOBRequest.respondWithNewView(view) ​

readableStreamBYOBRequest.view ​

Класс: WritableStream ​

new WritableStream([underlyingSink[, strategy]]) ​

writableStream.abort([reason]) ​

writableStream.close() ​

writableStream.getWriter() ​

writableStream.locked ​

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

Класс: WritableStreamDefaultWriter ​

new WritableStreamDefaultWriter(stream) ​

writableStreamDefaultWriter.abort([reason]) ​

writableStreamDefaultWriter.close() ​

writableStreamDefaultWriter.closed ​

writableStreamDefaultWriter.desiredSize ​

writableStreamDefaultWriter.ready ​

writableStreamDefaultWriter.releaseLock() ​

writableStreamDefaultWriter.write([chunk]) ​

Class: WritableStreamDefaultController ​

writableStreamDefaultController.error([error]) ​

writableStreamDefaultController.signal ​

Класс: TransformStream ​

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

transformStream.readable ​

transformStream.writable ​

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

Класс: TransformStreamDefaultController ​

transformStreamDefaultController.desiredSize ​

transformStreamDefaultController.enqueue([chunk]) ​

transformStreamDefaultController.error([reason]) ​

transformStreamDefaultController.terminate() ​

Класс: ByteLengthQueuingStrategy ​

new ByteLengthQueuingStrategy(init) ​

byteLengthQueuingStrategy.highWaterMark ​

byteLengthQueuingStrategy.size ​

Класс: CountQueuingStrategy ​

new CountQueuingStrategy(init) ​

countQueuingStrategy.highWaterMark ​

countQueuingStrategy.size ​

Web Streams API

Обзор

Пример `ReadableStream`

API

Class: `ReadableStream`

`new ReadableStream([underlyingSource [, strategy]])`

`readableStream.locked`

`readableStream.cancel([reason])`

`readableStream.getReader([options])`

`readableStream.pipeThrough(transform[, options])`

`readableStream.pipeTo(destination[, options])`

`readableStream.tee()`

`readableStream.values([options])`

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

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

`ReadableStream.from(iterable)`

Class: `ReadableStreamDefaultReader`

`new ReadableStreamDefaultReader(stream)`

`readableStreamDefaultReader.cancel([reason])`

`readableStreamDefaultReader.closed`

`readableStreamDefaultReader.read()`

`readableStreamDefaultReader.releaseLock()`

Класс: `ReadableStreamBYOBReader`

`new ReadableStreamBYOBReader(stream)`

`readableStreamBYOBReader.cancel([reason])`

`readableStreamBYOBReader.closed`

`readableStreamBYOBReader.read(view[, options])`

`readableStreamBYOBReader.releaseLock()`

Класс: `ReadableStreamDefaultController`

`readableStreamDefaultController.close()`

`readableStreamDefaultController.desiredSize`

`readableStreamDefaultController.enqueue([chunk])`

`readableStreamDefaultController.error([error])`

Класс: `ReadableByteStreamController`

`readableByteStreamController.byobRequest`

`readableByteStreamController.close()`

`readableByteStreamController.desiredSize`

`readableByteStreamController.enqueue(chunk)`

`readableByteStreamController.error([error])`

Класс: `ReadableStreamBYOBRequest`

`readableStreamBYOBRequest.respond(bytesWritten)`

`readableStreamBYOBRequest.respondWithNewView(view)`

`readableStreamBYOBRequest.view`

Класс: `WritableStream`

`new WritableStream([underlyingSink[, strategy]])`

`writableStream.abort([reason])`

`writableStream.close()`

`writableStream.getWriter()`

`writableStream.locked`

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

Класс: `WritableStreamDefaultWriter`

`new WritableStreamDefaultWriter(stream)`

`writableStreamDefaultWriter.abort([reason])`

`writableStreamDefaultWriter.close()`

`writableStreamDefaultWriter.closed`

`writableStreamDefaultWriter.desiredSize`

`writableStreamDefaultWriter.ready`

`writableStreamDefaultWriter.releaseLock()`

`writableStreamDefaultWriter.write([chunk])`

Class: `WritableStreamDefaultController`

`writableStreamDefaultController.error([error])`

`writableStreamDefaultController.signal`

Класс: `TransformStream`

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

`transformStream.readable`

`transformStream.writable`

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

Класс: `TransformStreamDefaultController`

`transformStreamDefaultController.desiredSize`

`transformStreamDefaultController.enqueue([chunk])`

`transformStreamDefaultController.error([reason])`

`transformStreamDefaultController.terminate()`

Класс: `ByteLengthQueuingStrategy`

`new ByteLengthQueuingStrategy(init)`

`byteLengthQueuingStrategy.highWaterMark`

`byteLengthQueuingStrategy.size`

Класс: `CountQueuingStrategy`

`new CountQueuingStrategy(init)`

`countQueuingStrategy.highWaterMark`

`countQueuingStrategy.size`