RxJS の Operators (3) - Observable のフィルター

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

# はじめに

今日は [ReactiveX の Operators の Filtering](http://reactivex.io/documentation/operators.html#transforming) について [RxJS の API ドキュメント](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/observable.md)やソースコードを見ていきます。

また [RxJS 4.0.7](https://github.com/Reactive-Extensions/RxJS/tree/v4.0.7) を対象にしています。

# Observable のフィルター

## `Observable.prototype.filter`

- [ReactiveX - Filter operator](http://reactivex.io/documentation/operators/filter.html)
- [`Observable.prototype.filter` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/where.md)
- [`Observable.prototype.filter` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/filter.js)

`Array.prototype.filter` でもおなじみのものです。動きについては説明は不要でしょう。

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

Observable
  .from([1, 2, 3, 4, 5, 6])
  .filter(value => value % 2 === 0)
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 2
// onNext: 4
// onNext: 6
// onCompleted
```

## `Observable.prototype.first` / `Observable.prototype.last`

- [ReactiveX - First operator](http://reactivex.io/documentation/operators/first.html)
- [`Observable.prototype.first` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/first.md)
- [`Observable.prototype.first` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/linq/observable/first.js)

- [ReactiveX - Last operator](http://reactivex.io/documentation/operators/last.html)
- [`Observable.prototype.last` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/last.md)
- [`Observable.prototype.last` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/linq/observable/last.js)

最初の値や最後の値のみにフィルターします。`Observable.prototype.filter` のように `predicate` をとることもできます。

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

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

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

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

前回 `Observable.prototype.scan` の説明でも書いたとおり、`onCompleted` を待たないケースもままあるので、そういう場面では `Observable.prototype.last` は使いづらいかもしれません。

また後述の `first` は `take(1)` や `elementAt(0)` と、`last` は `takeLast(1)` とはほとんど同じです (実装は別ものです) 。

## `Observable.prototype.elementAt`

- [ReactiveX - ElementAt operator](http://reactivex.io/documentation/operators/elementat.html)
- [`Observable.prototype.elementat` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/elementat.md)
- [`Observable.prototype.elementat` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/linq/observable/elementat.js)

n 番目の要素だけにフィルターします。`take` や `skip` と違い、引数は `count` ではなく `index` です (これは間違えないと思いますが) 。

## `Observable.prototype.skip` / `Observable.prototype.take`

- [ReactiveX - Skip operator](http://reactivex.io/documentation/operators/skip.html)
- [`Observable.prototype.skip` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/skip.md)
- [`Observable.prototype.skip` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/skip.js)

- [ReactiveX - Take operator](http://reactivex.io/documentation/operators/take.html)
- [`Observable.prototype.take` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/take.md)
- [`Observable.prototype.take` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/take.js)

既に前述の説明で紹介してしまいましたが、`skip` は n 個の要素をスキップし、`take` は n 個の要素をとる形でフィルターします。

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

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

`skip` および `take` はその派生版も便利だと思います。

たとえば [`Observable.prototype.skipWhile`](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/skipwhile.md) は `predicate` を取り、条件が真の間は値をスキップします。

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

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

個人的には `first` が `predicate` を取るなら `skip` も `predicate` を取ればいいのにと思うのですが (index も引数で取れますし……) 、どうなんでしょうね。

## `Observable.prototype.sample` / `Observable.prototype.debounce`

- [ReactiveX - Sample operator](http://reactivex.io/documentation/operators/sample.html)
- [`Observable.prototype.sample` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/sample.md)
- [`Observable.prototype.sample` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/linq/observable/sample.js)

- [ReactiveX - Debounce operator](http://reactivex.io/documentation/operators/debounce.html)
- [`Observable.prototype.debounce` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/debounce.md)
- [`Observable.prototype.debounce` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/linq/observable/debounce.js)

一定間隔ごとに最後の値を取ります。

正直なところ、ぼくはよく分かっていません。おそらく `debounce` は次の値まで指定した間隔があいた場合だけ流す、`sample` は指定した間隔ごとに流す、のような気がします。なので `debounce` だと指定した間隔よりも短い間隔で値が流れ続けるるとフィルターされて流れてこないし、 `sample` は指定した間隔ごとに流してくれるっぽいです。

## `Observable.prototype.distinct`

- [ReactiveX - Distinct operator](http://reactivex.io/documentation/operators/distinct.html)
- [`Observable.prototype.distinct` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/distinct.md)
- [`Observable.prototype.distinct` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/linq/observable/distinct.js)

重複していないものだけを流します。

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

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

派生版の `distinctUntilChanged` は直前のものから変化すると流れてきます。

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

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

`distinct` の重複チェックにはすこし注意が必要で、同一 `object` の内容を変更した場合には同一のものと見なされます。内部で直近の値 (参照) を保持していて `===` で比較しているからです。別 `object` にする (イミュータブルな `object` として運用する) か、`keySelector` に `object` の変更を検知するための hashCode などを指定すると良いと思います。

## `Observable.prototype.ignoreElements`

- [ReactiveX - IgnoreElements operator](http://reactivex.io/documentation/operators/ignoreelements.html)
- [`Observable.prototype.ignoreElements` API Document](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/doc/api/core/operators/ignoreelements.md)
- [`Observable.prototype.ignoreElements` Source Code](https://github.com/Reactive-Extensions/RxJS/blob/v4.0.7/src/core/perf/operators/ignoreelements.js)

`onNext` を流しません。すべての要素を無視します。 `onError` や `onCompleted` は流れます。

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

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

# おわりに

今日は `Observable` をフィルターする Operator を見てみました。