RxJS 缓冲直到改变

Question

是否有创建“更改前的缓冲区”功能的好方法？基本上，我有一个像 [1,1,1,2,2,2,4,5,5] 这样的排序的可观察对象，我想创建一个可观察对象的可观察对象(或者任何集合，真的)像 [[1,1,1], [2,2,2], [4], [5,5]]。

groupBy 理论上可以在这种情况下工作，但它会导致我的应用程序消耗大量内存，因为它不会立即关闭组可观察对象，这是不必要的，因为我知道原始可观察对象已排序。

Answer 1

也适用于即时 Observables (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