Обзор

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

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

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

Пример ReadableStream

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

Модули MJS

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

Модули CJS

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

readableStream.locked

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

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

readableStream.cancel([reason])

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

readableStream.getReader([options])

• options <Объект>
  • mode <строка> 'byob' или undefined
• Возвращает: <ReadableStreamDefaultReader> | <ReadableStreamBYOBReader>

Модули MJS

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

const stream = new ReadableStream();

const reader = stream.getReader();

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

Модули CJS

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

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

Приводит readableStream.locked к true во время активной операции конвейера.

Модули MJS

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

Модули CJS

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 <Объект>
  • preventAbort <логическое> Если true, ошибки в этом ReadableStream не приведут к прерыванию destination.
  • preventCancel <логическое> Если true, ошибки в destination не приведут к отмене этого ReadableStream.
  • preventClose <логическое> Если true, закрытие этого ReadableStream не приведёт к закрытию destination.
  • signal <AbortSignal> Позволяет отменить передачу данных с помощью <AbortController>.
• Возвращает: Промис, разрешаемый значением undefined

Приводит readableStream.locked к true во время активной операции конвейера.

readableStream.tee()

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

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

Приводит readableStream.locked к состоянию true.

readableStream.values([options])

• options <Объект>
  • preventCancel <булево> Если 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 <Итерируемый> Объект, реализующий протокол итерации Symbol.asyncIterator или Symbol.iterator.

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

Модули MJS

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'

Модули CJS

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

Класс: ReadableStreamDefaultReader

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

new ReadableStreamDefaultReader(stream)

• stream <ReadableStream>

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

readableStreamDefaultReader.cancel([reason])

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

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

readableStreamDefaultReader.closed

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

readableStreamDefaultReader.read()

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

Запрашивает следующий кусок данных из базового <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 <любой>
• Возвращает: промис, выполняемый со значением undefined.

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

readableStreamBYOBReader.closed

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

readableStreamBYOBReader.read(view)

• view <Буфер> | <Массив типизированных значений> | <DataView>
• Возвращает: промис, выполняемый с объектом:
  • value <ArrayBuffer>
  • done <булево>

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

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

readableStreamBYOBReader.releaseLock()

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

Класс: ReadableStreamDefaultController

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

readableStreamDefaultController.close()

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

readableStreamDefaultController.desiredSize

• Тип: <число>

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

readableStreamDefaultController.enqueue([chunk])

• chunk <любой>

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

readableStreamDefaultController.error([error])

• error <любой>

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

Класс: ReadableByteStreamController

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

readableByteStreamController.byobRequest

• Тип: <ReadableStreamBYOBRequest>

readableByteStreamController.close()

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

readableByteStreamController.desiredSize

• Тип: <число>

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

readableByteStreamController.enqueue(chunk)

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

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

readableByteStreamController.error([error])

• error <любой>

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

Класс: ReadableStreamBYOBRequest

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

readableStreamBYOBRequest.respond(bytesWritten)

• bytesWritten <число>

Указывает, что в 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 <Объект>
  • start <Функция> Пользовательская функция, которая вызывается сразу после создания WritableStream.
    • controller <WritableStreamDefaultController>
    • Возвращает: undefined или промис, выполненный с undefined.
  • write <Функция> Пользовательская функция, которая вызывается при записи фрагмента данных в WritableStream.
    • chunk <любой>
    • controller <WritableStreamDefaultController>
    • Возвращает: промис, выполненный с undefined.
  • close <Функция> Пользовательская функция, которая вызывается при закрытии WritableStream.
    • Возвращает: промис, выполненный с undefined.
  • abort <Функция> Пользовательская функция, вызываемая для прерывистого закрытия WritableStream.
    • reason <любой>
    • Возвращает: промис, выполненный с undefined.
  • type <любой> Параметр type зарезервирован для будущего использования и должен быть undefined.
• strategy <Объект>
  • highWaterMark <число> Максимальный размер внутренней очереди перед применением обратной связи о давлении.
  • size <Функция> Пользовательская функция для определения размера каждого фрагмента данных.
    • chunk <любой>
    • Возвращает: <число>

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 <Объект>
  • start <Функция> Пользовательская функция, которая вызывается немедленно при создании TransformStream.
    • controller <TransformStreamDefaultController>
    • Возвращает: undefined или промис, выполненный с undefined
  • transform <Функция> Пользовательская функция, которая получает и потенциально изменяет фрагмент данных, записанный в transformStream.writable, прежде чем передать его дальше transformStream.readable.
    • chunk <любой>
    • controller <TransformStreamDefaultController>
    • Возвращает: Промис, выполненный с undefined.
  • flush <Функция> Пользовательская функция, которая вызывается непосредственно перед закрытием записывающей части TransformStream, сигнализируя об окончании процесса преобразования.
    • controller <TransformStreamDefaultController>
    • Возвращает: Промис, выполненный с undefined.
  • readableType <любой> опция readableType зарезервирована для будущего использования и должна быть undefined.
  • writableType <любой> опция writableType зарезервирована для будущего использования и должна быть undefined.
• writableStrategy <Объект>
  • highWaterMark <число> Максимальный размер внутренней очереди перед применением обратной связи о загруженности.
  • size <Функция> Пользовательская функция, используемая для определения размера каждого фрагмента данных.
    • chunk <любой>
    • Возвращает: <число>
• readableStrategy <Объект>
  • highWaterMark <число> Максимальный размер внутренней очереди перед применением обратной связи о загруженности.
  • size <Функция> Пользовательская функция, используемая для определения размера каждого фрагмента данных.
    • chunk <любой>
    • Возвращает: <число>

transformStream.readable

• Тип: <Поток чтения>

transformStream.writable

• Тип: <Поток записи>

Передача с помощью 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

• Тип: <число>

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

transformStreamDefaultController.enqueue([chunk])

• chunk <любой>

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

transformStreamDefaultController.error([reason])

• reason <любой>

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

transformStreamDefaultController.terminate()

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

Класс: ByteLengthQueuingStrategy

new ByteLengthQueuingStrategy(init)

• init <Объект>
  • highWaterMark <число>

byteLengthQueuingStrategy.highWaterMark

• Тип: <число>

byteLengthQueuingStrategy.size

• Тип: <Функция>
  • chunk <любой>
  • Возвращает: <число>

Класс: CountQueuingStrategy

new CountQueuingStrategy(init)

• init <Объект>
  • highWaterMark <число>

countQueuingStrategy.highWaterMark

• Тип: <число>

countQueuingStrategy.size

• Тип: <Функция>
  • chunk <любой>
  • Возвращает: <число>

Класс: TextEncoderStream

new TextEncoderStream()

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

textEncoderStream.encoding

• Тип: <строка>

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

textEncoderStream.readable

• Тип: <ReadableStream>

textEncoderStream.writable

• Тип: <WritableStream>

Класс: TextDecoderStream

new TextDecoderStream([encoding[, options]])

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

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

textDecoderStream.encoding

• Тип: <строка>

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

textDecoderStream.fatal

• Тип: <логическое значение>

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

textDecoderStream.ignoreBOM

• Тип: <логическое значение>

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

textDecoderStream.readable

• Тип: <ReadableStream>

textDecoderStream.writable

• Тип: <WritableStream>

Класс: CompressionStream

new CompressionStream(format)

• format <строка> Одно из значений 'deflate', 'deflate-raw', или 'gzip'.

compressionStream.readable

• Тип: <ReadableStream>

compressionStream.writable

• Тип: <WritableStream>

Класс: DecompressionStream

new DecompressionStream(format)

• format <строка> Одно из значений 'deflate', 'deflate-raw', или 'gzip'.

decompressionStream.readable

• Тип: <ReadableStream>

decompressionStream.writable

• Тип: <WritableStream>

Функции-потребители утилит

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

К ним можно обратиться используя:

Модули MJS

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

Модули CJS

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

streamConsumers.arrayBuffer(stream)

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

Модули MJS

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

Модули CJS

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>, содержащим полное содержимое потока.

Модули MJS

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

Модули CJS

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>, содержащим полное содержимое потока.

Модули MJS

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

Модули CJS

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

Модули MJS

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

Модули CJS

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.

Модули MJS

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

Модули CJS

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