Reduxの概念をRxJSとTypeScriptで理解する ver.2

[Redux Advent Calendar 2016](http://qiita.com/advent-calendar/2016/redux) の15日目です。ちきさんです。

## 前置きと反省

この記事は僕がQiitaに投稿した中で最もストックされてしまった [Reduxの概念をRxJSとTypeScriptで理解する(初心者向け)](http://qiita.com/ovrmrw/items/89c79fae4a2acd8159fc) をリライトしたものです。

今振り返ると前回の記事にはいくつか問題がありました。

1. ステートとReducerにPromiseを持ち込むのはやはり筋が良くなかった。
1. Actionのdispatch順を無視して非同期が解決した順に処理が流れる構造になっていた。

1に関しては当初は問題ないと思っていたのですが僕も年を取り、その考えは若気の至りだったのではないかと考えるようになった次第です。
2は要件によってはそれで良いと思うのですが、それでも基本的にはdispatch順に処理する方がより期待される挙動であろうと思った次第です。

「RxJSでReduxを書く」というテーマは今回のリライトでようやく本質に近づけたかなと思います。



---

## ここから本編

GitHubリポジトリはこちら → [ovrmrw/understanding-redux-with-rxjs-2](https://github.com/ovrmrw/understanding-redux-with-rxjs-2)
`git clone`したら`npm install`して`npm start`ですぐに動かせます。

(注: Reactの話は一切出てきません)

## Reduxとは
[Redux公式](http://redux.js.org/)
アプリケーション全体で**状態(state)を一つのJSONツリーの構造**で持ち、それをActionがキックされる度に全体更新して配信するという概念。
僕は最初FluxとかReduxとかよくわからなかったけど、色々参考にしながら自分で書いてみたらようやく理解できました。

Middlewareという概念はよくわからなかったし今でもよくわかりません。
ロガーとか便利なものもあるらしいのですがそもそも本家Reduxを動かしてみたことがない。致命的な問題としてReduxはReducerの中で非同期処理を扱えないので、それをどう扱うかでMiddleware戦争が起きているぐらいです。


## Reduxを理解する
「しかしAngular派である俺達には最初からRxJSがある。だからRxJSをフルに使ってReduxっぽいの書いちゃおーよ！」という元ネタがこちら。→["Tackling State" by Victor Savkin](https://vsavkin.com/managing-state-in-angular-2-applications-caf78d123d02#.7zwj38sy1)
(Victor SavkinはAngularチームの中心メンバーです。彼のブログは購読する価値があるでしょう。)

で、当時RxJSの理解が浅かった(JSの理解も浅かった)僕はこれを理解するのにすごく時間がかかりました。Savkin流Reduxを自分なりに何回も書き直して書き直して、そしてようやく辿り着いた…
実は一枚の簡単な短いコードで全てを説明できるということがわかったのです。

![687474703a2f2f692e696d6775722e636f6d2f4149696d5138432e6a7067.jpeg](https://qiita-image-store.s3.amazonaws.com/0/74793/6c582eff-0424-3f65-c421-8dcbc931db4d.jpeg)

以下、GitHubリポジトリにあるコードそのまま。初心者でも簡単にReduxのコンセプトを理解できるはずです。
(RxJSがよくわからないという方は[公式ドキュメント](http://reactivex.io/rxjs/manual/overview.html)を読むとわかりやすいかと思います)

```ts:rxjs-redux.ts
import 'core-js';
import 'zone.js/dist/zone-node';
import * as lodash from 'lodash';
import { Observable, Subject, BehaviorSubject } from 'rxjs';
declare const Zone: any;


///////////////////////////////// Action
class IncrementAction {
  constructor(public num: number) { }
}

class OtherAction {
  constructor() { }
}

type Action = IncrementAction | OtherAction;


///////////////////////////////// State
interface IncrementState {
  counter: number;
}

interface OtherState {
  foo: string;
  bar: number;
}

interface AppState {
  increment: IncrementState;
  other?: OtherState;
}


const initialState: AppState = {
  increment: {
    counter: 0
  }
};


///////////////////////////////// Redux
Zone.current.fork({ name: 'myZone' }).runGuarded(() => {

  console.log('zone name:', Zone.current.name); /* OUTPUT> zone name: myZone */

  const dispatcher$ = new Subject<Action | Promise<Action> | Observable<Action>>(); // Dispatcher
  const provider$ = new BehaviorSubject<AppState>(initialState); // Provider


  const dispatcherQueue$ = // Queue
    dispatcher$
      .concatMap(action => { // async actions are resolved here.
        if (action instanceof Promise || action instanceof Observable) {
          return Observable.from(action);
        } else {
          return Observable.of(action);
        }
      })
      .share();


  Observable // ReducerContainer
    .zip(...[
      dispatcherQueue$.scan((state, action) => { // Reducer
        if (action instanceof IncrementAction) {
          return { counter: state.counter + action.num };
        } else {
          return state;
        }
      }, initialState.increment),

      (increment): AppState => { // projection
        return Object.assign<{}, AppState, {}>({}, initialState, { increment }); // always create new state object!
      }
    ])
    .subscribe(newState => {
      provider$.next(newState);
    });


  provider$
    .map(appState => appState.increment)
    .distinctUntilChanged((oldValue, newValue) => lodash.isEqual(oldValue, newValue)) // restrict same values to pass through.
    .subscribe(state => {
      console.log('counter:', state.counter); /* (First time) OUTPUT> counter: 0 */
    });


  /* 
    OUTPUT: 0 -> 1 -> 2 -> 4 -> 3 
    outputs are not determined by async resolution order but by action dispatched order.
  */
  dispatcher$.next(promiseAction(new IncrementAction(1), 100));  /* OUTPUT> counter: 1 */
  dispatcher$.next(promiseAction(new IncrementAction(1), 50));  /* OUTPUT> counter: 2 */
  dispatcher$.next(observableAction(new IncrementAction(0), 100));  /* OUTPUT> (restricted) */
  dispatcher$.next(observableAction(new IncrementAction(2), 50));  /* OUTPUT> counter: 4 */
  dispatcher$.next(new IncrementAction(-1)); /* OUTPUT> counter: 3 */
});



///////////////////////////////// Helper
function promiseAction(action: Action, timeout: number): Promise<Action> {
  return new Promise<Action>(resolve => {
    setTimeout(() => resolve(action), timeout);
  });
}

function observableAction(action: Action, timeout: number): Observable<Action> {
  return Observable.of(action).delay(timeout);
}
```

どうでしょうか。超簡単じゃないですか？

ちなみに今回のコードで[zone.js](https://github.com/angular/zone.js/)を使う必要はありませんでしたが、後々Angular上で動かす予定のあるコードはなるべくNode.js環境でも`Zone`を使って書いた方がいいです。
そうしないとNode.js環境で動いたコードがAngularで動かないとか、そういう現象に遭遇することがあるからです。(僕はただのAngular好きです)

## 要点1 - Subject
```ts
  dispatcher$.next(promiseAction(new IncrementAction(1), 100));
```
これがActionの起点です。ちなみに`dispatcher$`は`Subject`のインスタンスです。
この次にストリームがどこに流れているかわかるでしょうか。

正解は`dispatcherQueue$`です。

## 要点2 - concatMap
```ts
  const dispatcherQueue$ = // Queue
    dispatcher$
      .concatMap(action => { // async actions are resolved here.
        if (action instanceof Promise || action instanceof Observable) {
          return Observable.from(action);
        } else {
          return Observable.of(action);
        }
      })
      .share();
```

`dispatcher$`から流れてきたActionを`concatMap`オペレーターで受けています。何をしているかというと、

- PromiseかObservable、つまり非同期だったら`Obervable.from()`で非同期を解決させて返す。
- それ以外、つまり同期だったら`Observable.of()`でただ単にObservableに変換して返す。

これにより`Observable<Action | Promise<Action> | Observable<Action>>`だったActionの型は`Observable<Action>`に統一されます。

しかも`concatMap`の効能によりActionのdispatch順が遵守されます。素敵です。
このReducerの手前で非同期を解決させようというアプローチはredux-observableにも通じるものがあります。

<img width=640 src="http://reactivex.io/documentation/operators/images/concatMap.png">

## 要点3 - BehaviorSubject
```ts
const provider$ = new BehaviorSubject<AppState>(initialState);
```
ここは`Subject`ではなく`BehaviorSubject`であることに意味があります。もしこれを`Subject`に変えると最初の`"counter: 0"`が出力されなくなります。
初期値がすぐに配信される様子はマーブルダイアグラムで見るとわかりやすいかも知れません。
[Subjectのドキュメント](http://reactivex.io/documentation/subject.html)
<img width="640" alt="S.BehaviorSubject.png" src="https://qiita-image-store.s3.amazonaws.com/0/74793/ddbaa374-6810-129b-5ba1-d1162bd6902d.png">


## 要点4 - scan
```ts
      dispatcherQueue$.scan((state, action) => { // Reducer
        if (action instanceof IncrementAction) {
          return { counter: state.counter + action.num };
        } else {
          return state;
        }
      }, initialState.increment)
```
いわゆるReducer。
`scan`オペレーターはStore(Reducer)を構築する要です。これは時間をまたぐreduceであると理解できれば、それで大体OKかなと思います。この`scan`と事項の`zip`をしっかり理解できるかどうかが本記事理解の分かれ目となります。
[scanのマーブルダイアグラム](http://rxmarbles.com/#scan)

`dispatcher$.scan()`ではなく`dispatcherQueue$.scan()`であることが重要です。

## 要点5 - zip, projection
```ts
      (increment): AppState => { // projection
        return Object.assign<{}, AppState, {}>({}, initialState, { increment });
      }
```
`zip`オペレーターの最後の引数にprojectionと呼ばれる関数を入れて戻り値を整えます。
ちなみに`zip`の中がいくつもあるときはこのように書きます。

```ts
    .zip<AppState>(...[
      dispatcher$.scan(/* 省略 */), // state1
      dispatcher$.scan(/* 省略 */), // state2
      dispatcher$.scan(/* 省略 */), // state3

      (state1, state2, state3): AppState => { // projection
        return Object.assign<{}, AppState, {}>({}, initialState, { state1, state2, state3 });
      }
    ])
```
[zipのマーブルダイアグラム](http://rxmarbles.com/#zip)
`zip`と似たものに`combineLatest`というオペレーターがありますが、`zip`は内包する全てのObservableのnextを待つのに対し、`combineLatest`はどれかがnextする度に各々のObservableの最新の値を次に流します。
このためReduxの概念には`zip`が適しているということになります。
[combineLatestのマーブルダイアグラム](http://rxmarbles.com/#combineLatest)

## 要点6 - distinctUntilChanged
```ts
    .distinctUntilChanged((oldValue, newValue) => lodash.isEqual(oldValue, newValue))
```
`distinctUntilChanged`オペレーターは通過するストリームが同じ値の場合に抑制する機能です。
ただし今回のコードは前述のprojectionのところで

```ts
return Object.assign<{}, AppState, {}>({}, initialState, { increment });
```
こうしているため、単純に`.distinctUntilChanged()`と書いてしまうと流れてきたデータが`{counter: 2}`→`{counter: 2}`のように同じ値が続いたとしても両方通過します。これはオブジェクトの中身を見ていないからです。
オブジェクトの中身でチェックするためにはいわゆるdeepEqualの比較をしなければいけないため、comparerと呼ばれる関数を下記のように書く必要があります。

```ts
(oldValue, newValue) => lodash.isEqual(oldValue, newValue)
```
[distinctUntilChangedのドキュメント](http://reactivex.io/documentation/operators/distinct.html)
[lodash.isEqualのドキュメント](https://lodash.com/docs/#isEqual)

## Angularで使うには
DispatcherをDIコンテナに入れてComponentにインジェクトすれば、ComponentからActionをキックできるようになります。
またProviderのストリームの最後の方をなんとかしてComponentにインジェクトすれば、Componentで更新されたステートを取得できます。

ちなみに [Firebase-as-a-Store ～RxJSで作るFirebaseバックエンドのRedux～](http://qiita.com/ovrmrw/items/27d06475f405fd4ca9b4) で紹介しているAngularアプリはこの概念をそっくりそのまま使って構築しています。

## まとめ
要点はいくつかありますが、最も重要なのはReducerを構成する`scan`,`zip`オペレーターの部分です。
Reducerを増やしたければ`zip`オペレーターの中にどんどん`dispatcherQueue$.scan`を追加していきましょう。やることはそれだけです。