将 Promise.all 转换为 Observable答案

【问题标题】：Convert Promise.all into Observable将 Promise.all 转换为 Observable
【发布时间】：2016-10-04 18:44:13
【问题描述】：

我需要进行几个相互依赖的异步调用。我最初编写代码并使用Promise.all 逐步制作async。我遍历我的数据并创建了一个async 方法，以便将所有需要的操作放入一个数组中以传递给Promise.all()。这很好用，但我怎么能用 Observables 做同样的事情。我读过forkJoin 相当于Promise.all，但是我如何循环数据并包装我的async 函数，然后在移动到下一个flatMap 之前执行它？

public getMonthly(){
return this.http.get(url)
            .flatMap(response => {
                // need to convert this?
                let actions = response.json().map(this.asyncMonthlyRecord);
                return Promise.all(actions); 
            })
            .flatMap(()=> this.queryMonthly())
            .map(this.convertFromSQl)
            .catch((error:any)=> Observable.throw(error || 'Server Error'));
}

private asyncMonthlyRecord = (record):Promise<any> => {
      return this.setUsage(record,'HILowMonthly');
}

private queryMonthly(){
      return this.storage.query('SELECT * FROM HILowMonthly')
}

 getMonthly().subscribe(x => console.info(x)); // logs data from SQLite perfectly...

【问题讨论】：

为什么投反对票？

标签： javascript asynchronous rxjs observable

【解决方案1】：

我想你想要的是这样的

Rx.Observable.of({ itemIds: [1, 2, 3, 4, 5 ]})
  .mergeMap(response => {
    const promises = response.itemIds
      .map(id => {
        return new Promise((resolve, reject) => {
          // Wait for some random time, then resolve the promise.
          const timeout = Math.floor(Math.random() * 5000);
          setTimeout(() => {
            console.log(`Finished promise ${id}`); // debug only
            resolve({id, resolved: true}) 
          }, timeout);
        });
      });
    // Wait until all Promises have been resolved, then return all
    // of them in a single and ordered array.
    return Rx.Observable.forkJoin(promises);
  })
  .subscribe(x => console.log(x));

Working code on jsbin

请注意，promise 以任意顺序解析，但以正确的顺序返回。 jsbin 示例中的注释代码还表明，每个 Promise 都可以单独解析并合并回原始流，以防 Promise 的顺序不重要。

【讨论】：