fix(forkJoin): dispose the inner subscriptions when the outer subscri…

…ption is disposed
ReactiveX · Mar 18, 2016 · c7bf30c · c7bf30c
1 parent 2a1796a
commit c7bf30c
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 1 deletion.
diff --git a/spec/observables/forkJoin-spec.ts b/spec/observables/forkJoin-spec.ts
@@ -263,4 +263,20 @@ describe('Observable.forkJoin', () => {
 
     expectObservable(e1).toBe(expected);
   });
+
+  it('should allow unsubscribing early and explicitly', () => {
+    const e1 =   hot('--a--^--b--c---d-| ');
+    const e1subs =        '^        !    ';
+    const e2 =   hot('---e-^---f--g---h-|');
+    const e2subs =        '^        !    ';
+    const expected =      '----------    ';
+    const unsub =         '         !    ';
+
+    const result = Observable.forkJoin(e1, e2);
+
+    expectObservable(result, unsub).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
 });
diff --git a/src/observable/ForkJoinObservable.ts b/src/observable/ForkJoinObservable.ts
@@ -63,7 +63,8 @@ export class ForkJoinObservable<T> extends Observable<T> {
       if (isPromise(source)) {
         source = new PromiseObservable(<Promise<any>>source);
       }
-      (<Observable<any>>source).subscribe(new AllSubscriber(subscriber, i, context));
+      subscriber.add((<Observable<any>>source)
+        .subscribe(new AllSubscriber(subscriber, i, context)));
     }
   }
 }