You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
509 lines
19 KiB
509 lines
19 KiB
// Copyright (c) Microsoft, All rights reserved. See License.txt in the project root for license information. |
|
|
|
;(function (factory) { |
|
var objectTypes = { |
|
'function': true, |
|
'object': true |
|
}; |
|
|
|
function checkGlobal(value) { |
|
return (value && value.Object === Object) ? value : null; |
|
} |
|
|
|
var freeExports = (objectTypes[typeof exports] && exports && !exports.nodeType) ? exports : null; |
|
var freeModule = (objectTypes[typeof module] && module && !module.nodeType) ? module : null; |
|
var freeGlobal = checkGlobal(freeExports && freeModule && typeof global === 'object' && global); |
|
var freeSelf = checkGlobal(objectTypes[typeof self] && self); |
|
var freeWindow = checkGlobal(objectTypes[typeof window] && window); |
|
var moduleExports = (freeModule && freeModule.exports === freeExports) ? freeExports : null; |
|
var thisGlobal = checkGlobal(objectTypes[typeof this] && this); |
|
var root = freeGlobal || ((freeWindow !== (thisGlobal && thisGlobal.window)) && freeWindow) || freeSelf || thisGlobal || Function('return this')(); |
|
|
|
// Because of build optimizers |
|
if (typeof define === 'function' && define.amd) { |
|
define(['./rx.virtualtime', 'exports'], function (Rx, exports) { |
|
root.Rx = factory(root, exports, Rx); |
|
return root.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) { |
|
|
|
// Defaults |
|
var Observer = Rx.Observer, |
|
Observable = Rx.Observable, |
|
Notification = Rx.Notification, |
|
VirtualTimeScheduler = Rx.VirtualTimeScheduler, |
|
Disposable = Rx.Disposable, |
|
disposableEmpty = Disposable.empty, |
|
disposableCreate = Disposable.create, |
|
CompositeDisposable = Rx.CompositeDisposable, |
|
inherits = Rx.internals.inherits, |
|
defaultComparer = Rx.internals.isEqual; |
|
|
|
function OnNextPredicate(predicate) { |
|
this.predicate = predicate; |
|
} |
|
|
|
OnNextPredicate.prototype.equals = function (other) { |
|
if (other === this) { return true; } |
|
if (other == null) { return false; } |
|
if (other.kind !== 'N') { return false; } |
|
return this.predicate(other.value); |
|
}; |
|
|
|
function OnErrorPredicate(predicate) { |
|
this.predicate = predicate; |
|
} |
|
|
|
OnErrorPredicate.prototype.equals = function (other) { |
|
if (other === this) { return true; } |
|
if (other == null) { return false; } |
|
if (other.kind !== 'E') { return false; } |
|
return this.predicate(other.error); |
|
}; |
|
|
|
var ReactiveTest = Rx.ReactiveTest = { |
|
/** Default virtual time used for creation of observable sequences in unit tests. */ |
|
created: 100, |
|
/** Default virtual time used to subscribe to observable sequences in unit tests. */ |
|
subscribed: 200, |
|
/** Default virtual time used to dispose subscriptions in unit tests. */ |
|
disposed: 1000, |
|
|
|
/** |
|
* Factory method for an OnNext notification record at a given time with a given value or a predicate function. |
|
* |
|
* 1 - ReactiveTest.onNext(200, 42); |
|
* 2 - ReactiveTest.onNext(200, function (x) { return x.length == 2; }); |
|
* |
|
* @param ticks Recorded virtual time the OnNext notification occurs. |
|
* @param value Recorded value stored in the OnNext notification or a predicate. |
|
* @return Recorded OnNext notification. |
|
*/ |
|
onNext: function (ticks, value) { |
|
return typeof value === 'function' ? |
|
new Recorded(ticks, new OnNextPredicate(value)) : |
|
new Recorded(ticks, Notification.createOnNext(value)); |
|
}, |
|
/** |
|
* Factory method for an OnError notification record at a given time with a given error. |
|
* |
|
* 1 - ReactiveTest.onNext(200, new Error('error')); |
|
* 2 - ReactiveTest.onNext(200, function (e) { return e.message === 'error'; }); |
|
* |
|
* @param ticks Recorded virtual time the OnError notification occurs. |
|
* @param exception Recorded exception stored in the OnError notification. |
|
* @return Recorded OnError notification. |
|
*/ |
|
onError: function (ticks, error) { |
|
return typeof error === 'function' ? |
|
new Recorded(ticks, new OnErrorPredicate(error)) : |
|
new Recorded(ticks, Notification.createOnError(error)); |
|
}, |
|
/** |
|
* Factory method for an OnCompleted notification record at a given time. |
|
* |
|
* @param ticks Recorded virtual time the OnCompleted notification occurs. |
|
* @return Recorded OnCompleted notification. |
|
*/ |
|
onCompleted: function (ticks) { |
|
return new Recorded(ticks, Notification.createOnCompleted()); |
|
}, |
|
/** |
|
* Factory method for a subscription record based on a given subscription and disposal time. |
|
* |
|
* @param start Virtual time indicating when the subscription was created. |
|
* @param end Virtual time indicating when the subscription was disposed. |
|
* @return Subscription object. |
|
*/ |
|
subscribe: function (start, end) { |
|
return new Subscription(start, end); |
|
} |
|
}; |
|
|
|
/** |
|
* Creates a new object recording the production of the specified value at the given virtual time. |
|
* |
|
* @constructor |
|
* @param {Number} time Virtual time the value was produced on. |
|
* @param {Mixed} value Value that was produced. |
|
* @param {Function} comparer An optional comparer. |
|
*/ |
|
var Recorded = Rx.Recorded = function (time, value, comparer) { |
|
this.time = time; |
|
this.value = value; |
|
this.comparer = comparer || defaultComparer; |
|
}; |
|
|
|
/** |
|
* Checks whether the given recorded object is equal to the current instance. |
|
* |
|
* @param {Recorded} other Recorded object to check for equality. |
|
* @returns {Boolean} true if both objects are equal; false otherwise. |
|
*/ |
|
Recorded.prototype.equals = function (other) { |
|
return this.time === other.time && this.comparer(this.value, other.value); |
|
}; |
|
|
|
/** |
|
* Returns a string representation of the current Recorded value. |
|
* |
|
* @returns {String} String representation of the current Recorded value. |
|
*/ |
|
Recorded.prototype.toString = function () { |
|
return this.value.toString() + '@' + this.time; |
|
}; |
|
|
|
/** |
|
* Creates a new subscription object with the given virtual subscription and unsubscription time. |
|
* |
|
* @constructor |
|
* @param {Number} subscribe Virtual time at which the subscription occurred. |
|
* @param {Number} unsubscribe Virtual time at which the unsubscription occurred. |
|
*/ |
|
var Subscription = Rx.Subscription = function (start, end) { |
|
this.subscribe = start; |
|
this.unsubscribe = end || Number.MAX_VALUE; |
|
}; |
|
|
|
/** |
|
* Checks whether the given subscription is equal to the current instance. |
|
* @param other Subscription object to check for equality. |
|
* @returns {Boolean} true if both objects are equal; false otherwise. |
|
*/ |
|
Subscription.prototype.equals = function (other) { |
|
return this.subscribe === other.subscribe && this.unsubscribe === other.unsubscribe; |
|
}; |
|
|
|
/** |
|
* Returns a string representation of the current Subscription value. |
|
* @returns {String} String representation of the current Subscription value. |
|
*/ |
|
Subscription.prototype.toString = function () { |
|
return '(' + this.subscribe + ', ' + (this.unsubscribe === Number.MAX_VALUE ? 'Infinite' : this.unsubscribe) + ')'; |
|
}; |
|
|
|
var MockDisposable = Rx.MockDisposable = function (scheduler) { |
|
this.scheduler = scheduler; |
|
this.disposes = []; |
|
this.disposes.push(this.scheduler.clock); |
|
}; |
|
|
|
MockDisposable.prototype.dispose = function () { |
|
this.disposes.push(this.scheduler.clock); |
|
}; |
|
|
|
var MockObserver = (function (__super__) { |
|
inherits(MockObserver, __super__); |
|
|
|
function MockObserver(scheduler) { |
|
__super__.call(this); |
|
this.scheduler = scheduler; |
|
this.messages = []; |
|
} |
|
|
|
var MockObserverPrototype = MockObserver.prototype; |
|
|
|
MockObserverPrototype.onNext = function (value) { |
|
this.messages.push(new Recorded(this.scheduler.clock, Notification.createOnNext(value))); |
|
}; |
|
|
|
MockObserverPrototype.onError = function (e) { |
|
this.messages.push(new Recorded(this.scheduler.clock, Notification.createOnError(e))); |
|
}; |
|
|
|
MockObserverPrototype.onCompleted = function () { |
|
this.messages.push(new Recorded(this.scheduler.clock, Notification.createOnCompleted())); |
|
}; |
|
|
|
return MockObserver; |
|
})(Observer); |
|
|
|
function MockPromise(scheduler, messages) { |
|
var self = this; |
|
this.scheduler = scheduler; |
|
this.messages = messages; |
|
this.subscriptions = []; |
|
this.observers = []; |
|
for (var i = 0, len = this.messages.length; i < len; i++) { |
|
var message = this.messages[i], |
|
notification = message.value; |
|
(function (innerNotification) { |
|
scheduler.scheduleAbsolute(null, message.time, function () { |
|
var obs = self.observers.slice(0); |
|
|
|
for (var j = 0, jLen = obs.length; j < jLen; j++) { |
|
innerNotification.accept(obs[j]); |
|
} |
|
return disposableEmpty; |
|
}); |
|
})(notification); |
|
} |
|
} |
|
|
|
MockPromise.prototype.then = function (onResolved, onRejected) { |
|
var self = this; |
|
|
|
this.subscriptions.push(new Subscription(this.scheduler.clock)); |
|
var index = this.subscriptions.length - 1; |
|
|
|
var newPromise; |
|
|
|
var observer = Rx.Observer.create( |
|
function (x) { |
|
var retValue = onResolved(x); |
|
if (retValue && typeof retValue.then === 'function') { |
|
newPromise = retValue; |
|
} else { |
|
var ticks = self.scheduler.clock; |
|
newPromise = new MockPromise(self.scheduler, [Rx.ReactiveTest.onNext(ticks, undefined), Rx.ReactiveTest.onCompleted(ticks)]); |
|
} |
|
var idx = self.observers.indexOf(observer); |
|
self.observers.splice(idx, 1); |
|
self.subscriptions[index] = new Subscription(self.subscriptions[index].subscribe, self.scheduler.clock); |
|
}, |
|
function (err) { |
|
onRejected(err); |
|
var idx = self.observers.indexOf(observer); |
|
self.observers.splice(idx, 1); |
|
self.subscriptions[index] = new Subscription(self.subscriptions[index].subscribe, self.scheduler.clock); |
|
} |
|
); |
|
this.observers.push(observer); |
|
|
|
return newPromise || new MockPromise(this.scheduler, this.messages); |
|
}; |
|
|
|
var HotObservable = (function (__super__) { |
|
inherits(HotObservable, __super__); |
|
|
|
function HotObservable(scheduler, messages) { |
|
__super__.call(this); |
|
var message, notification, observable = this; |
|
this.scheduler = scheduler; |
|
this.messages = messages; |
|
this.subscriptions = []; |
|
this.observers = []; |
|
for (var i = 0, len = this.messages.length; i < len; i++) { |
|
message = this.messages[i]; |
|
notification = message.value; |
|
(function (innerNotification) { |
|
scheduler.scheduleAbsolute(null, message.time, function () { |
|
var obs = observable.observers.slice(0); |
|
|
|
for (var j = 0, jLen = obs.length; j < jLen; j++) { |
|
innerNotification.accept(obs[j]); |
|
} |
|
return disposableEmpty; |
|
}); |
|
})(notification); |
|
} |
|
} |
|
|
|
HotObservable.prototype._subscribe = function (o) { |
|
var observable = this; |
|
this.observers.push(o); |
|
this.subscriptions.push(new Subscription(this.scheduler.clock)); |
|
var index = this.subscriptions.length - 1; |
|
return disposableCreate(function () { |
|
var idx = observable.observers.indexOf(o); |
|
observable.observers.splice(idx, 1); |
|
observable.subscriptions[index] = new Subscription(observable.subscriptions[index].subscribe, observable.scheduler.clock); |
|
}); |
|
}; |
|
|
|
return HotObservable; |
|
})(Observable); |
|
|
|
var ColdObservable = (function (__super__) { |
|
inherits(ColdObservable, __super__); |
|
|
|
function ColdObservable(scheduler, messages) { |
|
__super__.call(this); |
|
this.scheduler = scheduler; |
|
this.messages = messages; |
|
this.subscriptions = []; |
|
} |
|
|
|
ColdObservable.prototype._subscribe = function (o) { |
|
var message, notification, observable = this; |
|
this.subscriptions.push(new Subscription(this.scheduler.clock)); |
|
var index = this.subscriptions.length - 1; |
|
var d = new CompositeDisposable(); |
|
for (var i = 0, len = this.messages.length; i < len; i++) { |
|
message = this.messages[i]; |
|
notification = message.value; |
|
(function (innerNotification) { |
|
d.add(observable.scheduler.scheduleRelative(null, message.time, function () { |
|
innerNotification.accept(o); |
|
return disposableEmpty; |
|
})); |
|
})(notification); |
|
} |
|
return disposableCreate(function () { |
|
observable.subscriptions[index] = new Subscription(observable.subscriptions[index].subscribe, observable.scheduler.clock); |
|
d.dispose(); |
|
}); |
|
}; |
|
|
|
return ColdObservable; |
|
})(Observable); |
|
|
|
/** Virtual time scheduler used for testing applications and libraries built using Reactive Extensions. */ |
|
Rx.TestScheduler = (function (__super__) { |
|
inherits(TestScheduler, __super__); |
|
|
|
function baseComparer(x, y) { |
|
return x > y ? 1 : (x < y ? -1 : 0); |
|
} |
|
|
|
function TestScheduler() { |
|
__super__.call(this, 0, baseComparer); |
|
} |
|
|
|
/** |
|
* Schedules an action to be executed at the specified virtual time. |
|
* |
|
* @param state State passed to the action to be executed. |
|
* @param dueTime Absolute virtual time at which to execute the action. |
|
* @param action Action to be executed. |
|
* @return Disposable object used to cancel the scheduled action (best effort). |
|
*/ |
|
TestScheduler.prototype.scheduleAbsolute = function (state, dueTime, action) { |
|
dueTime <= this.clock && (dueTime = this.clock + 1); |
|
return __super__.prototype.scheduleAbsolute.call(this, state, dueTime, action); |
|
}; |
|
/** |
|
* Adds a relative virtual time to an absolute virtual time value. |
|
* |
|
* @param absolute Absolute virtual time value. |
|
* @param relative Relative virtual time value to add. |
|
* @return Resulting absolute virtual time sum value. |
|
*/ |
|
TestScheduler.prototype.add = function (absolute, relative) { |
|
return absolute + relative; |
|
}; |
|
/** |
|
* Converts the absolute virtual time value to a DateTimeOffset value. |
|
* |
|
* @param absolute Absolute virtual time value to convert. |
|
* @return Corresponding DateTimeOffset value. |
|
*/ |
|
TestScheduler.prototype.toAbsoluteTime = function (absolute) { |
|
return new Date(absolute).getTime(); |
|
}; |
|
/** |
|
* Converts the TimeSpan value to a relative virtual time value. |
|
* |
|
* @param timeSpan TimeSpan value to convert. |
|
* @return Corresponding relative virtual time value. |
|
*/ |
|
TestScheduler.prototype.toRelativeTime = function (timeSpan) { |
|
return timeSpan; |
|
}; |
|
/** |
|
* Starts the test scheduler and uses the specified virtual times to invoke the factory function, subscribe to the resulting sequence, and dispose the subscription. |
|
* |
|
* @param create Factory method to create an observable sequence. |
|
* @param created Virtual time at which to invoke the factory to create an observable sequence. |
|
* @param subscribed Virtual time at which to subscribe to the created observable sequence. |
|
* @param disposed Virtual time at which to dispose the subscription. |
|
* @return Observer with timestamped recordings of notification messages that were received during the virtual time window when the subscription to the source sequence was active. |
|
*/ |
|
TestScheduler.prototype.startScheduler = function (createFn, settings) { |
|
settings || (settings = {}); |
|
settings.created == null && (settings.created = ReactiveTest.created); |
|
settings.subscribed == null && (settings.subscribed = ReactiveTest.subscribed); |
|
settings.disposed == null && (settings.disposed = ReactiveTest.disposed); |
|
|
|
var observer = this.createObserver(), source, subscription; |
|
|
|
this.scheduleAbsolute(null, settings.created, function () { |
|
source = createFn(); |
|
return disposableEmpty; |
|
}); |
|
|
|
this.scheduleAbsolute(null, settings.subscribed, function () { |
|
subscription = source.subscribe(observer); |
|
return disposableEmpty; |
|
}); |
|
|
|
this.scheduleAbsolute(null, settings.disposed, function () { |
|
subscription.dispose(); |
|
return disposableEmpty; |
|
}); |
|
|
|
this.start(); |
|
|
|
return observer; |
|
}; |
|
|
|
/** |
|
* Creates a hot observable using the specified timestamped notification messages either as an array or arguments. |
|
* @param messages Notifications to surface through the created sequence at their specified absolute virtual times. |
|
* @return Hot observable sequence that can be used to assert the timing of subscriptions and notifications. |
|
*/ |
|
TestScheduler.prototype.createHotObservable = function () { |
|
var len = arguments.length, args; |
|
if (Array.isArray(arguments[0])) { |
|
args = arguments[0]; |
|
} else { |
|
args = new Array(len); |
|
for (var i = 0; i < len; i++) { args[i] = arguments[i]; } |
|
} |
|
return new HotObservable(this, args); |
|
}; |
|
|
|
/** |
|
* Creates a cold observable using the specified timestamped notification messages either as an array or arguments. |
|
* @param messages Notifications to surface through the created sequence at their specified virtual time offsets from the sequence subscription time. |
|
* @return Cold observable sequence that can be used to assert the timing of subscriptions and notifications. |
|
*/ |
|
TestScheduler.prototype.createColdObservable = function () { |
|
var len = arguments.length, args; |
|
if (Array.isArray(arguments[0])) { |
|
args = arguments[0]; |
|
} else { |
|
args = new Array(len); |
|
for (var i = 0; i < len; i++) { args[i] = arguments[i]; } |
|
} |
|
return new ColdObservable(this, args); |
|
}; |
|
|
|
/** |
|
* Creates a resolved promise with the given value and ticks |
|
* @param {Number} ticks The absolute time of the resolution. |
|
* @param {Any} value The value to yield at the given tick. |
|
* @returns {MockPromise} A mock Promise which fulfills with the given value. |
|
*/ |
|
TestScheduler.prototype.createResolvedPromise = function (ticks, value) { |
|
return new MockPromise(this, [Rx.ReactiveTest.onNext(ticks, value), Rx.ReactiveTest.onCompleted(ticks)]); |
|
}; |
|
|
|
/** |
|
* Creates a rejected promise with the given reason and ticks |
|
* @param {Number} ticks The absolute time of the resolution. |
|
* @param {Any} reason The reason for rejection to yield at the given tick. |
|
* @returns {MockPromise} A mock Promise which rejects with the given reason. |
|
*/ |
|
TestScheduler.prototype.createRejectedPromise = function (ticks, reason) { |
|
return new MockPromise(this, [Rx.ReactiveTest.onError(ticks, reason)]); |
|
}; |
|
|
|
/** |
|
* Creates an observer that records received notification messages and timestamps those. |
|
* @return Observer that can be used to assert the timing of received notifications. |
|
*/ |
|
TestScheduler.prototype.createObserver = function () { |
|
return new MockObserver(this); |
|
}; |
|
|
|
return TestScheduler; |
|
})(VirtualTimeScheduler); |
|
|
|
return Rx; |
|
}));
|
|
|