fix: switchMap and exhaustMap behave correctly with re-entrant code.

- switchMap should unsubscribe previous inner sub when getting synchronously reentrance during subscribing the inner sub
- exhaustMap should ignore subsequent synchronous reentrances during subscribing the inner sub

spec/operators/exhaustMap-spec.ts

@@ -1,5 +1,5 @@
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
-import { concat, defer, Observable, of } from 'rxjs';
+import { concat, defer, Observable, of, BehaviorSubject } from 'rxjs';
 import { exhaustMap, mergeMap, takeWhile, map, take } from 'rxjs/operators';
 import { expect } from 'chai';
 import { asInteropObservable } from '../helpers/interop-helper';
@@ -451,4 +451,19 @@ describe('exhaustMap', () => {
 
     expect(sideEffects).to.deep.equal([0, 1, 2]);
   });
+
+  it('should ignore subsequent synchronous reentrances during subscribing the inner sub', () => {
+    const e = new BehaviorSubject(1);
+    const results: Array<number> = [];
+
+    e.pipe(
+      take(3),
+      exhaustMap(value => new Observable<number>(subscriber => {
+        e.next(value+1);
+        subscriber.next(value);
+      })),
+    ).subscribe(value => results.push(value));
+
+    expect(results).to.deep.equal([1]);
+  });
 });

spec/operators/switchMap-spec.ts

@@ -1,7 +1,7 @@
 import { expect } from 'chai';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
 import { switchMap, mergeMap, map, takeWhile, take } from 'rxjs/operators';
-import { concat, defer, of, Observable } from 'rxjs';
+import { concat, defer, of, Observable, BehaviorSubject } from 'rxjs';
 import { asInteropObservable } from '../helpers/interop-helper';
 
 /** @test {switchMap} */
@@ -460,4 +460,19 @@ describe('switchMap', () => {
 
     expect(sideEffects).to.deep.equal([0, 1, 2]);
   });
+
+  it('should unsubscribe previous inner sub when getting synchronously reentrance during subscribing the inner sub', () => {
+    const e = new BehaviorSubject(1);
+    const results: Array<number> = [];
+
+    e.pipe(
+      take(3),
+      switchMap(value => new Observable<number>(subscriber => {
+        e.next(value+1);
+        subscriber.next(value);
+      })),
+    ).subscribe(value => results.push(value));
+
+    expect(results).to.deep.equal([3]);
+  });
 });

src/internal/operators/exhaustMap.ts

@@ -113,7 +113,8 @@ class ExhaustMapSubscriber<T, R> extends SimpleOuterSubscriber<T, R> {
       const innerSubscriber = new SimpleInnerSubscriber(this);
       const destination = this.destination;
       destination.add(innerSubscriber);
-      this.innerSubscription = innerSubscribe(result, innerSubscriber);
+      this.innerSubscription = innerSubscriber;
+      innerSubscribe(result, innerSubscriber);
     }
   }
 

src/internal/operators/switchMap.ts

@@ -129,7 +129,8 @@ class SwitchMapSubscriber<T, R> extends SimpleOuterSubscriber<T, R> {
     }
     const innerSubscriber = new SimpleInnerSubscriber(this);
     this.destination.add(innerSubscriber);
-    this.innerSubscription = innerSubscribe(result, innerSubscriber);
+    this.innerSubscription = innerSubscriber;
+    innerSubscribe(result, innerSubscriber);
   }
 
   protected _complete(): void {