Estoy buscando construir una tubería rxjs que emita el siguiente evento solo si el actual ha pasado por toda la tubería (estructura FIFO).
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>Lo que está buscando me vuelve a ensamblar algún tipo de "contrapresión" en RxJS que ya no se mantiene y RxJS no proporciona ningún operador (o método de creación Observable) que haga exactamente lo que está describiendo.
Sin embargo, podría hacer lo mismo usando el operador zip() porque emitirá solo cuando todos los Observables de origen emitan la misma cantidad de next emisiones.
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); });Demostración en vivo: https://stackblitz.com/edit/rxjs-oqzeqx?devtoolsheight=60