import {Operator} from '../Operator'; import {Subscriber} from '../Subscriber'; import {Scheduler} from '../Scheduler'; import {Subscription} from '../Subscription'; import {async} from '../scheduler/async'; import {Observable} from '../Observable'; /** * @param delay * @param scheduler * @return {Observable|WebSocketSubject|Observable} * @method throttleTime * @owner Observable */ export function throttleTime(delay: number, scheduler: Scheduler = async): Observable { return this.lift(new ThrottleTimeOperator(delay, scheduler)); } export interface ThrottleTimeSignature { (dueTime: number, scheduler?: Scheduler): Observable; } class ThrottleTimeOperator implements Operator { constructor(private delay: number, private scheduler: Scheduler) { } call(subscriber: Subscriber, source: any): any { return source._subscribe(new ThrottleTimeSubscriber(subscriber, this.delay, this.scheduler)); } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ class ThrottleTimeSubscriber extends Subscriber { private throttled: Subscription; constructor(destination: Subscriber, private delay: number, private scheduler: Scheduler) { super(destination); } protected _next(value: T) { if (!this.throttled) { this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.delay, { subscriber: this })); this.destination.next(value); } } clearThrottle() { const throttled = this.throttled; if (throttled) { throttled.unsubscribe(); this.remove(throttled); this.throttled = null; } } } interface DispatchArg { subscriber: ThrottleTimeSubscriber; } function dispatchNext(arg: DispatchArg) { const { subscriber } = arg; subscriber.clearThrottle(); }