RxJSのswitchMapが実際には何を止めないのかをレビューで確認する

検索の入力ごとにAPIを呼ぶ実装で switchMap を使うのは定石になっている。新しい入力が来たら前の検索を打ち切る、という説明もよく見る。

この説明は半分正しい。switchMap が行うのは、以前の内部Observableからの購読解除である。購読を解除した結果として実処理まで止まるかどうかは、その内部Observableの実装によって決まる。ここを同じものとして扱うと、打ち切ったつもりの処理が最後まで走る。

購読解除が中断になる場合とならない場合

内部Observableが解除処理を実装していれば、購読解除はそのまま中断になる。

解除処理を持つObservable 
const load = (id: string) => new Observable<string>(subscriber => {
  const timer = setTimeout(() => {
    subscriber.next(`done-${id}`);
    subscriber.complete();
  }, 60);

  return () => clearTimeout(timer);
});

このObservableを switchMap に渡し、処理の途中で次の入力を流すと、clearTimeout が呼ばれて処理は実際に取り消される。HttpClient が通信を中断できるのも、同じ仕組みで解除処理が実装されているからだ。

一方、Promiseを渡した場合は事情が変わる。

Promiseを内部に置いた実装 
searchResults$ = this.keyword$.pipe(
  switchMap(keyword => from(this.searchApi.search(keyword))),

@Reviewer
`search()` がPromiseを返す場合、購読を解除してもPromiseの処理自体は止まりません。打ち切りを期待しているなら、中断可能な形へ変更する必要があります。
);

from(promise) の購読を解除しても、Promiseは既に走り始めている。RxJS 7.8 で実際に流してみると、先行分も後続分も両方が最後まで完了する。RxJS 側は結果を無視するだけで、処理そのものには関与しないからだ。

[Promise A] 実処理 開始
[Promise B] 実処理 開始
[Promise A] 実処理 完了（止まっていない）
[Promise B] 実処理 完了（止まっていない）
next done-B

購読者が受け取るのは done-B だけなので、画面上は期待どおりに見える。それでも、サーバーへのリクエストは2本とも届いている。

Comment 
@Reviewer: `switchMap` で前の処理を打ち切る想定に見えますが、内部がPromiseです。購読を解除しても処理は最後まで走り、サーバーには両方のリクエストが届きます。負荷の見積もりがこれで合っているか確認させてください。

サーバーが実行済みの更新はどうなるか

読み取りであれば、余分なリクエストが飛んでも結果を捨てるだけで済む。書き込みではそうはいかない。

switchMap が解除するのはクライアント側の購読であって、サーバーが既に実行した更新を取り消す機能ではない。リクエストがサーバーに届いて処理された後で購読を解除しても、更新は残る。

この区別は、連打対策を exhaustMap に任せている実装でも同じように関わってくる。

クライアント側だけで連打を抑えている実装 
this.submit$.pipe(
  exhaustMap(() => this.api.save(this.form.value)),
).subscribe();

exhaustMap は、内部Observableが complete するまでの間に来た入力を捨てる。ボタン連打はこれで抑えられる。

抑えられないものもある。捨てられた入力は後から再実行されないため、利用者から見ると押したはずの操作が消える。さらに、抑制が働くのは同じ購読の中だけなので、別タブからの操作や、再読み込み後に再送された操作は素通りする。クライアント側の競合制御とサーバー側の重複実行対策は別の問題であり、前者で後者を代替できない。

Comment 
@Reviewer: 連打対策が `exhaustMap` だけになっています。同じ購読の中でしか抑制されないため、別タブや再読み込み後の再送は防げません。サーバー側の冪等性はどう担保していますか。

4つのOperatorが答えている問い

高階Operatorの選択は、「非同期処理が重なったとき何をするか」への4通りの答えになっている。実際に流すと違いがはっきりする。

入力 A、B、C を20ミリ秒間隔で流し、処理時間をそれぞれ A=90ミリ秒、B=10ミリ秒、C=10ミリ秒とした場合の出力は次のようになる。

Operator 出力 重なったときの扱い
switchMap [B, C] 新しい入力が来たら前の購読を解除する
mergeMap [B, C, A] 並行して走らせ、終わった順に出す
concatMap [A, B, C] 前の内部Observableの complete を待って次を始める
exhaustMap [A] 実行中は後続の入力を捨てる

mergeMap の出力が [B, C, A] になっている点に注目したい。入力順は A、B、C だが、処理時間が違うため出力順は入れ替わる。既定では同時実行数に制限もない。順序が結果の正しさに関わる処理でこれを選ぶと、負荷が上がったときにだけ壊れる。

concatMap は順序を保つ代わりに、前が complete するまで次へ進まない。入力が処理速度より速いと待ち行列が伸び続ける。

switchMap の出力に A が含まれていないのは、A の処理中に B が来たためである。読み取りでは望ましい挙動だが、A の処理が書き込みだった場合、結果を受け取らないだけで更新は実行され得る。

Comment 
@Reviewer: 保存処理に `mergeMap` を使っているため、同時実行数に制限がなく、完了順も入力順とは限りません。順に処理する意図なら `concatMap`、実行中は捨てる意図なら `exhaustMap` が合います。

選択を決める3つの問い

どのOperatorを選ぶかは、次の順で決めていくと要件と対応づけやすい。

  1. 前の処理の結果は不要になるか。新しい入力が前の結果を無意味にするなら switchMap。検索、絞り込み、画面切り替えがこれにあたる
  2. 処理の順序が結果に影響するか。影響するなら concatMap。逐次のインポート、順番のある更新がこれにあたる
  3. 実行中の重複操作をどう扱うか。捨てたいなら exhaustMap、すべて処理したいなら mergeMap

どれを選んでも、書き込みであればサーバー側の重複実行対策が別途必要になる。この点だけはOperatorの選択では解けない。

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

高階Operatorを見たときの確認項目
  • 内部が Promise になっていないか。購読解除で実処理が止まる想定ではないか
  • 書き込み処理で、クライアント側の打ち切りをサーバー側の取り消しと混同していないか
  • mergeMap を使っている箇所で、出力順が入力順と違っても問題ないか
  • concatMap の待ち行列が、入力の頻度に対して伸び続けないか
  • exhaustMap で捨てた操作が、利用者に伝わらないまま消えていないか
  • 連打対策がクライアント側だけで完結していないか

おわりに

switchMap を打ち切りの道具として覚えると、内部に何を渡しているかを確認しなくなる。実際に確認すべきなのは、内部Observableが解除処理を持っているかどうかである。

レビューでは、Operatorの名前より先に内部を見たい。HttpClient なら中断まで届き、Promiseなら届かない。この差が分かっていれば、打ち切りを前提にした設計が成立するかどうかもその場で判断できる。