Web Streams API

[История]

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

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

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

Реализация стандарта потоков WHATWG.

Обзор

Стандарт потоков WHATWG (или «веб-потоки») определяет API для обработки потоковых данных. Он похож на API потоков 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

Класс: `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'. При установке ненулевого значения буфер представления автоматически выделяется для 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)
// Выведет: 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)
  // Выведет: 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>.
Возвращает: Promise, выполняющийся с undefined

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

`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) // Выведет: '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) // Выведет: 'a', 'b', 'c'
})()

Класс: `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 от выпущенного ридера.
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>.

Class: `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>

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

`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 или promise, выполняющийся с undefined.
- write <Function> Пользовательская функция, вызываемая при записи фрагмента данных в WritableStream.
- chunk <any>
- controller <WritableStreamDefaultController>
- Возвращает: promise, выполняющийся с undefined.
- close <Function> Пользовательская функция, вызываемая при закрытии WritableStream.
- Возвращает: promise, выполняющийся с undefined.
- abort <Function> Пользовательская функция, вызываемая для резкого закрытия WritableStream.
- reason <any>
- Возвращает: promise, выполняющийся с undefined.
- type <any> Опция type зарезервирована для будущего использования и должна быть неопределена.
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.

Прерывает WritableStream. Все ожидающие записи будут отменены, а связанные с ними Promise отклонены.

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

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

`transformStream.readable`

Добавлено в: 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>

Class: `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>

Class: `TextEncoderStream`

[История]

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

`new TextEncoderStream()`

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

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

`textEncoderStream.encoding`

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

Тип: <string>

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

`textEncoderStream.readable`

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

Тип: <ReadableStream>

`textEncoderStream.writable`

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

Тип: <WritableStream>

Класс: `TextDecoderStream`

[История]

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

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

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

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

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

`textDecoderStream.encoding`

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

Тип: <string>

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

`textDecoderStream.fatal`

Добавлено в: 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}`)
// Выводит: 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}`)
  // Выводит: 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}`)
// Выводит: 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}`)
  // Выводит: 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}`)
// Выводит: 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}`)
  // Выводит: 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}`)
// Выводит: 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}`)
  // Выводит: 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 ​

Класс: 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) ​

Класс: 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]) ​

Class: 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]) ​

Класс: 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 ​

Class: CountQueuingStrategy ​

new CountQueuingStrategy(init) ​

countQueuingStrategy.highWaterMark ​

countQueuingStrategy.size ​

Web Streams API

Обзор

Пример `ReadableStream`

API

Класс: `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)`

Класс: `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])`

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

Класс: `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`

Class: `CountQueuingStrategy`

`new CountQueuingStrategy(init)`

`countQueuingStrategy.highWaterMark`

`countQueuingStrategy.size`