import {Observable} from '../Observable'; import {Scheduler} from '../Scheduler'; import {Operator} from '../Operator'; import {PartialObserver} from '../Observer'; import {Subscriber} from '../Subscriber'; import {Notification} from '../Notification'; /** * @see {@link Notification} * * @param scheduler * @param delay * @return {Observable|WebSocketSubject|Observable} * @method observeOn * @owner Observable */ export function observeOn(scheduler: Scheduler, delay: number = 0): Observable { return this.lift(new ObserveOnOperator(scheduler, delay)); } export interface ObserveOnSignature { (scheduler: Scheduler, delay?: number): Observable; } export class ObserveOnOperator implements Operator { constructor(private scheduler: Scheduler, private delay: number = 0) { } call(subscriber: Subscriber, source: any): any { return source._subscribe(new ObserveOnSubscriber(subscriber, this.scheduler, this.delay)); } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ export class ObserveOnSubscriber extends Subscriber { static dispatch(arg: ObserveOnMessage) { const { notification, destination } = arg; notification.observe(destination); } constructor(destination: Subscriber, private scheduler: Scheduler, private delay: number = 0) { super(destination); } private scheduleMessage(notification: Notification): void { this.add(this.scheduler.schedule(ObserveOnSubscriber.dispatch, this.delay, new ObserveOnMessage(notification, this.destination))); } protected _next(value: T): void { this.scheduleMessage(Notification.createNext(value)); } protected _error(err: any): void { this.scheduleMessage(Notification.createError(err)); } protected _complete(): void { this.scheduleMessage(Notification.createComplete()); } } export class ObserveOnMessage { constructor(public notification: Notification, public destination: PartialObserver) { } }