feat(takeWhile): add takeWhile operator

closes ReactiveX#695

spec/operators/takeWhile-spec.js

@@ -0,0 +1,175 @@
+/* globals describe, it, expect, expectObservable, expectSubscriptions, hot */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.takeWhile()', function () {
+  it('should take all elements with predicate returns true', function () {
+    var e1 = hot('--a-^-b--c--d--e--|');
+    var sub =        '^             !';
+    var expected =   '--b--c--d--e--|';
+
+    expectObservable(e1.takeWhile(function () { return true; })).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it('should take all elements with truthy predicate', function () {
+    var e1 = hot('--a-^-b--c--d--e--|');
+    var sub =        '^             !';
+    var expected =   '--b--c--d--e--|';
+
+    expectObservable(e1.takeWhile(function () { return {}; })).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it('should skip all elements with predicate returns false', function () {
+    var e1 = hot('--a-^-b--c--d--e--|');
+    var sub =        '^ !';
+    var expected =   '--|';
+
+    expectObservable(e1.takeWhile(function () { return false; })).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it('should skip all elements with falsy predicate', function () {
+    var e1 = hot('--a-^-b--c--d--e--|');
+    var sub =        '^ !';
+    var expected =   '--|';
+
+    expectObservable(e1.takeWhile(function () { return null; })).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it ('should take all elements until predicate return false', function () {
+    var e1 = hot('--a-^-b--c--d--e--|');
+    var sub =        '^       !';
+    var expected =   '--b--c--|';
+
+    function predicate(value) {
+      return value !== 'd';
+    }
+
+    expectObservable(e1.takeWhile(predicate)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it ('should take elements with predicate when source does not complete', function () {
+    var e1 = hot('--a-^-b--c--d--e--');
+    var sub =        '^             ';
+    var expected =   '--b--c--d--e--';
+
+    expectObservable(e1.takeWhile(function () { return true; })).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it ('should not complete when source never completes', function () {
+    var e1 = Observable.never();
+    var expected =   '-';
+
+    expectObservable(e1.takeWhile(function () { return true; })).toBe(expected);
+  });
+
+  it ('should complete when source does not emit', function () {
+    var e1 = hot('--a-^------------|');
+    var sub =        '^            !';
+    var expected =   '-------------|';
+
+    expectObservable(e1.takeWhile(function () { return true; })).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it ('should complete when source is empty', function () {
+    var e1 = Observable.empty();
+    var expected =   '|';
+
+    expectObservable(e1.takeWhile(function () { return true; })).toBe(expected);
+  });
+
+  it ('should pass element index to predicate', function () {
+    var e1 = hot('--a-^-b--c--d--e--|');
+    var sub =        '^       !';
+    var expected =   '--b--c--|';
+
+    function predicate(value, index) {
+      return index < 2;
+    }
+
+    expectObservable(e1.takeWhile(predicate)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it ('should raise error when source raises error', function () {
+    var e1 = hot('--a-^-b--c--d--e--#');
+    var sub =        '^             !';
+    var expected =   '--b--c--d--e--#';
+
+    expectObservable(e1.takeWhile(function () { return true; })).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it('should raise error when source throws', function () {
+    var source = Observable.throw('error');
+    var expected = '#';
+
+    expectObservable(source.takeWhile(function () { return true; })).toBe(expected);
+  });
+
+  it ('should invoke predicate until return false', function () {
+    var e1 = hot('--a-^-b--c--d--e--|');
+    var sub =        '^       !';
+    var expected =   '--b--c--|';
+
+    var invoked = 0;
+    function predicate(value) {
+      invoked++;
+      return value !== 'd';
+    }
+
+    var source = e1.takeWhile(predicate).do(null, null, function () {
+      expect(invoked).toBe(3);
+    });
+    expectObservable(source).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it('should raise error if predicate throws', function () {
+    var e1 = hot('--a-^-b--c--d--e--|');
+    var sub =        '^ !';
+    var expected =   '--#';
+
+    function predicate(value) {
+      throw 'error';
+    }
+
+    expectObservable(e1.takeWhile(predicate)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it ('should pass element thisArg to predicate', function () {
+    var e1 = hot('--a-^-b--c--d--e--|');
+    var sub =        '^       !';
+    var expected =   '--b--c--|';
+
+    function predicate() {
+      this.take = function (value) {
+        return value !== 'd';
+      }
+    }
+
+    expectObservable(e1.takeWhile(function (v) { return this.take(v); }, new predicate())).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it ('should take elements until unsubscribed', function () {
+    var e1 = hot('--a-^-b--c--d--e--|');
+    var unsub =      '-----!';
+    var sub =        '^    !';
+    var expected =   '--b---';
+
+    function predicate(value) {
+      return value !== 'd';
+    }
+
+    expectObservable(e1.takeWhile(predicate), unsub).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+});

src/CoreOperators.ts

@@ -75,6 +75,7 @@ export interface CoreOperators<T> {
   switchMapTo?: <R>(observable: Observable<any>, projectResult?: (x: T, y: any, ix: number, iy: number) => R) => Observable<R>;
   take?: (count: number) => Observable<T>;
   takeUntil?: (notifier: Observable<any>) => Observable<T>;
+  takeWhile?: (predicate: (value: T, index: number) => boolean, thisArg?: any) => Observable<T>;
   throttle?: (durationSelector: (value: T) => Observable<any> | Promise<any>) => Observable<T>;
   throttleTime?: (delay: number, scheduler?: Scheduler) => Observable<T>;
   timeout?: (due: number | Date, errorToSend?: any, scheduler?: Scheduler) => Observable<T>;
@@ -89,4 +90,4 @@ export interface CoreOperators<T> {
   withLatestFrom?: <R>(...observables: Array<Observable<any> | ((...values: Array<any>) => R)>) => Observable<R>;
   zip?: <R>(...observables: Array<Observable<any> | ((...values: Array<any>) => R)>) => Observable<R>;
   zipAll?: <R>(project?: (...values: Array<any>) => R) => Observable<R>;
-}
+}

src/Observable.ts

@@ -228,6 +228,7 @@ export class Observable<T> implements CoreOperators<T>  {
   switchMapTo: <R>(observable: Observable<any>, projectResult?: (x: T, y: any, ix: number, iy: number) => R) => Observable<R>;
   take: (count: number) => Observable<T>;
   takeUntil: (notifier: Observable<any>) => Observable<T>;
+  takeWhile: (predicate: (value: T, index: number) => boolean, thisArg?: any) => Observable<T>;
   throttle: (durationSelector: (value: T) => Observable<any> | Promise<any>) => Observable<T>;
   throttleTime: (delay: number, scheduler?: Scheduler) => Observable<T>;
   timeout: (due: number | Date, errorToSend?: any, scheduler?: Scheduler) => Observable<T>;

src/Rx.KitchenSink.ts

@@ -285,6 +285,9 @@ observableProto.take = take;
 import {takeUntil} from './operators/takeUntil';
 observableProto.takeUntil = takeUntil;
 
+import {takeWhile} from './operators/takeWhile';
+observableProto.takeWhile = takeWhile;
+
 import {throttle} from './operators/throttle';
 observableProto.throttle = throttle;
 

src/Rx.ts

@@ -239,6 +239,9 @@ observableProto.take = take;
 import {takeUntil} from './operators/takeUntil';
 observableProto.takeUntil = takeUntil;
 
+import {takeWhile} from './operators/takeWhile';
+observableProto.takeWhile = takeWhile;
+
 import {throttle} from './operators/throttle';
 observableProto.throttle = throttle;
 

src/operators/takeWhile.ts

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