动态过滤 rxjs 流 [英] Dynamically filtering rxjs stream

问题描述

我正在使用 RXJS 并且希望动态过滤数据，但我遇到了问题:

I'm using RXJS and I'm looking to dynamically filter the data, but 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);  // <- 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，因为我希望对其进行任何更改以重播主题，但我认为我做错了.

What I am doing is subscribing to numberSource, that is the data I am interested in displaying. I am also subscribing to numberFilter, because I want any changes to that to replay the subject, but I think I have done that wrong.

我期待看到:

A
1
B
1
10

我看到了:

A
1
1
B
1
10

有人可以帮忙吗?

推荐答案

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

I think I understand what you're trying to do. You want to stack all value emitted by numberSource to be able to reemit and filter them when numberFilter changes.

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

The major problem in your implementation is that numberFilter is a BehaviorSubject which emits its default value (5 in this case) every time you subscribe to it which happens right on the line numberFilter.subscribe(newFilter => .... This callback subscribes to filteredData and then again right after console.log("A");. So you haven't even started emitting data to numberSource and you've already subscribed twice. That's why it gives you 1 two times.

简单的解决方案是使用经典的Subject，并记得unsubscribe()之前订阅filteredData:

Easy solution is to use classic Subject and remember to unsubscribe() the previous subscription to 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

现在它为您提供了您想要的输出.

Now it gives you the output you wanted.

这篇关于动态过滤 rxjs 流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！