如何组合多个rxjs BehaviourSubjects

Question

我正在构建一个Angular2应用程序,并且有两个BehaviourSubjects我想在逻辑上组合成一个订阅.我正在制作两个http请求,并希望在两个人都回来时发起一个事件.我期待在forkJoinVS combineLatest.似乎当更新behvaviorSubjects时,combineLatest将会触发,而forkJoin只会在所有behavoirSubjects更新后触发.它是否正确？必须有一个普遍接受的模式,这不是吗？

编辑
这是我的angular2组件订阅的一个behaviorSubjects的示例:

export class CpmService {

    public cpmSubject: BehaviorSubject<Cpm[]>;

    constructor(private _http: Http) {
        this.cpmSubject = new BehaviorSubject<Cpm[]>(new Array<Cpm>());
    }

    getCpm(id: number): void {
        let params: URLSearchParams = new URLSearchParams();
        params.set('Id', id.toString());

        this._http.get('a/Url/Here', { search: params })
            .map(response => <Cpm>response.json())
            .subscribe(_cpm => {
                this.cpmSubject.subscribe(cpmList => {
                    //double check we dont already have the cpm in the observable, if we dont have it, push it and call next to propigate new cpmlist everywheres
                    if (! (cpmList.filter((cpm: Cpm) => cpm.id === _cpm.id).length > 0) ) {
                        cpmList.push(_cpm);
                        this.cpmSubject.next(cpmList);
                    }
                })
            });
    }
}

以下是我的组件订阅的片段:

  this._cpmService.cpmSubject.subscribe(cpmList => {
      doSomeWork();
  });

但是,对于单个订阅,我不想触发doSomeWork(),而只是在cpmSubject和fooSubject触发时才触发doSomeWork().

Answer 1

您可以使用zip-operator,它的工作方式类似于combineLatest或forkJoin,但仅在两个流都发出时触发:http://reactivex.io/documentation/operators/zip.html

zip和之间的区别combineLatest是:Zip将仅触发"并行",而combineLatest将触发任何更新并发出每个流的最新值.因此,假设以下2个流:

streamA => 1--2--3
streamB => 10-20-30

用zip:

• "1,10"
• "2,20"
• "3,30"

用combineLatest:

• "1,10"
• "2,10"
• "2,20"
• "3,20"
• "3,30"

这也是一个实例:

const a = new Rx.Subject();
const b = new Rx.Subject();

Rx.Observable.zip(a,b)
  .subscribe(x => console.log("zip: " + x.join(", ")));
Rx.Observable.combineLatest(a,b)
  .subscribe(x => console.log("combineLatest: " + x.join(", ")));

a.next(1);
b.next(10);
a.next(2);
b.next(20);
a.next(3);
b.next(30);

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

另一个旁注:从来没有订阅过订阅.做这样的事情:

this._http.get('a/Url/Here', { search: params })
            .map(response => <Cpm>response.json())
            .withLatestFrom(this.cpmSubject)
            .subscribe([_cpm, cpmList] => {
                if (! (cpmList.filter((cpm: Cpm) => cpm.id === _cpm.id).length > 0) ) {
                    cpmList.push(_cpm);
                    this.cpmSubject.next(cpmList);
                }
            });