RxJS の Operators (9) - Conditional and Boolean Operators

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

はじめに

今日は ReactiveX の Conditional and Boolean Operators について RxJS の API ドキュメントを読んだりサンプルコードを書いたりしていきます。

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

Observable.prototype.every

• ReactiveX - All operator
• Observable.prototype.every API Document
• Observable.prototype.every Source Code

ReactiveX では All ですが、RxJS では every です。

Array.prototype.every と同様にすべての値が条件を満たしたときに onNext(true) します。ひとつでも違えば、その時点で onNext(false) し onCompleted します。

import { Observable } from 'rx';

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

あえて false になる例を。途中で打ち切っているのが分かるように do をはさんでいます。

Observable.amb / Observable.prototype.amb

• ReactiveX - Amb operator
• Observable.amb API Document
• Observable.amb Source Code
• Observable.prototype.amb API Document
• Observable.prototype.amb Source Code

複数の Observable を取り、最初に値を流した Observable の値を次に流します。

API ドキュメントによると amb は ambiguous の意味だそうです。

Observable.prototype.amb は自身ともうひとつの Observable を取り、Observable.amb と同様に、最初のひとつを選んで利用します。

import { Observable } from 'rx';

Observable
  .amb(
    Observable.timer(500).map(() => '500'),
    Observable.timer(300).map(() => '300')
  )
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 300
// onCompleted

Observable.prototype.includes

• ReactiveX - Includes operator
• Observable.prototype.includes API Document
• Observable.prototype.includes Source Code

ReactiveX では Contains ですが、RxJS では削除されており includes になっています。

指定した要素が含まれていれば true をなければ false を onNext します。

import { Observable } from 'rx';

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

ちなみに Observable.prototype.indexOf もあります。

Observable.prototype.defaultifempty

• ReactiveX - DefaultIfEmpty operator
• Observable.prototype.defaultifempty API Document
• Observable.prototype.defaultifempty Source Code

onCompleted のタイミングで一度も onNext されていない場合に defaultValue を流します。defaultValue はデフォルトで null です。下記の例では 123 を指定しています。

import { Observable } from 'rx';

Observable
  .empty()
  .defaultIfEmpty(123)
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 123
// onCompleted

Observable.prototype.sequenceEqual

• ReactiveX - SequenceEqual operator
• Observable.prototype.sequenceEqual API Document
• Observable.prototype.sequenceEqual Source Code

ふたつの Observable で前から順に onNext の値を比較してすべて一致すれば true を流します。

import { Observable } from 'rx';

Observable
  .from([1, 2, 3])
  .sequenceEqual(
    Observable
      .from([1, 2, 3])
      .delay(100)
  )
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: true
// onCompleted

上記の通り、タイミングがずれても (delay(100)) 、流れてくる値が同じなら true になります。

おわりに

ReactiveX の Conditional and Boolean Operators を見ていきました。この区分の意図はあまりよくわかりません。

リスト上は skipUntil / skipWhile / takeUntil / takeWhile があったのですが、以前触れたので割愛しました。