Fixing skipwhile and skipuntilwithtime

Author: mattpodwysocki

dist/rx.all.compat.js

@@ -5661,6 +5661,46 @@ Rx.Observable.prototype.flatMapLatest = function(selector, resultSelector, thisA
     if (count < 0) { throw new ArgumentOutOfRangeError(); }
     return new SkipObservable(this, count);
   };
+  var SkipWhileObservable = (function (__super__) {
+    inherits(SkipWhileObservable, __super__);
+    function SkipWhileObservable(source, fn) {
+      this.source = source;
+      this._fn = fn;
+      __super__.call(this);
+    }
+
+    SkipWhileObservable.prototype.subscribeCore = function (o) {
+      return this.source.subscribe(new SkipWhileObserver(o, this));
+    };
+
+    return SkipWhileObservable;
+  }(ObservableBase));
+
+  var SkipWhileObserver = (function (__super__) {
+    inherits(SkipWhileObserver, __super__);
+
+    function SkipWhileObserver(o, p) {
+      this._o = o;
+      this._p = p;
+      this._i = 0;
+      this._r = false;
+      __super__.call(this);
+    }
+
+    SkipWhileObserver.prototype.next = function (x) {
+      if (!this._r) {
+        var res = tryCatch(this._p._fn)(x, this._i++, this._p);
+        if (res === errorObj) { return this._o.onError(res.e); }
+        this._r = !res;
+      }
+      this._r && this._o.onNext(x);
+    };
+    SkipWhileObserver.prototype.error = function (e) { this._o.onError(e); };
+    SkipWhileObserver.prototype.completed = function () { this._o.onCompleted(); };
+
+    return SkipWhileObserver;
+  }(AbstractObserver));
+
   /**
    *  Bypasses elements in an observable sequence as long as a specified condition is true and then returns the remaining elements.
    *  The element's index is used in the logic of the predicate function.
@@ -5672,22 +5712,8 @@ Rx.Observable.prototype.flatMapLatest = function(selector, resultSelector, thisA
    * @returns {Observable} An observable sequence that contains the elements from the input sequence starting at the first element in the linear series that does not pass the test specified by predicate.
    */
   observableProto.skipWhile = function (predicate, thisArg) {
-    var source = this,
-        callback = bindCallback(predicate, thisArg, 3);
-    return new AnonymousObservable(function (o) {
-      var i = 0, running = false;
-      return source.subscribe(function (x) {
-        if (!running) {
-          try {
-            running = !callback(x, i++, source);
-          } catch (e) {
-            o.onError(e);
-            return;
-          }
-        }
-        running && o.onNext(x);
-      }, function (e) { o.onError(e); }, function () { o.onCompleted(); });
-    }, source);
+    var fn = bindCallback(predicate, thisArg, 3);
+    return new SkipWhileObservable(this, fn);
   };
 
   var TakeObservable = (function(__super__) {
@@ -10708,6 +10734,47 @@ Observable.fromNodeCallback = function (fn, ctx, selector) {
     }, source);
   };
 
+  var SkipUntilWithTimeObservable = (function (__super__) {
+    inherits(SkipUntilWithTimeObservable, __super__);
+    function SkipUntilWithTimeObservable(source, startTime, scheduler) {
+      this.source = source;
+      this._st = startTime;
+      this._s = scheduler;
+      __super__.call(this);
+    }
+
+    function scheduleMethod(s, state) {
+      state._open = true;
+    }
+
+    SkipUntilWithTimeObservable.prototype.subscribeCore = function (o) {
+      this._open = false;
+      return new BinaryDisposable(
+        this._s.scheduleFuture(this, this._st, scheduleMethod),
+        this.source.subscribe(new SkipUntilWithTimeObserver(o, this))
+      );
+    };
+
+    return SkipUntilWithTimeObservable;
+  }(ObservableBase));
+
+  var SkipUntilWithTimeObserver = (function (__super__) {
+    inherits(SkipUntilWithTimeObserver, __super__);
+
+    function SkipUntilWithTimeObserver(o, p) {
+      this._o = o;
+      this._p = p;
+      __super__.call(this);
+    }
+
+    SkipUntilWithTimeObserver.prototype.next = function (x) { this._p._open && this._o.onNext(x); };
+    SkipUntilWithTimeObserver.prototype.error = function (e) { this._o.onError(e); };
+    SkipUntilWithTimeObserver.prototype.completed = function () { this._o.onCompleted(); };
+
+    return SkipUntilWithTimeObserver;
+  }(AbstractObserver));
+
+
   /**
    *  Skips elements from the observable source sequence until the specified start time, using the specified scheduler to run timers.
    *  Errors produced by the source sequence are always forwarded to the result sequence, even if the error occurs before the start time.
@@ -10721,16 +10788,7 @@ Observable.fromNodeCallback = function (fn, ctx, selector) {
    */
   observableProto.skipUntilWithTime = function (startTime, scheduler) {
     isScheduler(scheduler) || (scheduler = defaultScheduler);
-    var source = this;
-    return new AnonymousObservable(function (o) {
-      var open = false;
-
-      return new BinaryDisposable(
-        scheduler.scheduleFuture(null, startTime, function () { open = true; }),
-        source.subscribe(
-          function (x) { open && o.onNext(x); },
-          function (e) { o.onError(e); }, function () { o.onCompleted(); }));
-    }, source);
+    return new SkipUntilWithTimeObservable(this, startTime, scheduler);
   };
 
   /**

dist/rx.all.compat.map

@@ -5523,6 +5523,46 @@ Rx.Observable.prototype.flatMapLatest = function(selector, resultSelector, thisA
     if (count < 0) { throw new ArgumentOutOfRangeError(); }
     return new SkipObservable(this, count);
   };
+  var SkipWhileObservable = (function (__super__) {
+    inherits(SkipWhileObservable, __super__);
+    function SkipWhileObservable(source, fn) {
+      this.source = source;
+      this._fn = fn;
+      __super__.call(this);
+    }
+
+    SkipWhileObservable.prototype.subscribeCore = function (o) {
+      return this.source.subscribe(new SkipWhileObserver(o, this));
+    };
+
+    return SkipWhileObservable;
+  }(ObservableBase));
+
+  var SkipWhileObserver = (function (__super__) {
+    inherits(SkipWhileObserver, __super__);
+
+    function SkipWhileObserver(o, p) {
+      this._o = o;
+      this._p = p;
+      this._i = 0;
+      this._r = false;
+      __super__.call(this);
+    }
+
+    SkipWhileObserver.prototype.next = function (x) {
+      if (!this._r) {
+        var res = tryCatch(this._p._fn)(x, this._i++, this._p);
+        if (res === errorObj) { return this._o.onError(res.e); }
+        this._r = !res;
+      }
+      this._r && this._o.onNext(x);
+    };
+    SkipWhileObserver.prototype.error = function (e) { this._o.onError(e); };
+    SkipWhileObserver.prototype.completed = function () { this._o.onCompleted(); };
+
+    return SkipWhileObserver;
+  }(AbstractObserver));
+
   /**
    *  Bypasses elements in an observable sequence as long as a specified condition is true and then returns the remaining elements.
    *  The element's index is used in the logic of the predicate function.
@@ -5534,22 +5574,8 @@ Rx.Observable.prototype.flatMapLatest = function(selector, resultSelector, thisA
    * @returns {Observable} An observable sequence that contains the elements from the input sequence starting at the first element in the linear series that does not pass the test specified by predicate.
    */
   observableProto.skipWhile = function (predicate, thisArg) {
-    var source = this,
-        callback = bindCallback(predicate, thisArg, 3);
-    return new AnonymousObservable(function (o) {
-      var i = 0, running = false;
-      return source.subscribe(function (x) {
-        if (!running) {
-          try {
-            running = !callback(x, i++, source);
-          } catch (e) {
-            o.onError(e);
-            return;
-          }
-        }
-        running && o.onNext(x);
-      }, function (e) { o.onError(e); }, function () { o.onCompleted(); });
-    }, source);
+    var fn = bindCallback(predicate, thisArg, 3);
+    return new SkipWhileObservable(this, fn);
   };
 
   var TakeObservable = (function(__super__) {
@@ -10477,6 +10503,47 @@ Observable.fromNodeCallback = function (fn, ctx, selector) {
     }, source);
   };
 
+  var SkipUntilWithTimeObservable = (function (__super__) {
+    inherits(SkipUntilWithTimeObservable, __super__);
+    function SkipUntilWithTimeObservable(source, startTime, scheduler) {
+      this.source = source;
+      this._st = startTime;
+      this._s = scheduler;
+      __super__.call(this);
+    }
+
+    function scheduleMethod(s, state) {
+      state._open = true;
+    }
+
+    SkipUntilWithTimeObservable.prototype.subscribeCore = function (o) {
+      this._open = false;
+      return new BinaryDisposable(
+        this._s.scheduleFuture(this, this._st, scheduleMethod),
+        this.source.subscribe(new SkipUntilWithTimeObserver(o, this))
+      );
+    };
+
+    return SkipUntilWithTimeObservable;
+  }(ObservableBase));
+
+  var SkipUntilWithTimeObserver = (function (__super__) {
+    inherits(SkipUntilWithTimeObserver, __super__);
+
+    function SkipUntilWithTimeObserver(o, p) {
+      this._o = o;
+      this._p = p;
+      __super__.call(this);
+    }
+
+    SkipUntilWithTimeObserver.prototype.next = function (x) { this._p._open && this._o.onNext(x); };
+    SkipUntilWithTimeObserver.prototype.error = function (e) { this._o.onError(e); };
+    SkipUntilWithTimeObserver.prototype.completed = function () { this._o.onCompleted(); };
+
+    return SkipUntilWithTimeObserver;
+  }(AbstractObserver));
+
+
   /**
    *  Skips elements from the observable source sequence until the specified start time, using the specified scheduler to run timers.
    *  Errors produced by the source sequence are always forwarded to the result sequence, even if the error occurs before the start time.
@@ -10490,16 +10557,7 @@ Observable.fromNodeCallback = function (fn, ctx, selector) {
    */
   observableProto.skipUntilWithTime = function (startTime, scheduler) {
     isScheduler(scheduler) || (scheduler = defaultScheduler);
-    var source = this;
-    return new AnonymousObservable(function (o) {
-      var open = false;
-
-      return new BinaryDisposable(
-        scheduler.scheduleFuture(null, startTime, function () { open = true; }),
-        source.subscribe(
-          function (x) { open && o.onNext(x); },
-          function (e) { o.onError(e); }, function () { o.onCompleted(); }));
-    }, source);
+    return new SkipUntilWithTimeObservable(this, startTime, scheduler);
   };
 
   /**

dist/rx.all.compat.min.js

@@ -5559,6 +5559,46 @@ Rx.Observable.prototype.flatMapLatest = function(selector, resultSelector, thisA
     if (count < 0) { throw new ArgumentOutOfRangeError(); }
     return new SkipObservable(this, count);
   };
+  var SkipWhileObservable = (function (__super__) {
+    inherits(SkipWhileObservable, __super__);
+    function SkipWhileObservable(source, fn) {
+      this.source = source;
+      this._fn = fn;
+      __super__.call(this);
+    }
+
+    SkipWhileObservable.prototype.subscribeCore = function (o) {
+      return this.source.subscribe(new SkipWhileObserver(o, this));
+    };
+
+    return SkipWhileObservable;
+  }(ObservableBase));
+
+  var SkipWhileObserver = (function (__super__) {
+    inherits(SkipWhileObserver, __super__);
+
+    function SkipWhileObserver(o, p) {
+      this._o = o;
+      this._p = p;
+      this._i = 0;
+      this._r = false;
+      __super__.call(this);
+    }
+
+    SkipWhileObserver.prototype.next = function (x) {
+      if (!this._r) {
+        var res = tryCatch(this._p._fn)(x, this._i++, this._p);
+        if (res === errorObj) { return this._o.onError(res.e); }
+        this._r = !res;
+      }
+      this._r && this._o.onNext(x);
+    };
+    SkipWhileObserver.prototype.error = function (e) { this._o.onError(e); };
+    SkipWhileObserver.prototype.completed = function () { this._o.onCompleted(); };
+
+    return SkipWhileObserver;
+  }(AbstractObserver));
+
   /**
    *  Bypasses elements in an observable sequence as long as a specified condition is true and then returns the remaining elements.
    *  The element's index is used in the logic of the predicate function.
@@ -5570,22 +5610,8 @@ Rx.Observable.prototype.flatMapLatest = function(selector, resultSelector, thisA
    * @returns {Observable} An observable sequence that contains the elements from the input sequence starting at the first element in the linear series that does not pass the test specified by predicate.
    */
   observableProto.skipWhile = function (predicate, thisArg) {
-    var source = this,
-        callback = bindCallback(predicate, thisArg, 3);
-    return new AnonymousObservable(function (o) {
-      var i = 0, running = false;
-      return source.subscribe(function (x) {
-        if (!running) {
-          try {
-            running = !callback(x, i++, source);
-          } catch (e) {
-            o.onError(e);
-            return;
-          }
-        }
-        running && o.onNext(x);
-      }, function (e) { o.onError(e); }, function () { o.onCompleted(); });
-    }, source);
+    var fn = bindCallback(predicate, thisArg, 3);
+    return new SkipWhileObservable(this, fn);
   };
 
   var TakeObservable = (function(__super__) {