동적 rxjs 필터링 스트림

Question

내가 RXJS을 사용하고 있는데 나는 동적으로 데이터를 필터링 할 찾고 있어요,하지만 난 문제가 발생이 : 내가 numberSource에 가입되어있는 중이 야

let numberSource: ReplaySubject<Number> = new ReplaySubject<Number>(); 
let numberFilter: BehaviorSubject<Number> = new BehaviorSubject<Number>(5); 

let filteredData = numberSource.filter(n => n < numberFilter.value); 
numberFilter.subscribe(newFilter => { 
    filteredData = numberSource.filter(n => n < newFilter); 
    filteredData.subscribe(console.log); // <- I think this is wrong 
}); 

console.log("A"); 
filteredData.subscribe(console.log); 

numberSource.next(1); 
numberSource.next(10); 
numberSource.next(100); 

console.log("B"); 
numberFilter.next(50); 

, 즉 데이터입니다 나는 전시에 관심이있다. 나는 numberFilter에도 가입하고 있는데, 그 이유는 피사체를 다시 재생하기 위해서 그 부분을 변경하기를 원하기 때문입니다. 그러나 나는 잘못했다고 생각합니다.

A 
1 
B 
1 
10 

을 내가보고 있어요 :

A 
1 
1 
B 
1 
10 

는 사람이 도와 줄 수

내가 볼 것으로 예상하고있다?

Answer 1

나는 당신이하려는 것을 이해하고 있다고 생각합니다. numberSource에 의해 방출 된 모든 값을 스택하여 numberFilter이 변경 될 때 다시 채우고 필터링 할 수 있습니다.

구현의 주요 문제는 numberFilter가 디폴트 값 (이 경우 5) 오른쪽 라인 numberFilter.subscribe(newFilter => ...에서 발생 당신이에 가입 할 때마다 방출하는 BehaviorSubject 것입니다. 이 콜백은 filteredData을 구독하고 console.log("A"); 바로 다음에 다시 구독합니다. 그래서 당신은 심지어 numberSource에 데이터를 방출하지 않았으며 이미 두 번 가입했습니다. 그것이 당신에게 1을 두 번주는 이유입니다. http://plnkr.co/edit/vOaD8tcWlLRdfzU14Ufw

지금 당신에게 당신이 원하는 출력을 제공합니다

let numberSource: ReplaySubject<Number> = new ReplaySubject<Number>(); 
let numberFilter: Subject<Number> = new Subject<Number>(); 

var subscription; 
numberFilter.subscribe(newFilter => { 
    if (subscription) { 
    subscription.unsubscribe(); 
    } 

    subscription = numberSource.filter(n => n < newFilter) 
    .subscribe(console.log); 
}); 

numberFilter.next(5); 

console.log("A"); 

numberSource.next(1); 
numberSource.next(10); 
numberSource.next(100); 

console.log("B"); 
numberFilter.next(50); 

라이브 데모를 참조하십시오

쉬운 해결책은 고전 Subject를 사용하고 filteredData에 unsubscribe() 이전 구독 기억하는 것입니다.