feat(takeWhile): add higher-order lettable version of takeWhile

src/operator/takeWhile.ts

@@ -1,7 +1,5 @@
-import { Operator } from '../Operator';
 import { Observable } from '../Observable';
-import { Subscriber } from '../Subscriber';
-import { TeardownLogic } from '../Subscription';
+import { takeWhile as higherOrder } from '../operators/takeWhile';
 
 /**
  * Emits values emitted by the source Observable so long as each value satisfies
@@ -40,49 +38,5 @@ import { TeardownLogic } from '../Subscription';
  * @owner Observable
  */
 export function takeWhile<T>(this: Observable<T>, predicate: (value: T, index: number) => boolean): Observable<T> {
- return this.lift(new TakeWhileOperator(predicate));
-}
-
-class TakeWhileOperator<T> implements Operator<T, T> {
- constructor(private predicate: (value: T, index: number) => boolean) {
- }
-
- call(subscriber: Subscriber<T>, source: any): TeardownLogic {
- return source.subscribe(new TakeWhileSubscriber(subscriber, this.predicate));
- }
-}
-
-/**
- * We need this JSDoc comment for affecting ESDoc.
- * @ignore
- * @extends {Ignored}
- */
-class TakeWhileSubscriber<T> extends Subscriber<T> {
- private index: number = 0;
-
- constructor(destination: Subscriber<T>,
- private predicate: (value: T, index: number) => boolean) {
- super(destination);
- }
-
- protected _next(value: T): void {
- const destination = this.destination;
- let result: boolean;
- try {
- result = this.predicate(value, this.index++);
- } catch (err) {
- destination.error(err);
- return;
- }
- this.nextOrComplete(value, result);
- }
-
- private nextOrComplete(value: T, predicateResult: boolean): void {
- const destination = this.destination;
- if (Boolean(predicateResult)) {
- destination.next(value);
- } else {
- destination.complete();
- }
- }
+ return higherOrder(predicate)(this);
 }

src/operators/index.ts

@@ -72,6 +72,7 @@ export { switchMapTo } from './switchMapTo';
 export { take } from './take';
 export { takeLast } from './takeLast';
 export { takeUntil } from './takeUntil';
+export { takeWhile } from './takeWhile';
 export { tap } from './tap';
 export { timestamp } from './timestamp';
 export { toArray } from './toArray';

src/operators/takeWhile.ts

@@ -0,0 +1,89 @@
+import { Operator } from '../Operator';
+import { Observable } from '../Observable';
+import { Subscriber } from '../Subscriber';
+import { TeardownLogic } from '../Subscription';
+import { MonoTypeOperatorFunction } from '../interfaces';
+
+/**
+ * Emits values emitted by the source Observable so long as each value satisfies
+ * the given `predicate`, and then completes as soon as this `predicate` is not
+ * satisfied.
+ *
+ * <span class="informal">Takes values from the source only while they pass the
+ * condition given. When the first value does not satisfy, it completes.</span>
+ *
+ * <img src="./img/takeWhile.png" width="100%">
+ *
+ * `takeWhile` subscribes and begins mirroring the source Observable. Each value
+ * emitted on the source is given to the `predicate` function which returns a
+ * boolean, representing a condition to be satisfied by the source values. The
+ * output Observable emits the source values until such time as the `predicate`
+ * returns false, at which point `takeWhile` stops mirroring the source
+ * Observable and completes the output Observable.
+ *
+ * @example <caption>Emit click events only while the clientX property is greater than 200</caption>
+ * var clicks = Rx.Observable.fromEvent(document, 'click');
+ * var result = clicks.takeWhile(ev => ev.clientX > 200);
+ * result.subscribe(x => console.log(x));
+ *
+ * @see {@link take}
+ * @see {@link takeLast}
+ * @see {@link takeUntil}
+ * @see {@link skip}
+ *
+ * @param {function(value: T, index: number): boolean} predicate A function that
+ * evaluates a value emitted by the source Observable and returns a boolean.
+ * Also takes the (zero-based) index as the second argument.
+ * @return {Observable<T>} An Observable that emits the values from the source
+ * Observable so long as each value satisfies the condition defined by the
+ * `predicate`, then completes.
+ * @method takeWhile
+ * @owner Observable
+ */
+export function takeWhile<T>(predicate: (value: T, index: number) => boolean): MonoTypeOperatorFunction<T> {
+ return (source: Observable<T>) => source.lift(new TakeWhileOperator(predicate));
+}
+
+class TakeWhileOperator<T> implements Operator<T, T> {
+ constructor(private predicate: (value: T, index: number) => boolean) {
+ }
+
+ call(subscriber: Subscriber<T>, source: any): TeardownLogic {
+ return source.subscribe(new TakeWhileSubscriber(subscriber, this.predicate));
+ }
+}
+
+/**
+ * We need this JSDoc comment for affecting ESDoc.
+ * @ignore
+ * @extends {Ignored}
+ */
+class TakeWhileSubscriber<T> extends Subscriber<T> {
+ private index: number = 0;
+
+ constructor(destination: Subscriber<T>,
+ private predicate: (value: T, index: number) => boolean) {
+ super(destination);
+ }
+
+ protected _next(value: T): void {
+ const destination = this.destination;
+ let result: boolean;
+ try {
+ result = this.predicate(value, this.index++);
+ } catch (err) {
+ destination.error(err);
+ return;
+ }
+ this.nextOrComplete(value, result);
+ }
+
+ private nextOrComplete(value: T, predicateResult: boolean): void {
+ const destination = this.destination;
+ if (Boolean(predicateResult)) {
+ destination.next(value);
+ } else {
+ destination.complete();
+ }
+ }
+}