From a1b18ec3478f8706137934b41a79d67a25aeb4fd Mon Sep 17 00:00:00 2001
From: Paul Taylor <paul.e.taylor@me.com>
Date: Mon, 12 Dec 2016 13:41:33 -0800
Subject: [PATCH] fix(takeUntil): If the notifier supplied to takeUntil
 synchronously emits a value, the source should

1360
---
 spec/operators/takeUntil-spec.ts |  9 +++++++++
 src/operator/takeUntil.ts        | 12 ++++++++----
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/spec/operators/takeUntil-spec.ts b/spec/operators/takeUntil-spec.ts
index 07622bba53..6b6cb62eaa 100644
--- a/spec/operators/takeUntil-spec.ts
+++ b/spec/operators/takeUntil-spec.ts
@@ -126,6 +126,15 @@ describe('Observable.prototype.takeUntil', () => {
     expectSubscriptions(e2.subscriptions).toBe(e2subs);
   });
 
+  it('should complete without subscribing to the source when notifier synchronously emits', () => {
+    const e1 =   hot('----a--|');
+    const e2 =  Observable.of(0);
+    const expected = '(|)     ';
+
+    expectObservable(e1.takeUntil(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe([]);
+  });
+
   it('should raise error if source raises error before notifier emits', () => {
     const e1 =     hot('--a--b--c--d--#     ');
     const e1subs =     '^             !     ';
diff --git a/src/operator/takeUntil.ts b/src/operator/takeUntil.ts
index 070d558433..094024e69a 100644
--- a/src/operator/takeUntil.ts
+++ b/src/operator/takeUntil.ts
@@ -49,7 +49,13 @@ class TakeUntilOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source.subscribe(new TakeUntilSubscriber(subscriber, this.notifier));
+    const takeUntilSubscriber = new TakeUntilSubscriber(subscriber);
+    const notifierSubscription = subscribeToResult(takeUntilSubscriber, this.notifier);
+    if (notifierSubscription && !notifierSubscription.closed) {
+      takeUntilSubscriber.add(notifierSubscription);
+      return source.subscribe(takeUntilSubscriber);
+    }
+    return takeUntilSubscriber;
   }
 }
 
@@ -60,10 +66,8 @@ class TakeUntilOperator<T> implements Operator<T, T> {
  */
 class TakeUntilSubscriber<T, R> extends OuterSubscriber<T, R> {
 
-  constructor(destination: Subscriber<any>,
-              private notifier: Observable<any>) {
+  constructor(destination: Subscriber<any>) {
     super(destination);
-    this.add(subscribeToResult(this, notifier));
   }
 
   notifyNext(outerValue: T, innerValue: R,