RxJSを用いた実装パターンの実例まとめ(前編)

小飼です。
今夏弊社では、クライアントサイドでの状態遷移・保持にReactiveProgrammingの考え方を用いたアプリケーションを制作していました。
その過程で見つけた、いくつかの実際的なストリーム作りのパターンをまとめておきます。
ReactiveProgrammingの(考え方を含めた)入門記事や、パラダイムの骨子を詳細に解説した記事は沢山あると思うのですが、実際のアプリケーションでどう使うのか?あるいはどう使ったのか?という辺りの情報はまだ少ないようにも思うので、その辺りの参考になればうれしいです。

DOM要素からストリームを作成する

まず基本となる、ユーザのUI操作からストリームを作成する方法です。
これは入門系の記事でも扱われている事が多いですが、ストリームの生成時にSubjectを作成して、直接DOM要素に渡す手法を導入として書いておきます。

import * as React from "react";
import { Observable, Subject } from "rxjs";

const { Component } = React;

interface RootState {
  greeting: string;
  onClick: (event: MouseEvent) => void;
}

// ストリームを生成する時に、Subjectのインスタンスを作成して、
// イベントリスナー(onClick関数)ごとStateに流してしまう
// Subjectは生成したものを注入するようにしておくと、テストしやすい
const root$ = (subject = new Subject<string>()): Observable<RootState> => {
  const onClick = (event: React.MouseEvent<HTMLDivElement>): void => subject.next("Hello, world");
  const greeting$ = subject.map(payload => `${payload}!!!`);
  const initialState = {
    greeting: "",
    onClick,
  };

  return subject
    .map(payload => payload + "!!!")
    .map(greeting => ({ greeting, onClick }))
    .startWith(initialState)
    ;
};


export class SampleApp extends Component<void, RootState> {
  componentWillMount() {
    root$().subscribe(root => this.setState(root));
  }

  render() {
    const { greeting, onClick } = this.state;
    return (
      <div onClick={ onClick }>{ greeting }</div>
    );
  }
}

この手法では、ストリームにイベントを作成する起点となるDOM要素ごとにSubjectを作成しています。
DOM要素ごとにストリームを作成するので構造としてはわかりやすいですが、起点となるDOM要素ごと(=毎ユーザがクリックできる要素と同じだけ)Subjectを作成することになるので、やや冗長な気もしてきます。
また、ルートとなる状態のストリームObservable<RootState>の中に、DOMに束縛される関数(RootState.onClick)も同梱して流してしまうので、実装が大きくなっていくに従って余計な荷物が一緒に流れているようにも思えてきます。
そこで、次項では単一のSubjectを使い回す手法をご紹介します。

単一のSubjectを使い回す

前項で見たように、イベントを発行するDOM要素ごとにSubjectを生成するのはやや冗長なきらいがあります。
例えばReduxでは、単一のdispatch関数を全てのDOM要素に紐付けて、dispatch関数の引数に『どういった操作が行われたのか』という情報(=ActionType)を載せることで、操作された要素を判別する、というアプローチになっています。

このアプローチを導入するとすると、前項の方法とは『イベント毎に別々のSubjectを用いるのか？』『単一のSubjectを荷物の種類によって分岐するのか』が違いますが、『ユーザが何をしたか』『ユーザがしたことによって生じるデータを観測できるか』は満たされたままですので、大きく構造を変えずに前項と同様の実装を作成できそうです。

import * as React from "react";
import { Observable, Subject } from "rxjs";

const { Component } = React;
const ACTION_ON_CLICK = "ACTION_ON_CLICK";

interface RootState {
  greeting: string;
}

interface Action<T> {
  type: string;
  payload: T;
}

const subject = new Subject<Action<any>>();

const onClick = (event: React.MouseEvent<HTMLDivElement>): void => subject.next({
  type: ACTION_ON_CLICK,
  payload: "Hello, world",
});

const root$ = (): Observable<RootState> => {
  const initialState = { greeting: "" };

  return subject
    .filter(({ type }) => type === ACTION_ON_CLICK)
    .map(({ payload }) => `${payload}!!!`)
    .map(greeting => ({ greeting }))
    .startWith(initialState)
    ;
};


export class SampleApp extends Component<void, RootState> {
  componentWillMount() {
    root$().subscribe(root => this.setState(root));
  }

  render() {
    const { greeting } = this.state;
    return (
      <div onClick={ onClick }>{ greeting }</div>
    );
  }
}

subjectがdispatcherを代替するようなアプローチです。
RxJSをはじめとするRP系ライブラリをラップするライブラリ(例えばredux-observable)では、これに近いアプローチが取られています。

このアプローチでは、.filter(({ type }) => type === ACTION_ON_CLICK)部分が頻出するはずですので、専用のオペレータがあると便利です。
前述のredux-observableでも同様のカスタムオペレータが実装されていますが、便利なオペレータなのでその部分の実装を抽出したパッケージを作りました。

// ofTypeオペレータを使うとこんな感じ

import "of-type-operator"
const root$ = (): Observable<RootState> => {
  const initialState = { greeting: "" };

  return subject
    .ofType<string>(ACTION_ON_CLICK)
    .map(payload => `${payload}!!!`)
    .startWith(initialState)
    ;
};

個人的にはこのアプローチだけで、ある程度の規模感のアプリケーションを構築可能だと考えています。

クリックしたらリクエスト

次に、クライアントサイドアプリケーションでは頻出のAPIサーバとの通信の実装をご紹介します。
特に複雑なことはしていませんが、入れ子になったストリーム(Observable<Observable>)をフラットなストリーム(Observable)に均すところで引っ掛かるかも知れません。

const { ajax } = Observable;
interface Response {
  greeting: string;
}

// ...

const root$ = (): Observable<RootState> => {
  const initialState = { greeting: "" };

  return subject
    .ofType<string>(ACTION_ON_CLICK)

    // mergeMapオペレータで入れ子になったストリームを均しています
    .mergeMap<Response>(payload => ajax.getJSON(`/path/to/endpoint?parameter=${payload}`))

    // 以下の２つのオペレータの組み合わせと等価です
    // .map(payload => ajax.getJSON(`/path/to/endpoint?parameter=${payload}`)) (= Observable<Observable<Response>>)
    // .mergeAll() (= Observable<Response>)

    .startWith(initialState)
    ;
};

条件が満たされたらリクエスト

前項の『クリックしたらリクエスト』の応用で、『条件が満たされた時にクリックされたらリクエストを発行する』ようなストリームを作ります。
これはwithLatestFromで、きっかけとなるストリーム(クリックのストリーム)と、条件となるストリーム（条件入力のストリーム）を合流することで実現できます。

// ...
const onClick = (event: React.MouseEvent<HTMLDivElement>): void => subject.next({
  type: ACTION_ON_CLICK,
});

const onInput = (event: React.FormEvent<HTMLInputElement>): void => subject.next({
  type: ACTION_ON_INPUT,
  payload: event.currentTarget.value,
});

const root$ = (): Observable<RootState> => {
  const initialState = { greeting: "" };

  const click$ = subject.ofType<void>(ACTION_ON_CLICK);
  const input$ = subject.ofType<string>(ACTION_ON_INPUT);

  // http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-withLatestFrom
  // 『クリックイベントが発行された時の、最後の条件入力ストリームのイベント群』を取りまとめられる

  return click$
    .withLatestFrom(input$, (clicked, input) => input)
    .mergeMap<Response>(input => ajax.getJSON(`/path/to/endpoint?parameter=${input}`))
    .startWith(initialState)
    ;
};


export class SampleApp extends Component<void, RootState> {
  // ...
  render() {
    const { greeting } = this.state;
    return (
      <div>
        <input onChange={ onInput } type="text" />
        <div onClick={ onClick }>{ greeting }</div>
      </div>
    );
  }
}

よく似たオペレータにcombineLatestがありますが、こちらは合成対象のストリーム(クリックのストリームと条件入力のストリーム)のいずれかでイベントが起こった時にイベントを発行するオペレータですので、この目的には適しません。(条件入力のたびにリクエストが発行されるような仕様になってしまいます)

と、ちょっと記事が長くなってきたので、ここで一旦区切ろうと思います。
後編では

• ドラッグストリームの作成
• 同期的なイベントのフィルタ
• ストリームの配列を配列のストリームに変換する

辺りを扱う予定です。