1import { async } from '../scheduler/async'; 2import { isDate } from '../util/isDate'; 3import { Operator } from '../Operator'; 4import { Subscriber } from '../Subscriber'; 5import { Subscription } from '../Subscription'; 6import { Notification } from '../Notification'; 7import { Observable } from '../Observable'; 8import { MonoTypeOperatorFunction, PartialObserver, SchedulerAction, SchedulerLike, TeardownLogic } from '../types'; 9 10/** 11 * Delays the emission of items from the source Observable by a given timeout or 12 * until a given Date. 13 * 14 * <span class="informal">Time shifts each item by some specified amount of 15 * milliseconds.</span> 16 * 17 * ![](delay.png) 18 * 19 * If the delay argument is a Number, this operator time shifts the source 20 * Observable by that amount of time expressed in milliseconds. The relative 21 * time intervals between the values are preserved. 22 * 23 * If the delay argument is a Date, this operator time shifts the start of the 24 * Observable execution until the given date occurs. 25 * 26 * ## Examples 27 * Delay each click by one second 28 * ```ts 29 * import { fromEvent } from 'rxjs'; 30 * import { delay } from 'rxjs/operators'; 31 * 32 * const clicks = fromEvent(document, 'click'); 33 * const delayedClicks = clicks.pipe(delay(1000)); // each click emitted after 1 second 34 * delayedClicks.subscribe(x => console.log(x)); 35 * ``` 36 * 37 * Delay all clicks until a future date happens 38 * ```ts 39 * import { fromEvent } from 'rxjs'; 40 * import { delay } from 'rxjs/operators'; 41 * 42 * const clicks = fromEvent(document, 'click'); 43 * const date = new Date('March 15, 2050 12:00:00'); // in the future 44 * const delayedClicks = clicks.pipe(delay(date)); // click emitted only after that date 45 * delayedClicks.subscribe(x => console.log(x)); 46 * ``` 47 * 48 * @see {@link debounceTime} 49 * @see {@link delayWhen} 50 * 51 * @param {number|Date} delay The delay duration in milliseconds (a `number`) or 52 * a `Date` until which the emission of the source items is delayed. 53 * @param {SchedulerLike} [scheduler=async] The {@link SchedulerLike} to use for 54 * managing the timers that handle the time-shift for each item. 55 * @return {Observable} An Observable that delays the emissions of the source 56 * Observable by the specified timeout or Date. 57 * @method delay 58 * @owner Observable 59 */ 60export function delay<T>(delay: number|Date, 61 scheduler: SchedulerLike = async): MonoTypeOperatorFunction<T> { 62 const absoluteDelay = isDate(delay); 63 const delayFor = absoluteDelay ? (+delay - scheduler.now()) : Math.abs(<number>delay); 64 return (source: Observable<T>) => source.lift(new DelayOperator(delayFor, scheduler)); 65} 66 67class DelayOperator<T> implements Operator<T, T> { 68 constructor(private delay: number, 69 private scheduler: SchedulerLike) { 70 } 71 72 call(subscriber: Subscriber<T>, source: any): TeardownLogic { 73 return source.subscribe(new DelaySubscriber(subscriber, this.delay, this.scheduler)); 74 } 75} 76 77interface DelayState<T> { 78 source: DelaySubscriber<T>; 79 destination: PartialObserver<T>; 80 scheduler: SchedulerLike; 81} 82 83/** 84 * We need this JSDoc comment for affecting ESDoc. 85 * @ignore 86 * @extends {Ignored} 87 */ 88class DelaySubscriber<T> extends Subscriber<T> { 89 private queue: Array<DelayMessage<T>> = []; 90 private active: boolean = false; 91 private errored: boolean = false; 92 93 private static dispatch<T>(this: SchedulerAction<DelayState<T>>, state: DelayState<T>): void { 94 const source = state.source; 95 const queue = source.queue; 96 const scheduler = state.scheduler; 97 const destination = state.destination; 98 99 while (queue.length > 0 && (queue[0].time - scheduler.now()) <= 0) { 100 queue.shift().notification.observe(destination); 101 } 102 103 if (queue.length > 0) { 104 const delay = Math.max(0, queue[0].time - scheduler.now()); 105 this.schedule(state, delay); 106 } else { 107 this.unsubscribe(); 108 source.active = false; 109 } 110 } 111 112 constructor(destination: Subscriber<T>, 113 private delay: number, 114 private scheduler: SchedulerLike) { 115 super(destination); 116 } 117 118 private _schedule(scheduler: SchedulerLike): void { 119 this.active = true; 120 const destination = this.destination as Subscription; 121 destination.add(scheduler.schedule<DelayState<T>>(DelaySubscriber.dispatch, this.delay, { 122 source: this, destination: this.destination, scheduler: scheduler 123 })); 124 } 125 126 private scheduleNotification(notification: Notification<T>): void { 127 if (this.errored === true) { 128 return; 129 } 130 131 const scheduler = this.scheduler; 132 const message = new DelayMessage(scheduler.now() + this.delay, notification); 133 this.queue.push(message); 134 135 if (this.active === false) { 136 this._schedule(scheduler); 137 } 138 } 139 140 protected _next(value: T) { 141 this.scheduleNotification(Notification.createNext(value)); 142 } 143 144 protected _error(err: any) { 145 this.errored = true; 146 this.queue = []; 147 this.destination.error(err); 148 this.unsubscribe(); 149 } 150 151 protected _complete() { 152 this.scheduleNotification(Notification.createComplete()); 153 this.unsubscribe(); 154 } 155} 156 157class DelayMessage<T> { 158 constructor(public readonly time: number, 159 public readonly notification: Notification<T>) { 160 } 161} 162