rxMethodの寿命とエラー処理をレビューで確認する

rxMethod は、メソッドのように呼べて中身は RxJS のパイプラインという形をしている。呼び出し側から見るとメソッドなので、メソッドと同じ感覚で扱われやすい。

実体は、渡された入力を処理し続ける1本のストリームである。そのため寿命とエラーの扱いが、通常のメソッドとは違う。

失敗した後の呼び出し

次の検索は、通信が1回失敗すると、以後は入力を変えても走らなくなる。

withMethods((store, api = inject(UserApi)) => ({
  search: rxMethod<string>(
    pipe(
      debounceTime(300),
      distinctUntilChanged(),
      switchMap(keyword => api.search(keyword)),
      tap(users => patchState(store, { users })),
@Reviewer
`api.search` が失敗すると未処理のエラーとしてパイプライン自体が終了します。以後この `search` を呼んでも反応しません。
    ),
  ),
})),

エラーが上流まで届くと、そのストリームは終了する。rxMethod が保持しているのはその1本なので、終了した後に呼び出しても処理される先が無い。

画面上は「検索しても何も起きない」状態になる。例外がコンソールに出るのは最初の1回だけで、2回目以降は無言になる。

捕捉は switchMap の内側へ置く。内側であれば、置き換えの対象は api.search(keyword) だけになり、外側のストリームは生き続ける。

withMethods((store, api = inject(UserApi)) => ({
  search: rxMethod<string>(
    pipe(
      debounceTime(300),
      distinctUntilChanged(),
      switchMap(keyword =>
        api.search(keyword).pipe(
          tap(users => patchState(store, { users })),
          catchError(() => {
            patchState(store, { error: '検索に失敗しました' });
            return EMPTY;
          }),
        ),
      ),
    ),
  ),
})),

@ngrx/operators の tapResponse を使うと、成功と失敗の扱いを1か所にまとめられる。内側に置いていることが形から読み取りやすくなる。

この構造は NgRx Effect と同じである。購読され続ける1本のストリームを、外側の置き換えや未処理のエラーで終わらせてしまう、という誤りが両方で起きる。

@Reviewer: この `rxMethod` は一度失敗すると以後反応しなくなります。捕捉を `switchMap` の内側へ移すか、`tapResponse` を使ってください。

購読が片付くタイミング

rxMethod に Signal や Observable を接続すると、その購読は SignalStore が作られたときの Injector の破棄に合わせて片付く。呼び出したコンポーネントの寿命ではない。

export class UserSearchComponent {
  readonly keyword = signal('');
  private readonly store = inject(UserSearchStore);

  constructor() {
    this.store.search(this.keyword);
  }
}

this.keyword を Signal のまま渡すと、以後の変更に追従する接続になる。この接続の寿命を決めるのは store の provider の位置である。

store をコンポーネントの providers に置いていれば、画面と一緒に破棄される。root に置いていれば、コンポーネントが破棄されても接続は残る。後者の場合、画面を離れた後も keyword の変更で検索が走り続ける。

@Reviewer: この store は root に提供されているため、コンポーネントが破棄されても `rxMethod` への接続が残ります。画面ローカルの検索であれば、store をコンポーネントの `providers` へ移してください。

括弧の有無で変わる意味

接続と1回渡しは、見た目がほとんど変わらない。

store.search(this.keyword);    // Signal を接続する。変更に追従する
store.search(this.keyword());  // その時点の値を1回渡すだけ

どちらも型が通り、どちらも動く。違いが出るのは2回目以降の変更があったときで、後者では検索が更新されない。

レビューでは、その rxMethod に何を期待しているかを確認する。初期化時に1回だけ実行したいのであれば値を渡す形が正しく、入力に追従させたいのであれば Signal をそのまま渡す。

Observable を渡した場合も接続になる。rxMethod は Signal、Observable、そして素の値のいずれも受け取れるため、渡しているものが何かを読み取る必要がある。

呼び出しが重なったときの扱い

rxMethod の中に高階Operatorが無い場合、複数回の呼び出しが重なったときの扱いが決まっていない。

save: rxMethod<Draft>(
  pipe(
    mergeMap(draft => api.save(draft)),
  ),
),

mergeMap は同時実行数に制限がなく、完了順も入力順とは限らない。保存であれば、後から始まった処理が先に終わる可能性を検討することになる。

検索のように最新の結果だけが要るなら switchMap、順に処理するなら concatMap、実行中の重複を捨てるなら exhaustMap を選ぶ。選択の基準は通常の RxJS と変わらない。

@Reviewer: 競合制御が指定されていません。保存処理なので、順に処理する `concatMap` か、実行中は捨てる `exhaustMap` のどちらが意図に合いますか。

レビュー観点チェックリスト

おわりに

rxMethod はメソッドの形をしているが、呼ぶたびに作り直されるものではない。1本のストリームを保持し続けている。

この前提を共有できていれば、エラーで止まる問題も寿命の取り違えも説明できる。レビューでは、そのストリームがいつ始まっていつ終わるのかを確認したい。