import { Observable } from '../Observable'; import { Scheduler } from '../Scheduler'; import { Operator } from '../Operator'; import { PartialObserver } from '../Observer'; import { Subscriber } from '../Subscriber'; import { Notification } from '../Notification'; import { TeardownLogic } from '../Subscription'; /** * @see {@link Notification} * * @param scheduler * @param delay * @return {Observable|WebSocketSubject|Observable} * @method observeOn * @owner Observable */ export function observeOn(this: Observable, scheduler: Scheduler, delay: number = 0): Observable { return this.lift(new ObserveOnOperator(scheduler, delay)); } export class ObserveOnOperator implements Operator { constructor(private scheduler: Scheduler, private delay: number = 0) { } call(subscriber: Subscriber, source: any): TeardownLogic { 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) { } }