From 7d4cb7ab618b7971217cb309c2940c3c253949c3 Mon Sep 17 00:00:00 2001
From: Andre Medeiros <andre.medeiros@futurice.com>
Date: Fri, 23 Oct 2015 13:27:52 +0300
Subject: [PATCH] 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       |  2 +-
 src/operators/takeUntil.ts | 43 +++++++++++++++++++++++++-------------
 2 files changed, 30 insertions(+), 15 deletions(-)

diff --git a/src/CoreOperators.ts b/src/CoreOperators.ts
index 09ffda5a0cb..cf8d226ac49 100644
--- a/src/CoreOperators.ts
+++ b/src/CoreOperators.ts
@@ -66,7 +66,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>;
diff --git a/src/operators/takeUntil.ts b/src/operators/takeUntil.ts
index e63d4026d5f..3f8239a6bc6 100644
--- a/src/operators/takeUntil.ts
+++ b/src/operators/takeUntil.ts
@@ -3,41 +3,56 @@ import Observer from '../Observer';
 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> {
-  constructor(destination: Subscriber<T>,
-              observable: Observable<any>) {
-    super(destination);
-    this.add(observable._subscribe(new TakeUntilInnerSubscriber(destination)));
+  private notificationSubscriber: TakeUntilInnerSubscriber<any> = null;
+
+  constructor(public destination: Subscriber<T>,
+              private notifier: Observable<any>) {
+    super(null);
+    this.notificationSubscriber = new TakeUntilInnerSubscriber(destination);
+    this.add(notifier.subscribe(this.notificationSubscriber));
+  }
+
+  _next(value: T) {
+    this.destination.next(value);
+  }
+
+  _error(err: any) {
+    this.destination.error(err);
+  }
+
+  _complete() {
+    this.destination.complete();
+    this.notificationSubscriber.unsubscribe();
   }
 }
 
 class TakeUntilInnerSubscriber<T> extends Subscriber<T> {
-  constructor(destination: Subscriber<T>) {
-    super(destination);
+  constructor(public destination: Subscriber<T>) {
+    super(null);
   }
+
   _next() {
     this.destination.complete();
   }
+
   _error(e) {
     this.destination.error(e);
   }
+
   _complete() {
   }
 }
\ No newline at end of file