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

この記事は bouzuya's RxJS Advent Calendar 2015 の 9 日目かつ RxJS Advent Calendar 2015 の 9 日目です。

はじめに

今日は ReactiveX の Observable Utility Operators について RxJS の API ドキュメントやサンプルコードを書いていきます。

また RxJS 4.0.7 を対象にしています。

Observable Utility Operators

Observable.prototype.subscribe

• ReactiveX - Subscribe operator
• Observable.prototype.subscribe API Document
• Observable.prototype.subscribe Source Code

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

Observable.prototype.subscribeOn

• ReactiveX - SubscribeOn operator
• Observable.prototype.subscribeOn API Document
• Observable.prototype.subscribeOn Source Code

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

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
• Observable.prototype.do API Document
• Observable.prototype.do Source Code

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

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
• Observable.prototype.delay API Document
• Observable.prototype.delay Source Code

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

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 のユーティリティを見ました。まだ途中なので続きは明日にします。