import {Observable} from '../Observable'; import {Operator} from '../Operator'; import {Subscriber} from '../Subscriber'; import {Scheduler} from '../Scheduler'; import {async} from '../scheduler/async'; /** * @param delay * @param scheduler * @return {Observable|WebSocketSubject|Observable} * @method sampleTime * @owner Observable */ export function sampleTime(delay: number, scheduler: Scheduler = async): Observable { return this.lift(new SampleTimeOperator(delay, scheduler)); } export interface SampleTimeSignature { (delay: number, scheduler?: Scheduler): Observable; } class SampleTimeOperator implements Operator { constructor(private delay: number, private scheduler: Scheduler) { } call(subscriber: Subscriber, source: any): any { return source._subscribe(new SampleTimeSubscriber(subscriber, this.delay, this.scheduler)); } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ class SampleTimeSubscriber extends Subscriber { lastValue: T; hasValue: boolean = false; constructor(destination: Subscriber, private delay: number, private scheduler: Scheduler) { super(destination); this.add(scheduler.schedule(dispatchNotification, delay, { subscriber: this, delay })); } protected _next(value: T) { this.lastValue = value; this.hasValue = true; } notifyNext() { if (this.hasValue) { this.hasValue = false; this.destination.next(this.lastValue); } } } function dispatchNotification(state: any) { let { subscriber, delay } = state; subscriber.notifyNext(); (this).schedule(state, delay); }