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