RxJS 缓冲区直到改变答案

【问题标题】：RxJS buffer until changedRxJS 缓冲区直到改变
【发布时间】：2023-03-07 01:21:01
【问题描述】：

有没有一种很好的方法来创建“缓冲区直到更改”功能？基本上，我有一个像[1,1,1,2,2,2,4,5,5] 这样的排序的可观察对象，我想创建一个像[[1,1,1], [2,2,2], [4], [5,5]] 这样的可观察对象（或任何集合，真的）的可观察对象。

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

【问题讨论】：

标签： rxjs

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

【讨论】：

我正在 RxJava 中寻找类似的。你有 RxJava 版本的解决方案吗

【解决方案2】：

您可以尝试将bufferWhen 与distinctUntilChanged 结合使用。像这样的：

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

【讨论】：

当源是Observable.of([1,1,2, etc.]) 时，这会导致堆栈溢出。我也试过source.buffer(source.distinctUntilChanged())，但这也不起作用，因为它依赖于时间。在一段时间内添加项目时它可能会起作用，但我宁愿不依赖时间（不想要竞争条件）。
可能堆栈溢出是由于 observable 是冷的。您可能需要多播源？