Angular 订阅两个 Observable 并返回两者的结果

Question

我正在尝试组合两个 Observable，并根据一个或两个的结果发出一个事件。

过度简化的代码：

export class MyDirective {
  // Filter properties
  filter = new ReplaySubject<any>();
}

export class MyComponent implements AfterContentInit {
  // First observable (my own creation)
  @ContentChildren(MyDirective) columns: MyDirective[];
  filterChange = Observable.merge(...this.columns.map(c => c.filter));

  // Second observable (EventEmitter from a component library)
  @ViewChild(MdSort) sort: MdSort;

  ngAfterContentInit() {
    // Now merge them and call a function with the results from both
    Observable.xxx([ this.sort.sortChange, this.filterChange ]).subscribe(res => {
      this.readData(res.sort, res.filter);
    });
  }

  readData(sort?: any, filter?: any) {
    // This method should receive the results from the observables
  }
}

解决方案尝试：

我尝试了不同的方法来合并这些：

Observable.merge

Observable.merge(this.sort.sortChange, this.filterChange).subscribe(res => {
  // Gives me no clue as to which of the observables has been fired
  // and thus I cannot process properties 
});

Observable.zip

Observable.zip(this.sort.sortChange, this.filterChange, (sort, filter) => {
  // I never reach this spot. Why?
  this.readData(sort, filter);
});

同时订阅两者

这可行，但我对此不满意。我希望以某种方式使用 Observable 合并来完成此操作...

// On any changes, read data
let sort; let filter;
this.sort.sortChange.subscribe(s => sort = s);
this.filterChange.subscribe(f => filter = f);
this.Observable.merge(this.sortChange, this.filterChange).subscribe(result => {
  // Fired whenever any of the two are changed
  this.readData(sort, filter);
});

Answer 1

您可以使用 distinctUntilchanged() 和 map() 或 mapTo() 来为您提供有关哪个流已发出值的信息

this.Observable.merge(this.sortChange.distinUntilChanged().mapTo('sort'), 
this.filterChange.distinUntilChanged().mapTo('filter').subscribe(result => {
  // Fired whenever any of the two are changed
 return this.readData(sort, filter);
});

取决于 sortchange 或 filterChange 是否发出单个值，您可能必须应用自定义比较器逻辑才能使 distinctUntilchanged 工作。例如

var source = Rx.Observable.of({value: 42}, {value: 42}, {value: 24}, {value: 24})
 .distinctUntilChanged(function (x) { return x.value; }, function (a,b) { 
return a !== b; });

参考：https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/distinctuntilchanged.md