typescript - 动态过滤rxjs流

我正在使用rxjs，并希望动态筛选数据，但我遇到了问题：

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

我所做的是订阅numbersource，这是我感兴趣显示的数据。我也订阅了numberfilter，因为我希望对它进行任何更改以重播主题，但我认为我做错了。
我希望看到：

A
1
B
1
10

我看到：

A
1
1
B
1
10

有人能帮忙吗？

最佳答案

我想我明白你想做什么。您希望对numberSource发出的所有值进行堆栈，以便在numberFilter更改时能够重新提交和过滤它们。
实现中的主要问题是numberFilter是一个BehaviorSubject，它在每次订阅它时都会发出它的默认值（在本例中为5，它发生在numberFilter.subscribe(newFilter => ...行上。此回调订阅filteredData，然后在console.log("A");之后再次订阅。因此，您甚至还没有开始向numberSource发送数据，而且已经订阅了两次。这就是为什么它会给你两次。
简单的解决方案是使用经典的1并记住Subject以前对unsubscribe()的订阅：

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

观看现场演示：http://plnkr.co/edit/vOaD8tcWlLRdfzU14Ufw
现在它给了你想要的输出。