rxjs-like-a-pro

プロのための RxJS

このスキルでは、慣用的で合成しやすく、よくある落とし穴を避けた RxJS コードの書き方を学べます。基本方針は、ロジックを Observable チェーンの中に保つことです。.subscribe() に手を伸ばすたびに、その処理を .pipe() 内の変換として表現できないか考えてみてください。

参考ファイル

詳しい例やパターンについては、該当する参考ファイルをお読みください。

• references/loading-state-patterns.md — チェーン内でローディング／エラー状態を導出する方法、カスタムオペレーター withLoadingState、および scan を使ってローディング状態の間も前回の結果を保持する方法。ローディングインジケーターが必要な非同期データ取得を扱う場合にお読みください。
• references/massive-observable.md — 肥大化した new Observable() コンストラクターを、小さくて目的の明確な要素にリファクタリングする方法。10 行を大きく超える new Observable コールバックを見つけた場合にお読みください。
• references/inner-observable-chains.md — タイミング、遅延、アニメーションのフェーズを含む、豊かな内部 Observable シーケンスの構築方法。複数ステップの非同期シーケンスを合成する場合や、setTimeout のパターンを置き換える場合にお読みください。
• references/custom-operators.md — OperatorFunction を使ったインラインおよび抽出済みのカスタムオペレーターの書き方。再利用可能なストリームロジックを抽出する場合にお読みください。

最大のアンチパターン：早すぎる Subscribe

RxJS で最もよくある間違いは、早い段階で subscribe し、コールバックの中で命令型の処理を行うことです。たとえば、変数で状態を追跡したり、副作用のある関数を呼び出したり、さらに悪いことに、コールバック内で別の Observable に subscribe したりすることです（「subscribe の入れ子」パターン）。

これが問題になる理由：早い段階で subscribe すると、リアクティブチェーンの力を失います。その処理を合成、再試行、キャンセル、デバウンス、共有できなくなります。宣言型の世界から抜け出して命令型のスパゲッティコードに入り、新しい要件（再試行の追加、タイムアウトの追加、別のストリームとの結合）が加わるたびに、手動での状態管理が増えていきます。

// ❌ Bad: subscribe-in-subscribe with manual state tracking
let currentData: Data | null = null;
let loading = false;

input$.subscribe(value => {
  loading = true;
  fetchData(value).subscribe(data => {
    currentData = data;
    loading = false;
  });
});

// ✅ Good: everything is in the chain
const data$ = input$.pipe(
  switchMap(value => fetchData(value)),
);

ローディング状態は startWith を使ってチェーン内で導出してください。詳しくは references/loading-state-patterns.md をご覧ください。

肥大化した new Observable() のアンチパターン

もう一つよくあるアンチパターンは、単一の new Observable(subscriber => { ... }) コンストラクターにプログラム全体を詰め込むことです。リスナーの設定、Promise の解決、別の Observable への subscribe、再試行状態の管理などを、すべて巨大なコールバックの中で行います。これは Observable の衣装を着た命令型コードです。

new Observable() コンストラクターは小さく、目的を絞ったものにするべきです。リアクティブではない単一のソースをリアクティブな世界につなぐ薄い橋渡しとして使います。Promise ベースのソースには、代わりに defer(() => promise) を使ってください。再試行ロジック、エラー処理、ソースの結合は、すべてオペレーターチェーンに含めます。

変更前と変更後の完全な例については、references/massive-observable.md をご覧ください。

適切なフラット化オペレーターの選択

オペレーター	動作	使用する場面
switchMap	新しい値が届くと、前の内部処理をキャンセルします	ユーザー入力、入力中の検索、ルート変更など、最新の値だけが重要な場合
mergeMap	すべての内部 Observable を並行して実行します	すべての結果が必要な独立した処理（ログ記録、結果を待たない処理）
concatMap	内部 Observable をキューに入れ、順番に実行します	順序が重要で、何も破棄してはいけない場合（逐次書き込み、キュー）
exhaustMap	内部処理の実行中は、新しい値を無視します	重複送信を防ぐ場合（フォームの送信ボタンクリック）

ほとんどの UI／リクエストの場面では、まず switchMap を選んでください。

内部 Observable は単一のリクエストである必要はありません。concat、merge、timer、delay を使って、イベントの一連のタイムライン全体を表すこともできます。アニメーションやタイミングの例については、references/inner-observable-chains.md をご覧ください。

エラー処理

外側のストリームを動作し続けさせたい場合は、内部 Observable に catchError を置いてください。エラー時にストリーム全体を本当に置き換えたい場合に限り、外側のストリームに置きます。

// ❌ Bad: catchError on outer stream kills it for good
source$.pipe(
  switchMap(value => fetchData(value)),
  catchError(err => of(fallback)),
);

// ✅ Good: catchError inside switchMap — outer stream survives
source$.pipe(
  switchMap(value =>
    fetchData(value).pipe(
      catchError(err => of(fallback)),
    ),
  ),
);

同じ原則は retry にも当てはまります。外側のストリーム全体ではなく、内部の処理を再試行してください。

source$.pipe(
  switchMap(value =>
    fetchData(value).pipe(
      retry({ count: 3, delay: 1000 }),
      catchError(err => of(fallback)),
    ),
  ),
);

メモリーリークを防ぐ

手動で購読する回数が少ないほど、リークが発生する可能性も低くなります。優先順位は次のとおりです。

1. 購読自体をしない — 可能な場合は、購読のライフサイクルをフレームワークに任せます
2. 自然に完了するオペレーターを使う — first()、take(n)、takeUntil(destroy$)
3. 通知元とともに takeUntil を使う：

const destroy$ = new Subject<void>();
someObservable$.pipe(
  takeUntil(destroy$),
).subscribe(value => /* ... */);

// In teardown: destroy$.next(); destroy$.complete();

takeUntil はパイプ内の最後のオペレーターにする必要があります。 その後にオペレーター（特にフラット化オペレーター）を置くと、takeUntil が把握できない内部購読が作成され、リークの原因になることがあります。

4. 複数のストリームを単一の購読にまとめる — 副作用を伴う独立したストリームが複数ある場合は、merge でまとめて一度だけ購読します。

ホットとコールド

• コールドなオブザーバブル（new Observable(...)、of()、HTTP リクエスト）は、購読者ごとに新しい実行を作成します
• ホットなオブザーバブル（Subject、fromEvent）は、単一の実行を共有します

コールドなオブザーバブルは、shareReplay({ bufferSize: 1, refCount: true }) で共有します。必ず refCount: true を使ってください。これがないと、すべての購読者が購読を解除した後もソースの購読が維持されます（メモリーリーク）。

状態をリアクティブに導出する

複数の購読から更新される可変変数を使う代わりに、ストリームから状態を導出します。

// ❌ Bad: mutable state, inconsistent windows
let items: Item[] = [];
let filter = '';
items$.subscribe(i => { items = i; recompute(); });
filter$.subscribe(f => { filter = f; recompute(); });

// ✅ Good: always consistent
const filteredItems$ = combineLatest([items$, filter$]).pipe(
  map(([items, filter]) => items.filter(item => item.name.includes(filter))),
);

combineLatest と withLatestFrom の違い：combineLatest は、いずれかの入力が値を出力すると値を出力します（すべての入力が出力を駆動します）。withLatestFrom は、ソースが値を出力したときにのみ値を出力します（1 つが駆動し、他はコンテキストになります）。

startWith：combineLatest は、すべての入力が少なくとも一度値を出力するまで値を出力しません。startWith を使って初期値を設定し、ストリームが動作できるようにします。

Subject は必要な場合にのみ使う

Subject、BehaviorSubject、ReplaySubject は、命令型コードとリアクティブコードを橋渡しするためのエスケープハッチです。イベントバスやコールバックの橋渡しには適しています。汎用的な状態コンテナーとして使うのは適切ではありません。Subject の状態を同期するために複数の箇所で .next() を呼び出している場合は、代わりに導出ストリームを使ってください。

カスタムオペレーター

作成をためらう必要はありません。単にシグネチャが (source: Observable<A>) => Observable<B> の関数です。繰り返し使う .pipe() のチェーンは、OperatorFunction<In, Out> を使った名前付きオペレーターに切り出します。インラインと切り出した例については、references/custom-operators.md を参照してください。

副作用は subscribe ではなく tap に記述する

目安として、.subscribe() に引数は渡しません。ロギング、DOM の更新、データベースへの書き込み、分析データの送信など、すべての副作用はチェーン内の tap に記述します。最後の .subscribe() はストリームを有効化するだけです。

// ❌ Bad: side effects crammed into subscribe
source$.pipe(
  switchMap(value => fetchData(value)),
).subscribe(
  data => {
    updateUI(data);
    logAnalytics('data_loaded', data);
    cache.set(data);
  },
  err => showError(err),
);

// ✅ Good: side effects in tap, subscribe just activates
source$.pipe(
  switchMap(value => fetchData(value)),
  tap(data => updateUI(data)),
  tap(data => logAnalytics('data_loaded', data)),
  tap(data => cache.set(data)),
  tap({ error: err => showError(err) }),
).subscribe();

これが重要な理由は、副作用をチェーン内に記述すると合成可能になるためです。副作用を追加、削除、並べ替えできます。その間に filter を挿入することもできます。ストリームを共有し、副作用のロジックを重複させずに異なる購読者を使うこともできます。すべてを .subscribe() に詰め込むと、これらの利点が失われます。

tap では、ライフサイクルフックを持つオブザーバーオブジェクトも受け取れます。特にデバッグに便利です。

source$.pipe(
  tap({
    subscribe: () => console.log('subscribed!'),
    next: value => console.log('value:', value),
    error: err => console.log('error:', err),
    complete: () => console.log('complete'),
    unsubscribe: () => console.log('unsubscribed'),
    finalize: () => console.log('finalized (complete or unsubscribe)'),
  }),
);

subscribe フックは、「ストリームが値を出力しないのはなぜか」をデバッグする際に特に便利です。実際に購読されているかどうかを確認できます。

不要な Promise への変換を避ける

firstValueFrom／lastValueFrom は、Promise ベースの API と一度だけ連携する場合に適しています。リアクティブな方法を学ばずに済ませるため、購読コールバック内で使うのはコードのにおいです。その処理は switchMap を使ってチェーン内に記述します。

クイックリファレンス：よくあるリファクタリングパターン

アンチパターン	リファクタリング
a$.subscribe(x => b$.subscribe(y => ...))	a$.pipe(switchMap(x => b$))（または mergeMap／concatMap／exhaustMap）
subscribe 内で更新される可変変数	scan() または combineLatest で状態を導出する
subscribe 内の setTimeout	delay()、timer()、または debounceTime()
値をスキップするための subscribe 内の if ガード	subscribe の前に filter() を使う
subscribe 内の try/catch	パイプ内で catchError() を使う
手動のリクエストキャンセルフラグ	switchMap（前の処理を自動的にキャンセルします）
同じコールドオブザーバブルへの複数の subscribe	shareReplay({ bufferSize: 1, refCount: true })
副作用を発生させるだけの .subscribe()	副作用には tap() を使い、チェーンを継続する
巨大な new Observable() コンストラクター	小さく焦点を絞ったコンストラクター + defer() + オペレーターの合成
subscribe 内の await firstValueFrom()	switchMap — チェーン内にとどめる