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

【问题标题】：How can I buffer an observable by a simple in Rxjs?如何通过简单的 Rxjs 缓冲可观察对象？
【发布时间】：2017-01-04 17:32:01
【问题描述】：

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

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] 的数组，但这似乎不是它的工作方式。缓冲区的所有文档都使用了像计时器这样的异步事件，但这不是我想要的。我只是想根据我决定的一些任意标准来获取最后 n 项。

非常感谢您的帮助！

【问题讨论】：

如果你想缓冲到每第三个项目，你可以使用bufferCount(3)
检查您的 JSBin 后，您的问题似乎与此处的此错误修复有关，该错误修复似乎与下一个次要版本一起发布：github.com/ReactiveX/rxjs/pull/2195
抱歉，要明确一点：3 只是一个例子，所以 bufferCount 不是我想要的（它可以是任何数字）。我不能让补丁做我想做的事，但无论如何 PR 似乎还没有登陆。 v4 似乎可以工作，但是它在完成时会发出一个最终的空数组。这是预期的吗？这里的垃圾箱：jsbin.com/pizugew/edit?html,js,console
我很确定这是我发布链接的错误，您的行为不是预期的
你可以使用scan，但是我建议你为此创建一个新问题，因为评论部分的长时间对话对可能有相同问题的其他人没有帮助跨度>

标签： javascript node.js rxjs reactive-programming rxjs5

【解决方案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);

【讨论】：

但这不会将缓冲的 observable 与订阅连接起来......还是我错过了什么？

【解决方案2】：

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

【讨论】：