fix(exhaustMap): stop listening to a synchronous inner-obervable when unsubscribed

Author: peaBerberian

spec/operators/exhaustMap-spec.ts

@@ -1,6 +1,6 @@
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
-import { Observable, of, from } from 'rxjs';
-import { exhaustMap, mergeMap } from 'rxjs/operators';
+import { concat, defer, Observable, of, from } from 'rxjs';
+import { exhaustMap, mergeMap, takeWhile } from 'rxjs/operators';
 import { expect } from 'chai';
 
 declare function asDiagram(arg: string): Function;
@@ -202,6 +202,31 @@ describe('exhaustMap', () => {
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = concat(
+      defer(() => {
+        sideEffects.push(1);
+        return of(1);
+      }),
+      defer(() => {
+        sideEffects.push(2);
+        return of(2);
+      }),
+      defer(() => {
+        sideEffects.push(3);
+        return of(3);
+      })
+    );
+
+    of(null).pipe(
+      exhaustMap(() => synchronousObservable),
+      takeWhile((x) => x != 2) // unsubscribe at the second side-effect
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([1, 2]);
+  });
+
   it('should switch inner cold observables, inner never completes', () => {
     const x = cold(     '--a--b--c--|                              ');
     const xsubs =    '   ^          !                              ';

src/internal/operators/exhaustMap.ts

@@ -106,15 +106,22 @@ class ExhaustMapSubscriber<T, R> extends OuterSubscriber<T, R> {
   }
 
   private tryNext(value: T): void {
+    let result: ObservableInput<R>;
     const index = this.index++;
-    const destination = this.destination;
     try {
-      const result = this.project(value, index);
-      this.hasSubscription = true;
-      this.add(subscribeToResult(this, result, value, index));
+      result = this.project(value, index);
     } catch (err) {
-      destination.error(err);
+      this.destination.error(err);
+      return;
     }
+    this.hasSubscription = true;
+    this._innerSub(result, value, index);
+  }
+
+  private _innerSub(result: ObservableInput<R>, value: T, index: number): void {
+    const innerSubscriber = new InnerSubscriber(this, undefined, undefined);
+    this.add(innerSubscriber);
+    subscribeToResult<T, R>(this, result, value, index, innerSubscriber);
   }
 
   protected _complete(): void {