web-utilities-rxjs

RxJS リアクティブプログラミングのパターン

クイックガイド： RxJS では、非同期処理をストリームとして組み立てます。Observable は遅延実行されるため、subscribe() するまでは何も実行されません。また、pipe() 内のオペレーターはパイプラインを記述するものであり、実行するものではありません。重要な判断は主に 3 つです。どのフラット化オペレーターを使うか（switchMap はキャンセル、mergeMap は並列化、concatMap はキューイング、exhaustMap は無視）、catchError をどこに置くか（内側のパイプ内に置くか、そうしないとエラーによって外側のストリームが恒久的に終了します）、そしてサブスクリプションをどのように終了させるかです。Observable の変数名には $ を付けます。

詳細なリソース：

• examples/core.md — 生成、defer による遅延実行、変換、エラー復旧、Promise との相互運用
• examples/higher-order.md — switchMap、mergeMap、concatMap、exhaustMap の具体的な使用例
• examples/subjects.md — 4 種類の Subject、マルチキャスト、shareReplay
• examples/combination.md — combineLatest、forkJoin、merge、concat、race
• examples/memory-leaks.md — takeUntil パターン、サブスクリプションのコレクション、自動的に完了するストリーム
• reference.md — オペレーターの表、シナリオ別オペレーター一覧、非推奨事項

---

該当するパス

クリーンアップの分岐は、ソースが自ら終了するかどうかで決まります。

• ソースが自ら完了する — of、from、ラップされた Promise、timer(n)、単一の HTTP レスポンス。購読解除は不要です。examples/core.md に進んでください。
• ソースが停止されるまで実行される — fromEvent、interval、任意の Subject、WebSocket。各サブスクリプションに終了処理が必要です。その方法は examples/memory-leaks.md に記載されています。
• ストリームから 1 つの値を取得する必要がある — firstValueFrom / lastValueFrom は await と連携し、ライフタイムを終了させます。examples/core.md を参照してください。

---

<critical_requirements>

RxJS のコードを書く前に

長期間存続するサブスクリプションには、必ず終了条件を設けてください — destroy Subject を使った takeUntil、Subscription のコレクション、または take や first などの自動的に完了するオペレーターを使用します。終了条件のない fromEvent や interval のサブスクリプションは、それを作成したものが終了しても存続し、クロージャーを保持し続けます。

takeUntil はパイプの最後に置いてください。 オペレーターが受け取るのはその前にあるものだけです。そのため、switchMap より前に takeUntil を置くと、外側のソースは終了しても内側の Observable は実行され続けます。正しくクリーンアップしているように見えるリークです。

catchError は外側ではなく、内側のパイプに置いてください。 エラーはサブスクライバーに伝播し、チェーンを完全に終了させます。内側の Observable でエラーから回復すれば、次の値に対して外側のストリームを継続できます。

必要な並行性に応じて、フラット化オペレーターを選んでください — 前の処理をキャンセルするなら switchMap、並列に実行するなら mergeMap、キューに入れるなら concatMap、処理中は無視するなら exhaustMap です。検索ボックスで安易に mergeMap を使うと、古いレスポンスが新しいレスポンスを上書きしてしまいます。

</critical_requirements>

---

自動検出： RxJS, rxjs, Observable, Subject, BehaviorSubject, ReplaySubject, AsyncSubject, subscribe, pipe, switchMap, mergeMap, concatMap, exhaustMap, combineLatest, forkJoin, withLatestFrom, fromEvent, defer, interval, timer, catchError, retry, debounceTime, throttleTime, distinctUntilChanged, takeUntil, shareReplay, firstValueFrom, lastValueFrom, Subscription

対象：

• デバウンス、スロットル、バッファー、重複排除など、合成が必要なイベントストリーム
• 複数の非同期ソースの調整 — 結合、フォークジョイン、競争、順次実行
• 新しい入力があると進行中のリクエストを中断する、キャンセル可能な処理
• ソケット、サーバー送信イベント、ポーリングなど、時間とともに届くリアルタイムデータ
• パイプラインの一部として表現されるリトライとバックオフのポリシー

別の場所で扱う内容：

• ストリームの性質を持たない単一のリクエスト — async／await なら一行で表現でき、Observable で包んでも何も得られません
• ローカルのコンポーネント状態 — 単純に読み書きする値に、ストリームは適切な形ではありません
• アプリケーションの状態管理 — RxJS は場所から場所へ値を運びますが、それらをどこに保持するかを決めるのは別の課題です
• フレームワークとのバインディング — ストリームをビュー層が描画する形式に変換するのは、その層独自の API の役割です

---

あらゆるものがストリームです。クリック、レスポンス、タイマー、ソケットのフレームなどです。Observable は処理そのものではなく、処理の記述です。そのため、pipe() チェーンでパイプラインを組み立て、subscribe() によって処理を開始します。これが、サブスクライバーのない pipe が何もしない理由であり、同じ Observable を二度サブスクライブすると通常は処理も二度実行される理由です。

この遅延性は、最もよくある意外な挙動の原因でもあります。チェーン内に置かれた即時実行される処理（defer の内部ではなく、構築時に評価される fetch 呼び出しなど）は、このモデルから外れて一度だけ実行され、誰もサブスクライブしていなくても共有されます。

---

基本パターン

パターン 1：生成と遅延性の維持

ソースに合った生成関数を選び、即時実行される処理は defer で包んで、サブスクライバーごとに独自の実行を行うようにします。

import {
  of,
  from,
  fromEvent,
  interval,
  timer,
  defer,
  EMPTY,
  throwError,
} from "rxjs";

of(1, 2, 3); // fixed values, then completes
from([1, 2, 3]); // any iterable, then completes
fromEvent(element, "click"); // never completes
interval(1000); // never completes
timer(2000); // one value, then completes

defer(() => from(fetch("/api/users"))); // fresh request per subscriber
from(fetch("/api/users")); // ❌ fires at construction, shared by all

コード全文：examples/core.md

---

パターン 2：パイプ内のオペレーターを使い、ネストした subscribe は使わない

const adults$ = users$.pipe(
  filter((user) => user.age >= 18),
  map((user) => user.name),
  distinctUntilChanged(),
  tap((name) => console.log(name)), // side effects live in tap
);

.subscribe() の中に .subscribe() を置くと、参照を保持するもののないサブスクリプションが作成されるため、終了させることもできません。フラット化オペレーターは、まさにこの形を置き換えるために存在します。

コード全文：examples/core.md

---

パターン 3：catchError によるエラーからの回復

const results$ = query$.pipe(
  switchMap((query) =>
    defer(() => from(search(query))).pipe(
      retry({ count: 3, delay: (_, n) => timer(1000 * 2 ** (n - 1)) }),
      catchError(() => of({ results: [], failed: true })), // inner — outer survives
    ),
  ),
);

catchError は Observable を返す必要があります。また、どこに置くかによって継続するものが決まります。内側のパイプに置けば外側のストリームが継続し、外側に置けば最初のエラーでチェーン全体が終了します。

コード全文：examples/core.md

---

パターン 4：レート制限

// Wait for a pause — user input
input$.pipe(debounceTime(300), distinctUntilChanged());

// Cap the rate — continuous events
fromEvent(window, "scroll").pipe(throttleTime(200));

debounceTime は値が途切れた後に最後の値を出力し、throttleTime は各時間枠の最初の値を出力して、残りを破棄します。デバウンスの後に distinctUntilChanged を置くことで、入力後に元に戻したとき、同じリクエストが再度発行されるのを防ぎます。

全コード：examples/higher-order.md の検索パイプライン内

オペレーターの比較：reference.md

---

パターン 5：フラット化オペレーター

内側の Observable がまだ実行中に新しい外側の値が届いたとき、どうするかが判断のポイントです。

Cancel the previous inner and switch    → switchMap    (search, navigation)
Run the new one alongside               → mergeMap     (uploads, fire-and-forget)
Queue it until the previous finishes    → concatMap    (ordered saves, message send)
Ignore it until the current one ends    → exhaustMap   (submit, login, refresh)

query$.pipe(switchMap((q) => search(q)));
files$.pipe(mergeMap((file) => upload(file), 3)); // 3 = concurrency limit
saves$.pipe(concatMap((data) => save(data)));
clicks$.pipe(exhaustMap(() => submit()));

全コード：examples/higher-order.md

---

パターン 6：サブジェクト

Subject は Observable と Observer の両方の性質を持つため、マルチキャストを行う方法であり、命令型コードから値をプッシュする方法でもあります。

new Subject<T>(); // no initial value, no replay
new BehaviorSubject<T>(initial); // always has a current value; .getValue()
new ReplaySubject<T>(n, windowMs); // replays the last n (optionally, recent ones)
new AsyncSubject<T>(); // the final value only, and only on complete()

常に値がある状態には BehaviorSubject、後から購読するユーザーに履歴が必要な場合は ReplaySubject、過去の値を持たないイベントには Subject、最後の結果を 1 つだけ返す場合は AsyncSubject を使います。

全コード：examples/subjects.md

---

パターン 7：takeUntil によるクリーンアップ

class Controller {
  private readonly destroy$ = new Subject<void>();

  start(): void {
    source$
      .pipe(
        switchMap((id) => this.load(id)),
        takeUntil(this.destroy$), // last — ends the inner subscription too
      )
      .subscribe((data) => this.render(data));
  }

  stop(): void {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

1 つの destroy$ で、それをパイプするすべてのサブスクリプションを終了できます。全コード： examples/memory-leaks.md

---

<red_flags>

注意すべき点

実行時に問題が発生するもの：

• エラーが処理されないと、チェーンは永久に終了します。それ以降の値はすべて黙って破棄され、再購読する以外にストリームを復帰させる方法はありません。
• catchError が Observable ではなく通常の値を返すと型エラーになります。of(fallback) を使えば解決できます。
• 高階オペレーターの前に takeUntil を置くと、外側のソースは終了しますが、内側のサブスクリプションは実行されたままになります。クリーンアップのコードは正しく見えるため、リークが気づかれないまま発生します。
• 入れ子になった .subscribe() 呼び出しでは、参照されないサブスクリプションが作られ、外側の値が届くたびに増えていきます。
• Observable が値を出力せずに完了すると、firstValueFrom は reject されます。非推奨の .toPromise() は代わりに undefined で resolve していたため、移行すると失敗時の動作が変わります。

予想外の動作：

• from(promise) は購読時ではなく、構築時に Promise を実行します。また、すべての購読者に同じ確定済みの結果が届きます。defer(() => from(promise)) を使うと、遅延実行に戻せます。
• shareReplay(1) では refCount の既定値が false のため、最後の購読者が離れた後もソースへの購読が続きます。意図しているのは、ほとんどの場合 shareReplay({ bufferSize: 1, refCount: true }) です。
• combineLatest は、すべてのソースが少なくとも 1 回値を出力するまで何も出力しません。値を出力しないソースが 1 つあると、組み合わせ全体が停止します。
• forkJoin は、すべてのソースが 完了 したときにのみ出力します。そのため、リストに Subject や interval が含まれていると、出力されません。
• retry はソースを先頭から再購読するため、冪等でないソースは副作用を再実行します。
• interval(0) も非同期です。スケジュールされるものであり、同期的に値を出力するものではありません。
• BehaviorSubject.getValue() は購読せずに現在の値を読み取るため、リアクティブなコードが気づかないうちにポーリングへ変わります。
• switchMap を使うべき箇所で mergeMap を使うと、競合が発生します。応答が順不同で届き、最も遅い応答が採用されます。

</red_flags>