點燈坊

失くすものさえない今が強くなるチャンスよ

使用 mergeMap() 攤平 Higher Order Observable

Sam Xiao's Avatar 2020-06-25

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

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 筆。

mergemap003

第 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 雖然較晚呼叫，但卻是先得到結果。

mergemap004

可發現 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

mergemap002

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

map() + mergeAll()

let { interval, of } = Rx
let { take, map, delay, mergeAll } = 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$),
  mergeAll()
)

19 行

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

使用了 mergeAll() 將 Inner Observable 攤平。

mergeAll()
將 Inner Observable 攤平並保持原 Inner Observable 的 timing 順序與個數

mergemap006

Marble diagram 斜線表示 Inner Observable，經過 mergeAll() 攤平後，完全維持原本 Observable 的 timing，且個數也相同。

mergemap000

mergeAll() 只負責攤平，因此原本 2 最先回傳的事實仍然保留。

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() 取代 map() + mergeAll()。

mergeMap()
將 Higher Order Observable 攤平並保持原 Inner Observable 的 timing 順序與個數

mergemap009

50 在 30 還沒結束就出現，因此 50 會提早出來，mergeMap() 會忠實保留原本 Inner Observable 的 timing 與個數。

mergemap001

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

  • mergeMap() 的 timing 與個數與 map() 完全相同，只是將 map() 攤平而已
  • map() + mergeAll() 與 mergeMap() 完全等效

Example

mergemap005

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

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

map()

<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, map, 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() {
  this.click$.pipe(
    pluck('data'),
    map(fetchBook$)
  ).subscribe(x => x.subscribe(x => this.title = x))
}

export default {
  name: 'app',
  domStreams: ['click$'],
  data: () => ({
    title: ''
  }),
  subscriptions,
}
</script>

第 3 行

<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>

3 個 button 都使用 click$ subject，唯 data 不一樣。

依次按下 1、2、3，但並不保證回傳結果順序如預期。

14 行

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

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

fetchBook$() 呼叫 API 回傳 title，特別加上 delayLut 與 delay() 模擬 API request 不見的先呼叫就先回傳。

fetchBook$() 相當於產生 Inner Observable。

26 行

this.click$.pipe(
  pluck('data'),
  map(fetchBook$)
).subscribe(x => x.subscribe(x => this.title = x))

click$ 相當於 Outer Observable，由 pluck('data') 取出資料，並在 map() 內呼叫 fetchBook$() 產生 Inner Observable。

但由於是 Higher Order Observable，因此需要兩次 subscribe() 才能取出 Observable 內部值，明顯是很糟糕寫法。

最後還會發現因為 map() 會保留 Inner Observable 原本時間產生順序，因此 title$ 實際顯示順序為：

  • RxJS in Action
  • FP in JavaScript
  • Speaking JavaScript

而非如預期的：

  • FP in JavaScript
  • RxJS in Action
  • Speaking JavaScript

mergemap007

map() 會引起 3 次 API request。

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$)
)

使用兩次 subscribe() 處理 Higher Order Observable 很糟糕，這與 Callback Hell 沒什麼兩樣，比較好的方式是改用 mergeMap() 將 Higher Order Observable 攤平處理。

mergemap008

mergeMap() 會引起 3 次 API request。

Conclusion

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

Reference

RxJS, map()
RxJs, mergeAll()
RxJS, mergeMap()

# RxJS