Web Streams API

[履歴]

バージョン	変更点
v21.0.0	実験的な段階ではなくなりました。
v18.0.0	この API の使用によるランタイム警告は表示されなくなりました。
v16.5.0	追加：v16.5.0

[安定版: 2 - 安定版]

安定版: 2 安定性: 2 - 安定版

WHATWG Streams Standardの実装です。

概要

WHATWG Streams Standard（または「web streams」）は、ストリーミングデータの処理のための API を定義しています。Node.js のStreams API に似ていますが、後に登場し、多くの JavaScript 環境でストリーミングデータのための「標準」API となっています。

3 つの主要なオブジェクトの種類があります。

ReadableStream - ストリーミングデータのソースを表します。
WritableStream - ストリーミングデータの宛先を表します。
TransformStream - ストリーミングデータを変換するためのアルゴリズムを表します。

`ReadableStream`の例

この例では、現在のperformance.now()タイムスタンプを 1 秒ごとに永遠にプッシュするシンプルなReadableStreamを作成します。非同期イテラブルを使用して、ストリームからデータを読み取ります。

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 または undefined で解決される Promise。
- pull <Function> ReadableStreamの内部キューがいっぱいでないときに繰り返し呼び出されるユーザー定義関数です。同期または非同期操作が可能です。非同期の場合は、以前に返された Promise が解決されるまで、この関数は再び呼び出されません。
- controller <ReadableStreamDefaultController> | <ReadableByteStreamController>
- 戻り値: undefined で解決される Promise。
- cancel <Function> ReadableStreamがキャンセルされたときに呼び出されるユーザー定義関数です。
- reason <any>
- 戻り値: undefined で解決される Promise。
- type <string> 'bytes'またはundefinedでなければなりません。
- autoAllocateChunkSize <number> typeが'bytes'の場合にのみ使用されます。0 以外の値に設定すると、ビューバッファがReadableByteStreamController.byobRequestに自動的に割り当てられます。設定しない場合、デフォルトのリーダーReadableStreamDefaultReaderを使用してストリームの内部キューを介してデータを転送する必要があります。
strategy <Object>
- highWaterMark <number> バックプレッシャーが適用される前の内部キューの最大サイズ。
- size <Function> データの各チャンクのサイズを識別するために使用されるユーザー定義関数。
- chunk <any>
- 戻り値: <number>

`readableStream.locked`

追加: v16.5.0

型: <boolean> この<ReadableStream>にアクティブなリーダーが存在する場合、trueに設定されます。

readableStream.lockedプロパティはデフォルトでfalseであり、ストリームのデータを読み込んでいるアクティブなリーダーが存在する間はtrueに切り替わります。

`readableStream.cancel([reason])`

追加: v16.5.0

reason <any>
戻り値: キャンセルが完了するとundefinedで解決される Promise。

`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> transform.writableがこのReadableStreamから受信した可能性のある変更されたデータをプッシュするReadableStream。
- writable <WritableStream> このReadableStreamのデータが書き込まれるWritableStream。
options <Object>
- preventAbort <boolean> trueの場合、このReadableStreamのエラーによってtransform.writableが中断されることはありません。
- preventCancel <boolean> trueの場合、宛先のtransform.writableのエラーによってこのReadableStreamがキャンセルされることはありません。
- preventClose <boolean> trueの場合、このReadableStreamを閉じてもtransform.writableは閉じられません。
- signal <AbortSignal> <AbortController>を使用してデータ転送のキャンセルを許可します。
戻り値: transform.readableからの<ReadableStream>。

この<ReadableStream>をtransform引数で指定された<ReadableStream>と<WritableStream>のペアに接続し、この<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> このReadableStreamのデータが書き込まれる <WritableStream>。
options <Object>
- preventAbort <boolean> trueの場合、このReadableStreamのエラーによってdestinationが中断されることはありません。
- preventCancel <boolean> trueの場合、destinationのエラーによってこのReadableStreamがキャンセルされることはありません。
- preventClose <boolean> trueの場合、このReadableStreamを閉じてもdestinationは閉じられません。
- signal <AbortSignal> <AbortController>を使用してデータ転送の中断を許可します。
戻り値：undefinedを解決する Promise

パイプ操作がアクティブな間、readableStream.lockedをtrueにします。

`readableStream.tee()`

[履歴]

バージョン	変更
v18.10.0, v16.18.0	読み取り可能なバイトストリームのティーイングをサポート
v16.5.0	追加：v16.5.0

戻り値：<ReadableStream[]>

このReadableStreamのデータが転送される、2 つの新しい <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>

与えられた<ReadableStream>にロックされた新しい<ReadableStreamDefaultReader>を作成します。

`readableStreamDefaultReader.cancel([reason])`

追加: v16.5.0

reason <any>
戻り値: undefinedで解決される Promise。

<ReadableStream>をキャンセルし、基礎となるストリームがキャンセルされたときに解決される Promise を返します。

`readableStreamDefaultReader.closed`

追加: v16.5.0

型: 関連付けられた<ReadableStream>が閉じられたときにundefinedで解決される<Promise>、ストリームにエラーが発生した場合、またはストリームが閉じ終わる前にリーダーのロックが解放された場合は拒否されます。

`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>（ReadableStreamの作成時にunderlyingSource.typeが'bytes'に設定されているもの）のための代替コンシューマーです。

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>

与えられた<ReadableStream>にロックされた新しいReadableStreamBYOBReaderを作成します。

`readableStreamBYOBReader.cancel([reason])`

追加日時: v16.5.0

reason <any>
戻り値: undefined で解決される Promise。

基底となる <ReadableStream> をキャンセルし、基底となるストリームがキャンセルされた時点で解決される Promise を返します。

`readableStreamBYOBReader.closed`

追加日時: v16.5.0

型: 関連付けられた <ReadableStream> が閉じられた場合に undefined で解決される <Promise> 。ストリームでエラーが発生した場合、またはストリームが閉じ終わる前にリーダーのロックが解放された場合は、Promise は拒否されます。

`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個の要素が利用可能になった時点で解決されます。設定されていない場合、少なくとも 1 つの要素が利用可能になった時点で解決されます。
戻り値: オブジェクトで解決される Promise:
- value <TypedArray> | <DataView>
- done <boolean>

基底となる <ReadableStream> から次のデータチャンクを要求し、データが利用可能になった時点で解決される Promise を返します。

このメソッドには、プールされた <Buffer> オブジェクトインスタンスを渡さないでください。プールされたBufferオブジェクトはBuffer.allocUnsafe()またはBuffer.from()を使用して作成されるか、様々なnode:fsモジュールのコールバックによって返されることがよくあります。これらのタイプのBufferは、すべてのプールされたBufferインスタンスのすべてのデータを含む共有の基底となる <ArrayBuffer> オブジェクトを使用します。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>にエラーを発生させ、閉じます。

クラス: `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、またはundefinedで解決される Promise。
- write <Function> データチャンクがWritableStreamに書き込まれたときに呼び出されるユーザー定義関数。
- chunk <any>
- controller <WritableStreamDefaultController>
- 戻り値: undefinedで解決される Promise。
- close <Function> WritableStreamが閉じられたときに呼び出されるユーザー定義関数。
- 戻り値: undefinedで解決される Promise。
- abort <Function> WritableStreamを強制的に閉じるときに呼び出されるユーザー定義関数。
- reason <any>
- 戻り値: undefinedで解決される Promise。
- type <any> typeオプションは将来の使用のために予約されており、必ず undefinedでなければなりません。
strategy <Object>
- highWaterMark <number> バックプレッシャーが適用される前の最大内部キューサイズ。
- size <Function> 各データチャンクのサイズを識別するために使用されるユーザー定義関数。
- chunk <any>
- 戻り値: <number>

`writableStream.abort([reason])`

追加日: v16.5.0

reason <any>
戻り値: undefined を解決する Promise。

WritableStream を突然終了します。すべてのキューイングされた書き込みはキャンセルされ、関連付けられた Promise は拒否されます。

`writableStream.close()`

追加日: v16.5.0

戻り値: undefined を解決する Promise。

追加の書き込みが期待されない場合にWritableStreamを閉じます。

`writableStream.getWriter()`

追加日: v16.5.0

戻り値: <WritableStreamDefaultWriter>

WritableStream にデータ書き込みに使用できる新しいライターインスタンスを作成して返します。

`writableStream.locked`

追加日: v16.5.0

型: <boolean>

writableStream.locked プロパティは、デフォルトで false であり、この WritableStream にアクティブなライターが接続されている間は true に切り替わります。

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

指定された WritableStream にロックされた新しい WritableStreamDefaultWriter を作成します。

`writableStreamDefaultWriter.abort([reason])`

追加日: v16.5.0

reason <any>
戻り値: undefined を解決する Promise。

WritableStream を突然終了します。すべてのキューイングされた書き込みはキャンセルされ、関連付けられた Promise は拒否されます。

`writableStreamDefaultWriter.close()`

追加: v16.5.0

戻り値: undefinedを解決する Promise。

追加の書き込みが予想されない場合にWritableStreamを閉じます。

`writableStreamDefaultWriter.closed`

追加: v16.5.0

型: <Promise> 関連付けられた<WritableStream>が閉じられた場合にundefinedを解決し、ストリームにエラーが発生した場合、またはストリームの閉じ完了前にライターのロックが解放された場合は拒否されます。

`writableStreamDefaultWriter.desiredSize`

追加: v16.5.0

型: <number>

<WritableStream>のキューを満たすために必要なデータ量。

`writableStreamDefaultWriter.ready`

追加: v16.5.0

型: <Promise> ライターが使用可能になったときにundefinedを解決する Promise。

`writableStreamDefaultWriter.releaseLock()`

追加: v16.5.0

基礎となる<ReadableStream>に対するこのライターのロックを解放します。

`writableStreamDefaultWriter.write([chunk])`

追加: v16.5.0

chunk: <any>
戻り値: undefinedを解決する Promise。

<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> <WritableStream> が中断された場合、保留中の書き込みまたはクローズ操作をキャンセルするために使用できる AbortSignal。

クラス: `TransformStream`

[履歴]

バージョン	変更
v18.0.0	このクラスはグローバルオブジェクトで公開されるようになりました。
v16.5.0	追加: v16.5.0

TransformStream は、WritableStream に書き込まれたデータが受信され、変換される可能性があり、ReadableStream のキューにプッシュされるように接続された <ReadableStream> と <WritableStream> で構成されます。

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 で解決される Promise
- transform <Function> transformStream.writable に書き込まれたデータのチャンクを受信し、変更する可能性があり、transformStream.readable に転送する前に変更するユーザー定義関数。
- chunk <any>
- controller <TransformStreamDefaultController>
- 戻り値: undefined で解決される Promise。
- flush <Function> TransformStream の書き込み可能な側が閉じられ、変換プロセスの終了を示す直前に呼び出されるユーザー定義関数。
- controller <TransformStreamDefaultController>
- 戻り値: undefined で解決される Promise。
- 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インスタンスがサポートするencodingを指定します。デフォルト: '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>

デコードエラーがTypeErrorをスローした場合trueになります。

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

クラス: 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]) ​

クラス: 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])`

クラス: `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`