// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.

;(function (factory) {

var objectTypes = {

'boolean': false,

'function': true,

'object': true,

'number': false,

'string': false,

'undefined': false

};

var root = (objectTypes[typeof window] && window) || this,

freeExports = objectTypes[typeof exports] && exports && !exports.nodeType && exports,

freeModule = objectTypes[typeof module] && module && !module.nodeType && module,

moduleExports = freeModule && freeModule.exports === freeExports && freeExports,

freeGlobal = objectTypes[typeof global] && global;

if (freeGlobal && (freeGlobal.global === freeGlobal || freeGlobal.window === freeGlobal)) {

root = freeGlobal;

}

// Because of build optimizers

if (typeof define === 'function' && define.amd) {

define(['rx'], function (Rx, exports) {

return factory(root, exports, Rx);

});

} else if (typeof module === 'object' && module && module.exports === freeExports) {

module.exports = factory(root, module.exports, require('./rx'));

} else {

root.Rx = factory(root, {}, root.Rx);

}

}.call(this, function (root, exp, Rx, undefined) {

// Refernces

var Observable = Rx.Observable,

observableProto = Observable.prototype,

AnonymousObservable = Rx.AnonymousObservable,

observableDefer = Observable.defer,

observableEmpty = Observable.empty,

observableNever = Observable.never,

observableThrow = Observable.throwException,

observableFromArray = Observable.fromArray,

timeoutScheduler = Rx.Scheduler.timeout,

SingleAssignmentDisposable = Rx.SingleAssignmentDisposable,

SerialDisposable = Rx.SerialDisposable,

CompositeDisposable = Rx.CompositeDisposable,

RefCountDisposable = Rx.RefCountDisposable,

Subject = Rx.Subject,

addRef = Rx.internals.addRef,

normalizeTime = Rx.Scheduler.normalize,

helpers = Rx.helpers,

isPromise = helpers.isPromise,

isScheduler = helpers.isScheduler,

observableFromPromise = Observable.fromPromise,

deprecate = helpers.deprecate;

function observableTimerDate(dueTime, scheduler) {

return new AnonymousObservable(function (observer) {

return scheduler.scheduleWithAbsolute(dueTime, function () {

observer.onNext(0);

observer.onCompleted();

});

});

}

function observableTimerDateAndPeriod(dueTime, period, scheduler) {

return new AnonymousObservable(function (observer) {

var d = dueTime, p = normalizeTime(period);

return scheduler.scheduleRecursiveWithAbsoluteAndState(0, d, function (count, self) {

if (p > 0) {

var now = scheduler.now();

d = d + p;

d <= now && (d = now + p);

}

observer.onNext(count);

self(count + 1, d);

});

});

}

function observableTimerTimeSpan(dueTime, scheduler) {

return new AnonymousObservable(function (observer) {

return scheduler.scheduleWithRelative(normalizeTime(dueTime), function () {

observer.onNext(0);

observer.onCompleted();

});

});

}

function observableTimerTimeSpanAndPeriod(dueTime, period, scheduler) {

return dueTime === period ?

new AnonymousObservable(function (observer) {

return scheduler.schedulePeriodicWithState(0, period, function (count) {

observer.onNext(count);

return count + 1;

});

}) :

observableDefer(function () {

return observableTimerDateAndPeriod(scheduler.now() + dueTime, period, scheduler);

});

}

/**

* Returns an observable sequence that produces a value after each period.

*

* @example

* 1 - res = Rx.Observable.interval(1000);

* 2 - res = Rx.Observable.interval(1000, Rx.Scheduler.timeout);

*

* @param {Number} period Period for producing the values in the resulting sequence (specified as an integer denoting milliseconds).

* @param {Scheduler} [scheduler] Scheduler to run the timer on. If not specified, Rx.Scheduler.timeout is used.

* @returns {Observable} An observable sequence that produces a value after each period.

*/

var observableinterval = Observable.interval = function (period, scheduler) {

return observableTimerTimeSpanAndPeriod(period, period, isScheduler(scheduler) ? scheduler : timeoutScheduler);

};

/**

* Returns an observable sequence that produces a value after dueTime has elapsed and then after each period.

* @param {Number} dueTime Absolute (specified as a Date object) or relative time (specified as an integer denoting milliseconds) at which to produce the first value.

* @param {Mixed} [periodOrScheduler] Period to produce subsequent values (specified as an integer denoting milliseconds), or the scheduler to run the timer on. If not specified, the resulting timer is not recurring.

* @param {Scheduler} [scheduler] Scheduler to run the timer on. If not specified, the timeout scheduler is used.

* @returns {Observable} An observable sequence that produces a value after due time has elapsed and then each period.

*/

var observableTimer = Observable.timer = function (dueTime, periodOrScheduler, scheduler) {

var period;

isScheduler(scheduler) || (scheduler = timeoutScheduler);

if (periodOrScheduler !== undefined && typeof periodOrScheduler === 'number') {

period = periodOrScheduler;

} else if (isScheduler(periodOrScheduler)) {

scheduler = periodOrScheduler;

}

if (dueTime instanceof Date && period === undefined) {

return observableTimerDate(dueTime.getTime(), scheduler);

}

if (dueTime instanceof Date && period !== undefined) {

period = periodOrScheduler;

return observableTimerDateAndPeriod(dueTime.getTime(), period, scheduler);

}

return period === undefined ?

observableTimerTimeSpan(dueTime, scheduler) :

observableTimerTimeSpanAndPeriod(dueTime, period, scheduler);

};

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);

};

/**

* Ignores values from an observable sequence which are followed by another value before dueTime.

* @param {Number} dueTime Duration of the debounce period for each value (specified as an integer denoting milliseconds).

* @param {Scheduler} [scheduler] Scheduler to run the debounce timers on. If not specified, the timeout scheduler is used.

* @returns {Observable} The debounced sequence.

*/

observableProto.debounce = observableProto.throttleWithTimeout = function (dueTime, scheduler) {

isScheduler(scheduler) || (scheduler = timeoutScheduler);

var source = this;

return new AnonymousObservable(function (observer) {

var cancelable = new SerialDisposable(), hasvalue = false, value, id = 0;

var subscription = source.subscribe(

function (x) {

hasvalue = true;

value = x;

id++;

var currentId = id,

d = new SingleAssignmentDisposable();

cancelable.setDisposable(d);

d.setDisposable(scheduler.scheduleWithRelative(dueTime, function () {

hasvalue && id === currentId && observer.onNext(value);

hasvalue = false;

}));

},

function (e) {

cancelable.dispose();

observer.onError(e);

hasvalue = false;

id++;

},

function () {

cancelable.dispose();

hasvalue && observer.onNext(value);

observer.onCompleted();

hasvalue = false;

id++;

});

return new CompositeDisposable(subscription, cancelable);

}, this);

};

/**

* @deprecated use #debounce or #throttleWithTimeout instead.

*/

observableProto.throttle = function(dueTime, scheduler) {

//deprecate('throttle', 'debounce or throttleWithTimeout');

return this.debounce(dueTime, scheduler);

};

/**

* Projects each element of an observable sequence into zero or more windows which are produced based on timing information.

* @param {Number} timeSpan Length of each window (specified as an integer denoting milliseconds).

* @param {Mixed} [timeShiftOrScheduler] Interval between creation of consecutive windows (specified as an integer denoting milliseconds), or an optional scheduler parameter. If not specified, the time shift corresponds to the timeSpan parameter, resulting in non-overlapping adjacent windows.

* @param {Scheduler} [scheduler] Scheduler to run windowing timers on. If not specified, the timeout scheduler is used.

* @returns {Observable} An observable sequence of windows.

*/

observableProto.windowWithTime = function (timeSpan, timeShiftOrScheduler, scheduler) {

var source = this, timeShift;

timeShiftOrScheduler == null && (timeShift = timeSpan);

isScheduler(scheduler) || (scheduler = timeoutScheduler);

if (typeof timeShiftOrScheduler === 'number') {

timeShift = timeShiftOrScheduler;

} else if (isScheduler(timeShiftOrScheduler)) {

timeShift = timeSpan;

scheduler = timeShiftOrScheduler;

}

return new AnonymousObservable(function (observer) {

var groupDisposable,

nextShift = timeShift,

nextSpan = timeSpan,

q = [],

refCountDisposable,

timerD = new SerialDisposable(),

totalTime = 0;

groupDisposable = new CompositeDisposable(timerD),

refCountDisposable = new RefCountDisposable(groupDisposable);

function createTimer () {

var m = new SingleAssignmentDisposable(),

isSpan = false,

isShift = false;

timerD.setDisposable(m);

if (nextSpan === nextShift) {

isSpan = true;

isShift = true;

} else if (nextSpan < nextShift) {

isSpan = true;

} else {

isShift = true;

}

var newTotalTime = isSpan ? nextSpan : nextShift,

ts = newTotalTime - totalTime;

totalTime = newTotalTime;

if (isSpan) {

nextSpan += timeShift;

}

if (isShift) {

nextShift += timeShift;

}

m.setDisposable(scheduler.scheduleWithRelative(ts, function () {

if (isShift) {

var s = new Subject();

q.push(s);

observer.onNext(addRef(s, refCountDisposable));

}

isSpan && q.shift().onCompleted();

createTimer();

}));

};

q.push(new Subject());

observer.onNext(addRef(q[0], refCountDisposable));

createTimer();

groupDisposable.add(source.subscribe(

function (x) {

for (var i = 0, len = q.length; i < len; i++) { q[i].onNext(x); }

},

function (e) {

for (var i = 0, len = q.length; i < len; i++) { q[i].onError(e); }

observer.onError(e);

},

function () {

for (var i = 0, len = q.length; i < len; i++) { q[i].onCompleted(); }

observer.onCompleted();

}

));

return refCountDisposable;

}, source);

};

/**

* Projects each element of an observable sequence into a window that is completed when either it's full or a given amount of time has elapsed.

* @param {Number} timeSpan Maximum time length of a window.

* @param {Number} count Maximum element count of a window.

* @param {Scheduler} [scheduler] Scheduler to run windowing timers on. If not specified, the timeout scheduler is used.

* @returns {Observable} An observable sequence of windows.

*/

observableProto.windowWithTimeOrCount = function (timeSpan, count, scheduler) {

var source = this;

isScheduler(scheduler) || (scheduler = timeoutScheduler);

return new AnonymousObservable(function (observer) {

var timerD = new SerialDisposable(),

groupDisposable = new CompositeDisposable(timerD),

refCountDisposable = new RefCountDisposable(groupDisposable),

n = 0,

windowId = 0,

s = new Subject();

function createTimer(id) {

var m = new SingleAssignmentDisposable();

timerD.setDisposable(m);

m.setDisposable(scheduler.scheduleWithRelative(timeSpan, function () {

if (id !== windowId) { return; }

n = 0;

var newId = ++windowId;

s.onCompleted();

s = new Subject();

observer.onNext(addRef(s, refCountDisposable));

createTimer(newId);

}));

}

observer.onNext(addRef(s, refCountDisposable));

createTimer(0);

groupDisposable.add(source.subscribe(

function (x) {

var newId = 0, newWindow = false;

s.onNext(x);

if (++n === count) {

newWindow = true;

n = 0;

newId = ++windowId;

s.onCompleted();

s = new Subject();

observer.onNext(addRef(s, refCountDisposable));

}

newWindow && createTimer(newId);

},

function (e) {

s.onError(e);

observer.onError(e);

}, function () {

s.onCompleted();

observer.onCompleted();

}

));

return refCountDisposable;

}, source);

};

/**

* Projects each element of an observable sequence into zero or more buffers which are produced based on timing information.

*

* @example

* 1 - res = xs.bufferWithTime(1000, scheduler); // non-overlapping segments of 1 second

* 2 - res = xs.bufferWithTime(1000, 500, scheduler; // segments of 1 second with time shift 0.5 seconds

*

* @param {Number} timeSpan Length of each buffer (specified as an integer denoting milliseconds).

* @param {Mixed} [timeShiftOrScheduler] Interval between creation of consecutive buffers (specified as an integer denoting milliseconds), or an optional scheduler parameter. If not specified, the time shift corresponds to the timeSpan parameter, resulting in non-overlapping adjacent buffers.

* @param {Scheduler} [scheduler] Scheduler to run buffer timers on. If not specified, the timeout scheduler is used.

* @returns {Observable} An observable sequence of buffers.

*/

observableProto.bufferWithTime = function (timeSpan, timeShiftOrScheduler, scheduler) {

return this.windowWithTime.apply(this, arguments).selectMany(function (x) { return x.toArray(); });

};

/**

* Projects each element of an observable sequence into a buffer that is completed when either it's full or a given amount of time has elapsed.

*

* @example

* 1 - res = source.bufferWithTimeOrCount(5000, 50); // 5s or 50 items in an array

* 2 - res = source.bufferWithTimeOrCount(5000, 50, scheduler); // 5s or 50 items in an array

*

* @param {Number} timeSpan Maximum time length of a buffer.

* @param {Number} count Maximum element count of a buffer.

* @param {Scheduler} [scheduler] Scheduler to run bufferin timers on. If not specified, the timeout scheduler is used.

* @returns {Observable} An observable sequence of buffers.

*/

observableProto.bufferWithTimeOrCount = function (timeSpan, count, scheduler) {

return this.windowWithTimeOrCount(timeSpan, count, scheduler).selectMany(function (x) {

return x.toArray();

});

};

/**

* Records the time interval between consecutive values in an observable sequence.

*

* @example

* 1 - res = source.timeInterval();

* 2 - res = source.timeInterval(Rx.Scheduler.timeout);

*

* @param [scheduler] Scheduler used to compute time intervals. If not specified, the timeout scheduler is used.

* @returns {Observable} An observable sequence with time interval information on values.

*/

observableProto.timeInterval = function (scheduler) {

var source = this;

isScheduler(scheduler) || (scheduler = timeoutScheduler);

return observableDefer(function () {

var last = scheduler.now();

return source.map(function (x) {

var now = scheduler.now(), span = now - last;

last = now;

return { value: x, interval: span };

});

});

};

/**

* Records the timestamp for each value in an observable sequence.

*

* @example

* 1 - res = source.timestamp(); // produces { value: x, timestamp: ts }

* 2 - res = source.timestamp(Rx.Scheduler.timeout);

*

* @param {Scheduler} [scheduler] Scheduler used to compute timestamps. If not specified, the timeout scheduler is used.

* @returns {Observable} An observable sequence with timestamp information on values.

*/

observableProto.timestamp = function (scheduler) {

isScheduler(scheduler) || (scheduler = timeoutScheduler);

return this.map(function (x) {

return { value: x, timestamp: scheduler.now() };

});

};

function sampleObservable(source, sampler) {

return new AnonymousObservable(function (observer) {

var atEnd, value, hasValue;

function sampleSubscribe() {

if (hasValue) {

hasValue = false;

observer.onNext(value);

}

atEnd && observer.onCompleted();

}

return new CompositeDisposable(

source.subscribe(function (newValue) {

hasValue = true;

value = newValue;

}, observer.onError.bind(observer), function () {

atEnd = true;

}),

sampler.subscribe(sampleSubscribe, observer.onError.bind(observer), sampleSubscribe)

);

}, source);

}

/**

* Samples the observable sequence at each interval.

*

* @example

* 1 - res = source.sample(sampleObservable); // Sampler tick sequence

* 2 - res = source.sample(5000); // 5 seconds

* 2 - res = source.sample(5000, Rx.Scheduler.timeout); // 5 seconds

*

* @param {Mixed} intervalOrSampler Interval at which to sample (specified as an integer denoting milliseconds) or Sampler Observable.

* @param {Scheduler} [scheduler] Scheduler to run the sampling timer on. If not specified, the timeout scheduler is used.

* @returns {Observable} Sampled observable sequence.

*/

observableProto.sample = observableProto.throttleLatest = function (intervalOrSampler, scheduler) {

isScheduler(scheduler) || (scheduler = timeoutScheduler);

return typeof intervalOrSampler === 'number' ?

sampleObservable(this, observableinterval(intervalOrSampler, scheduler)) :

sampleObservable(this, intervalOrSampler);

};

/**

* Returns the source observable sequence or the other observable sequence if dueTime elapses.

* @param {Number} dueTime Absolute (specified as a Date object) or relative time (specified as an integer denoting milliseconds) when a timeout occurs.

* @param {Observable} [other] Sequence to return in case of a timeout. If not specified, a timeout error throwing sequence will be used.

* @param {Scheduler} [scheduler] Scheduler to run the timeout timers on. If not specified, the timeout scheduler is used.

* @returns {Observable} The source sequence switching to the other sequence in case of a timeout.

*/

observableProto.timeout = function (dueTime, other, scheduler) {

(other == null || typeof other === 'string') && (other = observableThrow(new Error(other || 'Timeout')));

isScheduler(scheduler) || (scheduler = timeoutScheduler);

var source = this, schedulerMethod = dueTime instanceof Date ?

'scheduleWithAbsolute' :

'scheduleWithRelative';

return new AnonymousObservable(function (observer) {

var id = 0,

original = new SingleAssignmentDisposable(),

subscription = new SerialDisposable(),

switched = false,

timer = new SerialDisposable();

subscription.setDisposable(original);

function createTimer() {

var myId = id;

timer.setDisposable(scheduler[schedulerMethod](dueTime, function () {

if (id === myId) {

isPromise(other) && (other = observableFromPromise(other));

subscription.setDisposable(other.subscribe(observer));

}

}));

}

createTimer();

original.setDisposable(source.subscribe(function (x) {

if (!switched) {

id++;

observer.onNext(x);

createTimer();

}

}, function (e) {

if (!switched) {

id++;

observer.onError(e);

}

}, function () {

if (!switched) {

id++;

observer.onCompleted();

}

}));

return new CompositeDisposable(subscription, timer);

}, source);

};

/**

* Generates an observable sequence by iterating a state from an initial state until the condition fails.

*

* @example

* res = source.generateWithAbsoluteTime(0,

* function (x) { return return true; },

* function (x) { return x + 1; },

* function (x) { return x; },

* function (x) { return new Date(); }

* });

*

* @param {Mixed} initialState Initial state.

* @param {Function} condition Condition to terminate generation (upon returning false).

* @param {Function} iterate Iteration step function.

* @param {Function} resultSelector Selector function for results produced in the sequence.

* @param {Function} timeSelector Time selector function to control the speed of values being produced each iteration, returning Date values.

* @param {Scheduler} [scheduler] Scheduler on which to run the generator loop. If not specified, the timeout scheduler is used.

* @returns {Observable} The generated sequence.

*/

Observable.generateWithAbsoluteTime = function (initialState, condition, iterate, resultSelector, timeSelector, scheduler) {

isScheduler(scheduler) || (scheduler = timeoutScheduler);

return new AnonymousObservable(function (observer) {

var first = true,

hasResult = false,

result,

state = initialState,

time;

return scheduler.scheduleRecursiveWithAbsolute(scheduler.now(), function (self) {

hasResult && observer.onNext(result);

try {

if (first) {

first = false;

} else {

state = iterate(state);

}

hasResult = condition(state);

if (hasResult) {

result = resultSelector(state);

time = timeSelector(state);

}

} catch (e) {

observer.onError(e);

return;

}

if (hasResult) {

self(time);

} else {

observer.onCompleted();

}

});

});

};

/**

* Generates an observable sequence by iterating a state from an initial state until the condition fails.

*

* @example

* res = source.generateWithRelativeTime(0,

* function (x) { return return true; },

* function (x) { return x + 1; },

* function (x) { return x; },

* function (x) { return 500; }

* );

*

* @param {Mixed} initialState Initial state.

* @param {Function} condition Condition to terminate generation (upon returning false).

* @param {Function} iterate Iteration step function.

* @param {Function} resultSelector Selector function for results produced in the sequence.

* @param {Function} timeSelector Time selector function to control the speed of values being produced each iteration, returning integer values denoting milliseconds.

* @param {Scheduler} [scheduler] Scheduler on which to run the generator loop. If not specified, the timeout scheduler is used.

* @returns {Observable} The generated sequence.

*/

Observable.generateWithRelativeTime = function (initialState, condition, iterate, resultSelector, timeSelector, scheduler) {

isScheduler(scheduler) || (scheduler = timeoutScheduler);

return new AnonymousObservable(function (observer) {

var first = true,

hasResult = false,

result,

state = initialState,

time;

return scheduler.scheduleRecursiveWithRelative(0, function (self) {

hasResult && observer.onNext(result);

try {

if (first) {

first = false;

} else {

state = iterate(state);

}

hasResult = condition(state);

if (hasResult) {

result = resultSelector(state);

time = timeSelector(state);

}

} catch (e) {

observer.onError(e);

return;

}

if (hasResult) {

self(time);

} else {

observer.onCompleted();

}

});

});

};

/**

* Time shifts the observable sequence by delaying the subscription.

*

* @example

* 1 - res = source.delaySubscription(5000); // 5s

* 2 - res = source.delaySubscription(5000, Rx.Scheduler.timeout); // 5 seconds

*

* @param {Number} dueTime Absolute or relative time to perform the subscription at.

* @param {Scheduler} [scheduler] Scheduler to run the subscription delay timer on. If not specified, the timeout scheduler is used.

* @returns {Observable} Time-shifted sequence.

*/

observableProto.delaySubscription = function (dueTime, scheduler) {

return this.delayWithSelector(observableTimer(dueTime, isScheduler(scheduler) ? scheduler : timeoutScheduler), observableEmpty);

};

/**

* Time shifts the observable sequence based on a subscription delay and a delay selector function for each element.

*

* @example

* 1 - res = source.delayWithSelector(function (x) { return Rx.Scheduler.timer(5000); }); // with selector only

* 1 - res = source.delayWithSelector(Rx.Observable.timer(2000), function (x) { return Rx.Observable.timer(x); }); // with delay and selector

*

* @param {Observable} [subscriptionDelay] Sequence indicating the delay for the subscription to the source.

* @param {Function} delayDurationSelector Selector function to retrieve a sequence indicating the delay for each given element.

* @returns {Observable} Time-shifted sequence.

*/

observableProto.delayWithSelector = function (subscriptionDelay, delayDurationSelector) {

var source = this, subDelay, selector;

if (typeof subscriptionDelay === 'function') {

selector = subscriptionDelay;

} else {

subDelay = subscriptionDelay;

selector = delayDurationSelector;

}

return new AnonymousObservable(function (observer) {

var delays = new CompositeDisposable(), atEnd = false, done = function () {

if (atEnd && delays.length === 0) { observer.onCompleted(); }

}, subscription = new SerialDisposable(), start = function () {

subscription.setDisposable(source.subscribe(function (x) {

var delay;

try {

delay = selector(x);

} catch (error) {

observer.onError(error);

return;

}

var d = new SingleAssignmentDisposable();

delays.add(d);

d.setDisposable(delay.subscribe(function () {

observer.onNext(x);

delays.remove(d);

done();

}, observer.onError.bind(observer), function () {

observer.onNext(x);

delays.remove(d);

done();

}));

}, observer.onError.bind(observer), function () {

atEnd = true;

subscription.dispose();

done();

}));

};

if (!subDelay) {

start();

} else {

subscription.setDisposable(subDelay.subscribe(start, observer.onError.bind(observer), start));

}

return new CompositeDisposable(subscription, delays);

}, this);

};

/**

* Returns the source observable sequence, switching to the other observable sequence if a timeout is signaled.

* @param {Observable} [firstTimeout] Observable sequence that represents the timeout for the first element. If not provided, this defaults to Observable.never().

* @param {Function} timeoutDurationSelector Selector to retrieve an observable sequence that represents the timeout between the current element and the next element.

* @param {Observable} [other] Sequence to return in case of a timeout. If not provided, this is set to Observable.throwException().

* @returns {Observable} The source sequence switching to the other sequence in case of a timeout.

*/

observableProto.timeoutWithSelector = function (firstTimeout, timeoutdurationSelector, other) {

if (arguments.length === 1) {

timeoutdurationSelector = firstTimeout;

firstTimeout = observableNever();

}

other || (other = observableThrow(new Error('Timeout')));

var source = this;

return new AnonymousObservable(function (observer) {

var subscription = new SerialDisposable(), timer = new SerialDisposable(), original = new SingleAssignmentDisposable();

subscription.setDisposable(original);

var id = 0, switched = false;

function setTimer(timeout) {

var myId = id;

function timerWins () {

return id === myId;

}

var d = new SingleAssignmentDisposable();

timer.setDisposable(d);

d.setDisposable(timeout.subscribe(function () {

timerWins() && subscription.setDisposable(other.subscribe(observer));

d.dispose();

}, function (e) {

timerWins() && observer.onError(e);

}, function () {

timerWins() && subscription.setDisposable(other.subscribe(observer));

}));

};

setTimer(firstTimeout);

function observerWins() {

var res = !switched;

if (res) { id++; }

return res;

}

original.setDisposable(source.subscribe(function (x) {

if (observerWins()) {

observer.onNext(x);

var timeout;

try {

timeout = timeoutdurationSelector(x);

} catch (e) {

observer.onError(e);

return;

}

setTimer(isPromise(timeout) ? observableFromPromise(timeout) : timeout);

}

}, function (e) {

observerWins() && observer.onError(e);

}, function () {

observerWins() && observer.onCompleted();

}));

return new CompositeDisposable(subscription, timer);

}, source);

};

/**

* Ignores values from an observable sequence which are followed by another value within a computed throttle duration.

* @param {Function} durationSelector Selector function to retrieve a sequence indicating the throttle duration for each given element.

* @returns {Observable} The debounced sequence.

*/

observableProto.debounceWithSelector = function (durationSelector) {

var source = this;

return new AnonymousObservable(function (observer) {

var value, hasValue = false, cancelable = new SerialDisposable(), id = 0;

var subscription = source.subscribe(function (x) {

var throttle;

try {

throttle = durationSelector(x);

} catch (e) {

observer.onError(e);

return;

}

isPromise(throttle) && (throttle = observableFromPromise(throttle));

hasValue = true;

value = x;

id++;

var currentid = id, d = new SingleAssignmentDisposable();

cancelable.setDisposable(d);

d.setDisposable(throttle.subscribe(function () {

hasValue && id === currentid && observer.onNext(value);

hasValue = false;

d.dispose();

}, observer.onError.bind(observer), function () {

hasValue && id === currentid && observer.onNext(value);

hasValue = false;

d.dispose();

}));

}, function (e) {

cancelable.dispose();

observer.onError(e);

hasValue = false;

id++;

}, function () {

cancelable.dispose();

hasValue && observer.onNext(value);

observer.onCompleted();

hasValue = false;

id++;

});

return new CompositeDisposable(subscription, cancelable);

}, source);

};

observableProto.throttleWithSelector = function () {

//deprecate('throttleWithSelector', 'debounceWithSelector');

return this.debounceWithSelector.apply(this, arguments);

};

/**

* Skips elements for the specified duration from the end of the observable source sequence, using the specified scheduler to run timers.

*

* 1 - res = source.skipLastWithTime(5000);

* 2 - res = source.skipLastWithTime(5000, scheduler);

*

* @description

* This operator accumulates a queue with a length enough to store elements received during the initial duration window.

* As more elements are received, elements older than the specified duration are taken from the queue and produced on the

* result sequence. This causes elements to be delayed with duration.

* @param {Number} duration Duration for skipping elements from the end of the sequence.

* @param {Scheduler} [scheduler] Scheduler to run the timer on. If not specified, defaults to Rx.Scheduler.timeout

* @returns {Observable} An observable sequence with the elements skipped during the specified duration from the end of the source sequence.

*/

observableProto.skipLastWithTime = function (duration, scheduler) {

isScheduler(scheduler) || (scheduler = timeoutScheduler);

var source = this;

return new AnonymousObservable(function (o) {

var q = [];

return source.subscribe(function (x) {

var now = scheduler.now();

q.push({ interval: now, value: x });

while (q.length > 0 && now - q[0].interval >= duration) {

o.onNext(q.shift().value);

}

}, function (e) { o.onError(e); }, function () {

var now = scheduler.now();

while (q.length > 0 && now - q[0].interval >= duration) {

o.onNext(q.shift().value);

}

o.onCompleted();

});

}, source);

};

/**

* Returns elements within the specified duration from the end of the observable source sequence, using the specified schedulers to run timers and to drain the collected elements.

* @description

* This operator accumulates a queue with a length enough to store elements received during the initial duration window.

* As more elements are received, elements older than the specified duration are taken from the queue and produced on the

* result sequence. This causes elements to be delayed with duration.

* @param {Number} duration Duration for taking elements from the end of the sequence.

* @param {Scheduler} [scheduler] Scheduler to run the timer on. If not specified, defaults to Rx.Scheduler.timeout.

* @returns {Observable} An observable sequence with the elements taken during the specified duration from the end of the source sequence.

*/

observableProto.takeLastWithTime = function (duration, scheduler) {

var source = this;

isScheduler(scheduler) || (scheduler = timeoutScheduler);

return new AnonymousObservable(function (o) {

var q = [];

return source.subscribe(function (x) {

var now = scheduler.now();

q.push({ interval: now, value: x });

while (q.length > 0 && now - q[0].interval >= duration) {

q.shift();

}

}, function (e) { o.onError(e); }, function () {

var now = scheduler.now();

while (q.length > 0) {

var next = q.shift();

if (now - next.interval <= duration) { o.onNext(next.value); }

}

o.onCompleted();

});

}, source);

};

/**

* Returns an array with the elements within the specified duration from the end of the observable source sequence, using the specified scheduler to run timers.

* @description

* This operator accumulates a queue with a length enough to store elements received during the initial duration window.

* As more elements are received, elements older than the specified duration are taken from the queue and produced on the

* result sequence. This causes elements to be delayed with duration.

* @param {Number} duration Duration for taking elements from the end of the sequence.

* @param {Scheduler} scheduler Scheduler to run the timer on. If not specified, defaults to Rx.Scheduler.timeout.

* @returns {Observable} An observable sequence containing a single array with the elements taken during the specified duration from the end of the source sequence.

*/

observableProto.takeLastBufferWithTime = function (duration, scheduler) {

var source = this;

isScheduler(scheduler) || (scheduler = timeoutScheduler);

return new AnonymousObservable(function (o) {

var q = [];

return source.subscribe(function (x) {

var now = scheduler.now();

q.push({ interval: now, value: x });

while (q.length > 0 && now - q[0].interval >= duration) {

q.shift();

}

}, function (e) { o.onError(e); }, function () {

var now = scheduler.now(), res = [];

while (q.length > 0) {

var next = q.shift();

now - next.interval <= duration && res.push(next.value);

}