複数のrxjsを組み合わせる方法BehaviourSubjects

Question

私はAngular2アプリを構築しています。論理的に1つのサブスクリプションに結合する2つのBehaviourSubjectsがあります。私は2つのHTTP要求を出していて、両方が復帰したときにイベントを発生させたい。私はforkJoin対combineLatestを見ています。どちらかのbehvaviorSubjectsが更新されたときにcombineLatestが起動するように見えますが、forkJoinはすべてのbehavoirSubjectsが更新された後にのみ起動します。これは正しいです？これには一般的に受け入れられているパターンがありますか？ここで

EDIT
は私のangular2コンポーネントがに加入している私のbehaviorSubjectsの1の例である：ここで

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（）を起動するだけです。 http://reactivex.io/documentation/operators/zip.htmlが

zipとcombineLatestの違いは次のとおりです：

Answer 1

あなたがcombineLatestまたはforkJoinに似ていますが、両方のストリームが放出されている場合にのみ、トリガーzip演算子はを、使用することができますだけで」をトリガーします ジップパラレル "ですが、combineLatestは任意のアップデートでトリガーし、各ストリームの最新の値を出力します。

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

zipと：次の2つのストリームを想定したがって 、

• "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); 
       } 
      });