我如何通过 Rxjs 中的简单对象缓冲可观察对象？

Question

我想通过一些任意（但简单）的标准来缓冲可观察到的。我在这里设置了一个简单的例子：

const observable = Rx.Observable.from([1,2,3])
const filtered = observable.filter((n) => n === 3);
observable
  .buffer(filtered)
  .subscribe((n) => {
    // Why is this empty?
    console.log(n);
});

Bin here

尝试使用 filter 执行此操作只会生成一个空数组。我期待 [1,2,3] 的数组，但这似乎不是它的工作原理。 buffer 的所有文档都使用像计时器这样的异步事件，但这不是我想要的。我只想根据我决定的某个任意标准获取最后 n 个项目。

非常感谢您的帮助！

Answer 1

您的订阅是在值发出后进行的。如果你想让它工作：

const observable = new Rx.Subject()
const filtered = observable.filter((n) => n === 3);
filtered
  .buffer(filtered)
  .subscribe((n) => {
   console.log(n);
});
Rx.Observable.from([1,2,3]).subscribe(observable);

Answer 2

根据 olsn 的评论，这似乎是 Rxjs 5 中的一个错误。更改为 v4 有效地解决了这个问题。

我如何通过 Rxjs 中的简单对象缓冲可观察对象？

How can I buffer an observable by a simple in Rxjs?

javascript

reactive-programming

node.js

rxjs

rxjs5

Bin here