Angular RXJS拆分返回值，多次调用后重新加入

发布于10月11日

我订阅了一个可观察对象，得到了一个array.对于数组中的每个元素，我希望执行一次异步计算(可以使用From将其作为观察值返回)，当所有计算完成后，我希望发布所有答案.我该怎么做呢？

我猜它必须与forkJoin和mergeMap一起工作，但我似乎做不对.

myObservable.pipe(
  tap(data => {
    console.info(data);
    // [{id: 0}, {id: 1}, {id: 2}, ...]
  }),
  mergeMap(data => {
    // ??? Need help here
    const calculatedData = data.map(d => {...d, calculated: myCalculation(d.id).subscribe() });
    // or: const calculatedData = data.map(d => ({...d, calculated: myPromise(d.id)}));
    return forkJoin(calculatedData);
  }),
  tap(finalData => {
    console.info(finalData);
    // [{id: 0, calculated: "0000"}, {id: 1, calculated: "0001"}, {id: 2, calculated: "0002"}]
  }),
  switchMap(finalData => {
    return this.httpClient.post("/url", finalData);
 })
).subscribe();

推荐答案

如果你不想麻烦你--SwitchMap可以是异步的，它可以返回一个promise .

myObservable.pipe(
  tap((data) => {
    console.info(data);
    // [{id: 0}, {id: 1}, {id: 2}, ...]
  }),
  switchMap(async (data) => {
    const calculatedDataPromises = data.map(async (d) => {
      return {
        ...d,
        calculated: await myPromise(d.id)
      };
    });
    return Promise.all(calculatedDataPromises);
  }),
  tap((finalData) => {
    console.info(finalData);
    // [{id: 0, calculated: "0000"}, {id: 1, calculated: "0001"}, {id: 2, calculated: "0002"}]
  }),
  switchMap((finalData) => {
    return this.httpClient.post("/url", finalData);
  })
).subscribe();

没有promise 的例子:

myObservable$
  .pipe(
    tap((data) => {
      console.info(data);
      // [{id: 0}, {id: 1}, {id: 2}, ...]
    }),
    switchMap((data) =>
      combineLatest(
        data.map((d) =>
          myCalculation(d.id).pipe(
            catchError((err) => {
              console.warn(`Error loading calculated for ${d.id}`, err);
              return of(null);
            }),
            map((res) => ({ ...d, calculated: res }))
          )
        )
      )
    ),
    tap((finalData) => {
      console.info(finalData);
      // [{id: 0, calculated: "0000"}, {id: 1, calculated: "0001"}, {id: 2, calculated: "0002"}]
    }),
    switchMap((finalData) => {
      return this.httpClient.post("/url", finalData);
    })
  )
  .subscribe();