RxJSでFRP入門

この記事は [MICIN Advent Calendar 2022](https://adventar.org/calendars/7833) の14日目の記事です。
前回は原さんの[医療におけるプロダクトづくりのチャレンジあるある](https://note.micin.jp/n/nb984e4dd9959)でした。

---

プラットフォームチーム所属の竹内です。
主にビデオ通話基盤の開発を行なっています。
MICINのプロダクトにはビデオ通話機能を持ったプロダクトが複数存在しますが、プロダクトごとにビデオ通話機能を開発するのは効率が悪くなります。
そのため、プラットフォームチームでは、ビデオ通話機能を共通基盤システムとして提供しています。

ビデオ通話のクライアント実装においては、複雑な非同期イベントのハンドリングが要求されます。
この問題に対して、手続き的な手法で対処しようとすると、イベントの発生順に依存したアルゴリズムになりがちです。
その結果、開発環境とステージング環境のネットワークの差異によって、数回に一回正常に動作しないということが起こります。
そういったことを何度か経験し、もっと宣言的に書く方法はないか模索していました。
そこで、今注目しているのが**FRP**です。

***F**unctional **R**eactive **P**rogramming*（関数型リアクティブプログラミング）とはプログラミングパラダイムの1つで、関数型プログラミングとリアクティブプログラミングの両方の特徴を持ちます。
一言でいうと、FRPでは関数合成によって入力に対する振舞としてプログラムを表現します。

FRPのプログラムは原則としてステートを持ちません。
しかし、ローカルストレージの読み書きやHTTP通信のステーフルロジックなど、副作用を生じることを避けられない場面も存在します。
このような外界とのやりとりを **I/O（入出力）** と呼びます。

アプリケーションを開発する上で、必ずしもFRPだけでプログラムが完結するわけではなく、他のパラダイムと併用することになります。
そのような場合も、FRPと他のパラダイムを分離し境界を設けることで、FRPのメリットを十分に享受できます。

![FRPはFPとRPの積集合](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/72814/f97bc49e-37c1-554f-bd6c-ee8ff9893c1c.png)

まずは、関数型プログラミングについて紹介し、その後RxJSを利用したFRPの具体的な実践例を紹介します。

# 関数型プログラミングの作法

JavaScriptは関数型とオブジェクト指向の両方の機能を備えた言語です。
特に意識することなく、両者を使っているかもしれません。

オブジェクト指向プログラミングで出来上がるのは、オブジェクトに対する操作です。
一方、関数型プログラミングで出来上がるのは、宣言的に表現した式です。

これらの違いが分かると、より良いコードを書けるようになるはずです。

## プログラムの副作用

皆さんはアプリケーションが期待通りに動作しないとき、何を試しますか？
おそらく、多くの方はアプリケーションの再起動を行うでしょう。
それでもダメならOSごと再起動を試すかもしれません。
つまり、我々は再起動することによって異常なステートがクリアされることを経験的に理解しています。

では、そもそもなぜ、そのようなひどい状態になるのでしょうか。
それは、プログラムがステートに対して意図しない**副作用**を与えているからです。
至る所で複雑にステート管理されたプログラムはもはや人の手には負えません。

副作用を与える原因の一つとして値の**再代入**が挙げられます。
`var` や `let` を使って変数を宣言すると、再代入が可能となります。

```javascript
let a = 'Hello';
function foo() {
  a = 'Hi';
}
foo();
console.log(a); // "Hi"
```

一方で、 `const` で宣言された変数は再代入されず、基本的には値が一定です。

```javascript
const a = 'Hello';
function foo() {
  a = 'Hi'; // TypeError: Assignment to constant variable.
}
foo();
console.log(a);
```

ただし、 `const` で宣言してもオブジェクトの場合、そのプロパティを上書きすることは出来てしまうため、注意が必要です。

```javascript
const a = { key: 'Hello' };
function foo() {
  a.key = 'Hi';
}
foo();
console.log(a.key); // "Hi"
```

処理の中で副作用を発生させず、同じ入力に対して同じ出力を返すような関数は**純粋**であると言われ、このような性質のことを**参照透過性**と言います。

```javascript
// 参照透過性のない関数
let count = 0;
function nonTransparent() {
  count += 1;
  return count;
}
console.log(nonTransparent()); // 1
console.log(nonTransparent()); // 2

// 参照透過性のある関数
function transparent(count) {
  return count + 1;
}
console.log(transparent(0)); // 1
console.log(transparent(0)); // 1
```

オブジェクト指向ではメンバ変数に副作用を与えて状態を保持させるセッターなどのメソッドが多用されます。
そのため、参照透過性のないコードになりがちで、プログラマーはオブジェクトの状態を意識しながらコードを書く必要があります。

:::note info
**コラム: リトライ処理を入れればいい？**

再起動すれば元に戻るという発想は度々人々をリトライ処理の実装へと向かわせます。
原因分析をしないまま闇雲に「リトライ処理を入れてくれ」という類の発言には注意を払う必要があります。
「変数が未定義かどうか」で条件分岐してリトライして、値が出揃うのを待つというその場凌ぎを繰り返していてはコードが無駄に複雑化することは避けられません。
動作原理不明なフランケンシュタインのようなプログラムからは既存のコードを削除することができず、そのダークマターはどんどん膨張していきます。

過剰な防衛はせず、根本的な原因を明確化し、リトライ処理は最小限に収めましょう。
フォールトトレラントは結構ですが、耐える必要のないフォールトに寛容なシステムは堅牢とは言えません。
:::

## 関数合成

通常、プログラムは複数のモジュールによって構成され、適切に分割されたモジュールは凝集度が高く、疎結合です。
関数型プログラミングでも、あるタスクを実行するために単一の関数が全責務を担うのではなく、複数の関数に分割し関心を分離します。
そして、分割した関数を合成することで、複雑なタスクを単純な関数の組み合わせで実現することが出来ます。
関数合成は元々数学的な概念で、学生時代に習った方もいらっしゃると思いますが、ここで簡単におさらいしてみましょう。

例えば、次のような集合 *A*, *B*, *C* と写像 *f*, *g* があります。

![集合A, B, Cと写像f, gの関係](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/72814/94fbf7f1-8630-0277-76fd-42053532ff78.png)

```math

\displaylines{
f(x) = x + 1,\ g(x) = 2x^2のとき、合成関数は \\
\begin{eqnarray}
(g \circ f)(x) &=& g(f(x)) \\
&=& 2(x + 1)^2 \\
\end{eqnarray} \\
となる。 \\
A = \{1,2,3\}, a \in A,\ b \in B,\ c \in Cとすると、\\
\left\{
\begin{array}{l}
A = \{ a\ |\ 1,2,3 \} \\
B = \{ b\ |\ b = f(a) \} \\
C = \{ c\ |\ c = (g \circ f)(a) \}
\end{array}
\right.
}
```

これを手続き型プログラミングで表現すると、次のようになります。

```javascript
const A = [1, 2, 3];
const B = [];
const C = [];
for (let i = 0; i <= A.length; i++) {
  B.push(A[i] + 1);
  C.push(2 * B[i] ** 2);
}
```

これを関数型プログラミングで表現すると、次のようになります。

```javascript
const f = x => x + 1;
const g = x => 2 * x ** 2;
const A = [1, 2, 3];
const B = A.map(f);
const C = A.map(x => g(f(x)));
```

このように関数型では数学に近い形で表現できることが分かります。
`C` の変数宣言に着目してみると、手続き型では一番下まで読まないとどのような要素が格納されるのか分かりません。
一方、関数型では変数宣言の行を読むだけで、`C` は `f` と `g` の合成関数を `A` の各要素に適用した配列であることが分かります。

これが関数型が宣言的である所以です。
関数型の方が問題の本質を捉えており、バグの混入を防げそうです。

:::note info
**コラム: アロー関数はなぜ美しいのか？**

JavaScriptの言語仕様は度々批判の的となりますが、アロー関数のシンタックスについては受け入れられているように感じます。
アロー関数のシンタックスが支持されるのは数学的意味を端的に矢印で表現できているからだと思います。

例えば、次のような集合 *P* = { 1, 2, 3 }, *Q* = { *q* | *q* = *p*&sup2;, *p* &in; *P* }、写像 *f* があります。

![集合P, Qと写像f](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/72814/9bf865cd-699a-7fda-179d-355d83358d85.png)

これをアロー関数を用いて表現すると次のようになります。

```javascript
const P = [1, 2, 3];
const f = x => x ** 2;
const Q = P.map(f);
```

mapは日本語で写像という意味です。
つまり、集合のオブジェクト(Array)の写像というメソッド(map)にその振る舞いを決定する関数を引数として与える形です。
`=>` で左から右へ入力と出力の関係が分かりやすく表現されています。
アロー関数は関係ないですが、関数が写像に内包される構造も数学的で美しいです。
:::

# RxJSとは

[RxJS](https://rxjs.dev/)はJavaScriptでリアクティブプログラミングを実現するためのライブラリです。
MicrosoftによってC#用に開発されたRx.NETというライブラリが元となっています。
複数の言語に移植されており、言語を跨いで統一的なAPIが実装されています。

RxJSを学んで、まず疑問に思うのが「で、これってどこで使うの？」という点ではないでしょうか。
UIのステート管理の問題であればReactやVueが十分実用的な答を提示していますし、非同期処理もPromiseを使えば大抵シンプルに表現できます。
要するに、「RxJSは**オワコン**なんじゃないか」、と。
私見ですが、基本的にはUIのステート管理はUIフレームワークに任せて、複雑な非同期イベントハンドリング（バッファリングやマージなど）が必要な場合に有用だと考えています。

:::note info
**コラム: FRP関連ライブラリ**

[Cycle.js](https://cycle.js.org)

もしも、UIの実装にもRxJSを使ってFRPを実践したい場合なら、Cycle.jsというフレームワークもあります。
Cycle.jsはFluxに課題を感じていた [André Staltz](https://github.com/staltz) 氏を中心として開発されました。

[Marble.js](https://docs.marblejs.com)

Node.jsでRxJSベースのサーバーアプリケーションを構築したいなら、Marble.jsというフレームワークがあります。

[Lodash](https://lodash.com)

関数型プログラミングで利用されるユーティリティ関数を提供するライブラリ。

[Ramda.js](https://ramdajs.com)

lodashと同様にユーティリティライブラリですが、設計思想が異なります。
lodashは比較的パフォーマンスを重視しているのに対して、RamdaはシンプルなAPIで副作用が生じないことを重視します。

[Bacon.js](https://baconjs.github.io)

RxJSよりも一貫性を重視したFRPライブラリ。
エラーハンドリングの手法も異なります。

[kefir](https://kefirjs.github.io/kefir/)

速度と省メモリを重視した後発のリアクティブ・プログラミングライブラリ。
:::

## パイプ

ピュアなJavaScriptで関数合成を行うと、ネスト構造になってしまいます。

```typescript
const divideByThree = (x: number) => x / 3;
const addOne = (x: number) => x + 1;
const timesTwo = (x: number) => x * 2;
const result = divideByThree(addOne(timesTwo(1)));
console.log(result); // 1
```

内側から外側へ解釈する必要があり、これは不自然で読みづらいです。
この問題に対して、メソッドチェーンは可読性に優れた方法です。

```typescript
const divideByThree = (x: number) => x / 3;
const addOne = (x: number) => x + 1;
const timesTwo = (x: number) => x * 2;
const [result] = [1].map(timesTwo).map(addOne).map(divideByThree);
console.log(result); // 1
```

これで左から右へ順番に読むことが出来て、自然な順序で解釈が可能です。
しかし、4行目で不必要に配列化している点で、違和感があります。
メソッドというものは必然的にオブジェクトの型と密結合であるため、合成可能な関数に制約が生まれます。

例えば、次のようにチェーンの途中で型が変わる場合も、メソッドチェーンでは対応できません。

```typescript
const divideByThree = (x: number) => x / 3;
const addOne = (x: number) => x + 1;
const timesTwo = (x: number) => x * 2;
const plus = (a: number, b: number) => a + b;
const [result] = [1, 2.5].map(timesTwo).map(addOne).reduce(plus).map(divideByThree); // Error: Property 'map' does not exist on type 'number'.
```

RxJSでは `pipe` を使って、値をObservableという一貫した型で扱うことが出来ます。
`of` という関数はObservable型のデータを静的に生成する関数です。
`subscribe` については次節で説明します。

```typescript
import { of, map } from "rxjs";

const divideByThree = (x: number) => x / 3;
const addOne = (x: number) => x + 1;
const timesTwo = (x: number) => x * 2;
const observable = of(1).pipe(map(timesTwo), map(addOne), map(divideByThree));
observable.subscribe((result: number) => console.log(result)); // 1
```

途中で型が変わっても[^1]、同じように書けます。

```typescript
import { of, map, reduce } from "rxjs";

const divideByThree = (x: number) => x / 3;
const addOne = (x: number) => x + 1;
const timesTwo = (x: number) => x * 2;
const plus = (a: number, b: number) => a + b;
const observable = of(1, 2.5).pipe(map(timesTwo), map(addOne), reduce(plus), map(divideByThree));
observable.subscribe((result: number) => console.log(result)); // 3
```

Observableはこういった性質から**モナド**っぽいと言われます。

パイプを利用することで型と関数が疎結合になり、より柔軟な合成が可能となりました。
パイプラインの中の関数は引数の数と入出力の型が一致していれば、交換可能です。

![関数合成のパイプライン](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/72814/bc248dc6-813c-e1ee-c020-97dc59d614a1.png)

[^1]: 厳密にはObservableシーケンスは配列ではないので、reduceの箇所で型がArrayからNumberに変わったと捉えるのは誤りで、「値を束ねた」というのが正しいと思われます。

:::note info
**コラム: パイプライン演算子**

F#にはパイプが演算子として標準的に存在します。
JavaScriptでもTC39の[プロポーザル](https://github.com/tc39/proposal-pipeline-operator)として挙がっており、現在はStage2となっています。

当初はF#と同じ書き方 `|>` という演算子が提案されていました。
しかし、async/awaitとの仕様が競合したことから、Stage2ではHack Styleという仕様が採用されています。

[tc39/proposal-pipeline-operator の過去と現状](https://me1on.dev/blog/20211204)
:::

## ColdとHot

RxJSには**Cold**と**Hot**という概念があります。
言葉で説明するより、コードを見てもらうのが早そうです。

Cold Observableは `subscribe` が呼び出されると一連の関数を実行します。
前節の例はCold Observableにあたります。

```javascript
import { of, map } from "rxjs";

const divideByThree = (x: number) => x / 3;
const addOne = (x: number) => x + 1;
const timesTwo = (x: number) => x * 2;
const observable = of(1).pipe(map(timesTwo), map(addOne), map(divideByThree));

observable.subscribe((result: number) => console.log(result)); // 1
```

Hot Observableは `subscribe` が呼び出されて、次のイベントが発生した際に一連の関数を実行します。

```javascript
import { fromEvent, map } from "rxjs";

fromEvent(document, "mousedown")
  .pipe(map(({ clientX, clientY }) => clientX * clientY))
  .subscribe((result: number) => console.log(result));
```

## Observer

RxJSにおいて、ObserverはObservableが送る値を受け取るコンシューマーで、コールバック関数のセットです。

```typescript
import { of, map, Observer } from "rxjs";

const divideByThree = (x: number) => x / 3;
const addOne = (x: number) => x + 1;
const timesTwo = (x: number) => x * 2;
const observable = of(1).pipe(map(timesTwo), map(addOne), map(divideByThree));

const observer: Observer<number> = {
  next: (result: number) => console.log(result),
  error: (message: string) => console.error(message),
  complete: () => console.log("complete"),
};

observable.subscribe(observer); // 1, complete
```

通常は `next` が呼び出され、エラーが発生した場合は `error` が呼び出されます。
実行が完了したら最後に `complete` が呼び出されます。

各メソッドは省略可能で、次のように `next` のみでも正常に動作します。

```typescript
import { of, map, Observer } from "rxjs";

const divideByThree = (x: number) => x / 3;
const addOne = (x: number) => x + 1;
const timesTwo = (x: number) => x * 2;
const observable = of(1).pipe(map(timesTwo), map(addOne), map(divideByThree));

const observer: Observer<number> = {
  next: (result: number) => console.log(result),
};

observable.subscribe(observer); // 1
```

また、上記の `next` のみのサンプルは次のように更に省略して書くこともできます。

```typescript
import { of, map, Observer } from "rxjs";

const divideByThree = (x: number) => x / 3;
const addOne = (x: number) => x + 1;
const timesTwo = (x: number) => x * 2;
const observable = of(1).pipe(map(timesTwo), map(addOne), map(divideByThree));

const observer = (result: number) => console.log(result)

observable.subscribe(observer); // 1
```

前節までのサンプルコードはこの形式ですね。

:::note info
**コラム: RxJSにおけるエラーハンドリング**

FRPの例外処理においては、基本的にtry-catchやthrow文は使いません。
関数内でキャッチされ、戻り値に変換される分には問題ありませんが、FRPを正しく機能させるためには例外を外部に投げることは出来ません。
例外を投げると出口のパスが増えて参照透過性を失ってしまいます。

RxJSではエラーはObservableにラップされた値となっています。
`throwError` オペレーターや `error` メソッドによってエラーを送出します。
エラー発生時にObserverの `error` プロパティが省略されている場合、グローバルに例外が投げられます。

`catchError` オペレーターを利用して、エラーを抑制することも出来ます。
:::

## Subject

SubjectはObservableとObserverの両方の性質を持ちます。
Observableは `subscribe` で流れてくる値を取得できるのみですが、Subjectは任意のタイミングで `next` を呼び出して値を流すことができます。

```typescript
import { Subject, map } from "rxjs";

const divideByThree = (x: number) => x / 3;
const addOne = (x: number) => x + 1;
const timesTwo = (x: number) => x * 2;

const subject = new Subject<number>();
subject
  .pipe(map(timesTwo), map(addOne), map(divideByThree))
  .subscribe(console.log);

subject.next(1); // 1
subject.next(2.5); // 2
```

# 様々な非同期通信

ここからはRxJSを使ったFRPの実践例として、様々な非同期通信のパターンを考えてみたいと思います。

## 順次HTTP通信

サービスAから必要なデータを取得してから、サービスBにリクエストするパターンです。
これはCold Observableの例です。

```typescript
import { mergeMap } from "rxjs";
import { fromFetch } from "rxjs/fetch";

const getFrom = (baseUrl: string) => (path: string) =>
  fromFetch(`${baseUrl}${path}`, {
    selector: (response) => response.json(),
  });

const serviceA = "http://localhost:8080";
const serviceB = "http://localhost:8081";

const getFromA = getFrom(serviceA);
const getFromB = getFrom(serviceB);

getFromA("/user")
  .pipe(
    mergeMap(({ barUserId }: { barUserId: number }) =>
      getFromB(`/user/${barUserId}`)
    )
  )
  .subscribe(console.log);
```

ちなみに、このように多項関数をネストされた単項関数に分割することを**カリー化**と言います。
カリー化した関数を部分適用することで関数の再利用性が高まります。

## 並列HTTP通信

サービスAとサービスBに並列的にリクエストするパターンです。
これもCold Observableの例です。
`forkJoin` を使うと、Promise.allのように複数のリクエストを並列処理できます。

```typescript
import { forkJoin } from "rxjs";
import { fromFetch } from "rxjs/fetch";

const getFrom = (baseUrl: string) => (path: string) =>
  fromFetch(`${baseUrl}${path}`, {
    selector: (response) => response.json(),
  });

const serviceA = "http://localhost:8080";
const serviceB = "http://localhost:8081";

const getFromA = getFrom(serviceA);
const getFromB = getFrom(serviceB);

forkJoin([getFromA("/users"), getFromB("/users")]).subscribe(console.log);
```

## 並列WebSocket通信

最後に応用的な例として、複数のサービスに依存したチャットルームのようなアプリケーションを考えてみます。
サービスAとサービスBに並列的なWebSocket通信を行うパターンで、これはSubjectの例です。（例としてはやや特殊な状況ですが、実際に直面した問題です。）

入室時に両方のサービスに `JOIN` イベントを送信し、入室処理が完了すると `JOINED` イベントを受信します。
他のユーザーが入室したことも `JOINED` イベントで受信されます。
このとき、自分自身の入室イベントだけでなく、通信相手の入室イベントもハンドリングする必要があり、下図のようにユーザーイベントとサービスが多対多の関係にあります。

![ユーザーイベントとサービスの関係](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/72814/987d0f26-0b84-4788-ee88-7d0c3d50e5fe.png)

「ユーザーFooがサービスAとサービスBからJOINEDを受け取ったとき」というように、複数サービスから送られてくるイベント処理をユーザーごとにまとめて履行する必要があるとします。

最初に思いつくのがRxJSが提供している `combineLatest` というオペレーターを使う方法です。
`combineLatest` は複数のObservableから値を受け取るたびに最新の値のセットとして結合します。
公式ドキュメントのサンプルコードを掲載します。

```typescript
import { timer, combineLatest } from 'rxjs';

const firstTimer = timer(0, 1000); // emit 0, 1, 2... after every second, starting from now
const secondTimer = timer(500, 1000); // emit 0, 1, 2... after every second, starting 0,5s from now
const combinedTimers = combineLatest([firstTimer, secondTimer]);
combinedTimers.subscribe(value => console.log(value));
// Logs
// [0, 0] after 0.5s
// [1, 0] after 1s
// [1, 1] after 1.5s
// [2, 1] after 2s
```

`combineLatest` を使って、2つのWebSocketメッセージを結合してみます。
気になると思うので先に補足しておくと、Observableを表すために変数名の最後に `$` を付ける慣習があったりします。

```typescript
import { combineLatest, filter, map } from "rxjs";
import { webSocket } from "rxjs/webSocket";

type Payload = {
  uid: number;
};

type Message = {
  type: "JOIN" | "JOINED";
  payload: Payload;
};

const MY_UID = 123;

const wsA$ = webSocket<Message>("http://localhost:8080");
const wsB$ = webSocket<Message>("http://localhost:8081");

wsA$.next({ type: "JOIN", payload: { uid: MY_UID } });
wsB$.next({ type: "JOIN", payload: { uid: MY_UID } });

const joinedA$ = wsA$.pipe(filter(({ type }) => type === "JOINED"));
const joinedB$ = wsB$.pipe(filter(({ type }) => type === "JOINED"));

const joined$ = combineLatest([joinedA$, joinedB$]).pipe(
  map(([msgA, msgB]) => ({ msgA, msgB }))
);

const localJoined$ = joined$.pipe(
  filter(
    ({ msgA, msgB }) =>
      msgA.payload.uid === MY_UID && msgB.payload.uid === MY_UID
  )
);
const remoteJoined$ = joined$.pipe(
  filter(
    ({ msgA, msgB }) =>
      msgA.payload.uid !== MY_UID && msgB.payload.uid !== MY_UID
  )
);

localJoined$.subscribe(({ msgA, msgB }) =>
  console.log("Local Joined", msgA, msgB)
);
remoteJoined$.subscribe(({ msgA, msgB }) =>
  console.log("Remote Joined", msgA, msgB)
);
```

一見、問題なさそうですし、一定の条件下では期待通り動作します。
しかし、`combineLatest` から取得できるメッセージのペアはあくまで最新メッセージのペアであって、2つのメッセージが同一ユーザーに関するものであることは保証されていません。
例えば、ユーザーFooとユーザーBarが同じタイミングで入室した際に、`msgA` はFooのイベントで、 `msgB` はBarのイベントであるという、あべこべな状況が起こり得ます。
そのとき、上記のコードでは正常にイベントを検知することが出来ません。

`combineLatest` ではメッセージの同一性が担保できないのであれば、メッセージ内のユーザーIDをキーとしてイベントを結合出来ると良さそうです。
WebSocketのようにいつやってくるか分からない双方向通信のイベントを結合するには、先に受信したイベントをステートとしてメモリ上に保持せざるを得ません。
しかし、本稿の冒頭でも述べた通りFRPでは基本的にステートを保持すべきではありません。
FRPのロジックからステートを分離するために、次のように `innerJoin` という `combineLatest` を拡張したカスタムオペレーターを作成します。（`innerJoin` はSQLのINNER JOIN句から着想を得ています。）

工夫した点としては、パフォーマンスを最適化するため、ステート管理にはMapを利用しています。[^2]
また、二重ループ部分の可読性を高めるため、Generatorを利用しています。[^3]

```typescript
import { combineLatest, Observable, Subject } from 'rxjs';

type Relation = <T>(keyA: T, keyB: T) => boolean;

class State<T, U, V> {
  source: Map<V, T>;
  target: Map<V, U>;

  constructor() {
    this.source = new Map();
    this.target = new Map();
  }

  setSource(key: V, val: T) {
    this.source.set(key, val);
  }

  setTarget(key: V, val: U) {
    this.target.set(key, val);
  }

  *join(relation: Relation): Generator<[T, U, V, V]> {
    for (const [sKey, sVal] of this.source) {
      for (const [tKey, tVal] of this.target) {
        if (relation(sKey, tKey)) {
          yield [sVal, tVal, sKey, tKey];
          this.source.delete(sKey);
          this.target.delete(tKey);
        }
      }
    }
  }
}

/**
 * 内部結合
 * @param target 結合先
 * @param on 結合条件
 */
export const innerJoin =
  <T, U, V = any>(
    target: Observable<U>,
    on: (a: T, b: U) => [V, V, Relation]
  ) =>
  (source: Observable<T>) => {
    const state = new State<T, U, V>();
    const subject = new Subject<[T, U, V, V]>();
    combineLatest([source, target]).subscribe((vals: [T, U]) => {
      const [sourceKey, targetKey, relation] = on(...vals);
      state.setSource(sourceKey, vals[0]);
      state.setTarget(targetKey, vals[1]);
      for (const nextVals of state.join(relation)) {
        subject.next(nextVals);
      }
    });
    return subject;
  };
```

重要なことは次のようにFRPの世界で表示的意味論の合成性が損なわれないことです。
ステートを持っているのはFRPの外の世界であると捉えます。

```typescript
import { filter, map } from "rxjs";
import { webSocket } from "rxjs/webSocket";
import { innerJoin } from "./inner-join";

type Payload = {
  uid: number;
};

type Message = {
  type: "JOIN" | "JOINED";
  payload: Payload;
};

const MY_UID = 123;

const wsA$ = webSocket<Message>("http://localhost:8080");
const wsB$ = webSocket<Message>("http://localhost:8081");

wsA$.next({ type: "JOIN", payload: { uid: MY_UID } });
wsB$.next({ type: "JOIN", payload: { uid: MY_UID } });

const joinedA$ = wsA$.pipe(filter(({ type }) => type === "JOINED"));
const joinedB$ = wsB$.pipe(filter(({ type }) => type === "JOINED"));

const joined$ = joinedA$.pipe(
  innerJoin(joinedB$, (a, b) => [
    a.payload.uid,
    b.payload.uid,
    (keyA, keyB) => keyA === keyB,
  ]),
  map(([msgA, msgB, uid]) => ({ msgA, msgB, uid }))
);

const localJoined$ = joined$.pipe(filter(({ uid }) => uid === MY_UID));
const remoteJoined$ = joined$.pipe(filter(({ uid }) => uid !== MY_UID));

localJoined$.subscribe(({ msgA, msgB }) =>
  console.log("Local Joined", msgA, msgB)
);
remoteJoined$.subscribe(({ msgA, msgB }) =>
  console.log("Remote Joined", msgA, msgB)
);
```

これで、メッセージの順番が入れ替わっても正常に動作するようになりました。

オペレーター自体は純粋でなくとも、オペレーターに渡される関数は純粋であることで、FRPロジック内に副作用による不具合が含まれていないことが保証されます。
カスタムオペレーターは汎用的に作り過ぎると必要以上に技巧的になり、具体的に作り過ぎると再利用性が損なわれるので、さじ加減がなかなか難しいですね。

[^2]: Mapを利用すると、キーと値の組を頻繁に追加したり削除したりすることが求められるシナリオでは、性能がより良くなります。
[^3]: Generatorを用いて二重ループを一重ループに変換することで、呼び出し先のロジックが読みやすくなります。

# Marble Test

FRPロジックのユニットテストは容易で、パイプラインに組み込まれた関数について入出力のテストをすれば十分です。
副作用の考慮は必要ありませんし、わざわざユニットテストを書く必要もない場合も多いでしょう。
一方、カスタムオペレーターに対してはRxJSが提供するMarble Testというユニークな仕組みを使ってテストを書くことが出来ます。

Marble DiagramというDSLで対象のObservableの振る舞いを視覚的に表現します。
簡単な例を次に示します。

```typescript
import { combineLatest } from 'rxjs';
import { TestScheduler } from 'rxjs/testing';

describe('combineLatest', () => {
  it('should be sum of the set', () => {
    const scheduler = new TestScheduler((actual, expected) =>
      expect(actual).toEqual(expected)
    );
    scheduler.run(({ hot, expectObservable }) => {
      const t1 = hot('  a----b-----c---|', { a: 1, b: 2, c: 3 });
      const t2 = hot('  -----e------f--------|', { e: 1, f: 2 });
      const t3 = hot('  -------------h-----i---------|', { h: 1, i: 2 });
      const expected = '-------------y-----z---------|';
      const combined = combineLatest(t1, t2, t3, (a, b, c) => a + b + c);
      expectObservable(combined).toBe(expected, { y: 6, z: 7 });
    });
  });
});
```

このアスキーアートのような記法を初めて見る方は面食らうと思います。
左から右にフレームという単位で仮想的な時間が流れていると考えてください。
各字には次のような意味があります。

- `アルファベット` : Observable
- `スペース` : 垂直位置を合わせて見やすくするためのフィラー
- `-` : 1フレーム経過
- `|` : ストリームの終了

これはHot Observableを利用したテストで、hot関数の第一引数にMarble Diagramが渡されています。
hot関数の第二引数に渡されているのが、各アルファベットにバインドされる値です。

他にもシンタックスはありますが、長くなるためここでは割愛します。
これで、イベント発生順の違いによる振る舞いのテストも容易です。
副作用がカスタムオペレーターに閉じられていることで、集中的に効果的なテストが可能となります。

# おわりに

ここまで読んでいただき、ありがとうございます。
FRPの雰囲気をなんとなくご理解いただけたでしょうか？

当初はステートを一切持たずにアプリケーションを実装することが本当に可能なのか、という疑問がありました。
「I/Oを分離する」というコンセプトを理解することで、FRPが机上の空論ではなく、現実的な解であることが確信に変わってきました。
まだまだ勉強中の身ですが、複雑な非同期処理を要するアプリケーションの実装において、大きな可能性を感じることが出来ました。

なんとなくRxJSを使うのではなく、FRPを意識しながら、更なる実践を繰り返して理解を深めていきたいと思います。
圏論と型の関係などについても言及したかったのですが、記事に書くほど理解が及んでいないため、次の機会にしたいと思います。

# 参考資料

https://www.learnrxjs.io

- オペレーターが機能別・重要度別に分類されていて公式リファレンスより探しやすい
- サンプルコードがわかりやすい

https://reactive.how/

- 各オペレーターの違いがビジュアルで分かりやすく表現されている
- Illustrated RxJSというサイトにもっと豊富なコンテンツがあるが、お値段が高い

https://rxjs.dev/guide/glossary-and-semantics

- 用語の定義がまとまっている公式ドキュメント

https://blog.recruit.co.jp/rmp/front-end/rxjs-intro/

- リクルートのブログ記事
- wakamshaさんがいくつかFRPの記事を書いている

https://kentutorialbook.github.io/30minLearningJavaScriptMonad/

- 前半部分のリストモナドの説明が分かりやすい
- 何となくモナドの雰囲気がJavaScriptで分かる

https://kentutorialbook.github.io/functionalprogramming2022/

- モナドの記事の作者と同じ方の著作
- 賛否両論ありますが、個人的には示唆的な内容も多かった

https://www.shoeisha.co.jp/book/detail/9784798145563

- FRPとは何か、具体的に書かれている
- Javaで書かれてあるが、Javaが分からなくてもサンプルコードは読める
- RxJSについても少しだけ書かれている

https://book.impress.co.jp/books/1115101137

- JavaScriptで関数型プログラミング
- RxJSについても少しだけ書かれている

---

MICINではメンバーを大募集しています。
「とりあえず話を聞いてみたい」でも大歓迎ですので、お気軽にご応募ください！
MICIN採用ページ：[MICIN 採用情報](https://recruit.micin.jp/)