[S05E06] RxJS 運算子全面解析
https://www.youtube.com/watch?v=DPyZq74V60o&list=PL9LUW6O9WZqgUMHwDsKQf3prtqVvjGZ6S&index=17

今天由Will保哥分享，很多時間是大家在想operator的實際應用
這一集片長2個小時，所以拆成2天po

延申閱讀:

• Reactive.how
  reactive.how
• ReactiveX
  http://reactivex.io/
• RxMarbles
  https://rxmarbles.com/
  https://github.com/staltz/rxmarbles
  https://github.com/RxSwiftCommunity/RxMarbles

組合運算字(Combination)

組合多個Observable形成新的Observable
常用的例如:

• forkJoin
• concat
• combineLatest
• startWith:傳入一個Observable，再塞一組序列(值)在他前面

用圖型的方式了解operators

https://reactive.how/

combineLatest vs merge

https://reactive.how/combinelatest
有2個Observable，總是會組成1個Observable

• merge
  有其中1個Observable有值，就會merge值
  (範例中沒有展示，同時間有不同的值的case)

• combineLatest
  https://rxjs-dev.firebaseapp.com/api/index/function/combineLatest
  兩個observable都有值才會combine，而且只combine最新的值
  使用情境:
  1、當實現複合時的查詢OR排序變更時，只要其中1個變更，就重新觸發server load
  2、兩個form group要聯合做資料驗證，合併2個value change的observable做觸發驗證

保哥範例

從 RxJS 6 以後，import變成2個部分

import { defer, fromEvent, interval } from 'rxjs'; // 其他的
import { filter } from 'rxjs/operators'; // 跟序列處理相關的
// combineLatest(...observables: any[]): Observable<{}[]>
                 ^^^^^^^^^^^^^^ 要放幾個observables的陣列都可以
var ob = combineLastest(
    fromEvent(document,'click'), // 第1個Observable
    fromEvent(document,'dblclick'),
    (v1,v2) => ([v1,v2]) // 傳陣列 // resultSelector:(v1:{},v2:{})=>{}[]
    // (v1,v2) => ({v1,v2}) // 傳物件
        ^^ 第1個Observable丟進來的event
).pipe(做一些事).subscribe(([v1,v2]) => {
                           ^^^^^^^^ 或者寫v，就是([v1,v2])
    console.log(`v1.clientX = ${v1.clientX], v2.clientX = ${v2.clientX}`);
});

forkJoin()

https://rxjs-dev.firebaseapp.com/api/index/function/combineLatest
也是合併Observables的，看官網文件的圖比較好理解
Observables要complete才會吐出combine的東西

Combines multiple Observables to create an Observable whose values are calculated from the latest values of each of its input Observables.

combineLatest<O extends ObservableInput<any>, R>(...observables: (O | ((...values: ObservedValueOf<O>[]) => R) | SchedulerLike)[]): Observable<R>

var ob = forkJoin(
    fromEvent(document,'click').pipe(take(1)),
                                     ^^^^^^^只取1個，否則event stream不會停止
    fromEvent(document,'dblclick').pipe(take(1)),
).pipe(做一些事).subscribe(([v1,v2]) => {
                           ^^^^^^^^ 或者寫v，就是([v1,v2])
    console.log(`v1.clientX = ${v1.clientX], v2.clientX = ${v2.clientX}`);
});

startWith()

var ob=fromEvent(document,'click')
        .pipe(startWith(1,2,3)) // 多形最多6個，超過要放在陣列
              ^^^^^^^^^^^^^^^ 不用按click，就會先丟3個序列
        .subscribe(v=>{
            console.log(v);
        });

iif()

https://rxjs-dev.firebaseapp.com/api/index/function/iif
看官網的範例比較好理解

import { iif, of } from 'rxjs';
 
let subscribeToFirst;
const firstOrSecond = iif(
  () => subscribeToFirst, // 給一個callback function
  of('first'), // 如果為ture，就回傳第1個檔observable
  of('second'), // 如果為false，就回傳第2個檔observable
);
 
subscribeToFirst = true;
firstOrSecond.subscribe(value => console.log(value));
 
// Logs:
// "first"
 
subscribeToFirst = false;
firstOrSecond.subscribe(value => console.log(value));
 
// Logs:
// "second"

轉換運算子(Transformation)

常用的有

• concatMap
• map
• switchMap

map

把一個 序列 轉為 另一個序列
https://rxjs-dev.firebaseapp.com/api/operators/map

map(x=>10*x)
    ^  ^^^^

import { fromEvent } from 'rxjs';
import { map } from 'rxjs/operators';

const clicks = fromEvent(document, 'click');
const positions = clicks.pipe(map( (ev: MouseEvent) => ev.clientX));
                                    ^^                    ^^^^^^^ 轉成只要clientX
positions.subscribe(x => console.log(x));

var ob=fromEvent(document,'click')
        .pipe(startWith(1,2,3)) // 多形最多6個，超過要放在陣列
              ^^^^^^^^^^^^^^^ 不用按click，就會先丟3個序列
        .subscribe(v=>{
            console.log(v);
        });

scan()

輸入一個input stream、一個累加器(accumulator)、一個seed(初使化累加器的seed)

accepts 1 input stream, an accumulator, a seed (optional)
https://rxjs-dev.firebaseapp.com/api/operators/scan

看圖會比較好理解
https://reactive.how/scan

• scan() vs reduce()
  scan() 每收到1個就會回傳累加的結果
  reduce() 要全部跑完才會回傳結果

import { fromEvent } from 'rxjs';
import { scan, mapTo } from 'rxjs/operators';

const clicks = fromEvent(document, 'click');
const ones = clicks.pipe(mapTo(1));
                         ^^^^^^^^ 每click一次，ones就是1
const seed = 0;
const count = ones.pipe(   vvvvvvvvv The accumulator function
       scan( (acc, one) => acc + one, seed )
              ^^^ return的值是acc     ^^^^ 初使化accumulator，預設是undefine
    );
                              
count.subscribe(x => console.log(x));

concatMap() = map()+concat()

         ^^^ 看到map，就是回傳observable

https://rxjs-dev.firebaseapp.com/api/operators/concatMap
https://ithelp.ithome.com.tw/articles/10188387

import { fromEvent, interval } from 'rxjs';
import { concatMap, take } from 'rxjs/operators';
 
const ob = fromEvent(document, 'click')
    .pipe(
        map( (ev: MouseEvent) => ev.clientX) ),
        concat() // 如果是 merge() 就是mergeMap()
    )
            .subscribe(x => console.log(x));

exhaustMap() vs switchMap()

好像會cache每個observable的序列
只有在先前的observable有complete，才會合併到exhaustMap()要回傳的Observable中

Projects each source value to an Observable which is merged in the output Observable only if the previous projected Observable has completed.
https://rxjs-dev.firebaseapp.com/api/operators/exhaustMap

• switchMap()以最後1次為凖
• 如果資料流的順序很重要:用concatMap()

過濾運算子(Filtering)

常用的有:

• debounceTime
• debounce
• distinctUntilChanged
• filter
• first
• skip
• take
• takeUntil:接受observable
• takeWhile:接受一個function(回傳false時停止)
• throttle 截流:在一定的時間內只能通過n個
• throttleTime

audit()

由另一個observable來決定是否通行

Ignores source values for a duration determined by another Observable, then emits the most recent value from the source Observable, then repeats this process.

Emit clicks at a rate of at most one click per second

import { fromEvent, interval } from 'rxjs';
import { audit } from 'rxjs/operators'

const clicks = fromEvent(document, 'click');
const result = clicks.pipe( audit(ev => interval(1000)) );
                                  ^^^^^^^^^^^^^^^^^^^^ 1秒通過1個
result.subscribe(x => console.log(x));

Q&A

有A,B,C 3個observables
result B拿到資料，跟C做一些事情(doSomething)
A要跟result C做一些事情(doSomething)

條件:C要在B之後

const a$ = of('1');
const bc$ = b$.pipe(mergeMap(rb=> c$(rb));
                    ^^^^^^^
combineLatest(a$,bc$).subscribe(abc=>doSomething(abc));
^^^^^^^^^^^^^

多播運算子(Multicasting)

負責將1個Observable，廣播給多個Observer(得到的都是同1個Observable丟出來的資料)
(如果不用Multicasting，有多個subscribe()，就會有多個Observable)

常用運算子

• share = publish + refCount
• shareReplay= publishReplay + refCount
• multicast

錯誤處理運算子(Error Handling)

負責處理Observable觀察過程中出現的例外

常用運算子

• catchError
• retry
• retryWhen

Utility運算子

常用運算子

• tap (最常用)，中文翻譯:輕拍
  可以在觀察Observable的過程中，中途跑一些code
  https://rxjs-dev.firebaseapp.com/api/operators/tap

import { fromEvent } from 'rxjs';
import { tap, map } from 'rxjs/operators';

const clicks = fromEvent(document, 'click');
const positions = clicks.pipe(
  tap(ev => console.log(ev)),
  ^^^ 在map()之前，跑一點程式
  map(ev => ev.clientX),
);
positions.subscribe(x => console.log(x));

• delay (delay到裡面的條件為true時觸發，單位為ms)
  delay() + interval() 等於 timer()
  https://rxjs-dev.firebaseapp.com/api/index/function/timer

import { timer } from 'rxjs';

const numbers = timer(3000, 1000);
                      延遲  做的時間
// Emits ascending numbers, one every second (1000ms), starting after 3 seconds
// 延遲3秒，做interval 1秒
// 如果想要一開始就先做1秒，可用startWith()
// 或者time(0,多久觸發1次)
numbers.subscribe(x => console.log(x));

• delayWhen (function為true時觸發)
• finalize (complete時觸發)
• materialize (不常用，只是轉成meta資料)
• timeout (一個Observable到complete的時間，超過就報例外，例外用catchError接)

條件式與布林運算子(Conditional and Boolean)

都要等complete 才會運算

負責判斷布林值與條件式相關的運算子

常用運算子

• defaultIfEmpty
  https://rxjs-dev.firebaseapp.com/api/operators/defaultIfEmpty
  當Observable從開始到complete的過程都沒值產生時(都沒有資料流)，會送出預設值
  看圖比較好懂
  Emits a given value if the source Observable completes without emitting any next value, otherwise mirrors the source Observable.

• find

• findIndex

其他運算子

• every 傳入的每個東西在序列中都有出現，且有complete，才會為true
• isEmpty

Mathematical and Aggregate運算子

負責將Observable傳來的資料進行運算

常用運算子

• count
• max
• min
• reduce
  要complete才會運算
  不用complete就會運算的，要用scan()