Why doesn't this result in an infinite loop?

[Sawtaytoes]

Can someone explain why this isn't an infinite loop? Seems to be something about execution order with startWith. Might even be a bug; not sure.

This code has a subject calling .next on itself:

const subject$ = new Subject();

subject$
.pipe(
  startWith(null),
  switchMap(() => (
    throwError('error')
    .pipe(
      catchError(() => {
        subject$
        .next()

        return of(null);
      }),
    )
  )),
)
.subscribe()

There are three fixes to make this infinitely loop:

1. Add a delay after startWith:

startWith(null),
delay(0),

2. Wrap subject$ in merge and include of:

NOTE: This fix doesn't work with EMPTY or NEVER probably because of the same issue as startWith.

merge(
  of(null),
  subject$,
)

3. Add setTimeout before calling .next the first time through:

catchError(() => {
  if (isFirstTime) {
    isFirstTime = false

    setTimeout(() => {
      subject$
      .next()
    })
  }
  else {
    subject$
    .next()
  }

  return of(null);
}),

Looks like this might be related to an issue I had a few years back when trying to do the same thing; calling a subject's .next from in its pipeline: #5410.

[cartant]

My guess is that startWith emits before it subscribes to its source, so it's not subscribed at the time the next call is made. Feature. Not a bug.

[cartant]

Try:

.pipe(
  startWith(null),
  observeOn(queueScheduler),

[Sawtaytoes]

I used RxViz to test. Might not have the latest RxJS so who knows.

This is what it output, same as I had before; only the first value:

[cartant]

same as I had before; only the first value

That's correct. I should have given it some more thought. observeOn(queueScheduler) is going to be ineffectual. The queueScheduler works by queueing any actions that are scheduled whilst it's running an already-scheduled action. You could, however, use it to ensure the subject.next call is made after the entirety of the startWith implementation like this:

const subject = new Subject<void>();
subject
  .pipe(
    (source) =>
      new Observable((subscriber) => {
        const subscription = queueScheduler.schedule(() => {
          source.pipe(startWith(null)).subscribe(subscriber);
        });
        return subscription;
      }),
    switchMap(() =>
      throwError("error").pipe(
        catchError(() => {
          queueScheduler.schedule(() => {
            subject.next();
          });
          return of(null);
        })
      )
    )
  )
  .subscribe();

Not that I am recommending doing such a thing.

[Sawtaytoes]

I wonder though. Without a hidden setTimeout, this solves my problem and ensures the code runs as expected.

The observable in switchMap could execute synchronously or asynchronously and subject$.next will do what I'd expect.

I'd name that operator something like delayedStartWith. It's super niche, but is important for this particular case where I have a subject doing a .next on itself but also needs to immediately execute a value on subscribe.

---

What are the problems doing it this way?

[kwonoj]

All time recommendations from RxJS is don't reentrant. If you already aware problems and willing to deal with it, you could choose escape hatches for your purpose. Core api probably won't able to consider it to expose as public interface.

For escape hatches, there maybe problems / or not - but it depends on your reentrant usecase since there's no single approach applicable for all reentrant cases.

[backbone87]

are we just talking about behavior or is there a use case to this?

[cartant]

is there a use case to this?

IMO, no.