RxJSでHTTPリクエストの並列・逐次・一括処理をする

こんにちは。日々 RxJS と格闘している @ringtail003 です。

RxJSには [公式ドキュメント](https://rxjs-dev.firebaseapp.com/guide/overview) や [Learn RxJS](https://www.learnrxjs.io/) など学習コンテンツがあるのですが、それを読んでもなおマーブルダイアグラムが理解できない私です。なぜかって私が開発しているプロジェクトでは、クリックイベントを題材にしたコードや、数秒おきに `1 2 3...` を流すコードの使い所がなく、それらを参考にしても実践で扱いたい題材との乖離が激しくアンポンタンな頭でうまく転換する事ができないのです。

私が欲しいのは HTTP リクエスト！ HTTP リクエストのサンプルコード！！という事で、配列を並列・逐次・一括の HTTP リクエストにするサンプルコードを書いてみる事にしました。

# この記事を検証した環境

- [rxjs](https://www.npmjs.com/package/rxjs) v6.5.3
- [typescript](https://www.npmjs.com/package/typescript) v3.6.4

# はじめに

擬似的な HTTP リクエストとして、このようなサービスを作りました。

```typescript
import { of } from 'rxjs';
import { delay } from 'rxjs/operators';

export function httpService(id: string) {
  const response = [
    { id: 'a', delay: 3000 },
    { id: 'b', delay: 4000 },
    { id: 'c', delay: 1000 },
  ].find((item) => item.id === id);

  return of(response)
    .pipe(delay(response.delay))
  ;
}
```

`httpService()` 関数を呼び出すと、数秒遅延して擬似的なレスポンスを返します。

```typescript
httpService('a'); // レスポンス "{ id: 'a', delay: 1000 }"
```

`a` `b` `c` を一斉に呼び出すと `c > a > b` の順でストリームに値が流れます。

```typescript
httpService('a'); // 3 秒後
httpService('b'); // 4 秒後
httpService('c'); // 1 秒後
```

# 並列処理

一斉に POST をリクエストしてレスポンスが返ってきたものから順に処理したい、みたいなケースです。

```typescript
import { from } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

from(['a', 'b', 'c']).pipe(
    mergeMap((id) => httpService(id)),
  )
  .subscribe((value) => {...}})
;
```

[from](https://rxjs-dev.firebaseapp.com/api/index/function/from) はイテレーション可能な値をストリームに変換するオペレータです。 `from(['a', 'b', 'c'])` はストリームを作成し `a > b > c` の順に値を3回に分けて流します。

流れてきた値を `httpService()` 関数に渡すと HTTP レスポンスが `Observable` 型で返却されます。ストリームの中にストリームが流れている状態ですね。

```typescript
---Observable<response c>---Observable<response a>---Observable<response b>---
```

`mergeMap` を使うと内側のストリームの値を外側のストリームに放出する事ができます。

```typescript
---<response c>---<response a>---<response b>---
```

## 結果

レスポンスはこの順で `subscribe` に到着します。

```typescript
c (delay 1000) // 1秒後

a (delay 3000) // 3秒後

b (delay 4000) // 4秒後
```

https://stackblitz.com/edit/typescript-asnpud?file=index.ts

# 並列処理（最初に到達したものだけ欲しい）

フォールバックを含めた複数サーバーに一斉に GET をリクエストして最初に返ってきたレスポンスを使う、みたいなケースです。
`Promise.race()` の RxJS 版ですね。

```typescript
import { from } from 'rxjs';
import { switchMap } from 'rxjs/operators';

from(['a', 'b', 'c']).pipe(
    switchMap((id) => httpService(id)),
  )
  .subscribe((value) => {...}})
;
```

`switchMap` を使うと `subscribe` に到着する値は最初のひとつだけで、以降の値は無視する事になります。

## 結果

レスポンスはひとつだけ `subscribe` に到着します。

```typescript
c (delay 1000) // 1秒後
```

https://stackblitz.com/edit/typescript-ek2wz2?file=index.ts

# 逐次処理

複数の DELETE リクエストを順番に処理したい、みたいなケースです。

```typescript
import { from } from 'rxjs';
import { concatMap } from 'rxjs/operators';

from(['a', 'b', 'c']).pipe(
    concatMap((id) => httpService(id)),
  )
  .subscribe((value) => {...}})
;
```

`concatMap` は前の処理が終わるのを待ってから次の処理を実行します。

## 結果

レスポンスはこの順で `subscribe` に到着します。

```typescript
a (delay 3000) // 3秒後

b (delay 4000) // 3 + 4秒後

c (delay 1000) // 3 + 4 + 1秒後
```

https://stackblitz.com/edit/typescript-nenzef?file=index.ts

# 一括処理

複数の HTTP リクエストが全て完了するのを待って処理するケースです。
`Promise.all()`の RxJS 版ですね。

```typescript
import { forkJoin } from 'rxjs';

const sources = ['a', 'b', 'c'].map((id) => httpService(id));

forkJoin(sources)
  .subscribe((values) => {
    // forkJoin に渡した配列と同じ順番で values が返る
    values[0]; // a
    values[1]; // b
    values[2]; // c
  })
;
```

`forkJoin` は流れてきたストリームの全ての完了を待ち、それぞれのストリームで最後に流れた値を放出します。

## 結果

`subscribe` が呼ばれるのは 1 回です。全ての値が一括して到着します。

```typescript
 // 4秒後
[
  a (delay 3000),
  b (delay 4000),
  c (delay 1000),
]
```

https://stackblitz.com/edit/typescript-ncey9e?file=index.ts

# エラーハンドリング

HTTP 通信ではエラーハンドリングが欠かせません。私達はエラーの内容を解析して、システムのユーザーになぜエラーが起きたのかを伝え、その後取るべき操作に誘導する必要があります。

HTTP リクエストのエラーに似た状況を作るために `httpService()` 関数を修正してみましょう。

```typescript
import { Observable, of, from, timer, throwError } from 'rxjs';
import { delay, mergeMap } from 'rxjs/operators';

export function httpService(id: string) {
  // httpService('error') のように呼ばれたら 2 秒後にエラーを発生させる
  if (id === 'error') {
    const error = { id: 'error', delay: 2000 };

    return timer(error.delay).pipe(
      mergeMap(() => throwError(error)),
    );
  }
  
  ...
}
```

`a` `b` `c` `error` を一斉に呼び出すと `c > error > a > b` の順でストリームに値が流れます。

```typescript
httpService('a'); // 3 秒後
httpService('b'); // 4 秒後
httpService('c'); // 1 秒後
httpService('error'); // 2 秒後
```

これで準備ができました。
**並列処理** で登場した `mergeMap` のサンプルコードにエラーハンドリングを追加してみましょう。

# エラーハンドリング（無視）

複数の HTTP リクエストのうち、どれかでエラーが発生した場合に無視して次を実行するケースです。

```typescript
import { from, EMPTY } from 'rxjs';
import { mergeMap, tap, catchError } from 'rxjs/operators';

from(['a', 'b', 'c', 'error']).pipe(
    mergeMap((id) => httpService(id).pipe(
      catchError(() => EMPTY),
    )),
  )
  .subscribe((value) => {...}})
;
```

`catchError` でストリームにエラーハンドリングを追加する事ができます。この関数の返却値がエラーが発生した時の扱いを決定します。

`from` で作成したストリームにはこのように値が流れています。

```typescript
---Observable<c>---Observable<error>---Observable<a>---Observable<b>---
```

`catchError` でエラーがキャッチされます。`EMPTY` はストリームをただちに完了状態にし、それ以上値を流しません。

```typescript
---Observable<c>---(completed)---Observable<a>---Observable<b>---
```

`error` のストリームは購読できる状態ではなくなったため `subscribe` には到着しません。

```typescript
---Observable<c>---Observable<a>---Observable<b>---
```

## 結果

`subscribe` には `c` `a` `b` の順で到着します。

```typescript
c (delay 1000)

a (delay 3000)

b (delay 4000)
```

## 参考

実際のプロダクションコードで「エラーを完全に無視する = 握りつぶす」という実装はあんまりしないですよね。画面にアラートを表示しつつその後の処理を続行する、くらいが妥当だと思われます。

`catchError` のコールバック関数はエラーの詳細を得る事ができます。そしてこれらを使ってシステムに副作用を持たせる事ができます。

```typescript
from(['a', 'b', 'c', 'error']).pipe(
    tap((id) => log(`requested ${id}`)),
    mergeMap((id) => httpService(id).pipe(
      catchError((err, caught) => {
        console.log(err); // { id: 'error', delay: 2000 }
        console.log(caught); // Observable{...}
        
        // 実際の HTTP リクエストではステータスコードなどが拾える
        
        notifyService(err.id);
        
        return EMPTY;
      }),
    )),
  )
```

https://stackblitz.com/edit/typescript-ozfvx7?file=index.ts

# エラーハンドリング（続行）

`a` `b` `c` `error` の HTTP リクエストのうち、成功・エラーに関わらず全て購読するケースです。

```typescript
import { from, of } from 'rxjs';
import { mergeMap, tap, catchError } from 'rxjs/operators';

from(['a', 'b', 'c', 'error']).pipe(
    mergeMap((id) => httpService(id).pipe(
      catchError((err) => of(err)),
    )),
  )
  .subscribe((value) => {...}})
;
```

## 結果

`subscribe` には `c` `error` `a` `b` の順で到着します。

```typescript
c (delay 1000)

error (delay 2000)

a (delay 3000)

b (delay 4000)
```

https://stackblitz.com/edit/typescript-yphphc?file=index.ts

# エラーハンドリング（中断）

`a` `b` `c` `error` の HTTP リクエストのうち、どれかひとつでもエラーになったら購読を中断するケースです。

```typescript
import { from, EMPTY, throwError } from 'rxjs';
import { mergeMap, tap, catchError } from 'rxjs/operators';
import { httpService } from './http-service';
import { log } from './dom';

from(['a', 'b', 'c', 'error']).pipe(
    mergeMap((id) => httpService(id)),
    catchError((err) => EMPTY),
  )
  .subscribe((value) => {}})
;
```

さきほど登場した **エラーハンドリング（無視）** のサンプルコードとは `catchError` を指定するストリームが異なっているのでご注意ください。

`from` で作成した外側のストリームにはこのように値が流れています。`EMPTY` を流す事により完了状態になるため、それ以上値が流れる事はありません。

```typescript
---Observable<c>--- | (completed)
```

## 結果

`subscribe` には `c` のみ到着します。

```typescript
c (delay 1000)
```

https://stackblitz.com/edit/typescript-nusas4?file=index.ts

# おわり

以上です。RxJS はオペレータが非常に多いのとデバッグが困難で敷居が高く感じてしまいますが、知らなかったオペレータをあれこれ見つけ出して試すほど深みにはまっていく気がします。よく使う基本の処理さえ抑えておけば、それほど迷う事はないのかもしれませんね。