RxJS の Operators (7) - Observable Utility Operators (1)

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

# はじめに

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

## `Observable.prototype.subscribe`

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

リスト上は `subscribe` が挙げられていますが、もう特に書くことはないので割愛します。

## `Observable.prototype.subscribeOn`

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

指定した `scheduler` に従って値を流すようにします。

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

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

ただ `immediate` (同期的) -> `default` (非同期) はともかく、当然、その逆はできません。一度 `default` にすると `immediate` に戻すこともできません。

またドキュメントでは `Scheduler.timeout` になっていますが、このメソッドは既に削除されており `Scheduler.default` に名前が変更されています。(`Scheduler.default` がデフォルトじゃない `Scheduler.immediate` なのもよく分かりません)

## `Observable.prototype.do`

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

`do` は副作用を前提としたオペレーター。 `do` の戻り値に関係なく流れてきた値は次に流れます。 `subscribe` と同様のシグネチャを持つので、`Observer` を指定せずに `onNext` などを直接指定できます。

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

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

デバッグの際に流れる値を確認するのに便利かもしれません。

## `Observable.prototype.delay`

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

`delay` は遅らせて値を流します。絶対時間 (`Date`) や相対時間 (ms) や `Observable` での指定などができます。

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

Observable
  .from([1, 2, 3])
  .delay(1000)
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log(`e ${new Date().getTime()}`)
  );
console.log(`s ${new Date().getTime()}`);
// s 1449639923044
// onNext: 1
// onNext: 2
// onNext: 3
// e 1449639924038
```

当然、非同期になります。

# おわりに

今日は `Observable` のユーティリティを見ました。まだ途中なので続きは明日にします。