여러 rxjs BehaviourSubjects를 결합하는 방법

Question

저는 Angular2 응용 프로그램을 구축하고 논리적으로 하나의 가입으로 결합하려는 두 개의 BehaviourSubjects을 가지고 있습니다. 나는 두 가지 HTTP 요청을하고 있는데 둘 다 돌아올 때 이벤트를 시작하려고합니다. 나는 forkJoin 대 combineLatest을보고있다. behaviourSubjects가 업데이트되고 모든 behavoirSubjects가 업데이트 된 후에 만 ​​forkJoin이 시작되면 combineLatest가 시작되는 것 같습니다. 이 올바른지? 거기에 일반적으로 받아 들여지는 패턴이 있어야합니다.

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

하지만 그 대신 : 여기

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

내 구성 요소의 가입의 조각입니다 : 여기에 편집

내 angular2 구성 요소에 가입되어 내 behaviorSubjects 중 하나의 예입니다 단일 구독에 대해 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 :

,210

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