WebSocketStream: ストリームと WebSocket API の統合

バックプレッシャーを適用して、アプリが WebSocket メッセージで埋もれたり、WebSocket サーバーがメッセージでフラッディングされたりしないようにします。

Thomas Steiner

背景

WebSocket API は、WebSocket プロトコルへの JavaScript インターフェースを提供します。これにより、ユーザーのブラウザとサーバー間で双方向のインタラクティブな通信セッションを開くことができます。この API を使用すると、サーバーに応答をポーリングすることなく、サーバーにメッセージを送信し、イベント ドリブンのレスポンスを受信できます。

Streams API

Streams API を使用すると、JavaScript はネットワーク経由で受信したデータ チャンクのストリームにプログラムからアクセスし、必要に応じて処理できます。ストリームのコンテキストで重要なコンセプトは、バックプレッシャーです。これは、単一のストリームまたはパイプチェーンが読み取りまたは書き込みの速度を制御するプロセスです。ストリーム自体またはパイプライン チェーン内の後続ストリームがまだビジー状態で、より多くのチャンクを受け入れる準備ができていない場合は、必要に応じてチェーンを介してシグナルを逆方向に送信し、配信を遅らせます。

現在の WebSocket API の問題

受信したメッセージにバックプレッシャーを適用することはできません

現在の WebSocket API では、メッセージへの応答は WebSocket.onmessage で行われます。これは、サーバーからメッセージを受信すると EventHandler が呼び出されます。

たとえば、新しいメッセージを受信するたびに負荷の高いデータ クランチ オペレーションを実行する必要があるアプリケーションがあるとします。次のコードのようなフローをセットアップします。process() 呼び出しの結果を await なので、問題ないはずです。

// A heavy data crunching operation.
const process = async (data) => {
  return new Promise((resolve) => {
    window.setTimeout(() => {
      console.log('WebSocket message processed:', data);
      return resolve('done');
    }, 1000);
  });
};

webSocket.onmessage = async (event) => {
  const data = event.data;
  // Await the result of the processing step in the message handler.
  await process(data);
};

間違いです！現在の WebSocket API の問題は、バックプレッシャーを適用する方法がないことです。process() メソッドで処理できない速さでメッセージが到着すると、レンダリング プロセスはそれらのメッセージをバッファリングしてメモリをいっぱいにするか、CPU 使用率が 100% になって応答しなくなるか、またはその両方が発生します。

送信されたメッセージへのバックプレッシャーの適用は非人間工学的

送信されたメッセージにバックプレッシャーを適用することは可能ですが、WebSocket.bufferedAmount プロパティをポーリングする必要があるため、非効率的で人間工学に適していません。この読み取り専用プロパティは、WebSocket.send() の呼び出しを使用してキューに追加されているものの、まだネットワークに送信されていないデータのバイト数を返します。この値は、キューに格納されたデータがすべて送信されると 0 にリセットされますが、WebSocket.send() を呼び出し続けると増加し続けます。

WebSocketStream API とは

WebSocketStream API は、ストリームを WebSocket API と統合することで、バックプレッシャーが存在しない、または人間工学に適していないという問題に対処します。つまり、バックプレッシャーは追加費用なしで「無料で」適用されます。

WebSocketStream API の推奨ユースケース

この API を使用できるサイトの例は次のとおりです。

  • インタラクティビティ（特に動画や画面共有）を維持する必要がある高帯域幅の WebSocket アプリケーション。
  • 同様に、動画キャプチャなどのアプリケーションは、ブラウザで大量のデータを生成します。このデータはサーバーにアップロードする必要があります。バックプレッシャーにより、クライアントはデータをメモリに蓄積する代わりに、データの生成を停止できます。

現在のステータス

ステップ ステータス
1. 説明を作成する 完了
2. 仕様の初期ドラフトを作成する 作成中
3. フィードバックを収集してデザインを反復する 作成中
4. オリジン トライアル 完了
5. リリース 開始していません

WebSocketStream API の使用方法

WebSocketStream API は Promise ベースであるため、最新の JavaScript の世界で自然に扱うことができます。まず、新しい WebSocketStream を作成して、WebSocket サーバーの URL を渡します。次に、接続が opened になるまで待機します。これにより、ReadableStream または WritableStream が生成されます。

ReadableStream.getReader() メソッドを呼び出すと、最終的に ReadableStreamDefaultReader が取得されます。このオブジェクトから、ストリームが完了するまで（{value: undefined, done: true} 形式のオブジェクトが返されるまで）read() データを取得できます。

したがって、WritableStream.getWriter() メソッドを呼び出すと、最終的に WritableStreamDefaultWriter が取得され、write() にデータを渡すことができます。

  const wss = new WebSocketStream(WSS_URL);
  const {readable, writable} = await wss.opened;
  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);
  }

バックプレッシャー

約束のバックプレッシャー機能はどうなりますか？追加の手順は必要ありません。process() に余分な時間がかかる場合、パイプラインの準備ができてから次のメッセージが使用されます。同様に、WritableStreamDefaultWriter.write() ステップも安全に続行できる場合にのみ続行されます。

高度な例

WebSocketStream の 2 番目の引数は、将来の拡張を可能にするオプション バッグです。唯一のオプションは protocols です。これは、WebSocket コンストラクタの 2 番目の引数と同じように動作します。

const chatWSS = new WebSocketStream(CHAT_URL, {protocols: ['chat', 'chatv2']});
const {protocol} = await chatWSS.opened;

選択した protocol と潜在的な extensions は、WebSocketStream.opened Promise を介して利用できる辞書の一部です。接続が失敗した場合は関連性がないため、ライブ接続に関するすべての情報はこの Promise によって提供されます。

const {readable, writable, protocol, extensions} = await chatWSS.opened;

閉じられた WebSocketStream 接続に関する情報

WebSocket API の WebSocket.onclose イベントと WebSocket.onerror イベントから取得できた情報を、WebSocketStream.closed プロミスから取得できるようになりました。正常に終了しなかった場合、Promise は拒否されます。正常に終了しない場合は、サーバーから送信されたコードと理由に解決されます。

考えられるすべてのステータス コードとその意味については、CloseEvent ステータス コードのリストをご覧ください。

const {code, reason} = await chatWSS.closed;

WebSocketStream 接続を閉じる

WebSocketStream は AbortController で閉じることができます。したがって、AbortSignal を WebSocketStream コンストラクタに渡します。

const controller = new AbortController();
const wss = new WebSocketStream(URL, {signal: controller.signal});
setTimeout(() => controller.abort(), 1000);

代わりに WebSocketStream.close() メソッドを使用することもできますが、その主な目的は、サーバーに送信されるコードと理由を指定できるようにすることです。

wss.close({code: 4000, reason: 'Game over'});

段階的な機能強化と相互運用性

現在、Chrome は WebSocketStream API を実装する唯一のブラウザです。従来の WebSocket API との相互運用性を確保するため、受信したメッセージにバックプレッシャーを適用することはできません。送信されたメッセージにバックプレッシャーを適用することは可能ですが、これには WebSocket.bufferedAmount プロパティのポーリングが含まれます。これは非効率的であり、人間工学的ではありません。

機能検出

WebSocketStream API がサポートされているかどうかを確認するには、次のコマンドを使用します。

if ('WebSocketStream' in window) {
  // `WebSocketStream` is supported!
}

デモ

対応しているブラウザでは、埋め込まれた iframe で WebSocketStream API の動作を確認できます。また、Glitch で直接確認することもできます。

フィードバック

Chrome チームでは、WebSocketStream API の使用経験についてお聞かせください。

API 設計について

API で想定どおりに機能していないものはありますか？ または、アイデアを実装するために必要なメソッドやプロパティが不足している場合はどうすればよいですか？セキュリティ モデルに関するご質問やご意見がございましたら、対応する GitHub リポジトリで仕様の問題を提出するか、既存の問題に考えを追加します。

実装に関する問題を報告する

Chrome の実装にバグは見つかりましたか？ それとも、実装が仕様と異なるのでしょうか？new.crbug.com でバグを報告します。できるだけ詳細な情報を含め、再現手順を簡単に説明してください。[コンポーネント] ボックスに Blink>Network>WebSockets を入力します。Glitch は、再現ケースをすばやく簡単に共有するのに適しています。

API のサポートを表示する

WebSocketStream API を使用する予定はありますか？公開サポートは、Chrome チームが機能の優先順位を決める際に役立ち、他のブラウザ ベンダーに、サポートがどれほど重要であるかを示します。

ハッシュタグ #WebSocketStream を使用して @ChromiumDev にツイートを送信し、どこでどのように使用しているかをお知らせください。

関連情報

謝辞

WebSocketStream API は、Adam Rice と Yutaka Hirano によって実装されました。