typescript - Dynamically filtering rxjs stream -

i'm using rxjs , i'm looking dynamically filter data, i'm having problems:

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);  // <- think wrong });  console.log("a"); filtereddata.subscribe(console.log);  numbersource.next(1); numbersource.next(10); numbersource.next(100);  console.log("b"); numberfilter.next(50); 

what doing subscribing numbersource, data interested in displaying. subscribing numberfilter, because want changes replay subject, think have done wrong.

i expecting see:

a 1 b 1 10 

i seeing:

a 1 1 b 1 10 

can help?

i think understand you're trying do. want stack value emitted numbersource able reemit , filter them when numberfilter changes.

the major problem in implementation numberfilter behaviorsubject emits default value (5 in case) every time subscribe happens right on line numberfilter.subscribe(newfilter => .... callback subscribes filtereddata , again right after console.log("a");. haven't started emitting data numbersource , you've subscribed twice. that's why gives 1 2 times.

easy solution use classic subject , remember unsubscribe() previous subscription filtereddata:

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

see live demo: http://plnkr.co/edit/voad8tcwllrdfzu14ufw

now gives output wanted.