RxJS - 最多缓冲 n 个项目并发出这些缓冲区

Question

我正在尝试从流中缓冲多达 n 个（假设 5）个项目，但会发出部分填充的数组，直到缓冲区已满。

假设我有这样的流：

const stream = Rx.Observable.range(0, 6);

我要发射：

[0]
[0, 1]
[0, 1, 2]
[0, 1, 2, 3]
[0, 1, 2, 3, 4]
[1, 2, 3, 4, 5]

到目前为止，我有两种解决方案，我想知道哪一种更“Rx 方式”。

如果元素多于n，则扫描并移位累加器：

stream.scan((acc, current) => {
  acc.push(current);

  if (acc.length > 5) {
    acc.shift();
  }

  return acc;
}, []);

或者在元素不超过5个时扫描流并与缓冲区合并：

stream.scan((acc, current) => {
  acc.push(current);
  return acc;
}, [])
.takeWhile((x) => { return x.length < 5 })
.merge(stream.bufferWithCount(5));

哪种方法更好，更符合 Rx？根据 jsperf 的快速测试，在性能方面移动数组比合并快 60%。

或者也许有更好的解决方案？

Answer 1

第一种方法看起来相当不错。我还会将acc 视为不可变数组。简单地说，不使用push 或shift 做你正在做的事情。方法如下：

stream.scan((acc, current) => [...acc.slice(acc.length > 5 ? 1 : 0), current], []);