import { Operator } from '../Operator'; import { Subscriber } from '../Subscriber'; import { ArgumentOutOfRangeError } from '../util/ArgumentOutOfRangeError'; import { EmptyObservable } from '../observable/EmptyObservable'; import { Observable } from '../Observable'; import { TeardownLogic } from '../Subscription'; import { MonoTypeOperatorFunction } from '../interfaces'; /** * Emits only the first `count` values emitted by the source Observable. * * Takes the first `count` values from the source, then * completes. * * * * `take` returns an Observable that emits only the first `count` values emitted * by the source Observable. If the source emits fewer than `count` values then * all of its values are emitted. After that, it completes, regardless if the * source completes. * * @example Take the first 5 seconds of an infinite 1-second interval Observable * var interval = Rx.Observable.interval(1000); * var five = interval.take(5); * five.subscribe(x => console.log(x)); * * @see {@link takeLast} * @see {@link takeUntil} * @see {@link takeWhile} * @see {@link skip} * * @throws {ArgumentOutOfRangeError} When using `take(i)`, it delivers an * ArgumentOutOrRangeError to the Observer's `error` callback if `i < 0`. * * @param {number} count The maximum number of `next` values to emit. * @return {Observable} An Observable that emits only the first `count` * values emitted by the source Observable, or all of the values from the source * if the source emits fewer than `count` values. * @method take * @owner Observable */ export function take(count: number): MonoTypeOperatorFunction { return (source: Observable) => { if (count === 0) { return new EmptyObservable(); } else { return source.lift(new TakeOperator(count)); } }; } class TakeOperator implements Operator { constructor(private total: number) { if (this.total < 0) { throw new ArgumentOutOfRangeError; } } call(subscriber: Subscriber, source: any): TeardownLogic { return source.subscribe(new TakeSubscriber(subscriber, this.total)); } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ class TakeSubscriber extends Subscriber { private count: number = 0; constructor(destination: Subscriber, private total: number) { super(destination); } protected _next(value: T): void { const total = this.total; const count = ++this.count; if (count <= total) { this.destination.next(value); if (count === total) { this.destination.complete(); this.unsubscribe(); } } } }