Rxjs--組合操做符

Observable 組合(combine)、串行(concat)、並行(merge)、一對一合併(zip)、切換(switch)等

combineLatest

public combineLatest(other: ObservableInput, project: function): Observable

combineLatest 結合傳入的多個 Observables。 經過順序的訂閱每一個輸入Observable, 在每次任一輸入Observables發送的時候收集每一個輸入Observables最新的值組成一個數組, 而後要麼將這個數組傳給可選的投射函數併發送投射函數返回的結果, 或者在沒有提供投射函數時僅僅發出該數組。

combineAll

public combineAll(project: function): Observable

當高階 Observable 完成時(即源Observable發出了complete)，經過使用 combineLatest 將其打平

注意只有完成時纔打平執行訂閱!!!

let clicks = Rx.Observable.fromEvent(document, 'click');

    let higherOrder = clicks.map(ev =>
      Rx.Observable.interval(100).take(3)
    ).take(2);

    let result = higherOrder.combineAll();
    result.subscribe(x => console.log(x));

concat

public concat(other: ObservableInput, scheduler: Scheduler): Observable

順序的、串行的將全部輸入 Observable 的值合併給輸出 Observable

concatAll

public concatAll(): Observable

串行鏈接源(高階 Observable)所發出的每一個 Observable，只有當一個內部 Observable 完成的時候才訂閱下 一個內部 Observable，並將它們的全部值合併到返回的 Observable 中

經過順序地鏈接內部 Observable，將高階 Observable 轉化爲一階 Observable

var clicks = Rx.Observable.fromEvent(document, 'click');
    var higherOrder = clicks.map(ev => Rx.Observable.interval(1000).take(4));
    var firstOrder = higherOrder.concatAll();
    firstOrder.subscribe(x => console.log(x));

若是源發送很快則會遇到內存問題，由於傳入的 Observables 會在無界緩衝區中收集

exhaust

public exhaust(): Observable

exhaust 訂閱高階 Observable 。 每次觀察到這些已發出的內部 Observables 中的其中一個時，輸出 Observable 開始發出該內部 Observable 要發出的項, 然而若是前一個 Observable 還未完成的話，exhaust 會忽略每一個新的內部 Observable 。一旦完成，它將接受並打平下一個 內部 Observable ，而後重複此過程。（和concatAll相似，可是會拋棄掉在前一個高階Observable完成前發出的高階Observable）

merge

public static merge(observables: ...ObservableInput, concurrent: number, scheduler: Scheduler): Observable

merge 訂閱每一個給定的輸入 Observable (做爲參數)，而後只是將全部輸入 Observables 的全部值發送(不進行任何轉換)到輸出 Observable 。全部的輸入 Observable 都完成了，輸出 Observable 才 能完成。任何由輸入 Observable 發出的錯誤都會當即在輸出 Observalbe 上發出。

observables	...ObservableInput		合併到一塊兒的輸入Observables。
concurrent	number	可選的 默認值: Number.POSITIVE_INFINITY	能夠同時訂閱的輸入 Observables 的最大數量。
scheduler	Scheduler	可選的 默認值: null	調度器用來管理並行的輸入Observables。

mergeAll

mergeAll 訂閱高階 Observable 。 每當觀察到發出的內部 Observable 時，它會訂閱併發出輸出 Observable 上的這個 內部 Observable 的全部值。全部的內部 Observable 都完成了，輸出 Observable 才能完成。任何由內部 Observable 發出的錯誤都會當即在輸出 Observalbe 上發出。

 

race

public race(): Observable

返回 Observable，該 Observable 是源 Observable 和提供的 Observables 的組合中 第一個發出項的 Observable 的鏡像

startWith

public startWith(values: ...T, scheduler: Scheduler): Observable

返回的 Observable 會先發出做爲參數指定的項，而後再發出由源 Observable 所發出的項

switch

public switch(): Observable<T>

switch 訂閱高階 Observable。老是隻會訂閱最新的內部 Observable， 並取消訂閱以前的。

withLatestFrom

public withLatestFrom(other: ObservableInput, project: Function): Observable

withLatestFrom 結合源 Observablecombines（實例）和其餘輸入 Observables 的最新值，當且僅當 source 發出數據時, 可選的使用 project 函數以決定輸出 Observable 將要發出的值。 在輸出 Observable 發出值以前，全部的輸入 Observables 都必須發出至少一個值。

zip

public static zip(observables: *): Observable<R>

將多個 Observable 組合以建立一個 Observable，該 Observable 的值是由全部輸入 Observables 的值按順序計算而來的。

let age = Rx.Observable.of<number>(27, 25, 29);
    let name = Rx.Observable.of<string>('Foo', 'Bar', 'Beer');
    let isDev = Rx.Observable.of<boolean>(true, true, false);

    Rx.Observable
      .zip(age, name, isDev, (age: number, name: string, isDev: boolean) => ({age, name, isDev}))
      .subscribe(x => console.log(x));

zipAll

public zipAll(project: *): Observable<R> | WebSocketSubject<T> | Observable<T>