RxJS の Observable で Promise を扱う

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

# はじめに

今日は Promise と Obervable とを相互変換する方法について書きます。

# Observable と Promise

Promise は ES2015 で標準化され、各種 Web browser で実装されて、 JavaScript の非同期処理の代表的なものになっています。

RxJS の Observable は Promise からの変換や Promise への変換に対応しています。 Promise の扱いについて見ていきます。

## Promise to Observable

### Observable.fromPromise

- [`Observable.fromPromise` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/frompromise.md)
- [`Observable.fromPromise` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/frompromise.js)

`Promise` から `Observable` に変換します。

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

Observable
  .fromPromise(Promise.resolve(42))
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );

// onNext: 42
// onCompleted
```

`onNext` のあと `onCompleted` が流れてきます。

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

const promise = new Promise((resolve) => {
  setTimeout(() => resolve(100), 1000);
});

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

// onNext: 100
// onCompleted
```

`setTimeout` をはさんでも動きます。

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

const promise = new Promise((_, reject) => {
  setTimeout(() => reject(new Error('ERROR!')), 1000);
});

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

// onError: Error: ERROR!
```

`Error` も期待通りに動きます。

### Observable.prototype.flatMap / Observable.prototype.mergeAll

- [`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)
- [`FlatMapObservable` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/flatmapbase.js)
- [`Observable.prototype.mergeAll` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/mergeall.md)
- [`Observable.prototype.mergeAll` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/mergeall.js)

`flatMap` や `mergeAll` などの `Observable` を flat にする動作は `Observable` だけでなく `Promise` にも対応しています。`Observable` ではなく `Promise` が返された場合には、それを `Observable.fromPromise` で変換して `Observable` として処理してくれます。

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

Observable
  .from([1, 2, 3])
  // .map(x => Observable.fromPromise(Promise.resolve(x * 10)))
  .map(x => Promise.resolve(x * 10)) 
  .mergeAll()
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );

// onNext: 10
// onNext: 20
// onNext: 30
// onCompleted
```

URL を `fetch` すると `Promise` を返すことが多いので、この挙動はよく使います。

```js
import fetch from 'node-fetch';
import { Observable } from 'rx';

Observable
  .of('http://example.com')
  // fetch(url) returns Promise<Response>
  .flatMap(url => fetch(url))
  // response.text() returns Promise<String>
  .flatMap(response => response.text())
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );

// onNext: <!doctype html> ...
// onCompleted
```

### Observable.prototype.toPromise

- [`Observable.prototype.toPromise` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/topromise.md)
- [`Observable.prototype.toPromise` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/master/src/core/linq/observable/topromise.js)

`Observable` から `Promise` に変換します。`Observable` の `onCompleted` の際に最後の `onNext` の値が流れてきます。

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

Observable
  .from([1, 2, 3])
  .toPromise()
  .then(value => console.log(`fulfilled: ${value}`));
// fulfilled: 3
```

# おわりに

RxJS の Observable は Promise からの変換や Promise への変換を確認しました。

こういう単純な情報は意外と書かれていない気がしたので、改めて書いてみました。