fix(takeUntil): unsubscribe notifier when it completes

Fix operator takeUntil to automatically unsubscribe the notifier when it
completes. This is to conform RxJS Next with RxJS 4.

Somewhat related to issue #577.

src/CoreOperators.ts

@@ -67,7 +67,7 @@ export interface CoreOperators<T> {
   switchMap?: <R>(project: ((x: T, ix: number) => Observable<any>), projectResult?: (x: T, y: any, ix: number, iy: number) => R) => Observable<R>;
   switchMapTo?: <R>(observable: Observable<any>, projectResult?: (x: T, y: any, ix: number, iy: number) => R) => Observable<R>;
   take?: (count: number) => Observable<T>;
-  takeUntil?: (observable: Observable<any>) => Observable<T>;
+  takeUntil?: (notifier: Observable<any>) => Observable<T>;
   throttle?: (delay: number, scheduler?: Scheduler) => Observable<T>;
   timeout?: <T>(due: number|Date, errorToSend?: any, scheduler?: Scheduler) => Observable<T>;
   timeoutWith?: <T>(due: number|Date, withObservable: Observable<any>, scheduler?: Scheduler) => Observable<T>;

src/operators/takeUntil.ts

@@ -2,41 +2,48 @@ import Operator from '../Operator';
 import Observable from '../Observable';
 import Subscriber from '../Subscriber';
 
-export default function takeUntil<T>(observable: Observable<any>) {
-  return this.lift(new TakeUntilOperator(observable));
+export default function takeUntil<T>(notifier: Observable<any>) {
+  return this.lift(new TakeUntilOperator(notifier));
 }
 
 class TakeUntilOperator<T, R> implements Operator<T, R> {
-
-  observable: Observable<any>;
-
-  constructor(observable: Observable<any>) {
-    this.observable = observable;
+  constructor(private notifier: Observable<any>) {
   }
 
   call(subscriber: Subscriber<T>): Subscriber<T> {
-    return new TakeUntilSubscriber(subscriber, this.observable);
+    return new TakeUntilSubscriber(subscriber, this.notifier);
   }
 }
 
 class TakeUntilSubscriber<T> extends Subscriber<T> {
+  private notificationSubscriber: TakeUntilInnerSubscriber<any> = null;
+
   constructor(destination: Subscriber<T>,
-              observable: Observable<any>) {
+              private notifier: Observable<any>) {
     super(destination);
-    this.add(observable._subscribe(new TakeUntilInnerSubscriber(destination)));
+    this.notificationSubscriber = new TakeUntilInnerSubscriber(destination);
+    this.add(notifier.subscribe(this.notificationSubscriber));
+  }
+
+  _complete() {
+    this.destination.complete();
+    this.notificationSubscriber.unsubscribe();
   }
 }
 
 class TakeUntilInnerSubscriber<T> extends Subscriber<T> {
-  constructor(destination: Subscriber<T>) {
-    super(destination);
+  constructor(protected destination: Subscriber<T>) {
+    super(null);
   }
+
   _next() {
     this.destination.complete();
   }
+
   _error(e) {
     this.destination.error(e);
   }
+
   _complete() {
   }
 }