angular - 如何将 Subjects 和 Observables 与 Angular rxjs 结合起来

Question

我有以下两个订阅:

this.service1.source1.subscribe(countries => {
  this.filteredData =  [];
  countries.forEach(country => {
    this.filteredData = this.data.filter(user => {
      return user.country == country;
    });
  });
});

this.service2.source2.subscribe(companies => {
  companies.forEach(company => {
    this.filteredData =  [];
    this.filteredData = this.data.filter(user => {
      return user.company == company;
    });
  });
});

其中 source1 和 source2 是 BehaviorSubject 并且这两个事件都被 subscribe 捕获。

我想合并两个订阅，因为我想根据 subscribe

返回的两个结果过滤数据

我尝试了 forkJoin 和 zip

zip([this.service1.source1, this.service1.source2]).subscribe(results => {
  console.log('zip results:', results);
});

forkJoin([this.service1.source1, this.service1.source2]).subscribe(results => {
  console.log('forkJoin results:', results);
});

但是对于那些(zip 和 forkJoin)，我注意到我什至没有在控制台上得到任何东西，就好像 subscribe 在发出事件时没有被执行一样。

如何将两个来源的结果合并到一个订阅中？

Answer 1

forkJoin 可能不适合您的用例，因为它在所有可观察对象完成时从每个对象发出最后发出的值。

zip 也可能不会给你想要的行为，因为 它等待所有输入流都发出它们的第 n 个值，一旦满足这个条件，它就会结合所有这些第 n 个值并发出第 n 个组合值。

因此，在任何一种情况下，您都不会得到发射，直到两个可观察量都有发射。因为this.service1.source1 和this.service1.source2 是BehaviorSubject使用 zip 保证初始发射。但只有当两个 observable 都发出时才会发生后来的发射。

我建议使用 combineLatest因为每当任何输入流发出一个值时，它都会结合每个输入流发出的最新值。

combineLatest(this.service1.source1, this.service1.source2).subscribe(results => {
  console.log('combineLatest results:', results);
});

并且由于this.service1.source1 和this.service1.source2 是BehaviorSubject这样:

A variant of Subject that requires an initial value and emits its current value whenever it is subscribed to.

保证您在订阅它时以及任何可观察对象发出值时都会获得发射。