fix(repeatWhen): resulting observable will wait for the source to com…

…plete, even if a hot notifier completes first. (#2209)

After notifier completes wait for source observable to complete instead
of ending stream immediately

Closes #2054

spec/operators/repeatWhen-spec.ts

@@ -10,9 +10,9 @@ describe('Observable.prototype.repeatWhen', () => {
     const source =  cold('-1--2--|');
     const subs =        ['^      !                     ',
                        '             ^      !        ',
-                       '                          ^ !'];
+                       '                          ^      !'];
     const notifier = hot('-------------r------------r-|');
-    const expected =     '-1--2---------1--2---------1|';
+    const expected =     '-1--2---------1--2---------1--2--|';
 
     const result = source.repeatWhen((notifications: any) => notifier);
 
@@ -320,4 +320,4 @@ describe('Observable.prototype.repeatWhen', () => {
     expectObservable(result).toBe(expected);
     expectSubscriptions(source.subscriptions).toBe(subs);
   });
-});
+});

src/operator/repeatWhen.ts

@@ -27,16 +27,15 @@ import { subscribeToResult } from '../util/subscribeToResult';
  * @owner Observable
  */
 export function repeatWhen<T>(this: Observable<T>, notifier: (notifications: Observable<any>) => Observable<any>): Observable<T> {
-  return this.lift(new RepeatWhenOperator(notifier, this));
+  return this.lift(new RepeatWhenOperator(notifier));
 }
 
 class RepeatWhenOperator<T> implements Operator<T, T> {
-  constructor(protected notifier: (notifications: Observable<any>) => Observable<any>,
-              protected source: Observable<T>) {
+  constructor(protected notifier: (notifications: Observable<any>) => Observable<any>) {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source.subscribe(new RepeatWhenSubscriber(subscriber, this.notifier, this.source));
+    return source.subscribe(new RepeatWhenSubscriber(subscriber, this.notifier, source));
   }
 }
 
@@ -50,40 +49,39 @@ class RepeatWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
   private notifications: Subject<any>;
   private retries: Observable<any>;
   private retriesSubscription: Subscription;
+  private sourceIsBeingSubscribedTo: boolean = true;
 
   constructor(destination: Subscriber<R>,
               private notifier: (notifications: Observable<any>) => Observable<any>,
               private source: Observable<T>) {
     super(destination);
   }
 
-  complete() {
-    if (!this.isStopped) {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
+    this.source.subscribe(this);
+    this.sourceIsBeingSubscribedTo = true;
+  }
 
-      let notifications = this.notifications;
-      let retries: any = this.retries;
-      let retriesSubscription = this.retriesSubscription;
-
-      if (!retries) {
-        notifications = new Subject();
-        retries = tryCatch(this.notifier)(notifications);
-        if (retries === errorObject) {
-          return super.complete();
-        }
-        retriesSubscription = subscribeToResult(this, retries);
-      } else {
-        this.notifications = null;
-        this.retriesSubscription = null;
-      }
+  notifyComplete(innerSub: InnerSubscriber<T, R>): void {
+    if (this.sourceIsBeingSubscribedTo === false) {
+      return super.complete();
+    }
+  }
 
-      this.unsubscribe();
-      this.closed = false;
+  complete() {
+    this.sourceIsBeingSubscribedTo = false;
 
-      this.notifications = notifications;
-      this.retries = retries;
-      this.retriesSubscription = retriesSubscription;
+    if (!this.isStopped) {
+      if (!this.retries) {
+        this.subscribeToRetries();
+      } else if (this.retriesSubscription.closed) {
+        return super.complete();
+      }
 
-      notifications.next();
+      this.temporarilyUnsubscribe();
+      this.notifications.next();
     }
   }
 
@@ -100,10 +98,17 @@ class RepeatWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
     this.retries = null;
   }
 
-  notifyNext(outerValue: T, innerValue: R,
-             outerIndex: number, innerIndex: number,
-             innerSub: InnerSubscriber<T, R>): void {
+  private subscribeToRetries() {
+    this.notifications = new Subject();
+    const retries = tryCatch(this.notifier)(this.notifications);
+    if (retries === errorObject) {
+      return super.complete();
+    }
+    this.retries = retries;
+    this.retriesSubscription = subscribeToResult(this, retries);
+  }
 
+  private temporarilyUnsubscribe() {
     const { notifications, retries, retriesSubscription } = this;
     this.notifications = null;
     this.retries = null;
@@ -116,7 +121,6 @@ class RepeatWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
     this.notifications = notifications;
     this.retries = retries;
     this.retriesSubscription = retriesSubscription;
-
-    this.source.subscribe(this);
   }
+
 }