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

…ubscribing

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

spec/operators/mergeAll-spec.ts

@@ -415,4 +415,34 @@ describe('Observable.prototype.mergeAll', () => {
       },
       () => { done(new Error('should not be called')); });
   });
+
+  it('should finalize generators when merged if the subscription ends', () => {
+    const iterator1 = {
+      finalized: false,
+      next() {
+        return {value: 'duck', done: false};
+      },
+      return() {
+        this.finalized = true;
+      },
+      [Rx.Symbol.iterator]() {
+        return this;
+      }
+    };
+
+    const results = [];
+
+    const i1 = Rx.Observable.from(iterator1 as any);
+    Rx.Observable.merge(i1, i1)
+      .take(3)
+      .subscribe(
+        x => results.push(x),
+        null,
+        () => results.push('GOOSE!')
+      );
+
+    // never even get here
+    expect(results).to.deep.equal(['duck', 'duck', 'duck', 'GOOSE!']);
+    expect(iterator1.finalized).to.be.true;
+  });
 });

src/operator/mergeAll.ts

@@ -5,6 +5,7 @@ import { Subscription } from '../Subscription';
 import { OuterSubscriber } from '../OuterSubscriber';
 import { Subscribable } from '../Observable';
 import { subscribeToResult } from '../util/subscribeToResult';
+import { InnerSubscriber } from '../InnerSubscriber';
 
 export function mergeAll<T>(this: Observable<T>, concurrent?: number): T;
 export function mergeAll<T, R>(this: Observable<T>, concurrent?: number): Subscribable<R>;
@@ -83,7 +84,9 @@ export class MergeAllSubscriber<T> extends OuterSubscriber<Observable<T>, T> {
   protected _next(observable: Observable<T>) {
     if (this.active < this.concurrent) {
       this.active++;
-      this.add(subscribeToResult<Observable<T>, T>(this, observable));
+      const innerSubscriber = new InnerSubscriber(this, undefined, undefined);
+      this.add(innerSubscriber);
+      subscribeToResult<Observable<T>, T>(this, observable, undefined, undefined, innerSubscriber);
     } else {
       this.buffer.push(observable);
     }

src/util/subscribeToResult.ts

@@ -13,12 +13,14 @@ import { $$observable } from '../symbol/observable';
 export function subscribeToResult<T, R>(outerSubscriber: OuterSubscriber<T, R>,
                                         result: any,
                                         outerValue?: T,
-                                        outerIndex?: number): Subscription;
+                                        outerIndex?: number,
+                                        destination?: Subscriber<any>): Subscription;
 export function subscribeToResult<T>(outerSubscriber: OuterSubscriber<any, any>,
                                      result: ObservableInput<T>,
                                      outerValue?: T,
-                                     outerIndex?: number): Subscription {
-  let destination: Subscriber<any> = new InnerSubscriber(outerSubscriber, outerValue, outerIndex);
+                                     outerIndex?: number,
+                                     destination: Subscriber<any> = new InnerSubscriber(outerSubscriber, outerValue, outerIndex)
+): Subscription {
 
   if (destination.closed) {
     return null;