Queue operator for RxJS

Question

Is there an operator in RxJS that would allow me to buffer items and let them out one by one whenever a signal observable fires? Sort of like bufferWhen, but instead of dumping the whole buffer on each signal it would dump a certain number per signal. It could even dump the number that gets emitted by the signal observable.

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

cartant cartant · Accepted Answer · 2017-05-18T00:24:49

Yes, you can use zip to do what you want:

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>

In fact, zip is used as an example in this GitHub issue that pertains to buffering.

If you want to use the signal's emitted value to determine how many buffered values are to be released, you could do something like this:

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>

Queue operator for RxJS

2 Answers