Items are logged in the reverse of expected order in RxJS

Question

This behavior isn't bad in my particular use-case but it is very unexpected and I'd like to understand why this is happening.

I have the following code running in a sandbox (imported RxJS 6.6.0):

const Rx = rxjs;
const {map,tap, merge, take} = Rx.operators;

const sub = new Rx.ReplaySubject();

sub.next(1);
sub.pipe(
  tap(x => {
    if (x < 50) sub.next(x*2);
  })
)
.subscribe({
  next(x) {
    console.info('Subscriber A', x)
  },
  complete() {
    console.log('Subscriber A -- Complete!')
  }
});

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

I might be doing this entirely the wrong way, but I want intermediate functions to push items back into the original Subject/Observable while also emitting the original item(s) (a feedback loop of sorts). This is the output I'm getting:

Subscriber A64
Subscriber A32
Subscriber A16
Subscriber A8
Subscriber A4
Subscriber A2
Subscriber A1

I would've expected "Subscriber A1" first but it seems like next is adding items to the beginning of the stream instead of the end? Can anyone help make sense of this?

user_16158010 · Answer

The emitted value won't reach the subscriber before it has gone through all the pipes, and since you are calling next inside of your tap, this new value emitted gets handled before the previous tap finishes and thus reaches the subscriber first. Repeat for every calls of next.

Some visualization:

    const Rx = rxjs;
    const { map, tap, merge, take } = Rx.operators;

    const sub = new Rx.ReplaySubject();

    sub.next(1);
    sub.pipe(
        tap(x => {
            console.info("before next in tap x = ", x);
            if (x < 50) sub.next(x * 2);
            console.info("after next in tap x = ", x);
        })
    )
        .subscribe({
            next(x) {
                console.info('Subscriber A', x)
            },
            complete() {
                console.log('Subscriber A -- Complete!')
            }
        });

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

user_1224119 · Answer

My guess is you have a recursive effect happening due to the sub.next() being used inside the tap() operator. So it fills the call stack until it reaches the condition limit, then returns the stack from last-to-first.

However, if you need to have an accumulation effect inside an observable then scan() is the more optimal choice. It behaves like Array.reduce() but it runs the inner function for each emitted value. Then takeUntil() handles the conditional limit.

const Rx = rxjs;
const {interval} = Rx;
const { scan, tap, takeWhile, startWith} = Rx.operators;

const double$ = interval(1).pipe(
  scan((acc, cur)=>acc*2,1),
  startWith(1),
  takeWhile(value=>value<50, true),
  tap(value=>console.info(`Subscriber A${value}`)),
).subscribe();

Items are logged in the reverse of expected order in RxJS

2 Answers