RxJS の Operators (2) - Observable の変換

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

# はじめに

今日は [ReactiveX の Operators の Transform](http://reactivex.io/documentation/operators.html#transforming) について [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 の変換

## `Observable.prototype.map`

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

`Array.prototype.map` でもおなじみのものです。動きについては説明は不要でしょう。

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

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

せっかくなのでソースコードを眺めてみましょう。

```js
  observableProto.select = observableProto.map = function (selector, thisArg) {
    var selectorFn = isFunction(selector) ? bindCallback(selector, thisArg, 3) : function () { return selector; },
        source = this;
    return new AnonymousObservable(function (o) {
      var i = 0;
      return source.subscribe(function (x) {
        try {
          var result = selector(x, i++, source);
        } catch (e) {
          return observer.onError(e);
        }

        o.onNext(result);
      }, function (e) { o.onError(e); }, function () { o.onCompleted(); });
    }, source);
  };
```

[`AnonymousObservable`](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/anonymousobservable.js) という `Observable` を継承したクラスの、インスタンスを返しています。

ところで `selectorFn` が使われていないですね。バグっぽいです。→ このソースコードはドキュメントからリンクが貼られている [`/src/core/linq/observable/select.js`](https://github.com/Reactive-Extensions/RxJS/blob/master/src/core/linq/observable/select.js) ですが、ビルドスクリプトを確認すれば分かりますが、このソースコードはもう使われていません (おまけにバグがあります) 。現在は [`/src/core/perf/operators/map.js`](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/map.js) が使われているようです。

Pull Request 送ろうとしたところで気づいたのですが、このまま記録しておきます。

いきなり脱線しましたが、続けます。

## `Observable.prototype.scan`

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

`Observable.prototype.scan` は `Array.prototype.reduce` のように fold っぽい動きをします。ただ `Observable.prototype.reduce` とは違って途中の値も `onNext` で流します。

ちょうどいい記事があったので、こちらをどうぞ [Rxにおけるscanとreduceの違い - Qiita](http://qiita.com/izumin5210/items/952a23bf065648feabb7) 。

個人的な使用頻度としては `reduce` より `scan` のほうが高いです。

## `Observable.prototype.flatMap`

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

`Observable.prototype.flatMap` は `map` したものを flat にします。flat という名前からは入れ子になった Array を flat にするイメージを浮かべるでしょう。例えば `[[1, 2, 3], [4, 5, 6]]` が `[1, 2, 3, 4, 5, 6]` になるイメージです。これを Rx という前提で考えると `Observable.from([Observable.from([1, 2, 3]), Observable.from([4, 5, 6])])` は `Observable.from([1, 2, 3, 4, 5, 6])` になるでしょう。イメージはつかめたはずです。……あとは例を見てみましょう。

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

Observable
  .from([1, 2, 3])
  .flatMap(value => Observable.just(value * 2))
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log(`onCompleted`)
  );
// onNext: 2
// onNext: 4
// onNext: 6
// onCompleted
```

`flatMap` で返した `Observable` から流れてくる値が流れてきているのが分かります。この例なら `flatMap` である必要はまったくないのですが、例えば、これが非同期処理ならどうでしょう。複数の HTTP リクエストの戻り値をひとつの `Observable` にまとめることができるわけです。

また、ソースコードを見れば分かるのですが、`flatMap` は簡単に書くと次のようなものです。

```js
class Observable {
  // ...
  flatMap(selector) {
    return this.source.map(selector).mergeAll();
  }
}
```

`map` と `mergeAll` とを組み合わせたものです。 `Observable.prototype.mergeAll` はまた別の機会に紹介しますが、複数の `Observable` を merge ( flat に ) する、そんな動きです。

個人的には、この `flatMap` は `map` や `scan` と並んでよく使う Operator です。

## `Observable.prototype.groupBy`

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

`Observable.prototype.groupBy` は流れてくる値を `keySelector` でグループ分けした Observable に変換します。

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

Observable
  .from([1, 2, 3, 4, 5, 6])
  .groupBy(value => value % 2 == 0)
  .flatMap(observable => observable.toArray()) 
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log(`onCompleted`)
  );
// onNext: 1,3,5
// onNext: 2,4,6
// onCompleted
```

`keySelector` で偶数 / 奇数 (`true` / `false`) に分けています。`flatMap` (さっそく使っています) で `Observable` の `Observable` (`Observable<Observable<number>>` のイメージ) を、 `Array` の `Observable` (`Observable<Array<number>>` のイメージ) に変換してまとめています。

## `Observable.prototype.window`

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

`Observable.prototype.window` は `Observable.prototype.groupBy` と同様にグループ化して `Observable` の `Observable` を返すのですが、`groupBy` とは異なり流れてくる値以外でグループ化します。

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

Observable
  // 100 ms ごとに値を流す (延々と続く)
  .interval(100)
  // return した Observable が値を流すタイミング (300ms) で window を区切る
  .window(() => Observable.timer(300))
  // 延々と続くと困るので最初の 3 回を取る
  .take(3)
  .flatMap(observable => observable.toArray()) 
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log(`onCompleted`)
  );
// onNext: 0,1,2
// onNext: 3,4,5,6
// onNext: 7,8,9,10
// onCompleted
```

引数によって動きがかなり違うのでドキュメントを参照してください。

## `Observable.prototype.buffer`

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

`Observable.prototype.buffer` は `Observable.prototype.window` と似たような引数を取ってグループ化しますが、`Observable` の `Observable` ではなく、 `Array` の `Observable` を返します。

サンプルを見るよりもソースコードを読んだ方が分かりやすいと思います。

```js
  observableProto.buffer = function () {
    return this.window.apply(this, arguments)
      .flatMap(toArray);
  };
```

`window` して `flatMap(toArray)` しています。つまり、さきほどの `window` の例のように、区切ってまとめた値をほしい場合は `buffer` を使えばいいわけです。さきほどの例だと延々続いてしまうのでアレですが……。

# おわりに

今日は `Observable` を変換する Operator を見てみました。`map` / `scan` / `flatMap` は特によく使うものなので最初に覚えると良さそうです。