RxJSのOperatorsのコード まとめ

[RxJS](https://github.com/Reactive-Extensions/RxJS)を触り始めたので、勉強がてらOperators関連のコードを書いていきます＼(^o^)／

# 導入

```
npm install rx
```

または、

```
yarn add rx
```

# 生成(Creating Observables)

## [Create](http://reactivex.io/documentation/operators/create.html)

Observableを生成する。

```js
import Rx from 'rx';

Rx.Observable
  .create(observer => {
      observer.onNext(9);
      observer.onCompleted();
      return () => console.log('disposed'); 
  })
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 9
Completed
disposed
```

## [Empty](http://reactivex.io/documentation/operators/empty-never-throw.html)

OnCompleteだけ飛ばす。

```js
Rx.Observable.empty()
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Completed
```

## [Never](http://reactivex.io/documentation/operators/empty-never-throw.html)

何も呼ばれない空のObservable。

```js
Rx.Observable.never()
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```

```

## [Throw](http://reactivex.io/documentation/operators/empty-never-throw.html)

onErrorを流す。

```js
Rx.Observable.return(9)
  .selectMany(Rx.Observable.throw(new Error('error!')))
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Error: Error: error!
```

## [From](http://reactivex.io/documentation/operators/from.html)

IterableなオブジェクトからObservableを生成する。

```js
Rx.Observable
  .from(['apple', 'orange', 'peach'])
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: apple
Next: orange
Next: peach
Completed
```

## [Interval](http://reactivex.io/documentation/operators/interval.html)

一定時間ごとに値を流す。

```js
Rx.Observable
  .interval(1000)
  .timeInterval()
  .take(3)
  .subscribe(
    x => console.log(`Next: ${x.value}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0
Next: 1
Next: 2
Completed
```

## [Just](http://reactivex.io/documentation/operators/just.html)

値を直接指定して、Observableを生成する。

```js
Rx.Observable
  .just(88)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 88
Completed
```

## [Range](http://reactivex.io/documentation/operators/range.html)

始まりと終わりを指定し、その間の数字のObservableを生成する。

```js
Rx.Observable
  .range(0, 6)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0
Next: 1
Next: 2
Next: 3
Next: 4
Next: 5
Completed
```

## [Repeat](http://reactivex.io/documentation/operators/repeat.html)

繰り返す。

```js
Rx.Observable
  .repeat('Hello world', 3)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: Hello world
Next: Hello world
Next: Hello world
Completed
```

## [Start](http://reactivex.io/documentation/operators/start.html)

指定したスケジューラーのコンテキストで、非同期に関数を実行するっぽい？？。

```js
Rx.Observable
  .start(() => 'start', Rx.Scheduler.timeout)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: start
Completed
```

## [Timer](http://reactivex.io/documentation/operators/timer.html)

タイマー。

```js
Rx.Observable
  .timer(100, 1000)
  .timeInterval()
  .pluck('interval')
  .take(3)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 222
Next: 879
Next: 999
Completed
```

# 変換(Transforming Observables)

## [Buffer](http://reactivex.io/documentation/operators/buffer.html)

指定した数で、Observableのデータをまとめる。

```js
Rx.Observable
  .timer(0, 100)
  .buffer(() => Rx.Observable.timer(200))
  .take(3)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```js
Next: 
Next: 0,1
Next: 2,3
Completed
```

## [FlatMap](http://reactivex.io/documentation/operators/flatmap.html)

流れてくるアイテムを変換し、合成する。また、Observableを生成する。

```js
Rx.Observable
  .just('apple,orange,peach')
  .flatMap(x => Rx.Observable.from(x.split(',')))
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: apple
Next: orange
Next: peach
Completed
```

## [GroupBy](http://reactivex.io/documentation/operators/groupby.html)

グルーピング。

```js
Rx.Observable
  .range(0, 10)
  .groupBy(x => x % 2 == 0)
  .flatMap(group => group.toArray()) 
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0,2,4,6,8
Next: 1,3,5,7,9
Completed
```

## [Map] (http://reactivex.io/documentation/operators/map.html)

データの整形。

```js
Rx.Observable
  .from(['田中', '佐藤', '山田'])
  .map(x => `${x}さん`)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 田中さん
Next: 佐藤さん
Next: 山田さん
Completed
```

## [Scan](http://reactivex.io/documentation/operators/scan.html)

順番に足していく。

```js
Rx.Observable
  .from(['Hello', 'World', '!!'])
  .scan((acc, x) => `${acc} ${x}`)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: Hello
Next: Hello World
Next: Hello World !!
Completed
```

## [Window](http://reactivex.io/documentation/operators/window.html)

ストリームを束ねて、新たなストリームを生成する???


```js
Rx.Observable
  .timer(1, 50)
  .window(() => Rx.Observable.timer(100))
  .take(3)
  .flatMap(x => x.toArray())
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 
Next: 0,1
Next: 2,3
Completed
```

# フィルタリング(Filtering Observables)

## [Debounce](http://reactivex.io/documentation/operators/debounce.html)

指定した時間データが流れてこなければ、直前のデータでイベントが発火する。

```js
Rx.Observable
  .interval(1000)
  .timeInterval()
  .take(3)
  .debounce(1000)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 2
Completed
```

## [Distinct](http://reactivex.io/documentation/operators/distinct.html)

重複なく流す。

```js
Rx.Observable
  .from(['apple', 'orange', 'apple', 'apple', 'りんご'])
  .distinct()
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: apple
Next: orange
Next: りんご
Completed
```

## [Elementat](http://reactivex.io/documentation/operators/elementat.html)

指定した番号の要素だけ流す。

```js
Rx.Observable
  .from(['apple', 'orange', 'りんご'])
  .elementAt(1)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: orange
Completed
```

## [Filter](http://reactivex.io/documentation/operators/filter.html)

フィルター。

```js
Rx.Observable
  .range(0, 10)
  .filter(x => x % 2 == 0)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0
Next: 2
Next: 4
Next: 6
Next: 8
Completed
```
## [First]( http://reactivex.io/documentation/operators/first.html)

最初のやつを流す。

```js
Rx.Observable
  .range(0, 10)
  .first()
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0
Completed
```

## [IgnoreElements] (http://reactivex.io/documentation/operators/ignoreelements.html)

すべての要素を無視する。そして、onNextが呼ばれず、onError, OnCompletedのみ呼ばれる。

```js
Rx.Observable
  .range(0, 10)
  .ignoreElements()
  .subscribe(
    x => console.log(`onNext: ${x}`),
    err => console.log(`onError: ${err}`),
    () => console.log('onCompleted')
  );
```

```
Completed
```

## [Last](http://reactivex.io/documentation/operators/last.html)

最後のやつだけ流す。

```js
Rx.Observable
  .range(0, 10)
  .last()
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 9
Completed
```

## [Sample] (http://reactivex.io/documentation/operators/sample.html)

指定した間隔(一定間隔)で値を流す。

```js
Rx.Observable
  .interval(500)
  .sample(2000)
  .take(3)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 3
Next: 7
Next: 11
Completed
```

## [Skip](http://reactivex.io/documentation/operators/skip.html)

スキップ。

```js
Rx.Observable
  .range(0, 10)
  .skip(8)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 8
Next: 9
Completed
```

## [SkipLast](http://reactivex.io/documentation/operators/skiplast.html)

後ろから省く。

```js
Rx.Observable
  .range(0, 10)
  .skipLast(3)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0
Next: 1
Next: 2
Next: 3
Next: 4
Next: 5
Next: 6
Completed
```

## [Take](http://reactivex.io/documentation/operators/take.html)

指定した値だけ取り出す。

```js
Rx.Observable
  .range(0, 10)
  .take(2)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0
Next: 1
Completed
```

## [TakeLast](http://reactivex.io/documentation/operators/takelast.html)

後ろから指定した値だけ取り出す。

```js
Rx.Observable
  .range(0, 10)
  .takeLast(2)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 8
Next: 9
Completed
```

# 結合（Combining Observables）

## [And/Then/When](http://reactivex.io/documentation/operators/and-then-when.html)

複数のObservableを制御をしながら、結合してひとつのObservableを作るっぽい？？

```js
Rx.Observable
  .when(
    Rx.Observable.timer(200).and(Rx.Observable.timer(300)).thenDo((x, y) => 'first'),
    Rx.Observable.timer(400).and(Rx.Observable.timer(500)).thenDo((x, y) => 'second'))
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: first
Next: second
Completed
```

## [CombineLatest]( http://reactivex.io/documentation/operators/combinelatest.html)

簡単にいうと、Observable同士を合成するやつ。

```js
var source1 = Rx.Observable
  .interval(100)
  .map(x => `First: ${x}`);

var source2 = Rx.Observable
  .interval(150)
  .map(x => `Second: ${x}`);

source1
  .combineLatest(
    source2,
    (s1, s2) => `${s1} ${s2}`)
  .take(5)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: First: 0 Second: 0
Next: First: 1 Second: 0
Next: First: 2 Second: 0
Next: First: 2 Second: 1
Next: First: 3 Second: 1
Completed
```

## [Join](http://reactivex.io/documentation/operators/join.html)

２つのObservableをjoinする感じ？？

```js
var source1 = Rx.Observable
  .interval(100)
  .map(x => `First: ${x}`);

var source2 = Rx.Observable
  .interval(100)
  .map(x => `Second: ${x}`);

source1
  .join(
    source2,
    () => Rx.Observable.timer(0), 
    () => Rx.Observable.timer(0), 
    (x, y) => `${x}_${y}`)
  .take(5)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: First: 0_Second: 0
Next: First: 1_Second: 1
Next: First: 2_Second: 2
Next: First: 3_Second: 3
Next: First: 4_Second: 4
Completed
```

## [Merge](http://reactivex.io/documentation/operators/merge.html)

マージです。

```js
const source1 = Rx.Observable
  .range(0, 10)
  .filter(x => x % 2 == 0);

const source2 = Rx.Observable
  .range(0, 10)
  .filter(x => x % 2 == 1);

Rx.Observable
  .merge(
    source1,
    source2)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0
Next: 1
Next: 2
Next: 3
Next: 4
Next: 5
Next: 6
Next: 7
Next: 8
Next: 9
Completed
```

## [StartWith]( http://reactivex.io/documentation/operators/startwith.html)

与えた値を先に流してから、その後にObservableの値を流す。

```js

Rx.Observable
  .from(['ん', 'ご'])
  .startWith('り')
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: り
Next: ん
Next: ご
Completed
```

## [Switch](http://reactivex.io/documentation/operators/switch.html)

次のObservableへと、次々スイッチする。

```js
Rx.Observable
  .from([
    Rx.Observable.from(['まぐろ', 'あじ', 'ひらめ']),
    Rx.Observable.from(['りんご', 'みかん', 'おれんじ']),
    Rx.Observable.from(['いぬ', 'ねこ', 'しか']),
  ])
  .switch()
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: まぐろ
Next: りんご
Next: いぬ
Next: ねこ
Next: しか
Completed
```

## [Zip](http://reactivex.io/documentation/operators/zip.html)

2つ以上のObservableから流れてきた値を合成して新しい一つの値を作る。

```js
const source1 = Rx.Observable.range(0, 3);
const source2 = Rx.Observable.range(5, 8);

Rx.Observable
  .zip(
    source1,
    source2,
    (x, y) => `${x} + ${y} = ${x + y}` 
  )
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0 + 5 = 5
Next: 1 + 6 = 7
Next: 2 + 7 = 9
Completed
```

# エラーハンドリング(Error Handling Operators)

## [Catch](http://reactivex.io/documentation/operators/catch.html)

onErrorの時に代わりに流すObservableなどを指定する。

```js
Rx.Observable
  .from([0, 1, 2])
  .map(x => {
    if(x > 0) return x;
    throw new Error('error');
  })
  .catch(e => {
    console.log(`catcht: ${e}`);
    return Rx.Observable.just(2);
  })
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
catcht: Error: error
Next: 2
Completed
```

## [Retry](http://reactivex.io/documentation/operators/retry.html)

例外が発生しても指定の回数リトライする。

```js
Rx.Observable
  .range(0, 5)
  .map(x => {
    if(x > 3) {
      throw new Error('error: x > 3');
    }
    return x;
  })
  .retry(2)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0
Next: 1
Next: 2
Next: 3
Next: 0
Next: 1
Next: 2
Next: 3
Error: Error: error: x > 3
```

# Observable Utility Operators

## [Delay](http://reactivex.io/documentation/operators/delay.html)

delay。

```js
Rx.Observable
  .range(0, 5)
  .delay(1000)
  .subscribe(
    x => console.log(`Next: ${x}`),
    err => console.log(`Error: ${err}`),
    () => console.log('Completed')
  );
```

```
Next: 0
Next: 1
Next: 2
Next: 3
Next: 4
Completed
```

残りは、また今度書きます...