Observable 的運算子(Operator)

什麼是 Operator? Operator 是 Observable 的 method, 例如 .map(...), .filter(...)等等，它有一個很重要的特性是它基於輸入的 Observable 產生一個新的 Observable, 這個特性讓我們可以用 Javascript 的 chain method, 例如 test.add().del()

造一個自己的 Operator

為了理解 Operator 的概念，我們自己來打造一個 map Operator, 這是一個對資料流用 someCallback 加工, 為了解說方便，將一切從簡，拿掉所有的 error handling

function myOperator(someCallback) {
  return Rx.Observable.create(observer => {
    var subscription =  this.subscribe(
      data => {
        observer.next(someCallback(data));
      }
    );
    return subscription;
  })
}

// use own operator
Rx.Observable.prototype.myOperator = myOperator;
var input1 = Rx.Observable.from([1,2,3,4]).myOperator(v => 10 * v);
input1.subscribe(x => console.log('from myOperator', x));

// use map
var input2 = Rx.Observable.from([5,6,7,8]).map(v => 10 * v)
input2.subscribe(x => console.log('from map', x));

codepen
簡單解說一下

1. 這個函數會用 Rx.Observable.create()產生一個新的 Observable 並返回它
2. function(observer) 是針對原本資料流加工 observer.next(v => 10 * v), 這裡的 this 就是原本的 Observable，因為用了 Javascript 的 chain mathod, 所以 data 是從原來 Observable 的資料來的.
3. 當然它的作用如同 .map(v => 10 * v) 一般。

單一資料流的 Operator

Observable 的 Operator 群非常強大，歸類的方式也有很多種，這裡只取我們會用到的，先前其實在產生 Observable 時已經看過一些 Operator, 像是 .from, .of, .interval, .fromPromise..., 這些由 Data Provider 產生 Observable 的 Operator, 還有上面已經提過很重要的 .map Operator，接下來是針對單一資料流的 Operator

take

對於資料流只取參數值的個數

const source$ = Rx.Observable.interval(1000);

source$.take(5)
.subscribe(v => console.log(v));

codepen
印出 0, 1, 2, 3, 4

filter

根據參數裡的判斷條件，過濾資料流，返回一個新的 Observable

const source$ = Rx.Observable.interval(1000);

source$.take(10)
.filter(x => x % 2 === 0)
.subscribe(v => console.log(v));    

codepen
因為我們的資料流是連續不斷的，這個例子用 .take(10) 取 10 個值，用 .filter 來過濾偶數
印出 0, 2, 4, 6, 8

reduce

對資料流每個元素做累積，產生單一值的 Observable

const source$ = Rx.Observable.interval(1000);
source$.take(10)
  .reduce((acc, x) => acc + x)
  .subscribe( v => console.log(v));

codepen
從 0 加到 9, 最終印出一個值45

scan

對資料流每個元素做累積如同 reduce, 但會產生中間經過的這些所有值的 Observable

const source$ = Rx.Observable.interval(1000);

ource$.take(10)
.scan((acc, x) =>  acc + x)
.subscribe(v => console.log(v));

codepen
印出1,3,6,10,15,21,28,36,45

下次再來介紹針對多個資料流的 Operator