RxJS の Operators (4) - Observable の結合 (1)

この記事は [bouzuya's RxJS Advent Calendar 2015](http://www.adventar.org/calendars/1200) の 6 日目かつ RxJS Advent Calendar 2015 の 6 日目です。

# はじめに

今日は [ReactiveX の Operators の Combining](http://reactivex.io/documentation/operators.html#combining) について [RxJS の API ドキュメント](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/observable.md)やソースコードを見ていきます。

また [RxJS 4.0.7](https://github.com/Reactive-Extensions/RxJS/tree/v4.0.7) を対象にしています。

# Observable の結合 (1)

## `Observable.merge`

- [ReactiveX - Merge operator](http://reactivex.io/documentation/operators/merge.html)
- [`Observable.merge` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/merge.md)
- [`Observable.merge` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/linq/observable/merge.js)

可変長引数で `observable` を取ります。各 `Observable` に値が流れてくるたびに、それをそのまま流します。

```js
import { Observable } from 'rx';

Observable
  .merge(
    Observable.from([1, 2, 3]),
    Observable.from(['A', 'B'])
  )
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 1
// onNext: A
// onNext: 2
// onNext: B
// onNext: 3
// onCompleted
```

ちなみに名前の似ている [`Observable.prototype.mergeAll`](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/mergeall.md) は動きが違います。

ただ、`Observable.merge` のソースコードを見ると、実装は `Observable.from([...]).mergeAll()` だと分かります (実際には `Observable.from` ではなく `Observable.ofWithScheduler(Scheduler.immediate, ...)` なのですが、「Observable の生成」で `of` に触れておらず、Scheduler の説明もまだなので仮に `Observable.from` で説明します) 。

つまり、次のコードはほとんど同じです (厳密には異なります)。

```js
Observable
  .merge(
    Observable.from([1, 2, 3]),
    Observable.from(['A', 'B'])
  )
```

```js
Observable
  .from([
    Observable.from([1, 2, 3]),
    Observable.from(['A', 'B'])
  ])
  .mergeAll()
```

## `Observable.prototype.merge`

- [ReactiveX - Merge operator](http://reactivex.io/documentation/operators/merge.html)
- [`Observable.prototype.merge` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/mergeproto.md)
- [`Observable.prototype.merge` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/mergeconcat.js)

引数が `Observable` である (実装は `number` でない) ときは `Observable.merge(...)` と同じで、引数が `number` のときは並行性の最大を制限します。

引数によって挙動が大幅に違うあたりが RxJS らしいと感じます。`merge` の引数の違いについては別で Qiita 記事を書きました。『[RxJS 4 の merge の引数の差異 (concurrent 指定) と RxJS 5 での実装 - Qiita](http://qiita.com/bouzuya/items/6d0730c126bf842feec7)』

以下は等価のコードを並べたものです。

```js
import { Observable, Scheduler } from 'rx';

const source1 = Observable
  .ofWithScheduler(
    Scheduler.immediate,
    Observable.from([1, 2, 3]),
    Observable.from(['A', 'B'])
  )
  .mergeAll();
const source2 = Observable
  .merge(
    Observable.from([1, 2, 3]),
    Observable.from(['A', 'B'])
  );
const source3 = Observable
  .from([1, 2, 3])
  .merge(Observable.from(['A', 'B']));

const subscribe = (source) =>
  source.subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );

subscribe(source1);
subscribe(source2);
subscribe(source3);
// onNext: 1
// onNext: A
// onNext: 2
// onNext: B
// onNext: 3
// onCompleted
// onNext: 1
// onNext: A
// onNext: 2
// onNext: B
// onNext: 3
// onCompleted
// onNext: 1
// onNext: A
// onNext: 2
// onNext: B
// onNext: 3
// onCompleted
```

## `Observable.prototype.startWith`

- [ReactiveX - StartWith operator](http://reactivex.io/documentation/operators/startwith.html)
- [`Observable.prototype.startWith` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/startwith.md)
- [`Observable.prototype.startWith` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/linq/observable/startwith.js)

与えた引数を流してから、その後に現在の observable の値を流すような動きです。`Observable.prototype.startWith` は prepend で、 `Observable.prototype.concat` は append のイメージですね。実装は `Observable.from([..., this]).concat(...)` を期待したのですが、`Enumerable.of(...).concat()` のようです。

```js
import { Observable } from 'rx';

Observable
  .from([1, 2, 3])
  .startWith(0)
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 0
// onNext: 1
// onNext: 2
// onNext: 3
// onCompleted
```

後述の `combineLatest` のために必ず値を流すようにしたい場合などにも使えます。

## `Observable.combineLatest`

- [ReactiveX - CombineLatest operator](http://reactivex.io/documentation/operators/combinelatest.html)
- [`Observable.combineLatest` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/combinelatest.md)
- [`Observable.combineLatest` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/combinelatest.js)

可変長引数で `observable` を取り、最後の引数で `resultSelector` を取ります。各 `Observable` に値が流れてくるたびに、各 `Observable` の直近の値が入った `Array` を `resultSelector` に渡してくれて、その戻り値が次へと流れる `Observable` を返します。

この日本語の説明ではきっと「何を言っているのか わからねーと思うが」状態だと思うのですが、ぼくもこれで誤読ないように説明できている自信がまったくありません。これに限らず Rx のメソッドの中でも複雑なものは言葉で説明してもよく分からないものが多いのでソースコードを読むか試すかしたほうがいいです。

ちなみに `combineLatest` に限れば [ReactiveX の combineLatest の marble diagram](http://reactivex.io/documentation/operators/combinelatest.html) を見ると、とても分かりやすいとぼくは思います。

```js
import { Observable } from 'rx';

// -0-1-2-3->
const observable1 = Observable
  .interval(100)
  .take(4)
  .map((_, index) => index);
// --A---B-->
const observable2 = Observable
  .interval(150)
  .take(2)
  .map((_, index) => 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'[index]);

// -0-1-2-3->
// --A---B-->
// --xx-xxx->
Observable
  .combineLatest(
    observable1,
    observable2,
    (value1, value2) => value1 + value2
  )
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 0A
// onNext: 1A
// onNext: 2A
// onNext: 2B
// onNext: 3B
// onCompleted
```

注意しておきたいのが各 `Observable` の最初の値が流れてこないと、次へ値を流さないという点。直近の値が出揃ってはじめて次に流れるということです。何も流さない `Observable.never` を使ってみるとこんな風になります。

```js
import { Observable } from 'rx';

const observable1 = Observable.never();
const observable2 = Observable
  .interval(150)
  .take(2)
  .map((_, index) => 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'[index]);

Observable
  .combineLatest(
    observable1,
    observable2,
    (value1, value2) => value1 + value2
  )
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// (no output)
```

## `Observable.zip`

- [ReactiveX - Zip operator](http://reactivex.io/documentation/operators/zip.html)
- [`Observable.zip` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/zip.md)
- [`Observable.zip` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/linq/observable/zip.js)

`Observable.zip` は zip です (適当) 。複数の `Observable` で流れた要素をペアにして流します。`combineLatest` は各 `Observable` に値が流れるたびに値を流したのに対して、`zip` はすべての `Observable` が値を流すまで値を流しません。`combineLatest` は最初の値で各 `Observable` の値が出揃うのを待つように、`zip` は各値で各 `Observable` の値が出揃うのを待つようです。

また実装としては、最初の引数 (`Observable`) の `Observable.prototype.zip` に残りの引数を渡しているだけです。

```js
import { Observable } from 'rx';

Observable
  .zip(
    Observable.from([0, 1, 2]),
    Observable.from([3, 4, 5]),
    Observable.from([6, 7, 8])
  )
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 0,3,6
// onNext: 1,4,7
// onNext: 2,5,8
// onCompleted
```

## `Observable.prototype.zip`

- [ReactiveX - Zip operator](http://reactivex.io/documentation/operators/zip.html)
- [`Observable.prototype.zip` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/zipproto.md)
- [`Observable.prototype.zip` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/zip.js)

前述のとおり、`Observable.zip` とほとんど同じ挙動なので割愛します。

## `Observable.prototype.switch`

- [ReactiveX - Switch operator](http://reactivex.io/documentation/operators/switch.html)
- [`Observable.prototype.switch` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/switch.md)
- [`Observable.prototype.switch` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/switch.js)

`Observable.prototype.switch` は `mergeAll` (`flatMap`) のように `Observable` の `Observalbe` を flat にします。`mergeAll` のようにすべての `Observable` の値を merge するのとは違い、次の `Observable` へ次々と switch していきます。例を示します。

```js
import { Observable } from 'rx';

Observable
  .from([
    Observable.from([0, 1, 2]),
    Observable.from([3, 4, 5]),
    Observable.from([6, 7, 8])
  ])
  .switch()
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 0
// onNext: 3
// onNext: 6
// onNext: 7
// onNext: 8
// onCompleted
```

もちろん 2 個目の Observable のタイミングを遅らせると `[0, 1, 2]` がすべて流れてきます。

ちなみに `switch` の公式の例は相当にわかりづらいので腹が立ちます。

# おわりに

今日は `Observable` を結合する Operator を見てみました。まだリスト的には終わっていないのですが、時間の都合でここで区切ります。