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. * * Takes values from the source only while they pass the * condition given. When the first value does not satisfy, it completes. * * * * `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 Emit click events only while the clientX property is greater than 200 * 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} 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(predicate: (value: T, index: number) => boolean): MonoTypeOperatorFunction { return (source: Observable) => source.lift(new TakeWhileOperator(predicate)); } class TakeWhileOperator implements Operator { constructor(private predicate: (value: T, index: number) => boolean) { } call(subscriber: Subscriber, source: any): TeardownLogic { return source.subscribe(new TakeWhileSubscriber(subscriber, this.predicate)); } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ class TakeWhileSubscriber extends Subscriber { private index: number = 0; constructor(destination: Subscriber, 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(); } } }