typescript - RxJS：等待 Subscription 接收下一个值

我在一个 Angular 项目中使用 RxJS，并且我有一个对定期发出值的 Observable 的 Subscription。

在同一个组件中，我也使用 RxJS（结合 Angular http 客户端）来进行网络调用并订阅它。

我想做的很简单：当我进行网络调用并返回结果时，我想等待 Subscription 接收到下一个值之后再处理结果。让我们通过一个代码示例更好地看看:

假设我有这个订阅，它以固定间隔从可观察对象接收新值：

tenantStatusSub: Subscription;

我想做的事情在评论中解释如下：

this.httpClient.get<any>(...).subscribe({
  next: result => {
      /* I want to wait here for tenantStatusSub to receive a value before proceeding */
  },
  error: err => {
    //...
  }
});

我相信一定有一种直接的方法可以做到这一点，但我还在学习 RxJS，所以我不确定哪种方式是最好的/最干净的。

答案＃1

我认为你不能直接等待 tenantStatusSub-subscription 本身。然而，你可以等待 tenantStatusSub-subscription 所指向的可观察对象的下一次发射。我建议你使用 switchMap() 直接将你的 HTTP 请求链到接收连续发射的可观察对象上。此外，你可以使用 first()，在第一个值到达后触发“取消订阅”:

tenantStatusObs$: Observable<any> = observableThatReceivesAFlowOfValues;

ngOnInit(): void {
    this.httpClient.get<any>('http://www.some-url.com').pipe(
        switchMap(resultFromApi => {
            return this.tenantStatusObs$.pipe(
                first(), // unsubscribes after the first value has been emitted
                tap(tenantStatus => {
                    console.log('Tenant Status:', tenantStatus);
                    console.log('Result from API:', resultFromApi);
                })
            )
        })
    )
    .subscribe();
}

答案＃2

CombineLatest 可以做到这一点，但效果不太好。如果 tenantStatusSub 在获取响应的时间间隔内定期发出数据，这个方法是可行的。即便如此，下面使用 switchMap 的方法（虽然稍微复杂一些）应该会更稳健。

combineLatest([
  tenantStatusSub,
  this.httpClient.get<any>(...)
]).pipe(
  skip(1),
  take(1)
).subscribe({
  next: result => {
    // both values in result here
  },
  error: err => {
    //...
  }
});

更新：

一旦你获得了 HTTP 响应，获取下一个租户状态并将两个值合并为单一输出。由于切换到 tenantStatusObs$ 是长期的，你需要通过 take(1) 取消订阅。

this.httpClient.get<any>(...).pipe(
  switchMap(httpRes => this.tenantStatusObs$.pipe(
    map(tenantStatus => ({ httpRes, tenantStatus })
  )),
  take(1)
).subscribe({
  next: ({ httpRes, tenantStatus }) => {
    // both values available here
  },
  error: err => {
    //...
  }
});

旁白：

在这种情况下，你可以在 map 或 switchMap 操作符之后写 take(1)。它们会产生相同的输出。然而，从语义上讲，它们的含义是不同的。

After map 的意思是：“每个 tenantStatusSub 发射最多一个（配对的）发射”
switchMap 之后意味着：“最多一次（配对）发射”