使用 RXJS 取消嵌套订阅

Question

我花了太多时间试图找出解决此嵌套订阅的最佳解决方案。我尝试过没有成功的mergeMap、flatMap 和switchMap。不幸的是，找到的示例并不是我所需要的，所以我最终只得到一个结果，或者未定义或错误。要修复的代码是：

this.myService.getAll().subscribe(res => {

// res.result is an array of 20 objects
res.result.forEach(m => {
    // for each of them I call 2 different endpoints adding a key with the response
    this.myService.checkFirst(m.id).subscribe(result => {
        m.first = result;
    });
    this.myService.checkSecond(m.id).subscribe(result => {
        m.second = result;
    });
});
// once all subscriptions are fulfilled I would like to return the mapped array
this.dataLength = res.total;
});

Answer 1

试试

this.myService.getAll().pipe(
  switchMap(res => {
    const obs$ = res.result.map(m => {
      return this.myService.checkFirst(m.id).pipe(
        map(first => ({...m, first})),
      );
    });
  
    return forkJoin(obs$).pipe(
      map(result => ({...res, result})),
    ),
  }),
  switchMap(res => {
    const obs$ = res.result.map(m => {
      return this.myService.checkSecond(m.id).pipe(
        map(second => ({...m, second})),
      );
    });
  
    return forkJoin(obs$).pipe(
      map(result => ({...res, result})),
    ),
  }),
).subscribe(/* ... */);

Answer 2

如果我理解你的问题，我会这样处理。

我假设this.myService.getAll() 是某种http 调用，所以您想要的是在此调用返回并且相关的Observable 完成后立即执行某些操作。为此，要使用的运算符是 concatMap，它允许您在源代码（在本例中为 this.myService.getAll()）完成后立即使用后续的 Observable。

现在，一旦您检索到 this.myService.getAll() 的结果，您需要为返回的数组中的每个项目发出 2 次调用。这样的调用可以并行运行，并且每个都有更新项目的某些属性的副作用。

对于并行运行 2 个调用，您可以使用 forkJoin 函数，该函数返回一个 Observable，在两个调用完成后立即发出，并发出一个包含每次调用结果的数组。换句话说，这段代码应该为单个项目完成工作

forkJoin([this.myService.checkFirst(m.id), this.myService.checkSecond(m.id)]).pipe(
   tap(([first, second]) => {
     m.first = first;
     m.second = second;
   })
)

由于您的数组中有 20 个项目，因此您需要运行上述逻辑的 20 次，可能是并行运行。如果是这种情况，您可以再次使用forkJoin 对数组中的每个项目运行上述请求。

因此，将它们拼接在一起，您的解决方案可能看起来像这样

this.myService.getAll().pipe(
  concatMap(res => {
    // reqestForItems is an array of Observables, each Observable created by calling the forkJoin that allows us to run the 2 calls in parallel
    const reqestForItems = res.result.map(m => 
      forkJoin([this.myService.checkFirst(m.id), this.myService.checkSecond(m.id)]).pipe(
        tap(([first, second]) => {
          m.first = first;
          m.second = second;
        })
      )
    )
    // return the result of the execution of requests for the items
    return forkJoin(reqestForItems).pipe(
      // since what is requested as result is the array with each item enriched with the data retrieved, you return the res object which has been modified by the above logic
      map(() => res)
    )
  })
)
.subscribe(res => // the res.result is an array of item where each item has been enriched with data coming from the service)

如果您必须处理 Observble 和 http 用例，您可能会发现 this article about Observable and http patterns 很有趣。

Answer 3

RxJS 的一个很好的特性是您可以任意深度地嵌套流。因此，如果您可以构建一个丰富单个对象的流，那么您可以嵌套其中的 20 个来丰富整个数组。

因此，对于一个丰富的对象，将丰富的对象打印到控制台的流可能如下所示：

const oneObject = getObject();
forkJoin({
  firstResult: this.myService.checkFirst(oneObject.id),
  secondResult: this.myService.checkSecond(oneObject.id)
}).pipe(
  map(({firstResult, secondResult}) => {
    oneObject.first = firstResult;
    oneObject.second = secondResult;
    return oneObject;
  })
).subscribe(
  console.log
);

如果oneObject 本身是从可观察对象返回的，那么同样的事情会是什么样子？是一样的，只是现在我们将对象合并或切换到我们在上面创建的同一流中。

this.myService.getOneObject().pipe(
  mergeMap(oneObject => 
    forkJoin({
      firstResult: this.myService.checkFirst(oneObject.id),
      secondResult: this.myService.checkSecond(oneObject.id)
    }).pipe(
      map(({firstResult, secondResult}) => {
        oneObject.first = firstResult;
        oneObject.second = secondResult;
        return oneObject;
      })
    )
  )
).subscribe(
  console.log
);

现在，还剩一步。为整个对象数组执行所有这些操作。为了实现这一点，我们需要一种方法来运行一组可观察对象。幸运的是，我们有 forkJoin - 我们用来同时运行 checkFirst 和 checkSecond 的运算符。它也可以将整个事物连接在一起。可能看起来像这样：

this.myService.getAll().pipe(
  map(allRes =>
    allRes.result.map(m => 
      forkJoin({
        first: this.myService.checkFirst(m.id),
        second: this.myService.checkSecond(m.id)
      }).pipe(
        map(({first, second}) => {
          m.first = first;
          m.second = second;
          return m;
        })
      )
    )
  ),
  // forkJoin our array of streams, so that your 40 service calls (20 for 
  // checkFirst and 20 for checkSecond) are all combined into a single stream.
  mergeMap(mArr => forkJoin(mArr)),
).subscribe(resultArr => {
  // resultArr is an aray of length 20, with objects enriched with a .first
  // and a .second
  // Lets log the result for he first object our array.
  console.log(resultArr[0].first, resultArr[0].second)
});

这是我将map 和mergeMap 合并为一个mergeMap 的相同解决方案：

this.myService.getAll().pipe(
  mergeMap(allRes =>
    forkJoin(allRes.result.map(m => 
      forkJoin({
        first: this.myService.checkFirst(m.id),
        second: this.myService.checkSecond(m.id)
      }).pipe(
        map(({first, second}) => {
          m.first = first;
          m.second = second;
          return m;
        })
      )
    ))
  )
).subscribe(console.log);

如果您不确定checkFirst 和checkSecond 是否完整，您可以使用zip 代替forkJoin，然后使用take(1) 或first() 取消订阅

this.myService.getAll().pipe(
  mergeMap(allRes =>
    forkJoin(allRes.result.map(m => 
      zip(
        this.myService.checkFirst(m.id),
        this.myService.checkSecond(m.id)
      ).pipe(
        first(),
        map(([first, second]) => {
          m.first = first;
          m.second = second;
          return m;
        })
      )
    ))
  )
).subscribe(console.log);