分别为第二个 Observable 调用 RxJS concatMap 订阅

Question

我正在编写一个 Angular 6 应用程序。我希望它在路由更改以及任何未决的 http 请求时显示加载动画。所以我有 2 个 Observables 看起来像这样。对于 httpPendingRequests，我使用 Angular 的 HttpInterceptor 编写的计数器来实现；对于 locationChanged，我已经订阅了 Router 的 NavigationStart/NavigationEnd 事件。

httpPendingRequests: BehaviorSubject<number>;
locationChanged: BehaviorSubject<boolean>;

我正在使用 concatMap 订阅这 2 个 Observable，代码如下所示：

this.locationChanged.pipe(
  concatMap(changed => {
    console.log('location change:' + changed);
    if (changed) {
      return this.httpPendingRequests;
    } else {
      return of(0);
    }
  }),      
  map(count => count > 0)
).subscribe(val => { 
  console.log('isloading: ' + val);
});

所以我希望这仅在位置已更改时将“正在加载”记录到控制台，以及是否有任何待处理的请求。它在这种情况下确实有效。但是我发现当只有待处理的 http 请求但位置没有改变时，它也会记录“正在加载”消息。这让我很困惑，我认为运营商正在确保按顺序订阅 Observables？如果第一个（位置更改）没有发出，那么第二个（挂起的请求）不应该被触发吗？我对这个概念的理解有误吗？

此外.. 我还尝试了其他方法来组合 Observables、zip、forkJoin、combineLatest - 在我的情况下，它们都只触发了一次订阅，所以我也不太确定出了什么问题。

如果需要，我很高兴提供更多信息。提前致谢

Answer 1

您可能想看看您使用的是BehaviorSubject。

BehaviorSubject 始终需要在创建时发出一个值。这意味着locationChanged 总是发出至少一个通知。

如果您更愿意使用Subject，您可以控制何时发出第一个通知。

在以下示例中，直接取自您的代码，您看不到任何记录，因为 locationChanged 从未发出。

const httpPendingRequests = new BehaviorSubject<number>(0);
const locationChanged = new Subject<boolean>();

locationChanged.pipe(
    concatMap(changed => {
      console.log('location change:' + changed);
      if (changed) {
        return httpPendingRequests;
      } else {
        return of(0);
      }
    }),      
    map(count => count > 0)
  ).subscribe(val => { 
    console.log('isloading: ' + val);
});

setTimeout(() => {
    httpPendingRequests.next(1);
}, 100);
setTimeout(() => {
    httpPendingRequests.next(2);
}, 200);
setTimeout(() => {
    httpPendingRequests.next(3);
}, 300);

Answer 2

可以通过使用combineLatest observable 和map 运算符来解决。这是演示：https://stackblitz.com/edit/so-rxjs-concat。

代码在app.component.ts。查看console.log 值以了解更多信息。