rxjs 常用的管道操作符

操作符文檔
api 列表

do -> tap
catch -> catchError
switch -> switchAll
finally -> finalize

map switchMap mergeMap

mep 類似於 Array.prototype.map()
switchMap switchMap 會停止發出先前發出的內部 Observable 並開始發出新的內部 Observable 的值。(可以停止上一次發出的ajax)
mergeMap 將每個值映射到Observable，必須返回一個Observable

scan 和 reduce

reduce 只返回最后的值

// res: 12, 15
from([2, 3]).pipe(
  scan((acc, item) => acc += item, 10))
  .subscribe(v => console.log(v))

// res: 15
from([2, 3]).pipe(
  reduce((acc, item) => acc += item, 10))
  .subscribe(v => console.log(v))

filter 和 partition

filter 返回你想要的數據
partition 返回兩個 Observables [0] 符合斷言， [1] 不符合斷言

from([2, 3, 4]).pipe(
    filter(item => item <= 3))
    .subscribe(v => console.log(v))

pairwise()

將當前值和前一個值作為數組放在一起，然后將其發出

of(1, 2, 3)
  .pipe(
    pairwise()).subscribe(v => console.log(v))
[1,2]
[2,3]

min，max，count

都可以接收一個函數作為參數

from([1,2]).pipe(max()).subscribe(l) // 2
from([1,2]).pipe(min()).subscribe(l) // 1
from([1,2]).pipe(count()).subscribe(l) // 2

distinct(lambda) distinctUntilChanged([(Prev, current)=>{}]) 和 distinctUntilKeyChanged(key)

過濾掉重復的項

from([1, 2, 2, 3, 2])
  .pipe(distinct())
  .subscribe(l); // 1,2,3

elementAt

只發出第n個值, 然后完成 ,從0開始

from([1, 2])
  .pipe(elementAt(0, "default value"))
  .subscribe(l);

ignoreElements()

忽略源所發送的所有項，只傳遞 complete 或 error

skip(count: Number)，skipLast(count: number)，skipWhile(lambda)

skip 跳過源發出的前N個值
skipLast 跳過源最后發出的的N個值
skipWhile 跳過lambda返回true的值

take(count: number)， takeLast(count: number)，takeUntil(notifier: Observable)，takeWhile(lambda)

take 接收源 最初的N個值
takeLast 接收源最后N個值
takeUntil notifier發出值， 源斷流
takeWhile lambda返回true，才發出源的值

throttle(lambda: Observable)和 throttleTime(number)

先發出最新的值， 在忽略

audit(lambda: Observable)和 auditTime(number)

先忽略， 在發出最新的值

    interval(500)
      .pipe(auditTime(1000))
      .subscribe(l); // 1s后發出 2, 1被忽略

debounce(lambda: Observable) 和 debounceTime(number)

延時發送源發出的值, 如果期間源發出了新值的話，返回的值為最新的值，上一個會被丟棄掉（避免高消費事件時使用）

sample(Observable) 和 sampleTime(number)

在周期時間間隔內發出源最新值。

find 和 findIndex

類似 Array.prototype.find()

toPromise

把 Observable 轉化為 promise

  click = async e => {
    let res = await ajax('http://localhost:1995/a').pipe(map(res => res.response)).toPromise();
    l(res)
  }

buffer bufferCount bufferTime bufferToggle bufferWhen

buffer系列，將過去的值作為數組收集，在事件觸發時發出收集好的值

const send$= fromEvent(document, 'click');
const interval = interval(1000);

const buffered = interval.pipe(buffer(send$));
buffered.subscribe(l);

defaultIfEmpty

如果源Observable完成而沒有發出任何next值，則發出給定值 ，否則鏡像源Observable

const { of, from, empty } = require("rxjs");
const { mergeMap, defaultIfEmpty } = require("rxjs/operators");

from([1, 2, 2, 3, 2])
  .pipe(
    mergeMap(el => (el > 10 ? of(el) : empty())),
    defaultIfEmpty("not data"),
  )
  .subscribe(l); // not data

delay delayWhen

延遲來自源Observable的項目的發射

endWith

from([1, 2])
  .pipe(endWith("源觀察完成后，附加一個發射，然后完成。"))
  .subscribe(l); // 1, 2, "源觀察完成后，附加一個發射，然后完成。"

不常用的

pluck(properties: ...string); 取一個對象的屬性類似 obj.xxx.xxx
toArray(); 把流發出的值塞在Array后，返回Array