RxJS 6: オペレータをつくってみる

RxJS 6公式サイトはベータ段階のようで、まだチュートリアル(Overview)の情報が「Introduction」しかありません。そこで、バージョン5の「Overview」を下じきに、バージョン6に対応させたうえ、コードや解説も改めて「RxJS 6入門」01-07を書きました。

中でも「RxJS 6入門 06: Observableをつくる関数とオペレータ」は、ほぼ書き直しでした。そこで本稿では、RxJS 6のオペレータの使い方をかいつまんでおさらいし、簡単なオペレータを自作してみます。オペレータの仕組みを理解するのに役立つでしょう。

ライブラリをCDNから読み込む

今回は手軽に試せるように、RxJSライブラリはCDNから読み込みます。GitHubの「Installation and Usage」の「CDN」にしたがって、<script>要素につぎのようにsrc属性を定めてください。

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

前にあったRxJS 5.5のCDNは消されたようです。RxJS 6公式サイトの「Installation Instructions」のページも「CDN」の情報がバージョン5.0から更新されていません。

クラスと関数およびオペレータの読み込みパス

バージョン6でまず気をつけなければならないのは、ライブラリの読み込みパスです。おもなクラスと関数は名前空間がrxjsにまとめられ、オペレータはrxjs.operatorsになりました(「Installation and Usage」参照)。

const {Observable, Subject, ReplaySubject, from, of, range} = rxjs;
const {map, filter, switchMap} = rxjs.operators;

npmでECMAScript 2015(ES6)やCommonJSを使う場合は、それぞれリンク先のGitHubのページをご覧ください(なお「Import paths」参照)。

pipe()メソッドでオペレータを使う

RxJS 6では、オペレータはpipe()メソッドの引数に渡さなければなりません。このメソッドそのものは、バージョン5.5で備わったものです(「Operator pipe syntax」)。けれどこのときは、Observableのメソッドとしてオペレータを呼び出すバージョン5の書き方もできました。それがpipe()メソッド一択になったということです。

つぎのコードは、from()関数が配列からつくったObservableに、pipe()メソッドでmap()オペレータを適用しています。送られた値はオペレータが2乗して、コンソールに出力するという流れです。

const {from} = rxjs;
const {map} = rxjs.operators;
const observable = from([1, 2, 3, 4, 5])
.pipe(
	map((x) => x ** 2)
)
.subscribe(
	(x) => console.log(`value: ${x}`)
);

value: 1
value: 4
value: 9
value: 16
value: 25

• from(): 配列や配列型(array-like)オブジェクト、Promise、反復可能(iterable)オブジェクト、あるいはObservableから新たなObservableをつくって返します。
• map(): もとのObservableから送られた値に引数のプロジェクト関数を適用し、その戻り値が新たな値となるObservableにして返します。
• **: べき乗演算子(ECMAScript 2016)。左オペランド(被演算子)を右オペランドで累乗します。
• ``: テンプレート文字列(ECMAScript 2015)。組み込み式${}を扱うことができる文字列リテラルです。

オペレータの働きをする関数

オペレータが何をしているのか、簡単な関数をつくって確かめてみましょう。つぎの関数(square)は、送られた値を2乘するオペレータです。引数にObservableを受け取って、新たなObservableにして返します。その戻り値のオブジェクトに、オペレータの処理(2乗)をサブスクライブするのです。

const {Observable, from} = rxjs;
function square(input) {
	return Observable.create((observer) => {
		input.subscribe({
			next(value) {observer.next(value ** 2);},
			error(err) {observer.error(err);},
			complete() {observer.complete();}
		});
	});
}
const input = from([1, 2, 3, 4, 5]);
square(input)
.subscribe(
	(x) => console.log(`value: ${x}`)
);

もっとも、この関数はオペレータの働きを示しただけです。標準のオペレータのようにpipe()の引数には渡せません。

関数をpipe()メソッドに渡す

前掲の関数(square())は、引数にObservableを受け取って、新たなObservableにして返しました。pipe()メソッドがやっているのは、オペレータにObservableを渡すことです。そのため、引数はObservableが受け取れる関数を返さなければならないのです。

したがって、つぎのようにObservableを受け取って前掲関数が返される無名関数にすれば、pipe()メソッドの引数に与えられます。

// const input = 
from([1, 2, 3, 4, 5])  // ;
.pipe(
	(input) => square(input)
)
// square(input)
.subscribe(
	(x) => console.log(`value: ${x}`)
);

pipe()メソッドに渡せるオペレータを定める

つまり、前掲関数(square())は入れ子にして、つぎのように関数を戻り値にすれば標準のオペレータと同じ使い方ができるのです。

// function square(input) {
function square() {
	return (input) =>
	/* return */ Observable.create((observer) => {
		input.subscribe({
			next(value) {observer.next(value ** 2);},
			error(err) {observer.error(err);},
			complete() {observer.complete();}
		});
	});
}
from([1, 2, 3, 4, 5])
.pipe(
	// (input) => square(input)
	square()
)
.subscribe(
	(x) => console.log(`value: ${x}`)
);

オペレータをつくるというお題は、RxJSドキュメントの「Pipeable Operators」に「Build Your Own Operators Easily」として例が掲げられています。ご興味があればご覧ください。

RxJS 6入門

• 01: RxJSを使ってみる
• 02: Observable
• 03: Observer
• 04: Subscription
• 05: Subject
• 06: Observableをつくる関数とオペレータ
• 07: Scheduler