RxJS缓冲器直到更改

Question

是否有一个很好的方法来创建一个“缓冲区直到改变”函数？基本上，我有一个排序的可观察的类似[1,1,1,2,2,2,4,5,5]，我想创建一个可观察的可观测的(或任何集合，真的)像[[1,1,1], [2,2,2], [4], [5,5]]。

理论上，groupBy可以在这种情况下工作，但是它会在我的应用程序中造成巨大的内存消耗，因为它没有立即关闭组可观测值，这是不必要的，因为我知道原始的可观测性是排序的。

Answer 1

也适用于即时可观测的解决方案(rxjs 5)：

source
  .concat(Observable.of('~'))
  .scan(({ buffer }, item) => buffer.length === 0 || buffer.includes(item) ? 
        ({ buffer: [...buffer, item] }) : 
        ({ buffer: [item], output: buffer }), { buffer: [] })
  .pluck('output')
  .filter(x => x);

http://jsbin.com/ruyozinali/1/edit?js,console

Answer 2

您可以尝试使用bufferWhen和distinctUntilChanged相结合。就像这样：

source.bufferWhen(() => source.distinctUntilChanged())