RxJSとWebSocket

2025-07-14

RxJSとは

RxJS（Reactive Extensions for JavaScript）はリアクティブプログラミングのためのライブラリ。非同期処理やイベントをストリームとして扱い、宣言的に記述できる。

中心となる概念は Observable・Observer・Subject の3つ。

ObservableとObserver

Observable はデータを流すストリーム。Observer はそのストリームを購読（subscribe）し、データが流れてくるたびに処理を実行する。

observable$.subscribe(value => {
  console.log(value); // データが来るたびに呼ばれる
});

WebSocketSubjectとは

RxJSの webSocket() が返すのは WebSocketSubject というオブジェクト。Subject は Observable を継承しており、受信（Observable）と送信（Observer）の両方の性質を持つ。

socket$.subscribe(...) → Observableとして、サーバーからのメッセージを受信する
socket$.next(...) → Observerとして、サーバーへメッセージを送信する

この1つのオブジェクトで双方向通信が完結するのが Subject の特徴。

実装コード

const { webSocket } = rxjs.webSocket;

// WebSocketSubjectを生成
const socket$ = webSocket("wss://ws.postman-echo.com/raw");

// subscribeしてサーバーからのメッセージを受信
socket$.subscribe(msg => {
  document.getElementById("log").innerHTML += `<p>Received: ${msg}</p>`;
});

// next()でサーバーへメッセージを送信
const txt = document.getElementById("textForWebsocket").value;
socket$.next(txt);

デモ

テキストを入力してSendを押すと、エコーサーバーに送信され、そのまま返ってきたメッセージが下に表示される。