From 6f0d98f671d98a3c05fad54755c104d1561e8033 Mon Sep 17 00:00:00 2001
From: kwonoj <kwon.ohjoong@gmail.com>
Date: Wed, 14 Oct 2015 00:31:35 -0700
Subject: [PATCH] fix(skipUntil): update skipUntil behavior with error,
 completion

- update skipUntil operator behavior
- add test coverage

closes #518
---
 spec/operators/skipUntil-spec.js | 112 ++++++++++++++++++++++++++++++-
 src/operators/skipUntil.ts       |  37 +++++++---
 2 files changed, 137 insertions(+), 12 deletions(-)

diff --git a/spec/operators/skipUntil-spec.js b/spec/operators/skipUntil-spec.js
index beb272e320..829b30a661 100644
--- a/spec/operators/skipUntil-spec.js
+++ b/spec/operators/skipUntil-spec.js
@@ -4,10 +4,118 @@ var Observable = Rx.Observable;
 
 describe('Observable.prototype.skipUntil()', function () {
   it('should skip values until another observable notifies', function () {
-    var source = hot('--a--b--c--d--e--|');
+    var e1 =     hot('--a--b--c--d--e--|');
     var skip =   hot('-------------x--|');
     var expected =  ('--------------e--|');
 
-    expectObservable(source.skipUntil(skip)).toBe(expected);
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should skip value and raises error until another observable raises error', function () {
+    var e1 =   hot('--a--b--c--d--e--|');
+    var skip = hot('-------------#');
+    var expected = '-------------#';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should skip all element when another observable does not emit and completes early', function () {
+    var e1 =   hot('--a--b--c--d--e--|');
+    var skip = hot('------------|');
+    var expected = '-----------------|';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should skip all element when another observable is empty', function () {
+    var e1 =   hot('--a--b--c--d--e--|');
+    var skip = Observable.empty();
+    var expected = '-----------------|';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should not complete if source observable does not complete', function () {
+    var e1 =   hot('-');
+    var skip = hot('-------------x--|');
+    var expected = '-';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should not complete if source observable never completes', function () {
+    var e1 = Observable.never();
+    var skip = hot('-------------x--|');
+    var expected = '-';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should raise error if source does not completes when another observable raises error', function () {
+    var e1 =   hot('-');
+    var skip = hot('-------------#');
+    var expected = '-------------#';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should raise error if source never completes when another observable raises error', function () {
+    var e1 = Observable.never();
+    var skip = hot('-------------#');
+    var expected = '-------------#';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should skip all element and does not complete when another observable never completes', function () {
+    var e1 =   hot('--a--b--c--d--e--|');
+    var skip = Observable.never();
+    var expected = '-';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should skip all element and does not complete when another observable does not completes', function () {
+    var e1 =   hot('--a--b--c--d--e--|');
+    var skip = hot('-');
+    var expected = '-';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should skip all element and does not complete when another observable completes after source', function () {
+    var e1 =   hot('--a--b--c--d--e--|');
+    var skip = hot('------------------------|');
+    var expected = '-';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should not completes if source does not completes when another observable does not emit', function () {
+    var e1 =   hot('-');
+    var skip = hot('--------------|');
+    var expected = '-';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should not completes if source and another observable both does not complete', function () {
+    var e1 =   hot('-');
+    var skip = hot('-');
+    var expected = '-';
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
+  });
+
+  it('should skip all element when another observable unsubscribed early before emit', function () {
+    var e1 =   hot('--a--b--c--d--e--|');
+    var skip = hot('-------------x--|');
+    var expected = '-';
+
+    e1.subscribe(function () {
+      skip.unsubscribe();
+    });
+
+    expectObservable(e1.skipUntil(skip)).toBe(expected);
   });
 });
\ No newline at end of file
diff --git a/src/operators/skipUntil.ts b/src/operators/skipUntil.ts
index dee57b571b..ae4ef693d7 100644
--- a/src/operators/skipUntil.ts
+++ b/src/operators/skipUntil.ts
@@ -11,35 +11,52 @@ class SkipUntilOperator<T, R> implements Operator<T, R> {
   constructor(private notifier: Observable<any>) {
   }
 
-  call(subscriber: Subscriber<R>): Subscriber<T> {
+  call(subscriber: Subscriber<T>): Subscriber<T> {
     return new SkipUntilSubscriber(subscriber, this.notifier);
   }
 }
 
 class SkipUntilSubscriber<T> extends Subscriber<T> {
-  private notificationSubscriber: NotificationSubscriber<any> = new NotificationSubscriber();
+  private notificationSubscriber: NotificationSubscriber<any> = null;
 
   constructor(destination: Subscriber<T>, private notifier: Observable<any>) {
     super(destination);
+    this.notificationSubscriber = new NotificationSubscriber(this);
     this.add(this.notifier.subscribe(this.notificationSubscriber));
   }
 
-  _next(x) {
-    if (this.notificationSubscriber.hasNotified) {
-      this.destination.next(x);
+  _next(value: T) {
+    if (this.notificationSubscriber.hasValue) {
+      this.destination.next(value);
     }
   }
+
+  _complete() {
+    if (this.notificationSubscriber.hasCompleted) {
+      this.destination.complete();
+    }
+    this.notificationSubscriber.unsubscribe();
+  }
 }
 
 class NotificationSubscriber<T> extends Subscriber<T> {
-  hasNotified: boolean = false;
+  hasValue: boolean = false;
+  hasCompleted: boolean = false;
 
-  constructor() {
+  constructor(private parent: SkipUntilSubscriber<any>) {
     super(null);
   }
 
-  _next() {
-    this.hasNotified = true;
-    this.unsubscribe();
+  _next(unused: T) {
+    this.hasValue = true;
+  }
+
+  _error(err) {
+    this.parent.error(err);
+    this.hasValue = true;
+  }
+
+  _complete() {
+    this.hasCompleted = true;
   }
 }