RxJs——在每次事件爆发后重播所有事件

Question

你是怎么做到的?RxJs对我来说仍然是个谜。

我试着这样做:

filterChanges
   .delay(400)
   .replay()
   .reduce(function(acc,x) { return acc.concat(x) }, [])
   .subscribe(function(changes) {
      console.log(changes);
      ...

或

 filterChanges.subscribe(function() {
     filterChanges.aggregate(function(changes) {
          ...

我真的很迷路了。我想要的行为是:某些用户操作导致多个过滤器更改。我不想一次一个地处理它们，而只想在变化结束时处理它们。但是当我处理它们时，我想要从流开始的所有过滤器更改。

现在我写了这篇文章，我意识到最好是在一次变化之后得到所有的过滤器，而不是所有的变化本身，所以我只需要抓住一次变化的结束。我想知道这两个问题的答案，因为我认为这会帮助我更好地理解RxJs。

1. 订阅一系列事件的结束。
2. 订阅事件爆发的结束，并从事件中捕获所有事件

Answer 1

如果我理解正确的话，应该将buffer与debounce结合使用。缓冲区会给你在缓冲区窗口期间发生的事件列表，debounce将定义何时缓冲区窗口应该关闭，这就是你如何在喷射中获得所有事件。为了从流的最开始获取所有事件，您需要scan从缓冲区中累积所有列表。

var hotFilterChanges = filterChanges.share();
hotFilterChanges
  .buffer(hotFilterChanges.debounce(200))
  .scan((acc, curr) => acc.concat(curr))
  .subscribe(x => console.log(x));

我们使用.share()来使Observable为"hot"。这是必要的，所以hotFilterChange.buffer( ... )和hotFilterChanges.debounce(200)指向完全相同的Observable执行。