RxJS - 使用 forEach 的多个请求并等待全部完成

Question

我有这个案例：

• 首先我调用一个服务，并得到一个项目列表。（对象数组）
• 对于此列表中的每个项目，我调用另一个服务，它们实际上都可以并行触发。
• 我必须等待所有的答案。当我有了所有的答案时，我有了一个最终的逻辑。

我现在有这样的事情，而没有正确使用 RxJS：

this.service.readArray().subscribe((array: Object[]) => {
  if (array.length > 0) {
    array.forEach((item, index) => {

      this.service2.readItem(item.id)
        .subscribe(details => {
          item.details = details;
          // manually finally logic
          if (index === array.length - 1) { // if the last iteration
            ...
          }
        }, (response: HttpErrorResponse) => {
            ...
          // manually finally logic also for error part
          if (index === array.length - 1) { // if the last iteration
          ... 
          }
        });

    });
  } else {
    ... logic for no items in list
  }
}, (error) => {
  ...
});

我如何在 Rxjs (5) 语句中表示这一点？

Answer 1

您可以使用 forkJoin 等待所有调用完成。看起来您正在使用 rxjs 5 [正如您在问题中提到的] 所以让我们像这样更改代码 [请参阅代码注释中的描述]：

this.service.readArray()
        .switchMap(array => {
          //lets map the array member to the respective observable
          const obs$ = array.map(item => {

             return this.service2.readItem(item.id)
                        .pipe(
                           catchError(err => {

                             //Do whatever you want to do with this error
                             //make sure to return an observable as per your logic. For this example, I am simply returning the err wrapped in an observable. Having catchError operator will gracefully handle the exception and make sure to emit the value as part of forkJoin.
                             return of(err);

                           })
                         )
          });
          //forkJoin will wait for all the readItem calls get finished.
          return forkJoin(obs$);
        })
        .subscribe((finalArray) => {
          //finalArray will be the array of object [an response of this.service2.readItem(item.id)]
          console.log(finalArray);
          //do whatever you want to do with the array
        });

编辑 - 按照 OP 的要求 - 掌握 readArray

this.service.readArray()
        .switchMap(array => {
          //lets map the array member to the respective observable
          const obs$ = array.map(item => {

             return this.service2.readItem(item.id)
                        .pipe(
                           catchError(err => {

                             //Do whatever you want to do with this error
                             //make sure to return an observable as per your logic. For this example, I am simply returning the err wrapped in an observable. Having catchError operator will gracefully handle the exception and make sure to emit the value as part of forkJoin.
                             return of(err);

                           })
                         )
          });
          //forkJoin will wait for all the readItem calls get finished.
          return forkJoin(obs$)
                  .pipe(
                    //return the original array along with joined using of
                    mergeMap((joined) => {
                      return of([array, joined]); 
                    })
                  );
        })
        .subscribe((finalArray) => {
          //finalArray will have readArray API response [i.e. array] at 0 index and on 1st index it will have joined array
          console.log(finalArray);
          //do whatever you want to do with the array
        });