function observableDelayTimeSpan(source, dueTime, scheduler) { return new AnonymousObservable(function (observer) { var active = false, cancelable = new SerialDisposable(), exception = null, q = [], running = false, subscription; subscription = source.materialize().timestamp(scheduler).subscribe(function (notification) { var d, shouldRun; if (notification.value.kind === 'E') { q = []; q.push(notification); exception = notification.value.exception; shouldRun = !running; } else { q.push({ value: notification.value, timestamp: notification.timestamp + dueTime }); shouldRun = !active; active = true; } if (shouldRun) { if (exception !== null) { observer.onError(exception); } else { d = new SingleAssignmentDisposable(); cancelable.setDisposable(d); d.setDisposable(scheduler.scheduleRecursiveWithRelative(dueTime, function (self) { var e, recurseDueTime, result, shouldRecurse; if (exception !== null) { return; } running = true; do { result = null; if (q.length > 0 && q[0].timestamp - scheduler.now() <= 0) { result = q.shift().value; } if (result !== null) { result.accept(observer); } } while (result !== null); shouldRecurse = false; recurseDueTime = 0; if (q.length > 0) { shouldRecurse = true; recurseDueTime = Math.max(0, q[0].timestamp - scheduler.now()); } else { active = false; } e = exception; running = false; if (e !== null) { observer.onError(e); } else if (shouldRecurse) { self(recurseDueTime); } })); } } }); return new CompositeDisposable(subscription, cancelable); }, source); } function observableDelayDate(source, dueTime, scheduler) { return observableDefer(function () { return observableDelayTimeSpan(source, dueTime - scheduler.now(), scheduler); }); } /** * Time shifts the observable sequence by dueTime. The relative time intervals between the values are preserved. * * @example * 1 - res = Rx.Observable.delay(new Date()); * 2 - res = Rx.Observable.delay(new Date(), Rx.Scheduler.timeout); * * 3 - res = Rx.Observable.delay(5000); * 4 - res = Rx.Observable.delay(5000, 1000, Rx.Scheduler.timeout); * @memberOf Observable# * @param {Number} dueTime Absolute (specified as a Date object) or relative time (specified as an integer denoting milliseconds) by which to shift the observable sequence. * @param {Scheduler} [scheduler] Scheduler to run the delay timers on. If not specified, the timeout scheduler is used. * @returns {Observable} Time-shifted sequence. */ observableProto.delay = function (dueTime, scheduler) { isScheduler(scheduler) || (scheduler = timeoutScheduler); return dueTime instanceof Date ? observableDelayDate(this, dueTime.getTime(), scheduler) : observableDelayTimeSpan(this, dueTime, scheduler); };