Fix operators that fail the synchronous firehose test

[cartant]

The following operators currently fail tests that were introduced in #5652 and those failing tests are currently skipped:

• catchError - see fix(catchError): inner synchronous observables will properly terminate #5655
• exhaust - see test: enable now-passing firehose tests #5743
• exhaustMap - see test: enable now-passing firehose tests #5743
• multicast
• raceWith - see test: enable now-passing firehose tests #5743
• refCount
• repeat - see Issue/5739 smaller operator creation #5742
• repeatWhen - see Issue/5739 smaller operator creation #5742
• retry - see Issue/5739 smaller operator creation #5742
• retryWhen - see test: enable now-passing firehose tests #5743
• share
• shareReplay
• single - see test: enable now-passing firehose tests #5743
• skipLast - see test: enable remaining synchronous firehose tests #5749
• throttle - see test: enable remaining synchronous firehose tests #5749
• timeout - see test(timeout): enable skipped test #5748
• timeoutWith - see test(timeout): enable skipped test #5748

The tests are marked with // TODO: fix firehose unsubscription comments.

Update: it's not going to be possible to make the multicast-based tests pass.

[kobelb]

I've noticed some oddities with the behavior of exhaustMap when there's a synchronous subscription. I'm curious whether this behavior is related to this bug, or something else entirely.

Synchronous Subscription

stackblitz

From within a synchronous subscription to the exhaustMap, the inner observable isn't marked as completed when the next callback is called:

const subject = new BehaviorSubject<number>(0);
const obs$ = subject.pipe(
  tap(() => console.log('before')),
  exhaustMap((val) => {
    return Promise.resolve(val * 100);
  }),
  tap(() => console.log('after')),
);

let i = 0;
obs$.subscribe(x => { 
  i++;
  console.log(x)
  if (i === 1) {
    subject.next(i); // the `exhaustMap` causes this to have no effect
  }
});

// output:
//
// before
// after
// 0
// before

Asynchronous Subscription

stackblitz

However, when there's an asynchronous subscription to the exhaustMap, the inner observable is marked as completed when the next callback is called:

const subject = new BehaviorSubject<number>(0);
const obs$ = subject.pipe(
  tap(() => console.log('before')),
  exhaustMap((val) => {
    return Promise.resolve(val * 100);
  }),
  switchMap((val) => {
    return Promise.resolve(val);
  }),
  tap(() => console.log('after')),
);

let i = 0;
obs$.subscribe(x => { 
  i++;
  console.log(x)
  if (i === 1) {
    subject.next(i);
  }
});

// output:
//
// before
// after
// 0
// before
// after
// 100

Cause

I believe that this is caused by the implementation of subscribeToPromise. When the promise is resolved, it first calls subscriber.next(value) immediately before calling subscriber.complete(). When the subscription is asynchronous, there's an opportunity to mark the subscriber as complete but there isn't when the subscription is synchronous. It feels like these two code-paths should behave the same, otherwise, the "implementation details" of how the observable is created is leaked to all consumers.

[josepot]

👋 @kobelb

This is the behavior that I would expect. Because exhaustMap will ignore any emissions from its source until the inner-observable completes. In the code of the first example: a value is being pushed into the subject before the inner observable has had a chance to complete. That's why I think that this should be the expected behavior.

the inner observable isn't marked as completed when the next callback is called

Keep in mind that once an observable has completed, then it can no longer emit. So, it's not possible to complete before emitting the value. First it emits and then it completes.

One way to get the behavior that you are expecting would be to observe the resulting observable of exhaustMap through the asapScheduler, like this:

const subject = new BehaviorSubject<number>(0);
const obs$ = subject.pipe(
  tap(() => console.log('before')),
  exhaustMap((val) => {
    return Promise.resolve(val * 100);
  }),
  observeOn(asapScheduler),
  tap(() => console.log('after')),
);

let i = 0;
obs$.subscribe(x => { 
  i++;
  console.log(x)
  if (i === 1) {
    subject.next(i); // the `exhaustMap` causes this to have no effect
  }
});

It feels like these two code-paths should behave the same

I disagree. In the second snipped, the inner observable of switchMap creates a micro-task that ensures that when the resulting observable emits, the inner observable of exhaustMap has completed. It would be equivalent, if the inner-observable of switchMap was synchronous:

const subject = new BehaviorSubject<number>(0);
const obs$ = subject.pipe(
  tap(() => console.log('before')),
  exhaustMap((val) => {
    return Promise.resolve(val * 100);
  }),
  switchMap((val) => {
    return of(val);
  }),
  tap(() => console.log('after')),
);

[miginmrs]

I had to make a change in ConnectableObservable by adding an internal method prepare to prepare the connect subscription before calling connect. I didn't understand why "it's not going to be possible to make the multicast-based tests pass".