Fix teardown issues with synchronous inner-observables

[peaBerberian]

This PR fixes teardown issues that arised when unsubscribing from an observable emitting values from inner-observable(s) synchronously after subscription, like reported in #4025. An even simpler example is available here.
Basically, the problem is that even after the unsubscription, the inner-observable(s) will continue to run until an asynchronous step is reached.

This was simply because most OuterSubscribers first call subscribeToResult on an inner-Observable and THEN call this.add on the returned Subscription. This sounds fair, but when we're dealing with an unsubscription happening synchronously after the subscription, this.add will not have been called yet, and so the teardown will happen without unsubscribing from those inner-observables.

A fix has already been brought in 40852ff71, but only for the MergeMapSubscriber. In this fix, an InnerSubscriber is created, registered through this.add and then given as a last argument to subscribeToResult, which will use it instead of creating one.
It looks like it worked well for mergeMap and other related operators for some time now, so I brought the same fix for the following OuterSubscribers (I linked an example showing problematic real use-cases for each of them):

• CatchSubscriber
• ExhaustMapSubscriber
• MergeScanSubscriber
• OnErrorResumeNextSubscriber
• SkipUntilSubscriber
• SwitchMapSubscriber

I was hesitant to also bring the same fix to other OuterSubscribers with the same problems.

For example, with combineLatest the following code:

combineLatest(
  of(1), // should be subscribed to
  throwError(new Error()), // error! -> Teardown
  defer(() => { // should not run
    console.log("side-effect 1");
    return of(2).pipe(
      tap(() => console.log("side-effect 2")) // just to insist a little
    );
  })
).subscribe(
  (i) => { console.log("next", i); },
  (e) => { console.log("error!"); },
  () => { console.log("complete"); },
);

outputs:

error
side-effect 1
side-effect 2

instead of just:

error

But here, it's not that problematic, considering that we can just think that every Observable given to a CombineLatest run at the same time (or at least, that's a mental model I have when I'm using it).
Still, code run for nothing there and the code logic looks wrong. What do you think we should do?

The same problematic logic ("adding" the return of a subscribeToResult) is also written for many other OuterSubscribers.
Namely for the BufferSubscriber, BufferToggleSubscriber, DistinctSubscriber (for its flushes argument, but unsubscribing/erroring from that observable just completes/errors the observable its applied on), ExpandSubscriber, SwitchFirstSubscriber, ThrottleSubscriber, WithLatestFromSubscriber, WindowSubscriber and WindowToggleSubscriber but I did not yet succeed to reproduce a real use-case with any of them where that would be problematic. For this reason, I did not update the logic there.

[cartant]

Thanks for the PR. This looks good. I'll review it tomorrow.

[coveralls]

Coverage decreased (-0.2%) to 96.794% when pulling 1c257db on peaBerberian:fix-teardown-synchronous-issues into 030eee7 on ReactiveX:master.

[benlesh]

This LGTM. I'll wait for a second opinion from @cartant or @kwonoj before I merge.

[kwonoj]

I recall few places I added checker for return value of subscribeToResult for synchronous, as sync case it can complete and does not return any values for it. Do we possibly need audit same to check / guard around those for new changes? or it's logically prevented?

[cartant]

LGTM.

The places that @kwonoj mentioned where the return value of subscribeToResult is checked don't appear to be involved in this PR's changes. Rather, those places are referred to in the last paragraph of the OP's PR.

Also, I agree with the OP that whether or not combineLatest should receive similar treatment is best discussed/implemented elsewhere. That change might be more contentious.

[benlesh]

An important PR. Thanks, @peaBerberian