ストリーム - 決定版ガイド

Streams API でストリームの読み取り、書き込み、変換を行う方法について説明します。

Thomas Steiner

Streams API を使用すると、ネットワーク経由で受信したデータのストリームにプログラムからアクセスできます。 なんらかの方法でローカルに作成されたり JavaScript で処理します。ストリーミングでは、受信、送信、または変換するリソースの分類が必要です。 分割し、少しずつ処理するということです。ストリーミングは ウェブページに表示する HTML や動画などのアセットを受け取ると、ブラウザは 2015 年にストリームを使用した fetch が導入されるまでは、JavaScript で機能を使用できませんでした。

注: XMLHttpRequest ではストリーミングは技術的には可能ですが、実際には不十分ではありませんでした。

以前は、なんらかのリソース（動画やテキスト ファイルなど）を処理する場合、 ファイル全体をダウンロードして、適切な形式にシリアル化解除されるのを待つ必要があります。 処理しますストリーミングは これがすべて変更されますJavaScript を使用して元データを段階的に処理し、 クライアントで使用できるようになるとすぐに発行されます。バッファ、文字列、blob を生成する必要はありません。 これによりさまざまなユースケースが可能になります。その一部を以下に示します。

• ビデオ エフェクト: 読み取り可能な動画ストリームを、エフェクトを適用する変換ストリームにパイプします。 リアルタイムで分析できます
• データ（解凍）: 選択的に変換ストリームにファイル ストリームをパイプで渡す 解凍します。
• 画像のデコード: バイトをデコードする変換ストリームに HTTP レスポンス ストリームをパイプ処理します。 ビットマップ データに変換してから、ビットマップを PNG に変換する別の変換ストリームで変換します。条件 Service Worker の fetch ハンドラ内にインストールされるため、 新しい画像形式に対応できます

ブラウザ サポート

ReadableStream と WritableStream

対応ブラウザ

• <ph type="x-smartling-placeholder"></ph> 43
• <ph type="x-smartling-placeholder"></ph> 14 歳
• <ph type="x-smartling-placeholder"></ph> 65
• <ph type="x-smartling-placeholder"></ph> 10.1

ソース

TransformStream

対応ブラウザ

• <ph type="x-smartling-placeholder"></ph> 67
• <ph type="x-smartling-placeholder"></ph> 79
• <ph type="x-smartling-placeholder"></ph> 102 回
• <ph type="x-smartling-placeholder"></ph> 14.1

ソース

注: 追加のサポートとして、ポリフィルを利用できます。可能であれば、組み込み機能が利用できない場合にのみ、条件付きでポリフィルを読み込みます。

基本コンセプト

さまざまなタイプのストリームについて詳しく説明する前に、基本コンセプトをいくつか紹介します。

チャンク

チャンクとは、ストリームに対して読み書きされる単一のデータです。任意の値を指定できます。 type;ストリームにはさまざまなタイプのチャンクを含めることができますほとんどの場合、チャンクはアトミックな 単位を指定します。たとえば、バイト ストリームは 16 個のチャンクで構成される 1 バイトではなく KiB Uint8Array 単位。

読み取り可能なストリーム

読み取り可能ストリームは、読み取り可能なデータのソースを表します。言い換えれば、データは ストリームから取り出します具体的には、読み取り可能なストリームは ReadableStream のインスタンスです。 クラスです。

書き込み可能なストリーム

書き込み可能なストリームは、書き込み可能なデータの宛先を表します。言い換えれば、データは 入るということです。具体的には、書き込み可能なストリームは、 WritableStream クラス。

変換ストリーム

変換ストリームは、ストリームのペアで構成されます。ストリームのペアは、書き込み可能側と呼ばれる書き込み可能なストリームです。 読み取り可能ストリーム（読み取り側）があります これを実際にたとえれば、 同時通訳 ある言語を別の言語にリアルタイムに翻訳できます 変換ストリームに固有の方法で、 新しいデータが BigQuery から読み取ることが あります。具体的には、writable プロパティと readable プロパティを持つオブジェクトはすべて配信可能です。 変換ストリームとして受け取りますただし、標準の TransformStream クラスを使用すると、 適切に絡み合ったペアになります

パイプチェーン

ストリームは主に、互いに「パイプ」で処理する場合に使用します。読み取り可能なストリームは、 読み取り可能なストリームの pipeTo() メソッドを使用して書き込み可能なストリームにエクスポートするか、 読み取り可能なストリームの pipeThrough() メソッドを使用します。 パイプでつながれた複数のストリームをパイプ チェーンと呼びます。

バックプレッシャー

パイプ チェーンが構築されると、チャンクの流速に関するシグナルが伝播される できます。チェーン内のステップでまだチャンクを受け入れることができない場合、シグナルを逆方向に伝播する 最終的に元のソースに対してチャンクの生成を停止するよう指示されるまで、 迅速に提供できます。フローを正規化するこのプロセスは、バックプレッシャーと呼ばれます。

ティーイング

読み取り可能なストリームは、tee() メソッドを使用して ted（大文字の「T」の形にちなんで名付けられます）できます。 この操作を行うと、ストリームがロックされ、直接使用できなくなります。ただし、新しい ストリーム（ブランチと呼ばれます）で、個別に使用できます。 ライブ配信の巻き戻しや再開ができないため、ティーイングは重要です。これについては後述します。

<ph type="x-smartling-placeholder">

</ph>

パイプチェーン。

読み取り可能なストリームの仕組み

読み取り可能ストリームとは、JavaScript で ReadableStream オブジェクトで ソースからデータを取り込みます。「 ReadableStream() コンストラクタは、指定されたハンドラから読み取り可能なストリーム オブジェクトを作成して返します。2 つのモデル 基になるソースのタイプ:

• push ソースは、アクセスすると常にデータを push します。ソースのソースは ストリームへのアクセスを開始、一時停止、キャンセルできます。たとえば、ライブ動画ストリーム、サーバー送信イベント、 WebSocket です
• pull ソースは、接続後に明示的にデータをリクエストする必要があります。例 fetch() または XMLHttpRequest 呼び出しによる HTTP オペレーションが含まれます。

ストリーム データは、チャンクと呼ばれる小さな断片で順次読み取られます。 ストリームに配置されたチャンクは「エンキューされる」と呼ばれます。これはキューで待機していることを意味します 読み取る準備が整います内部キューは、まだ読み取られていないチャンクを追跡します。

キューイング戦略は、 その内部キューの状態を確認できます。キューイング戦略では、各チャンクにサイズを割り当て、 キュー内のすべてのチャンクの合計サイズを指定された数まで増やすことをハイ ウォーターマークと呼びます。

ストリーム内のチャンクはリーダーによって読み取られます。このリーダーは、データをチャンクごとに 1 つずつ取得し、 それに対してどんな操作を実行できるでしょうか。リーダーと 一緒に作成される処理コードをコンシューマと呼びます。

次の構文はコントローラと呼ばれます。読み取り可能な各ストリームには その名のとおり、ストリームをコントロールできます。

ストリームを読み取ることができるリーダーは一度に 1 つのみです。リーダーが作成され、ストリームの読み取りを開始したとき （つまり、「アクティブなリーダー」になると「ロック」されます。別の読者に引き継ぐ場合 ストリームを読み取る場合、通常は他の処理を行う前に、最初のリーダーを解放する必要があります。 （ただし、ライブ配信は可能です）。

読み取り可能なストリームの作成

コンストラクタを呼び出して、読み取り可能なストリームを作成します。 ReadableStream()。 このコンストラクタには、オブジェクトを表すオプションの引数 underlyingSource があります。 作成されたストリーム インスタンスの動作を定義するメソッドとプロパティが含まれています。

underlyingSource

これには、次のデベロッパー定義のメソッド（省略可）を使用できます。

• start(controller): オブジェクトが作成されるとすぐに呼び出されます。「 メソッドはストリーム ソースにアクセスし、 ストリーミング機能の設定に必要です。このプロセスを非同期で行う場合、メソッドは 成功または失敗を通知する Promise を返します。このメソッドに渡される controller パラメータは次のとおりです。 ReadableStreamDefaultController。
• pull(controller): より多くのチャンクがフェッチされるたびに、ストリームを制御するために使用できます。これは、 ストリームの内部のチャンクのキューがいっぱいでなければ、キュー ハイウォーターマークに達しますpull() の呼び出し結果が Promise の場合、 その Promise が履行されるまで、pull() は再度呼び出されることはありません。 Promise が拒否された場合、ストリームはエラーになります。
• cancel(reason): ストリーム コンシューマがストリームをキャンセルしたときに呼び出されます。

const readableStream = new ReadableStream({
  start(controller) {
    /* … */
  },

  pull(controller) {
    /* … */
  },

  cancel(reason) {
    /* … */
  },
});

ReadableStreamDefaultController は、次のメソッドをサポートしています。

• ReadableStreamDefaultController.close() 関連するストリームを閉じます
• ReadableStreamDefaultController.enqueue() 特定のチャンクを関連するストリームのキューに追加します。
• ReadableStreamDefaultController.error() 関連するストリームでの以降の操作でエラーが発生します。

/* … */
start(controller) {
  controller.enqueue('The first chunk!');
},
/* … */

queuingStrategy

同様にオプションの ReadableStream() コンストラクタの 2 番目の引数は queuingStrategy です。 これは、オプションでストリームのキューイング戦略を定義するオブジェクトで、 parameters:

• highWaterMark: このキュー戦略を使用したストリームのハイ ウォーターマークを示す負でない数値。
• size(chunk): 指定されたチャンク値の負でない有限サイズを計算して返す関数。 結果はバックプレッシャーを判断するために使用され、適切な ReadableStreamDefaultController.desiredSize プロパティによって示されます。 また、基になるソースの pull() メソッドが呼び出されるタイミングも管理されます。

で確認できます。

const readableStream = new ReadableStream({
    /* … */
  },
  {
    highWaterMark: 10,
    size(chunk) {
      return chunk.length;
    },
  },
);

注: 独自のカスタム queuingStrategy を定義することも、このオブジェクトの値に ByteLengthQueuingStrategy または CountQueuingStrategy のインスタンスを使用することもできます。

queuingStrategy が指定されていない場合、デフォルトで 1 の highWaterMark を持つ CountQueuingStrategy が使用されます。

getReader() メソッドと read() メソッド

読み取り可能なストリームから読み取るには、リーダーが必要です。リーダーは ReadableStreamDefaultReader。 ReadableStream インターフェースの getReader() メソッドはリーダーを作成し、ストリームをロックします。 できます。ストリームがロックされている間は、このリーダーがリリースされるまで他のリーダーを取得できません。

read() ReadableStreamDefaultReader インターフェースの メソッドは、次のリソースへのアクセスを提供する Promise を返します。 分割されます。実行状態に応じて、結果を返すか拒否するかを クリックします。次のようなさまざまな可能性があります。

• チャンクが利用可能な場合、Promise は
  形式のオブジェクトで履行されます。 { value: chunk, done: false }。
• ストリームがクローズされると、Promise は
  形式のオブジェクトで履行されます。 { value: undefined, done: true }。
• ストリームがエラーになると、Promise は関連するエラーで拒否されます。

const reader = readableStream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) {
    console.log('The stream is done.');
    break;
  }
  console.log('Just read a chunk:', value);
}

locked プロパティ

読み取り可能なストリームがロックされているかどうかは、 ReadableStream.locked プロパティです。

const locked = readableStream.locked;
console.log(`The stream is ${locked ? 'indeed' : 'not'} locked.`);

読み取り可能なストリームのコードサンプル

以下のコードサンプルは、動作中のすべてのステップを示しています。まず、ReadableStream を作成します。 underlyingSource 引数（つまり TimestampSource クラス）は start() メソッドを定義します。 このメソッドは、ストリームの controller に以下を指示します。 10 秒間、1 秒ごとにタイムスタンプを enqueue() する。 最後に、コントローラにストリームを close() するように指示します。消費額: getReader() メソッドを使用してリーダーを作成し、ストリームが完了するまで read() を呼び出すことで、ストリームをストリーミングできます。 done。

class TimestampSource {
  #interval

  start(controller) {
    this.#interval = setInterval(() => {
      const string = new Date().toLocaleTimeString();
      // Add the string to the stream.
      controller.enqueue(string);
      console.log(`Enqueued ${string}`);
    }, 1_000);

    setTimeout(() => {
      clearInterval(this.#interval);
      // Close the stream after 10s.
      controller.close();
    }, 10_000);
  }

  cancel() {
    // This is called if the reader cancels.
    clearInterval(this.#interval);
  }
}

const stream = new ReadableStream(new TimestampSource());

async function concatStringStream(stream) {
  let result = '';
  const reader = stream.getReader();
  while (true) {
    // The `read()` method returns a promise that
    // resolves when a value has been received.
    const { done, value } = await reader.read();
    // Result objects contain two properties:
    // `done`  - `true` if the stream has already given you all its data.
    // `value` - Some data. Always `undefined` when `done` is `true`.
    if (done) return result;
    result += value;
    console.log(`Read ${result.length} characters so far`);
    console.log(`Most recently read chunk: ${value}`);
  }
}
concatStringStream(stream).then((result) => console.log('Stream complete', result));

非同期反復処理

read() ループの反復処理ごとにストリームが done かどうかを確認することは、最も便利な API ではない場合があります。 幸い、非同期の反復処理というより優れた方法がまもなく提供されます。

for await (const chunk of stream) {
  console.log(chunk);
}

注意: 非同期の反復処理は、一部のブラウザではまだ実装されていません。

現在非同期イテレーションを使用する際の回避策は、ポリフィルでの動作を実装することです。

if (!ReadableStream.prototype[Symbol.asyncIterator]) {
  ReadableStream.prototype[Symbol.asyncIterator] = async function* () {
    const reader = this.getReader();
    try {
      while (true) {
        const {done, value} = await reader.read();
        if (done) {
          return;
          }
        yield value;
      }
    }
    finally {
      reader.releaseLock();
    }
  }
}

読み取り可能なストリームを作成する

tee() メソッド（ ReadableStream インターフェースは、現在読み取り可能なストリームをティースし、2 要素の配列を返します。 結果として得られる 2 つのブランチを、新しい ReadableStream インスタンスとして格納しています。これにより、 ストリームを同時に読み取ることができますたとえば Service Worker で サーバーからレスポンスを取得してブラウザにストリーミングするだけでなく、 Service Worker のキャッシュです。レスポンスの本文は複数回使用できないため、2 つのコピーが必要 できます。ストリームをキャンセルするには、生成された両方のブランチをキャンセルする必要があります。小川のティーイング 通常、その間はロックされ、他の読者によってロックされません。

const readableStream = new ReadableStream({
  start(controller) {
    // Called by constructor.
    console.log('[start]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // Called `read()` when the controller's queue is empty.
    console.log('[pull]');
    controller.enqueue('d');
    controller.close();
  },
  cancel(reason) {
    // Called when the stream is canceled.
    console.log('[cancel]', reason);
  },
});

// Create two `ReadableStream`s.
const [streamA, streamB] = readableStream.tee();

// Read streamA iteratively one by one. Typically, you
// would not do it this way, but you certainly can.
const readerA = streamA.getReader();
console.log('[A]', await readerA.read()); //=> {value: "a", done: false}
console.log('[A]', await readerA.read()); //=> {value: "b", done: false}
console.log('[A]', await readerA.read()); //=> {value: "c", done: false}
console.log('[A]', await readerA.read()); //=> {value: "d", done: false}
console.log('[A]', await readerA.read()); //=> {value: undefined, done: true}

// Read streamB in a loop. This is the more common way
// to read data from the stream.
const readerB = streamB.getReader();
while (true) {
  const result = await readerB.read();
  if (result.done) break;
  console.log('[B]', result);
}

読み取り可能なバイト ストリーム

バイトを表すストリームの場合は、読み取り可能なストリームの拡張バージョンが 特にコピーを最小化することで、バイト効率が向上します。バイト ストリームでは、お客様所有のバッファを使用できます。 （BYOB）読者の獲得を目指します。デフォルトの実装では、次のようなさまざまな出力を提供できます。 WebSocket の場合、文字列や配列バッファとして使用できますが、バイト ストリームはバイト出力が保証されます。 さらに、BYOB リーダーには安定性に関する利点もあります。これは、 バッファがデタッチしても、同じバッファに 2 回書き込まないことが保証されるため、 それによって競合状態を回避できますBYOB リーダーにより、ブラウザの実行回数を減らせる バッファの再利用が可能なため、ガベージ コレクションに対して行いません。

読み取り可能なバイト ストリームの作成

読み取り可能なバイト ストリームを作成するには、追加の type パラメータを ReadableStream() コンストラクタ。

new ReadableStream({ type: 'bytes' });

underlyingSource

読み取り可能なバイト ストリームの基になるソースには、ReadableByteStreamController が渡されます。 あります。ReadableByteStreamController.enqueue() メソッドは chunk 引数を受け取り、その値は ArrayBufferView である。プロパティ ReadableByteStreamController.byobRequest は、 BYOB pull リクエスト。存在しない場合は null。最後に、ReadableByteStreamController.desiredSize プロパティは、制御対象のストリームの内部キューに必要なサイズを返します。

queuingStrategy

同様にオプションの ReadableStream() コンストラクタの 2 番目の引数は queuingStrategy です。 これは、オプションでストリームのキューイング戦略を定義するオブジェクトで、 parameter:

• highWaterMark: このキューイング戦略を使用するストリームのハイ ウォーターマークを示す、負でないバイト数。 これはバックプレッシャーを判断するために使用され、適切な ReadableByteStreamController.desiredSize プロパティによって表されます。 また、基になるソースの pull() メソッドが呼び出されるタイミングも管理されます。

で確認できます。

注: 他のストリーム タイプのキューイング戦略とは異なり、読み取り可能なバイト ストリームのキューイング戦略には size(chunk) 関数がありません。各チャンクのサイズは、常にその byteLength プロパティで決定されます。

で確認できます。

注: queuingStrategy が指定されていない場合、デフォルトで 0 の highWaterMark が使用されます。

getReader() メソッドと read() メソッド

次に、mode パラメータを適切に設定して、ReadableStreamBYOBReader にアクセスできるようにします。 ReadableStream.getReader({ mode: "byob" })。これにより、バッファリングをきめ細かく コピーを回避します。バイト ストリームから読み取るには、 ReadableStreamBYOBReader.read(view)（view は ArrayBufferView。

読み取り可能なバイト ストリームのコードサンプル

const reader = readableStream.getReader({ mode: "byob" });

let startingAB = new ArrayBuffer(1_024);
const buffer = await readInto(startingAB);
console.log("The first 1024 bytes, or less:", buffer);

async function readInto(buffer) {
  let offset = 0;

  while (offset < buffer.byteLength) {
    const { value: view, done } =
        await reader.read(new Uint8Array(buffer, offset, buffer.byteLength - offset));
    buffer = view.buffer;
    if (done) {
      break;
    }
    offset += view.byteLength;
  }

  return buffer;
}

次の関数は読み取り可能なバイト ストリームを返します。これにより、 ランダムに生成された配列です。事前に決められたチャンクサイズである 1,024 を使用する代わりに、 開発者が提供するバッファに格納されるため、完全に制御できます。

const DEFAULT_CHUNK_SIZE = 1_024;

function makeReadableByteStream() {
  return new ReadableStream({
    type: 'bytes',

    pull(controller) {
      // Even when the consumer is using the default reader,
      // the auto-allocation feature allocates a buffer and
      // passes it to us via `byobRequest`.
      const view = controller.byobRequest.view;
      view = crypto.getRandomValues(view);
      controller.byobRequest.respond(view.byteLength);
    },

    autoAllocateChunkSize: DEFAULT_CHUNK_SIZE,
  });
}

書き込み可能なストリームの仕組み

書き込み可能なストリームはデータの書き込み先であり、JavaScript で WritableStream オブジェクト。この 「基盤となるシンク」（基盤となるシンク）を抽象化 元データが書き込まれます。

データは、ライターによって一度に 1 つのチャンクでストリームに書き込まれます。チャンクは さまざまな形式で生成できます。作成した任意のコードを使用して チャンクに分割されます。ライターとそれに関連するコードはプロデューサーと呼ばれます。

作成されたライターがストリーム（アクティブなライター）への書き込みを開始すると、 ロックされます。書き込み可能なストリームに一度に書き込めるのは 1 つのライターのみです。別のゾーンに 書き込みを開始する場合、通常はそれを解放してからアタッチの前に 書き込みます。

内部キューは、ストリームに書き込まれても、まだ書き込まれていないチャンクを追跡します。 基になるシンクによって処理されます

キューイング戦略は、 その内部キューの状態を確認できます。キューイング戦略では、各チャンクにサイズを割り当て、 キュー内のすべてのチャンクの合計サイズを指定された数まで増やすことをハイ ウォーターマークと呼びます。

最後の構造はコントローラと呼ばれます。各書き込み可能なストリームには、 ストリームを制御（中止するなど）できます。

書き込み可能なストリームの作成

次の WritableStream インターフェース: Streams API は、ストリーミング データを宛先に書き込むための標準的な抽象化を提供します。 使用できますこのオブジェクトには、バックプレッシャーとキューイングが組み込まれています。書き込み可能なストリームを作成するには、 そのコンストラクタを呼び出す WritableStream()。 省略可能な underlyingSink パラメータがあり、オブジェクトを表す 作成されたストリーム インスタンスの動作を定義するメソッドとプロパティが含まれています。

underlyingSink

underlyingSink には、デベロッパー定義の以下のオプションのメソッドを含めることができます。controller いくつかのメソッドに渡されるパラメータは、 WritableStreamDefaultController。

• start(controller): このメソッドは、オブジェクトが作成されるとすぐに呼び出されます。「 このメソッドの内容は、基盤となるシンクへのアクセスを目的としている必要があります。このプロセスを 成功または失敗を通知する Promise を返すことができます。
• write(chunk, controller): このメソッドは、新しいデータチャンク（ chunk パラメータなど）を基盤となるシンクに書き込む準備が整っています。次の Promise を返すことができます。 書き込みオペレーションが成功したか失敗したかを示します。このメソッドは、前のステップの後でのみ 書き込みは成功し、ストリームがクローズまたは中止された後は決して成功しません。
• close(controller): このメソッドは、書き込みが完了したことをアプリが通知した場合に呼び出されます。 分割されます。コンテンツでは、アプリケーションへの書き込みのファイナライズに そのシンクへのアクセス権を 解放する必要がありますこのプロセスが非同期の場合は、 成功または失敗を知らせることを約束します。このメソッドは、キューに格納されたすべての書き込みの後にのみ呼び出されます 成功したことになります。
• abort(reason): アプリが突然終了することを通知した場合に、このメソッドが呼び出されます。 ストリームをエラー状態にします次のように、保持されているリソースをクリーンアップできます。 close()。ただし、書き込みがキューに入っている場合でも abort() が呼び出されます。これらのチャンクは 遠ざかってしまいます。このプロセスが非同期の場合は、Promise を返して成功または失敗を知らせることができます。「 reason パラメータには、ストリームが中止された理由を示す DOMString が含まれます。

const writableStream = new WritableStream({
  start(controller) {
    /* … */
  },

  write(chunk, controller) {
    /* … */
  },

  close(controller) {
    /* … */
  },

  abort(reason) {
    /* … */
  },
});

「 WritableStreamDefaultController Streams API のインターフェース: WritableStream の状態を制御できるコントローラ 書き込みのためにより多くのチャンクが送信されるため、または書き込みの最後で、セットアップ中や書き込み中に検出されます。作成時に WritableStream。基盤となるシンクに、対応する WritableStreamDefaultController が与えられます。 指定します。WritableStreamDefaultController のメソッドは 1 つだけです。 WritableStreamDefaultController.error(), これにより、関連するストリームでの以降の操作でエラーが発生します。 WritableStreamDefaultController は、signal プロパティもサポートしています。 AbortSignal、 必要に応じて WritableStream オペレーションを停止できます。

/* … */
write(chunk, controller) {
  try {
    // Try to do something dangerous with `chunk`.
  } catch (error) {
    controller.error(error.message);
  }
},
/* … */

queuingStrategy

同様にオプションの WritableStream() コンストラクタの 2 番目の引数は queuingStrategy です。 これは、オプションでストリームのキューイング戦略を定義するオブジェクトで、 parameters:

• highWaterMark: このキュー戦略を使用したストリームのハイ ウォーターマークを示す負でない数値。
• size(chunk): 指定されたチャンク値の負でない有限サイズを計算して返す関数。 結果はバックプレッシャーを判断するために使用され、適切な WritableStreamDefaultWriter.desiredSize プロパティによって示されます。

で確認できます。

注: 独自のカスタム queuingStrategy を定義することも、このオブジェクト値に ByteLengthQueuingStrategy または CountQueuingStrategy のインスタンスを使用することもできます。

queuingStrategy が指定されていない場合、デフォルトで 1 の highWaterMark を持つ CountQueuingStrategy が使用されます。

getWriter() メソッドと write() メソッド

書き込み可能なストリームに書き込むには、ライターが必要です。ライターは WritableStreamDefaultWriter。WritableStream インターフェースの getWriter() メソッドは、 WritableStreamDefaultWriter の新しいインスタンスが作成され、そのインスタンスへのストリームがロックされます。一方、 ストリームがロックされているため、現在の書き込みが解放されるまで他の書き込みを取得できません。

write() メソッドを WritableStreamDefaultWriter インターフェースは、渡されたデータのチャンクを WritableStream とその基盤となるシンクに書き込み、 書き込みオペレーションの成功または失敗を示すために解決される Promise。なお、 "success"基盤となるシンクに委ねられますチャンクが受け入れられたことを示す場合もありますが、 必ずしも最終的な宛先に安全に保存されるとは限りません

const writer = writableStream.getWriter();
const resultPromise = writer.write('The first chunk!');

locked プロパティ

書き込み可能なストリームがロックされているかどうかは、 WritableStream.locked プロパティです。

const locked = writableStream.locked;
console.log(`The stream is ${locked ? 'indeed' : 'not'} locked.`);

書き込み可能なストリームのコードサンプル

以下のコードサンプルは、動作中のすべてのステップを示しています。

const writableStream = new WritableStream({
  start(controller) {
    console.log('[start]');
  },
  async write(chunk, controller) {
    console.log('[write]', chunk);
    // Wait for next write.
    await new Promise((resolve) => setTimeout(() => {
      document.body.textContent += chunk;
      resolve();
    }, 1_000));
  },
  close(controller) {
    console.log('[close]');
  },
  abort(reason) {
    console.log('[abort]', reason);
  },
});

const writer = writableStream.getWriter();
const start = Date.now();
for (const char of 'abcdefghijklmnopqrstuvwxyz') {
  // Wait to add to the write queue.
  await writer.ready;
  console.log('[ready]', Date.now() - start, 'ms');
  // The Promise is resolved after the write finishes.
  writer.write(char);
}
await writer.close();

読み取り可能なストリームを書き込み可能なストリームにパイプする

読み取り可能なストリームは、読み取り可能なストリームの pipeTo() メソッドを使用します。 ReadableStream.pipeTo() は、現在の ReadableStream を指定された WritableStream にパイプし、 Promise は、パイピング プロセスが正常に完了した場合に履行され、エラーが発生した場合は拒否します。 発生します。

const readableStream = new ReadableStream({
  start(controller) {
    // Called by constructor.
    console.log('[start readable]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // Called when controller's queue is empty.
    console.log('[pull]');
    controller.enqueue('d');
    controller.close();
  },
  cancel(reason) {
    // Called when the stream is canceled.
    console.log('[cancel]', reason);
  },
});

const writableStream = new WritableStream({
  start(controller) {
    // Called by constructor
    console.log('[start writable]');
  },
  async write(chunk, controller) {
    // Called upon writer.write()
    console.log('[write]', chunk);
    // Wait for next write.
    await new Promise((resolve) => setTimeout(() => {
      document.body.textContent += chunk;
      resolve();
    }, 1_000));
  },
  close(controller) {
    console.log('[close]');
  },
  abort(reason) {
    console.log('[abort]', reason);
  },
});

await readableStream.pipeTo(writableStream);
console.log('[finished]');

変換ストリームの作成

Streams API の TransformStream インターフェースは、変換可能なデータのセットを表します。マイページ コンストラクタ TransformStream() を呼び出して変換ストリームを作成します。これにより、 指定されたハンドラの変換ストリーム オブジェクト。TransformStream() コンストラクタは、 最初の引数に、transformer を表すオプションの JavaScript オブジェクトを指定します。このようなオブジェクトは 次のメソッドのいずれかを含む:

transformer

• start(controller): このメソッドは、オブジェクトが作成されるとすぐに呼び出されます。通常 これは、controller.enqueue() を使用して接頭辞チャンクをキューに追加するために使用されます。これらのチャンクは 書き込み可能側からの書き込みには依存しません。このイニシャルが たとえば、プレフィックス チャンクの取得に多少の手間がかかるため、 関数は、成功または失敗を通知する Promise を返すことができます。拒否された Promise が 。例外がスローされると、TransformStream() コンストラクタによって再スローされます。
• transform(chunk, controller): このメソッドは、最初に新しいチャンクがテーブルに書き込まれたときに呼び出されます。 変換の準備ができました。ストリームの実装では、この関数が 前の変換が成功した場合にのみ呼び出され、start() が成功するまでは呼び出されません。 flush() の呼び出し後に行われます。この関数は実際の変換を実行し、 変換ストリームの処理が行われます。controller.enqueue() を使用して結果をキューに追加できます。この これにより、書き込み可能側に書き込まれる 1 つのチャンクが 0 または複数のチャンクに controller.enqueue() が何回呼び出されたかによって、可読性が高くなります。このプロセスが 変換は非同期であるため、この関数は Promise を返し、 必要があります。Promise が拒否された場合、読み取り可能な側と書き込み可能な側の両方でエラーが 使用します。transform() メソッドを指定しない場合は、恒等変換が使用されます。 は、書き込み可能側から読み取り可能側に変更されずにチャンクをキューに登録します。
• flush(controller): このメソッドは、書き込み可能側に書き込まれたすべてのチャンクが transform() を正常に渡すことで変換され、書き込み可能側が 閉じています。通常これは、接尾辞のチャンクを読み取り可能側にキューに追加するときに使用されます。 閉じられます。フラッシュ プロセスが非同期の場合、関数は Promise を返すことができます。 成功または失敗のシグナルを送ります。結果は呼び出し元に通知されます。 stream.writable.write()。さらに、Promise が拒否された場合、読み取り可能なエラーと ストリームの書き込み可能な側に 配置されます例外のスローは、拒否されたリクエストを返す場合と同様に扱われます。 あります。

const transformStream = new TransformStream({
  start(controller) {
    /* … */
  },

  transform(chunk, controller) {
    /* … */
  },

  flush(controller) {
    /* … */
  },
});

writableStrategy と readableStrategy のキューイング戦略

TransformStream() コンストラクタの 2 番目と 3 番目のオプション パラメータは省略可能です。 writableStrategy と readableStrategy のキューイング戦略。定義については、 読み取り可能ストリームと書き込み可能ストリーム セクションがあります。

変換ストリームのコードサンプル

次のコードサンプルは、単純な変換ストリームの動作を示しています。

// Note that `TextEncoderStream` and `TextDecoderStream` exist now.
// This example shows how you would have done it before.
const textEncoderStream = new TransformStream({
  transform(chunk, controller) {
    console.log('[transform]', chunk);
    controller.enqueue(new TextEncoder().encode(chunk));
  },
  flush(controller) {
    console.log('[flush]');
    controller.terminate();
  },
});

(async () => {
  const readStream = textEncoderStream.readable;
  const writeStream = textEncoderStream.writable;

  const writer = writeStream.getWriter();
  for (const char of 'abc') {
    writer.write(char);
  }
  writer.close();

  const reader = readStream.getReader();
  for (let result = await reader.read(); !result.done; result = await reader.read()) {
    console.log('[value]', result.value);
  }
})();

変換ストリームを介して読み取り可能ストリームをパイプする

pipeThrough() ReadableStream インターフェースのメソッドを使用すると、チェーン可能な方法で現在のストリームをパイプ処理できます。 変換ストリームまたはその他の書き込み可能/読み取り可能なペアを経由します。通常、ストリームをパイピングすると 他のリーダーがロックするのを防ぎます。

const transformStream = new TransformStream({
  transform(chunk, controller) {
    console.log('[transform]', chunk);
    controller.enqueue(new TextEncoder().encode(chunk));
  },
  flush(controller) {
    console.log('[flush]');
    controller.terminate();
  },
});

const readableStream = new ReadableStream({
  start(controller) {
    // called by constructor
    console.log('[start]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // called read when controller's queue is empty
    console.log('[pull]');
    controller.enqueue('d');
    controller.close(); // or controller.error();
  },
  cancel(reason) {
    // called when rs.cancel(reason)
    console.log('[cancel]', reason);
  },
});

(async () => {
  const reader = readableStream.pipeThrough(transformStream).getReader();
  for (let result = await reader.read(); !result.done; result = await reader.read()) {
    console.log('[value]', result.value);
  }
})();

次のコードサンプル（やや不自然）は、「シャウト」を実装する方法を示しています。fetch() のバージョン 返されたレスポンスの Promise を使用してすべてのテキストを大文字にします。 ストリームとして 分割して分割しますこの方法の利点は、インフラストラクチャの ドキュメント全体がダウンロードされるため、サイズの大きいファイルを扱う場合には大きな違いを生む可能性があります。

function upperCaseStream() {
  return new TransformStream({
    transform(chunk, controller) {
      controller.enqueue(chunk.toUpperCase());
    },
  });
}

function appendToDOMStream(el) {
  return new WritableStream({
    write(chunk) {
      el.append(chunk);
    }
  });
}

fetch('./lorem-ipsum.txt').then((response) =>
  response.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(upperCaseStream())
    .pipeTo(appendToDOMStream(document.body))
);

デモ

以下のデモは、読み取り、書き込み、変換が可能なストリームの動作を示しています。他にも例や pipeThrough() と pipeTo() のパイプ チェーンに対応し、tee() も示しています。必要に応じて 別のウィンドウでデモを見るか、 ソースコードをご覧ください。

ブラウザで便利なストリームを利用可能

ブラウザには、便利なストリームが多数組み込まれています。簡単に blob の ReadableStream。Blob インターフェースの stream() メソッドが返す 読み取り時に blob に含まれるデータを返す ReadableStream。また、 File オブジェクトは、 Blob であり、blob で可能なあらゆるコンテキストで使用できます。

const readableStream = new Blob(['hello world'], { type: 'text/plain' }).stream();

TextDecoder.decode() と TextEncoder.encode() のストリーミング バリアントは、 TextDecoderStream、 それぞれ TextEncoderStream。

const response = await fetch('https://streams.spec.whatwg.org/');
const decodedStream = response.body.pipeThrough(new TextDecoderStream());

ファイルの圧縮や解凍は、 CompressionStream、 DecompressionStream 変換ストリーム できます。以下のコードサンプルは、Streams の仕様をダウンロードして圧縮（gzip）する方法を示しています。 圧縮ファイルを直接ディスクに書き込むことができます。

const response = await fetch('https://streams.spec.whatwg.org/');
const readableStream = response.body;
const compressedStream = readableStream.pipeThrough(new CompressionStream('gzip'));

const fileHandle = await showSaveFilePicker();
const writableStream = await fileHandle.createWritable();
compressedStream.pipeTo(writableStream);

File System Access API の FileSystemWritableFileStream 試験運用版の fetch() リクエスト ストリームは次のとおりです。 実際の書き込み可能なストリームの例

Serial API は、読み取り可能なストリームと書き込み可能なストリームの両方を多用します。

// Prompt user to select any serial port.
const port = await navigator.serial.requestPort();
// Wait for the serial port to open.
await port.open({ baudRate: 9_600 });
const reader = port.readable.getReader();

// Listen to data coming from the serial device.
while (true) {
  const { value, done } = await reader.read();
  if (done) {
    // Allow the serial port to be closed later.
    reader.releaseLock();
    break;
  }
  // value is a Uint8Array.
  console.log(value);
}

// Write to the serial port.
const writer = port.writable.getWriter();
const data = new Uint8Array([104, 101, 108, 108, 111]); // hello
await writer.write(data);
// Allow the serial port to be closed later.
writer.releaseLock();

最後に、WebSocketStream API はストリームを WebSocket API と統合します。

const wss = new WebSocketStream(WSS_URL);
const { readable, writable } = await wss.connection;
const reader = readable.getReader();
const writer = writable.getWriter();

while (true) {
  const { value, done } = await reader.read();
  if (done) {
    break;
  }
  const result = await process(value);
  await writer.write(result);
}

関連リソース

• ストリームの仕様
• 付随するデモ
• ストリーム ポリフィル
• 2016 年 - ウェブ ストリームの年
• 非同期イテレータと非同期ジェネレータ
• ストリーム ビジュアライザー

謝辞

この記事をレビューしたユーザー: Jake Archibald 氏、 François Beaufort Sam Dutton Mattias Buelens 氏 Surma Joe Medley Adam Rice。 Jake Archibald のブログ投稿は、私自身が理解を深めるうえで大いに役立ちました 使用できます。一部のコードサンプルは GitHub ユーザーにヒントを得たものです @bellbind のデータ探索と ここは、文章の書き方を ストリームでの MDN ウェブ ドキュメント。「 Streams Standard の 著者は 説明します。ヒーロー画像（Ryan Lara） スプラッシュを解除します。