结合 Observable 和 Promise

Question

我有一个服务 MyService，它有一个方法 getList()，它 return 是一个项目数组的 Observable，类似于

getList(): Observable<Item[]>

获取列表后，每个Item都需要经过MyService、fillItem(Item)的第二种方法。此方法向某些后端发出请求，其中 return 是一个 Promise，类似于：

fillItem(item: Item) {
   BackEnd.retrieveDetails(item.id)
          .then(data => item.data = data);
}

我需要的是结合以上两种方法，以便 MyService 能够 return 一个 Observable，一旦所有的项目都被填充了数据（即一旦所有的 Promises 都有）已解决）。

到目前为止我所尝试的是这些方面的东西

fillAll(items: Item[]) {
   const promises = [];
   for (let i = 0; i < items.length; i++) {
      const item = items[i];
      promises.push(this.fillItem(item));
   }
   const subject = new Subject<Item[]>();
   Promise.all(promises).then(res => {
                subject.next(items);
            },
            err => {
                subject.error(err);
            });
   return subject.asObservable();
}

getListWithFilledItems() {
   return this.getList()
              .map(items => this.fillAll(items))
}

不幸的是，如果我这样做，方法 getListWithFilledItems() return 是一个 Observable<Observable<Item[]>> 而不是我希望的 Observable<Item[]>.

任何关于如何更改代码以便方法 getListWithFilledItems() returns 和 Observable<Item[]> 的建议将不胜感激。

Answer 1

我认为您可以只使用 mergeMap():

而不是 .map(items => this.fillAll(items))

.mergeMap(items => this.fillAll(items))

这将订阅从 fillAll() 返回的 Observable，并在准备就绪时重新发出它的值。

顺便说一句，请注意您没有在内部 subject 上调用 complete()，因此 Observable 链未正确处理（但是在您的用例中它可能无关紧要） .

结合 Observable 和 Promise

combine Observable and Promise

reactive-programming

system.reactive

rxjs

rx-java