fix(mergeAll): add source subscription to composite before actually subscribing

mpodlasin · cartant · commit f5970bf48d82 · 2018-07-07T21:10:38.000+10:00
Add subscriptions for source Observables to mergeAll composite subscription before actually subscribing to any of these Observables, so that if source Observable emits synchronously and consumer of mergeAll unsubscribes at that moment (for example `take` operator), subscription to source is unsubscribed as well and Observable stops emitting. Closes #2476
diff --git a/spec/operators/mergeAll-spec.ts b/spec/operators/mergeAll-spec.ts
@@ -1,5 +1,5 @@
 import { expect } from 'chai';
-import { mergeAll, mergeMap } from 'rxjs/operators';
+import { mergeAll, mergeMap, take } from 'rxjs/operators';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
 import { throwError, from, of, Observable } from 'rxjs';
 
@@ -413,6 +413,36 @@ describe('mergeAll oeprator', () => {
       () => { done(new Error('should not be called')); });
   });
 
+  it('should finalize generators when merged if the subscription ends', () => {
+    const iterable = {
+      finalized: false,
+      next() {
+        return {value: 'duck', done: false};
+      },
+      return() {
+        this.finalized = true;
+      },
+      [Symbol.iterator]() {
+        return this;
+      }
+    };
+
+    const results: string[] = [];
+
+    const iterableObservable = from<string>(iterable as any);
+    of(iterableObservable).pipe(
+      mergeAll(),
+      take(3)
+    ).subscribe(
+      x => results.push(x),
+      null,
+      () => results.push('GOOSE!')
+    );
+
+    expect(results).to.deep.equal(['duck', 'duck', 'duck', 'GOOSE!']);
+    expect(iterable.finalized).to.be.true;
+  });
+
   type(() => {
     /* tslint:disable:no-unused-variable */
     const source1 = of(1, 2, 3);
diff --git a/src/internal/operators/mergeMap.ts b/src/internal/operators/mergeMap.ts
@@ -139,7 +139,9 @@ export class MergeMapSubscriber<T, R> extends OuterSubscriber<T, R> {
   }
 
   private _innerSub(ish: ObservableInput<R>, value: T, index: number): void {
-    this.add(subscribeToResult<T, R>(this, ish, value, index));
+    const innerSubscriber = new InnerSubscriber(this, undefined, undefined);
+    this.add(innerSubscriber);
+    subscribeToResult<T, R>(this, ish, value, index, innerSubscriber);
   }
 
   protected _complete(): void {
diff --git a/src/internal/util/subscribeToResult.ts b/src/internal/util/subscribeToResult.ts
@@ -1,19 +1,26 @@
-
 import { ObservableInput } from '../types';
 import { Subscription } from '../Subscription';
 import { InnerSubscriber } from '../InnerSubscriber';
 import { OuterSubscriber } from '../OuterSubscriber';
+import { Subscriber } from '../Subscriber';
 import { subscribeTo } from './subscribeTo';
 
-export function subscribeToResult<T, R>(outerSubscriber: OuterSubscriber<T, R>,
-                                        result: any,
-                                        outerValue?: T,
-                                        outerIndex?: number): Subscription;
-export function subscribeToResult<T>(outerSubscriber: OuterSubscriber<any, any>,
-                                     result: ObservableInput<T>,
-                                     outerValue?: T,
-                                     outerIndex?: number): Subscription | void {
-  const destination = new InnerSubscriber(outerSubscriber, outerValue, outerIndex);
-
+export function subscribeToResult<T, R>(
+  outerSubscriber: OuterSubscriber<T, R>,
+  result: any,
+  outerValue?: T,
+  outerIndex?: number,
+  destination?: Subscriber<any>
+): Subscription;
+export function subscribeToResult<T, R>(
+  outerSubscriber: OuterSubscriber<T, R>,
+  result: any,
+  outerValue?: T,
+  outerIndex?: number,
+  destination: Subscriber<any> = new InnerSubscriber(outerSubscriber, outerValue, outerIndex)
+): Subscription | void {
+  if (destination.closed) {
+    return;
+  }
   return subscribeTo(result)(destination);
 }

Original file line number	Diff line number	Diff line change
`@@ -139,7 +139,9 @@ export class MergeMapSubscriber<T, R> extends OuterSubscriber<T, R> {`
`139`	`139`	`}`
`140`	`140`
`141`	`141`	`private _innerSub(ish: ObservableInput<R>, value: T, index: number): void {`
`142`		`- this.add(subscribeToResult<T, R>(this, ish, value, index));`
	`142`	`+ const innerSubscriber = new InnerSubscriber(this, undefined, undefined);`
	`143`	`+ this.add(innerSubscriber);`
	`144`	`+ subscribeToResult<T, R>(this, ish, value, index, innerSubscriber);`
`143`	`145`	`}`
`144`	`146`
`145`	`147`	`protected _complete(): void {`