rxjs - RxJS 的队列操作符

Question

RxJS 中是否有一个运算符可以让我缓冲项目并在信号可观察对象触发时将它们一个一个地放出？有点像bufferWhen，但不是在每个信号上转储整个缓冲区，而是每个信号转储一定数量。它甚至可以转储信号 observable 发出的数字。

Input observable:  >--a--b--c--d--|
Signal observable: >------1---1-1-|
Count in buffer:   !--1--21-2-121-|
Output observable: >------a---b-c-|

Answer 1

是的，您可以使用 zip 做你想做的事:

const input = Rx.Observable.from(["a", "b", "c", "d", "e"]);
const signal = new Rx.Subject();
const output = Rx.Observable.zip(input, signal, (i, s) => i);
output.subscribe(value => console.log(value));
signal.next(1);
signal.next(1);
signal.next(1);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

事实上， zip在 this GitHub issue 中用作示例这与缓冲有关。

如果您想使用信号的发射值来确定要释放多少缓冲值，您可以执行以下操作:

const input = Rx.Observable.from(["a", "b", "c", "d", "e"]);
const signal = new Rx.Subject();
const output = Rx.Observable.zip(
  input,
  signal.concatMap(count => Rx.Observable.range(0, count)),
  (i, s) => i
);
output.subscribe(value => console.log(value));
signal.next(1);
signal.next(2);
signal.next(1);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>