RxJS / wait for an event to go through the entire pipeline before emitting the next

user_4099454 0
Viewed 167

I am looking to build an rxjs pipeline that emits the next event only if the current one has gone through the entire pipeline (FIFO struct).

const { from } = rxjs;
const { map, filter } = rxjs.operators;

// the event `2` should only be emitted
// when the event `1` has reached the end of the pipeline.
// and so on and so fort
from([1, 2, 3, 4, 5, 6, 7]).pipe(
  map((n) => n ** 2),
  filter((n) => n % 2),
).subscribe(console.log);


/**
-1-------|
-2-------|-1
...
**/
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.4.0/rxjs.umd.js" integrity="sha512-DRDXreq+zyiPhlKTfJ5pgzWRn+6SgJ7cPoRxMNksyHmUEOjKiKIoqvssNYNwknpvVbQsVN5hlhh3sp0rxbU7Bg==" crossorigin="anonymous" referrerpolicy="no-referrer"></script>

rxjs6 rxjs queue javascript
1 Answers

What you're looking for reassembles me some kind of "backpressure" in RxJS which is not maintained any more and RxJS doesn't provide any operator (or Observable creation method) that would do exactly what you're describing.

However, you could do the same using zip() operator because it'll emit only when all source Observables emit the same number of next emissions.

import { Subject, zip, from, map, startWith } from 'rxjs';

const done$ = new Subject<void>();

zip(
  from([1, 2, 3, 4, 5, 6, 7]),
  done$.pipe(
    startWith(void 0), // trigger the first emission
  ),
)
  .pipe(
    map(([value]) => value),
  )
  .subscribe(value => {
    console.log(value);
    setTimeout(() => done$.next(void 0), 500);
  });

Live demo: https://stackblitz.com/edit/rxjs-oqzeqx?devtoolsheight=60

user_3107260
Related