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>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