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

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

Author: mpodlasin

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(newError('should not be called'));});
 });
 
+it('should finalize generators when merged if the subscription ends',()=>{
+constiterable={
+finalized: false,
+next(){
+return{value: 'duck',done: false};
+},
+return(){
+this.finalized=true;
+},
+[Symbol.iterator](){
+returnthis;
+}
+};
+
+constresults: string[]=[];
+
+constiterableObservable=from<string>(iterableasany);
+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 */
 constsource1=of(1,2,3);

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));
+constinnerSubscriber=newInnerSubscriber(this,undefined,undefined);
+this.add(innerSubscriber);
+subscribeToResult<T,R>(this,ish,value,index,innerSubscriber);
 }
 
 protected_complete(): void{

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';
 
-exportfunctionsubscribeToResult<T,R>(outerSubscriber: OuterSubscriber<T,R>,
-result: any,
-outerValue?: T,
-outerIndex?: number): Subscription;
-exportfunctionsubscribeToResult<T>(outerSubscriber: OuterSubscriber<any,any>,
-result: ObservableInput<T>,
-outerValue?: T,
-outerIndex?: number): Subscription|void{
-constdestination=newInnerSubscriber(outerSubscriber,outerValue,outerIndex);
-
+exportfunctionsubscribeToResult<T,R>(
+outerSubscriber: OuterSubscriber<T,R>,
+result: any,
+outerValue?: T,
+outerIndex?: number,
+destination?: Subscriber<any>
+): Subscription;
+exportfunctionsubscribeToResult<T,R>(
+outerSubscriber: OuterSubscriber<T,R>,
+result: any,
+outerValue?: T,
+outerIndex?: number,
+destination: Subscriber<any>=newInnerSubscriber(outerSubscriber,outerValue,outerIndex)
+): Subscription|void{
+if(destination.closed){
+return;
+}
 returnsubscribeTo(result)(destination);
 }