RxJS 6: Observableで非同期の処理を行う

[RxJS](https://rxjs-dev.firebaseapp.com)は`Observable`オブジェクトを軸にしたリアクティブプログラミングのJavaScriptライブラリです。非同期の処理やイベントにもとづくコードが簡単に組み立てられます。本稿では、RxJSの`Observable`でweb APIのデータを読み込んでみます。なお、RxJSについて、詳しくは「[RxJS 6入門](http://www.fumiononaka.com/Business/index.html#RxJS_6)」01〜07をお読みください。

#01 RxJSライブラリをCDNで読み込む
RxJSを手っ取り早く試すには、CDNから読み込むのが手軽でしょう。HTMLドキュメントの`<head>`要素に、`<script>`要素をつぎのように加えます。本稿執筆時の最新リリースバージョンはRxJS 6.2.2です。

```html
<script src="https://unpkg.com/rxjs/bundles/rxjs.umd.min.js"></script>
```

npmなどを用いたそのほかのインストールの仕方についてはGitHubのRxJS 6「[Installation and Usage](https://github.com/ReactiveX/rxjs/tree/stable#installation-and-usage)」または「[Installation](https://rxjs-dev.firebaseapp.com/guide/installation)」をご覧ください。ただし、後者のCDNはバージョン5.0.0のままリンクが切れているようです。

#02 簡単な非同期の処理にObservableを使ってみる
まずは、`Observable`を使わない非同期の処理です。つぎの関数(async())に、[`setTimeout()`メソッド](https://developer.mozilla.org/docs/Web/API/WindowTimers/setTimeout)で時間待ちの処理を定めました。関数の引数に渡されたミリ秒ののちに、`console.log()`メソッドで文字列が出力されます。

```js
function async(arg) {
	console.log('start');
	setTimeout(() =>
		console.log(`${Math.floor(arg / 1000)} second delay is up`)
	, arg);
}
async(1000);

// コンソール出力
start
1 second delay is up
```

RxJSの処理は[`Observable`オブジェクト](http://fumiononaka.com/Business/html5/FN1805014.html)をつくることから始まります。そのためのメソッドは[`Observable.create()`](https://rxjs-dev.firebaseapp.com/api/index/Observable#create)です。渡す関数には、データをどう処理して送るのか定めます。データを送るメソッドが、関数の引数([`Observer`オブジェクト](http://fumiononaka.com/Business/html5/FN1806001.html))に対して呼び出す[`Observer.next()`](https://rxjs-dev.firebaseapp.com/api/index/Observer#next)です。

`Observable`の処理を実行するには、[`Observable.subscribe()`メソッド](https://rxjs-dev.firebaseapp.com/api/index/Observable#subscribe)を呼び出さなければなりません。引数に渡す関数が、`Observer.next()`により送られてくるデータにどう処理を加えるのか定めます。

```js
const {Observable} = rxjs;
function async(arg) {
    return Observable.create((observer) => {
        console.log('start');
        setTimeout(() => {
            observer.next(arg);
        }, arg);
    });
}
async(1000)
.subscribe(
    (arg) => console.log(`${Math.floor(arg / 1000)} second delay is up`)
);
```

```console
// コンソール出力
start
1 second delay is up
```

#03 XMLHttpRequestとObservableでweb APIのデータを読み込む
つぎは、webで公開されているAPIを`XMLHttpRequest`オブジェクトで読み込んでみます。APIとして用いるのは、「[Open Notify](http://open-notify.org/)」の「[How Many People Are In Space Right Now](http://open-notify.org/Open-Notify-API/People-In-Space/)」です。NASAのデータにもとづいて、今宇宙でどれだけの人たちが働いているのかわかります。`XMLHttpRequest`クラスの使い方について、詳しくは「[XMLHttpRequest の利用](https://developer.mozilla.org/docs/Web/API/XMLHttpRequest/Using_XMLHttpRequest)」をお読みください。

`XMLHttpRequest`オブジェクトでデータを読み込み終えたときの処理は、`load`イベントのリスナーとして定めます。[`XMLHttpRequest.response`プロパティ](https://developer.mozilla.org/docs/Web/API/XMLHttpRequest/response)でJSONデータが得られますので、調べたいプロパティ値を取り出せばよいでしょう。

```js
const url = 'http://api.open-notify.org/astros.json';
function getSpacePeople(url) {
	return Rx.Observable.create((observer) => {
		const request = new XMLHttpRequest();
		request.addEventListener('load', (event) =>
			observer.next(request.response)
		);
		request.open('GET', url);
		request.send();
	});
}
getSpacePeople(url)
.subscribe(
	(data) => console.log(JSON.parse(data).people)
);
```

#04 イベント待ちのオブジェクトをObservableにする
[`fromEvent()`関数](https://rxjs-dev.firebaseapp.com/api/index/fromEvent)は、イベントを待つオブジェクトの`Observable`をつくります。引数は、オブジェクトとイベントのふたつです。前掲コードの`XMLHttpRequest`オブジェクトが`load`イベントを待つ処理はつぎのように書き替えられます。

```js
const {Observable, fromEvent} = rxjs;
const url = 'http://api.open-notify.org/astros.json';
function getSpacePeople(url) {
    return Observable.create((observer) => {
        const request = new XMLHttpRequest();
        /* request.addEventListener('load', (event) =>
            observer.next(request.response)
        ); */
        fromEvent(request, 'load')
        .subscribe((event) => observer.next(request.response));
        request.open('GET', url);
        request.send();
    });
}
getSpacePeople(url)
.subscribe(
    (data) => console.log(JSON.parse(data).people)
);
```

#05 処理の失敗を送る
`Observable`の処理の失敗は、[`Observer. error()`](https://rxjs-dev.firebaseapp.com/api/index/Observer#error)で送ります。受け取ったエラーは、`Observable.subscribe()`メソッドの第2引数に渡す関数が扱います。前掲コードにエラーの処理を加えたのがつぎのコードです。実際の動きを確かめてコードが試せるように、以下のサンプル001をjsdo.itに掲げました。

```js
const {Observable, fromEvent} = rxjs;
const url = 'http://api.open-notify.org/astros.json';
function getSpacePeople(url) {
    return Observable.create((observer) => {
        const request = new XMLHttpRequest();
        fromEvent(request, 'load')
        .subscribe((event) => observer.next(request.response));
        fromEvent(request, 'error')
        .subscribe((event) => observer.error(request));
        request.open('GET', url);
        request.send();
    });
}
getSpacePeople(url)
.subscribe(
    (data) => console.log(JSON.parse(data).people),
    (request) => console.error('エラー:', request.status)
);
```

####サンプル001■RxJS + ES6: Getting number of people in space with Promise object
[![es6_promise.png](https://qiita-image-store.s3.amazonaws.com/0/128949/6a9f1eab-6811-6a7d-0f68-0eb99ec4dc8f.png)
\>\> jsdo.itへ](http://jsdo.it/FumioNonaka/i3Bl)