feat(operator): add skipWhile operator

spec/operators/skipWhile-spec.js

@@ -0,0 +1,142 @@
+/* globals describe, it, expect, expectObservable, hot */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.skipWhile()', function () {
+  it('should skip all elements with a true predicate', function () {
+    var source = hot('-1-^2--3--4--5--6--|');
+    var expected =      '----------------|';
+
+    expectObservable(source.skipWhile(function () { return true; })).toBe(expected);
+  });
+
+  it('should skip all elements with a truthy predicate', function () {
+    var source = hot('-1-^2--3--4--5--6--|');
+    var expected =      '----------------|';
+
+    expectObservable(source.skipWhile(function () { return {}; })).toBe(expected);
+  });
+
+  it('should not skip any element with a false predicate', function () {
+    var source = hot('-1-^2--3--4--5--6--|');
+    var expected =      '-2--3--4--5--6--|';
+
+    expectObservable(source.skipWhile(function () { return false; })).toBe(expected);
+  });
+
+  it('should not skip any elements with a falsy predicate', function () {
+    var source = hot('-1-^2--3--4--5--6--|');
+    var expected =      '-2--3--4--5--6--|';
+
+    expectObservable(source.skipWhile(function () { return undefined; })).toBe(expected);
+  });
+
+  it('should skip all elements until predicate is true', function () {
+    var source = hot('-1-^2--3--4--5--6--|');
+    var expected =      '-------4--5--6--|';
+
+    var predicate = function (v) {
+      return +v < 4;
+    };
+
+    expectObservable(source.skipWhile(predicate)).toBe(expected);
+  });
+
+  it('should skip elements on hot source', function () {
+    var source = hot('--1--2-^-3--4--5--6--7--8--');
+    var expected =          '--------5--6--7--8--';
+
+    var predicate = function (v) {
+      return +v < 5;
+    };
+
+    expectObservable(source.skipWhile(predicate)).toBe(expected);
+  });
+
+  it('should be possible to skip using the element\'s index', function () {
+    var source = hot('--a--b-^-c--d--e--f--g--h--|');
+    var expected =          '--------e--f--g--h--|';
+
+    var predicate = function (v, index) {
+      return index < 2;
+    };
+
+    expectObservable(source.skipWhile(predicate)).toBe(expected);
+  });
+
+  it('should skip using index with source unsubscribes early', function () {
+    var source = hot('--a--b-^-c--d--e--f--g--h--|');
+    var unsub =             '-----------!';
+    var expected =          '-----d--e---';
+
+    var predicate = function (v, index) {
+      return index < 1;
+    };
+
+    expectObservable(source.skipWhile(predicate), unsub).toBe(expected);
+  });
+
+  it('should skip using value with source throws', function () {
+    var source = hot('--a--b-^-c--d--e--f--g--h--#');
+    var expected =          '-----d--e--f--g--h--#';
+
+    var predicate = function (v) {
+      return v !== 'd';
+    };
+
+    expectObservable(source.skipWhile(predicate)).toBe(expected);
+  });
+
+  it('should invoke predicate while its false and never again', function () {
+    var source = hot('--a--b-^-c--d--e--f--g--h--|');
+    var expected =          '--------e--f--g--h--|';
+
+    var invoked = 0;
+    var predicate = function (v) {
+      invoked++;
+      return v !== 'e';
+    };
+
+    expectObservable(
+      source.skipWhile(predicate).do(null, null, function () {
+        expect(invoked).toBe(3);
+      })
+    ).toBe(expected);
+  });
+
+  it('should accept a thisArg', function () {
+    var source = hot('-1-^--2--3--4--5--6--|');
+    var expected =      '---------4--5--6--|';
+
+    function Skiper() {
+      this.doSkip = function (v) { return +v < 4; };
+    }
+
+    var skiper = new Skiper();
+
+    expectObservable(
+      source.skipWhile(function (v) { return this.doSkip(v); }, skiper)
+    ).toBe(expected);
+  });
+
+  it('should handle Observable.empty', function () {
+    var source = Observable.empty();
+    var expected = '|';
+
+    expectObservable(source.skipWhile(function () { return true; })).toBe(expected);
+  });
+
+  it('should handle Observable.never', function () {
+    var source = Observable.never();
+    var expected = '-';
+
+    expectObservable(source.skipWhile(function () { return true; })).toBe(expected);
+  });
+
+  it('should handle Observable.throw', function () {
+    var source = Observable.throw(new Error('oh no!'));
+    var expected = '#';
+
+    expectObservable(source.skipWhile(function () { return true; })).toBe(expected, undefined, new Error('oh no!'));
+  });
+});

src/CoreOperators.ts

@@ -68,6 +68,7 @@ export interface CoreOperators<T> {
   single?: (predicate?: (value: T, index: number) => boolean, thisArg?: any) => Observable<T>;
   skip?: (count: number) => Observable<T>;
   skipUntil?: (notifier: Observable<any>) => Observable<T>;
+  skipWhile?: (predicate: (x: T, index: number) => boolean, thisArg?: any) => Observable<T>;
   startWith?: (x: T) => Observable<T>;
   subscribeOn?: (scheduler: Scheduler, delay?: number) => Observable<T>;
   switch?: () => Observable<T>;

src/Rx.KitchenSink.ts

@@ -258,6 +258,9 @@ observableProto.skip = skip;
 import {skipUntil} from './operators/skipUntil';
 observableProto.skipUntil = skipUntil;
 
+import {skipWhile} from './operators/skipWhile';
+observableProto.skipWhile = skipWhile;
+
 import {startWith} from './operators/startWith';
 observableProto.startWith = startWith;
 

src/Rx.ts

@@ -218,6 +218,9 @@ observableProto.skip = skip;
 import {skipUntil} from './operators/skipUntil';
 observableProto.skipUntil = skipUntil;
 
+import {skipWhile} from './operators/skipWhile';
+observableProto.skipWhile = skipWhile;
+
 import {startWith} from './operators/startWith';
 observableProto.startWith = startWith;
 
@@ -313,4 +316,4 @@ export {
     Notification,
     EmptyError,
     ArgumentOutOfRangeError
-};
+};

src/operators/skipWhile.ts

@@ -0,0 +1,47 @@
+import {Operator} from '../Operator';
+import {Subscriber} from '../Subscriber';
+import {tryCatch} from '../util/tryCatch';
+import {errorObject} from '../util/errorObject';
+import {bindCallback} from '../util/bindCallback';
+
+export function skipWhile<T>(predicate: (x: T, index: number) => boolean, thisArg?: any) {
+  return this.lift(new SkipWhileOperator(predicate, thisArg));
+}
+
+class SkipWhileOperator<T, R> implements Operator<T, R> {
+  predicate: (x: T, index: number) => boolean;
+
+  constructor(predicate: (x: T, index: number) => boolean, thisArg?: any) {
+    this.predicate = <(x: T, index: number) => boolean>bindCallback(predicate, thisArg, 2);
+  }
+
+  call(subscriber: Subscriber<T>): Subscriber<T> {
+    return new SkipWhileSubscriber(subscriber, this.predicate);
+  }
+}
+
+class SkipWhileSubscriber<T> extends Subscriber<T> {
+  private skipping: boolean = true;
+  private index: number = 0;
+
+  constructor(destination: Subscriber<T>,
+              private predicate: (x: T, index: number) => boolean) {
+    super(destination);
+  }
+
+  _next(value: T): void {
+    const destination = this.destination;
+    if (this.skipping === true) {
+      const index = this.index++;
+      const result = tryCatch(this.predicate)(value, index);
+      if (result === errorObject) {
+        destination.error(result.e);
+      } else {
+        this.skipping = Boolean(result);
+      }
+    }
+    if (this.skipping === false) {
+      destination.next(value);
+    }
+  }
+}