使用 concatMap() 維持 Outer Observable 順序 | 點燈坊

Front-end 兩大 Asynchronous 就是 DOM Event 與 API Request，實務上常將這兩個 Asynchronous 視為 Observable 一起處理，這導致了 Higher Order Observable 出現，RxJS 提供了 mergeMap()、concatMap()、switchMap() 與 exhaustMap() 處理，本文探討 concatMap()。

Version

macOS Catalina 10.15.5
WebStorm 2020.1.2
Vue 2.6.11
RxJS 6.5.5

map()

let { interval, of } = Rx
let { take, map, delay, mergeMap } = RxOperators

let fetchOuter$ = t => interval(t).pipe(
  map(x => x + 1), 
  take(3)
)

let delayLut = {
  1: 3000,
  2: 1000,
  3: 2000
}

let fetchInner$ = n => of(n).pipe(
  delay(delayLut[n])
)

fetchOuter$(1000).pipe(
  map(fetchInner$)
)

第 4 行

let fetchOuter$ = t => interval(t).pipe(
  map(x => x + 1), 
  take(3)
)

fetchOuter$() 產生 Outer Observable，可傳入 t 決定 interval() 時間，並只取前 3 筆。

第 9 行

let delayLut = {
  1: 3000,
  2: 1000,
  3: 2000
}

let fetchInner$ = n => of(n).pipe(
  delay(delayLut[n])
)

fetchInner$() 產生 Inner Observable，其值由 n 傳入，並由 of() 包成 Observable 回傳。

較特殊的是尚有 delayLut Object 決定其 delay 時間，主要是要模擬 asynchronous 回傳時間並不固定，先呼叫不一定保證先回傳結果，如 1 回傳需 3 秒，但 2 回傳僅需 1 秒，也就是會發現 2雖然較晚呼叫，但卻是先得到結果。

可發現 2 時間最短只需 1 秒，其次為 3 需 2 秒，1 則需要 3 秒最晚回傳。

19 行

fetchOuter$(1000).pipe(
  map(fetchInner$)
)

由 fetchOuter$() 產生間隔為 1 秒的 Outer Observable，且每個值又呼叫 fetchInner$() 產生 Inner Observable。

原本應該為 map(x => fetchInner$(x))，可省略 x 使其 point-free

黑色為 Outer Observable，每個值都產生 Inner Observable，因此出現了 Observable of Observable，也就是 Higher Order Observable。

mergeMap()

let { interval, of } = Rx
let { take, map, delay, mergeMap } = RxOperators

let fetchOuter$ = t => interval(t).pipe(
  map(x => x + 1), 
  take(3)
)

let delayLut = {
  1: 3000,
  2: 1000,
  3: 2000
}

let fetchInner$ = n => of(n).pipe(
  delay(delayLut[n])
)

fetchOuter$(1000).pipe(
  mergeMap(fetchInner$)
)

19 行

fetchOuter$(1000).pipe(
  mergeMap(fetchInner$)
)

可改用 mergeMap() 將 Inner Observable 攤平。

concatMap()

let { interval, of } = Rx
let { take, map, delay, concatMap } = RxOperators

let fetchOuter$ = t => interval(t).pipe(
  map(x => x + 1), 
  take(3)
)

let delayLut = {
  1: 3000,
  2: 1000,
  3: 2000
}

let fetchInner$ = n => of(n).pipe(
  delay(delayLut[n])
)

fetchOuter$(1000).pipe(
  concatMap(fetchInner$)
)

mergeMap() 雖然能將 Inner Observable 攤平，但也發現了其問題：mergeMap() 會保留原本 Inner Observable 的 timing 順序，如前例回傳為 2、1、3。

若希望最後結果與 Outer Observable 的 timing 相同，也就是 1、2、3，就要改用 concatMap()。

雖然結果如預期為 1、2 、3，但 timing 已經與 mergeMap() 不一樣，且 3 與 2 的距離似乎隔的特別遠 ? 也就是 concatMap() 斯乎會改變原本 Inner Observable 的 timing ?

map() + concantAll()

let { interval, of } = Rx
let { take, map, delay, concatAll } = RxOperators

let fetchOuter$ = t => interval(t).pipe(
  map(x => x + 1), 
  take(3)
)

let delayLut = {
  1: 3000,
  2: 1000,
  3: 2000
}

let fetchInner$ = n => of(n).pipe(
  delay(delayLut[n])
)

fetchOuter$(1000).pipe(
  map(fetchInner$),
  concatAll()
)

19 行

fetchOuter$(1000).pipe(
  map(fetchInner$),
  concatAll()
)

事實上 concatMap() 就是 map() + concatAll()。

這就要談到 mergeAll() 與 concatAll() 的不同。

mergeAll() 會直接將 Inner Observable 依照原本 Inner Observable 的 timing 攤平。

而 concatAll() 會依照 Outer Observable 的 timing 重新 subscribe Inner Observable，如本例 1 會 delay 3 秒，雖然 2 只 delay 1 秒， concatAll() 會等到 1 回傳後才 subscribe 2，由於 2 只 delay 1 秒，因此在 1 的下一秒會顯示 2。

3 為什麼會延遲這麼久才顯示呢 ? 因為 concatAll() 要等 2 顯示完才 subscribe 3，因此會在 2 之後 delay 3 秒才顯示 3，而非在 Outer Observable 的 3 delay 3 秒後顯示 3。

特別將 concatMap()、mergeMap()、map() + concatAll() 與 map() 的 timing 都放在一起比較：

• mergeMap() 的 timing 與個數與 map() 完全相同，只是將 map() 攤平而已
• concatMap() 則與 Outer Observable 的順序完全一樣，但會改變原 Inner Observable 的 timing
• map() + concatAll() 與 concatMap() 完全等效

Example

Higher Order Observable 最常見應用是 DOM Event 與 API Request 結合，由於 DOM Event 為 asynchronous，而 API request 亦為 asynchronous，兩者都可視為 Observable，但由於當時網路速度與 server 忙碌程度，並不保證先呼叫的 API request 就會先回傳。

當 DOM Evnet 與 API Request 一起運用時，就會出現 Higher Order Observable。

mergeMap()

<template>
  <div>
    <button v-stream:click="{ subject: click$, data: 1 }">1</button>
    <button v-stream:click="{ subject: click$, data: 2 }">2</button>
    <button v-stream:click="{ subject: click$, data: 3 }">3</button>
    <h1>{{ title$ }}</h1>
  </div>
</template>

<script>
import { ajax } from 'rxjs/ajax'
import { pluck, mergeMap, delay } from 'rxjs/operators'

let delayLut = {
  1: 3000,
  2: 1000,
  3: 2000
}

let fetchBook$ = x => ajax(`http://localhost:3000/books/${x}`).pipe(
  pluck('response', 'title'),
  delay(delayLut[x])
)

let subscriptions = function() {
  let title$ = this.click$.pipe(
    pluck('data'),
    mergeMap(fetchBook$)
  )

  return { title$ }
}

export default {
  name: 'app',
  domStreams: ['click$'],
  subscriptions,
}
</script>

26 行

let title$ = this.click$.pipe(
  pluck('data'),
  mergeMap(fetchBook$)
)

使用 mergeMap() 會發現儘管一次按下 1、2、3，但結果是 RxJS in Action、FP in JavaScript、Speaking JavaScript，並非依照 Outer Observable 順序，而是依照 Inner Observable 順序。

mergeMap() 會引起 3 次 API request。

concatMap()

<template>
  <div>
    <button v-stream:click="{ subject: click$, data: 1 }">1</button>
    <button v-stream:click="{ subject: click$, data: 2 }">2</button>
    <button v-stream:click="{ subject: click$, data: 3 }">3</button>
    <h1>{{ title$ }}</h1>
  </div>
</template>

<script>
import { ajax } from 'rxjs/ajax'
import { pluck, concatMap, delay } from 'rxjs/operators'

let delayLut = {
  1: 3000,
  2: 1000,
  3: 2000
}

let fetchBook$ = x => ajax(`http://localhost:3000/books/${x}`).pipe(
  pluck('response', 'title'),
  delay(delayLut[x])
)

let subscriptions = function() {
  let title$ = this.click$.pipe(
    pluck('data'),
    concatMap(fetchBook$)
  )

  return { title$ }
}

export default {
  name: 'app',
  domStreams: ['click$'],
  subscriptions,
}
</script>

26 行

let title$ = this.click$.pipe(
  pluck('data'),
  concatMap(fetchBook$)
)

改用 concatMap() 之後，雖然依然依次按下 1、2、3，但結果為 FP in JavaScript、RxJS in Action 與 Speaking JavaScript，也就是結果會依 Outer Observable 順序顯示。

concatMap() 會引起 3 次 API request。

Conclusion

• Higher Order Observable 為學習 RxJS 無法逃避課題，而初學者常常在 mergeMap()、concatMap()、exhaustMap() 與 switchMap() 之間有選擇障礙
• 由於 concatMap() 本質是 map() + concatAll()，因此會保留原本 Outer Observable 的時間順序，這也導致了 Inner Observable 的 timing 順序會完全改變
• 若要確保結果與 Inner Observable 順序相同，可改用 mergeMap()

Reference

RxJS, map()
RxJS, concatAll()
RxJS, concatMap()