takeLast 似乎杀死了可观察的流

Question

我正在尝试设置一个可观察的流，在其中将最近的设置值与上一个设置值进行比较，并且我想根据两者之间的一些变化进行过滤。我在服务中设置了一个 BehaviorSubject（Angular 2，如果这很重要）并创建了返回它的函数：

getFilters() {
    return this.filtersSubject.asObservable();
}

然后在使用它的组件中，我正在尝试这个：

this.eventFilterService.getFilters()
    .takeLast(2)
    .subscribe((data) => console.log(data));

但我没有得到控制台日志。如果我删除 takeLast，我会看到我的数据返回。从文档中，我的假设是，它只会触发两次。我的目标是将takeLast 输入到reduce 然后是一个过滤器。

我使用 takeLast 是不是错了？

Answer 1

takeLast(x) 将等待 Observable 完成，然后才能发出最后的 x 发射。由于您使用主题，因此您需要调用 complete() 才能使其工作。

但您确实想计算两个发射之间的差异。为此，请使用 .buffer() 运算符

Rx.Observable.from([2,3,5,8])
     .bufferCount(2,1)
     .map(buff =>  buff[1] - buff[0])
     .subscribe(delta => console.log(delta))