RXJS5 与 Promise.all答案

【问题标题】：RXJS5 vs Promise.allRXJS5 与 Promise.all
【发布时间】：2016-03-07 15:04:51
【问题描述】：

有没有一个 Promise.all 的等价物？

let prom1 = doA(); // some promise
let prom2 = doB(); // another promise

// wait for both promises to complete.
Promise.all([prom1, prom2], values => {
    // do something;
});

无法从文档中将其拼凑起来，各种文章都建议使用 ForkJoin，但无法使其正常工作......

let behaviour1 = new BehaviourSubject(0);
let behaviour2 = new BehaviourSubject(1);
let allObserver = new ForkJoinObservable(behaviour1, behaviour2);

behaviour1.subscribe( () => console.log('i work'));
behaviour2.subscribe( () => console.log('i work'));
allObserver.subscribe( () => console.log('i dont work'));

可能只是切换回理智的承诺世界。

【问题讨论】：

标签： javascript rxjs5

【解决方案1】：

Rx.Observable 有一个toArray 函数，可用于复制Promise.all 行为：它存储流的所有发出值并等待底层流的onComplete 事件触发。发出所有底层项目后，生成的流将发出单个项目：

// Instead of Promises, we can model our async actions as observables
const operation1$ = Rx.Observable.just(1);
const operation2$ = Rx.Observable.just(2);

// Merge all our async results into a single stream
const result$ = Rx.Observable.merge(operation1$, operation2$)

// Finally, call toArray to combine all results
result$
    .toArray()
    .subscribe(x => console.log(x));
// >> [1, 2]

【讨论】：

谢谢，试试看！

【解决方案2】：

import Rx, { Observable } from 'rxjs' 
import axios from 'axios'

const promiseA = axios.get('https://jsonplaceholder.typicode.com/users/1')
    , promiseB = axios.get('https://jsonplaceholder.typicode.com/users/2')

const promiseStream$ = Observable   
       .of(promiseA, promiseB)       // promises go here
       .flatMap(promise=>promise)    // resolve the promise under the hood         
       .map(response=>response.data)   
       .map(user=>user.name)   
       .subscribe(
           name=>console.log(`name is ${name}`)
           // name is Ervin Howell
           // name is Leanne Graham   
       )

flatMap(promise=>promise)flapMap 将帮助您解决底层的承诺

【讨论】：

【解决方案3】：

一种有点俗气的做法是使用toPromise

Promise.all([behaviour1.toPromise(), behaviour2.toPromise()])

toPromise 将返回一个在底层可观察对象完成时解析的承诺。

但是，由于大多数 observable 在完成之前会发出超过 1 个值，因此像 zip、combineLatest 和 withLatestFrom 这样的操作可能更接近您要查找的内容：

压缩

behavior1 的每个值都被压缩为 behavior2 的值。如果任一输入的值用完，它就会停止，直到该输入再次有值为止。

Observable.zip(behavior1, behavior2)
  .subscribe(([b1, b2]) => console.log(b1, b2))

另见：Docs, Interactive marble diagram

结合最新

与 zip 类似，但每次 behavior1 或 behavior2 发出一个值时都会发出一个值，并将重用另一个 observable 的最新值进行配对。

Observable.combineLatest(behavior1, behavior2)
  .subscribe(([b1, b2]) => console.log(b1, b2))

另见：Docs, Interactive marble diagram

withLatestFrom

与 combineLatest 类似，但只有一个 observable 决定何时发出值。

behavior1.withLatestFrom(behavior2)
  .subscribe(([b1, b2]) => console.log(b1, b2))

另见：Docs, Interactive marble diagram

【讨论】：