使用 combineLatest() 結合多個 Observable 最新值 | 點燈坊

若兩個 Observable 各自產生而非相依，但又必須隨時間一起合作時，可使用 combineLatest() 各自取得其最新值然後一起合作。

Version

macOS Catalina 10.15.4
WebStorm 2020.1
Vue 2.6.11
Vue-rx 6.2
RxJS 6.5.5

Browser

當選擇不同 ID 時，下方會即時顯示不同 title。

Data

[
    {
      "id": 1,
      "title": "FP in JavaScript",
      "price": 100,
      "categoryId": 1,
      "image": "fpjs.jpg"
    },
    {
      "id": 2,
      "title": "RxJS in Action",
      "price": 200,
      "categoryId": 2,
      "image": "rxjs.jpg"
    },
    {
      "id": 3,
      "title": "Speaking JavaScript",
      "price": 300,
      "categoryId": 3,
      "image": "spjs.jpg"
    }
]

http://localhost:3000/books 回傳以上 Object Array。

DOM Event Stream

<template>
  <div>
    ID:
    <select v-stream:change="select$">
      <option v-for="x in options" :value="x.value" :key="x.id">
        {{ x.text }}
      </option>
    </select>
    <p> {{ result$ }} </p>
  </div>
</template>

<script>
import { ajax } from 'rxjs/ajax'
import { combineLatest } from 'rxjs'
import { pluck, map } from 'rxjs/operators'

let fetchBooks$ = ajax(`http://localhost:3000/books/`).pipe(
  pluck('response')
)

let subscriptions = function() {
  let books$ = fetchBooks$

  let index$ = this.select$.pipe(
    pluck('event', 'target', 'value')
  )

  let result$ = combineLatest(books$, index$).pipe(
    map(([books, i]) => books[i]),
    pluck('title')
  )

  return { result$ }
}

export default {
  name:'app',
  data:() => ({
    selectValue:0,
    options:[
      { text:'1', value:'0' },
      { text:'2', value:'1' },
      { text:'3', value:'2' },
    ]
  }),
  domStreams:['select$'],
  subscriptions
}
</script>

分別使用 DOM Event Stream 與 watchAsObservable() 兩種方式使用 combineLatest()。

第 4 行

<select v-stream:change="select$">
  <option v-for="x in options" :value="x.value" :key="x.id">
    {{ x.text }}
  </option>
</select>
<p> {{ result$ }} </p>

<option> 使用 v-for 根據 options data 組合而成。

select$ 則由 <select> 的 change event 產生。

23 行

let books$ = fetchBooks$

由 fetchBooks$ 回傳 books$ Observable。

25 行

let index$ = this.select$.pipe(
  pluck('event', 'target', 'value')
)

index$ 由 select$ 取出 event.target.value，也就是 <option> 所選取的 value。

可發現將 DOM event 轉成 Observable 時，亦可讀取 DOM 的 event.target.value ，這使的 event stream 有更多應用

29 行

let result$ = combineLatest(books$, index$).pipe(
  map(([books, i]) => books[i]),
  pluck('title')
)

目前 book$ 與 select$ 都是 Observable，我們該如何結合這兩個 asynchronous 資料呢 ?

books$ 每次回傳為一整個 array，而 index$ 每次回傳為最新 selected value，套用 combineLatest() 後的 map() 會同時取得最新的 books array 與 i selected value，可使用這兩個值再產生新的 Observable。

combineLatest()
取得多個 Observable 的最新值重組成新 Obsevable

每當 Observable 有新資料來時，都會以各自 Observable 目前最新值組合出新 Observable。

watchAsObservable()

<template>
  <div>
    ID:
    <select v-model="selectValue">
      <option v-for="x in options" :value="x.value" :key="x.id">
        {{ x.text }}
      </option>
    </select>
    <p> {{ result$ }} </p>
  </div>
</template>

<script>
import { ajax } from 'rxjs/ajax'
import { combineLatest } from 'rxjs'
import { pluck, map } from 'rxjs/operators'

let fetchBook$ = ajax(`http://localhost:3000/books/`).pipe(
  pluck('response')
)

let subscriptions = function() {
  let books$ = fetchBook$

  let index$ = this.$watchAsObservable('selectValue').pipe(
    pluck('newValue')
  )

  let result$ = combineLatest(books$, index$).pipe(
    map(([books, i]) => books[i]),
    pluck('title')
  )

  return { result$ }
}

export default {
  name:'app',
  data:() => ({
    selectValue:0,
    options:[
      { text:'1', value:'0' },
      { text:'2', value:'1' },
      { text:'3', value:'2' },
    ]
  }),
  domStreams:['select$'],
  subscriptions
}
</script>

也可使用 data binding 方式，將 data 轉成 Observable。

第 4 行

<select v-model="selectValue">
  <option v-for="x in options" :value="x.value" :key="x.id">
     {{ x.text }}
  </option>
</select>
<p> {{ result$ }} </p>

<select> 改用 v-model 綁定到 selectValue。

25 行

let index$ = this.$watchAsObservable('selectValue').pipe(
  pluck('newValue')
)

使用 watchAsObservable() 將 selectValue data 轉成 index$ Observable。

後續 combineLatest() 用法完全一樣。

DOM event stream 可由 event 產生，亦可由 watchAsObservable() 產生

Conclusion

• 要將 DOM event stream 轉成 Observable 時，可使用 event 配合 event.target.value，也可使用 watchAsObservable() 將 data 轉成 Observable
• combineLatest() 適合兩個 stream 各自變動時，典型就是 DOM event stream 與 API 回傳資料兩個 Observable 要一起搭配時

Reference

John Lindquist, Map Vue.js Components to Remote Data Streams with RxJS
RxJS, combineLatest()