rxjs - 在 rxjs 中，如何创建连续值组？

Question

我有一个这种形式的值流:

[1,2,3,1,2,1,1,1,2...]

我想将其转换为以下形式的组流:

[[1,2,3],[1,2],[1],[1],[1,2]...]

每次值变为 1 时都应创建新组。

Answer 1

您可以使用 bufferWhen()运算符收集发出的值，直到 observable 发出一个值。在这个例子中，我使用一个主题将流的浅拷贝发送到缓冲区。

缓冲区将在数字 1 时发出被发射。如果流以 1 开头然后发出一个空数组。所以我过滤掉了。

const {from, Subject} = rxjs;
const {filter, bufferWhen, tap, skip} = rxjs.operators;

const stream = from([1,2,3,1,2,1,1,1,2]);
const trigger = new Subject();

stream.pipe(
   tap(v => trigger.next(v)),
   bufferWhen(() => trigger.pipe(filter(v => v === 1))),
   filter(v => v.length)
).subscribe(x => console.log(x));

<script src="https://unpkg.com/@reactivex/rxjs@6.x/dist/global/rxjs.umd.js"></script>

您可以使用 scan()如果您想在值比前一个值减少时发出触发器。这在逻辑上可能更好一点，但使用 1因为触发器适合问题。