typescript - 动态过滤 rxjs 流

Question

我正在使用 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

有人能帮忙吗？

Answer 1

我想我明白你想做什么。您希望堆叠 numberSource 发出的所有值，以便能够在 numberFilter 更改时重新发出和过滤它们。

您的实现中的主要问题是 numberFilter 是一个 BehaviorSubject，每次您发出默认值(在本例中为 5)订阅它发生在 numberFilter.subscribe(newFilter => ... 行。此回调订阅 filteredData 然后在 console.log 之后再次订阅("A");。所以你甚至还没有开始向 numberSource 发送数据，你已经订阅了两次。这就是为什么它给你 1两次。

简单的解决方案是使用经典的 Subject 并记住 unsubscribe() 之前对 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);

查看现场演示:http://plnkr.co/edit/vOaD8tcWlLRdfzU14Ufw

现在它会为您提供所需的输出。