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.
343 lines
11 KiB
343 lines
11 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'], 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) { |
|
|
|
// Aliases |
|
var Observable = Rx.Observable, |
|
observableProto = Observable.prototype, |
|
AnonymousObservable = Rx.AnonymousObservable, |
|
observableThrow = Observable.throwError, |
|
observerCreate = Rx.Observer.create, |
|
SingleAssignmentDisposable = Rx.SingleAssignmentDisposable, |
|
CompositeDisposable = Rx.CompositeDisposable, |
|
AbstractObserver = Rx.internals.AbstractObserver, |
|
noop = Rx.helpers.noop, |
|
inherits = Rx.internals.inherits, |
|
isFunction = Rx.helpers.isFunction; |
|
|
|
var errorObj = {e: {}}; |
|
|
|
function tryCatcherGen(tryCatchTarget) { |
|
return function tryCatcher() { |
|
try { |
|
return tryCatchTarget.apply(this, arguments); |
|
} catch (e) { |
|
errorObj.e = e; |
|
return errorObj; |
|
} |
|
}; |
|
} |
|
|
|
var tryCatch = Rx.internals.tryCatch = function tryCatch(fn) { |
|
if (!isFunction(fn)) { throw new TypeError('fn must be a function'); } |
|
return tryCatcherGen(fn); |
|
}; |
|
|
|
function thrower(e) { |
|
throw e; |
|
} |
|
|
|
var Map = root.Map || (function () { |
|
function Map() { |
|
this.size = 0; |
|
this._values = []; |
|
this._keys = []; |
|
} |
|
|
|
Map.prototype['delete'] = function (key) { |
|
var i = this._keys.indexOf(key); |
|
if (i === -1) { return false; } |
|
this._values.splice(i, 1); |
|
this._keys.splice(i, 1); |
|
this.size--; |
|
return true; |
|
}; |
|
|
|
Map.prototype.get = function (key) { |
|
var i = this._keys.indexOf(key); |
|
return i === -1 ? undefined : this._values[i]; |
|
}; |
|
|
|
Map.prototype.set = function (key, value) { |
|
var i = this._keys.indexOf(key); |
|
if (i === -1) { |
|
this._keys.push(key); |
|
this._values.push(value); |
|
this.size++; |
|
} else { |
|
this._values[i] = value; |
|
} |
|
return this; |
|
}; |
|
|
|
Map.prototype.forEach = function (cb, thisArg) { |
|
for (var i = 0; i < this.size; i++) { |
|
cb.call(thisArg, this._values[i], this._keys[i]); |
|
} |
|
}; |
|
|
|
return Map; |
|
}()); |
|
|
|
/** |
|
* @constructor |
|
* Represents a join pattern over observable sequences. |
|
*/ |
|
function Pattern(patterns) { |
|
this.patterns = patterns; |
|
} |
|
|
|
/** |
|
* Creates a pattern that matches the current plan matches and when the specified observable sequences has an available value. |
|
* @param other Observable sequence to match in addition to the current pattern. |
|
* @return {Pattern} Pattern object that matches when all observable sequences in the pattern have an available value. |
|
*/ |
|
Pattern.prototype.and = function (other) { |
|
return new Pattern(this.patterns.concat(other)); |
|
}; |
|
|
|
/** |
|
* Matches when all observable sequences in the pattern (specified using a chain of and operators) have an available value and projects the values. |
|
* @param {Function} selector Selector that will be invoked with available values from the source sequences, in the same order of the sequences in the pattern. |
|
* @return {Plan} Plan that produces the projected values, to be fed (with other plans) to the when operator. |
|
*/ |
|
Pattern.prototype.thenDo = function (selector) { |
|
return new Plan(this, selector); |
|
}; |
|
|
|
function Plan(expression, selector) { |
|
this.expression = expression; |
|
this.selector = selector; |
|
} |
|
|
|
function handleOnError(o) { return function (e) { o.onError(e); }; } |
|
function handleOnNext(self, observer) { |
|
return function onNext () { |
|
var result = tryCatch(self.selector).apply(self, arguments); |
|
if (result === errorObj) { return observer.onError(result.e); } |
|
observer.onNext(result); |
|
}; |
|
} |
|
|
|
Plan.prototype.activate = function (externalSubscriptions, observer, deactivate) { |
|
var joinObservers = [], errHandler = handleOnError(observer); |
|
for (var i = 0, len = this.expression.patterns.length; i < len; i++) { |
|
joinObservers.push(planCreateObserver(externalSubscriptions, this.expression.patterns[i], errHandler)); |
|
} |
|
var activePlan = new ActivePlan(joinObservers, handleOnNext(this, observer), function () { |
|
for (var j = 0, jlen = joinObservers.length; j < jlen; j++) { |
|
joinObservers[j].removeActivePlan(activePlan); |
|
} |
|
deactivate(activePlan); |
|
}); |
|
for (i = 0, len = joinObservers.length; i < len; i++) { |
|
joinObservers[i].addActivePlan(activePlan); |
|
} |
|
return activePlan; |
|
}; |
|
|
|
function planCreateObserver(externalSubscriptions, observable, onError) { |
|
var entry = externalSubscriptions.get(observable); |
|
if (!entry) { |
|
var observer = new JoinObserver(observable, onError); |
|
externalSubscriptions.set(observable, observer); |
|
return observer; |
|
} |
|
return entry; |
|
} |
|
|
|
function ActivePlan(joinObserverArray, onNext, onCompleted) { |
|
this.joinObserverArray = joinObserverArray; |
|
this.onNext = onNext; |
|
this.onCompleted = onCompleted; |
|
this.joinObservers = new Map(); |
|
for (var i = 0, len = this.joinObserverArray.length; i < len; i++) { |
|
var joinObserver = this.joinObserverArray[i]; |
|
this.joinObservers.set(joinObserver, joinObserver); |
|
} |
|
} |
|
|
|
ActivePlan.prototype.dequeue = function () { |
|
this.joinObservers.forEach(function (v) { v.queue.shift(); }); |
|
}; |
|
|
|
ActivePlan.prototype.match = function () { |
|
var i, len, hasValues = true; |
|
for (i = 0, len = this.joinObserverArray.length; i < len; i++) { |
|
if (this.joinObserverArray[i].queue.length === 0) { |
|
hasValues = false; |
|
break; |
|
} |
|
} |
|
if (hasValues) { |
|
var firstValues = [], |
|
isCompleted = false; |
|
for (i = 0, len = this.joinObserverArray.length; i < len; i++) { |
|
firstValues.push(this.joinObserverArray[i].queue[0]); |
|
this.joinObserverArray[i].queue[0].kind === 'C' && (isCompleted = true); |
|
} |
|
if (isCompleted) { |
|
this.onCompleted(); |
|
} else { |
|
this.dequeue(); |
|
var values = []; |
|
for (i = 0, len = firstValues.length; i < firstValues.length; i++) { |
|
values.push(firstValues[i].value); |
|
} |
|
this.onNext.apply(this, values); |
|
} |
|
} |
|
}; |
|
|
|
var JoinObserver = (function (__super__) { |
|
inherits(JoinObserver, __super__); |
|
|
|
function JoinObserver(source, onError) { |
|
__super__.call(this); |
|
this.source = source; |
|
this.onError = onError; |
|
this.queue = []; |
|
this.activePlans = []; |
|
this.subscription = new SingleAssignmentDisposable(); |
|
this.isDisposed = false; |
|
} |
|
|
|
var JoinObserverPrototype = JoinObserver.prototype; |
|
|
|
JoinObserverPrototype.next = function (notification) { |
|
if (!this.isDisposed) { |
|
if (notification.kind === 'E') { |
|
return this.onError(notification.error); |
|
} |
|
this.queue.push(notification); |
|
var activePlans = this.activePlans.slice(0); |
|
for (var i = 0, len = activePlans.length; i < len; i++) { |
|
activePlans[i].match(); |
|
} |
|
} |
|
}; |
|
|
|
JoinObserverPrototype.error = noop; |
|
JoinObserverPrototype.completed = noop; |
|
|
|
JoinObserverPrototype.addActivePlan = function (activePlan) { |
|
this.activePlans.push(activePlan); |
|
}; |
|
|
|
JoinObserverPrototype.subscribe = function () { |
|
this.subscription.setDisposable(this.source.materialize().subscribe(this)); |
|
}; |
|
|
|
JoinObserverPrototype.removeActivePlan = function (activePlan) { |
|
this.activePlans.splice(this.activePlans.indexOf(activePlan), 1); |
|
this.activePlans.length === 0 && this.dispose(); |
|
}; |
|
|
|
JoinObserverPrototype.dispose = function () { |
|
__super__.prototype.dispose.call(this); |
|
if (!this.isDisposed) { |
|
this.isDisposed = true; |
|
this.subscription.dispose(); |
|
} |
|
}; |
|
|
|
return JoinObserver; |
|
} (AbstractObserver)); |
|
|
|
/** |
|
* Creates a pattern that matches when both observable sequences have an available value. |
|
* |
|
* @param right Observable sequence to match with the current sequence. |
|
* @return {Pattern} Pattern object that matches when both observable sequences have an available value. |
|
*/ |
|
observableProto.and = function (right) { |
|
return new Pattern([this, right]); |
|
}; |
|
|
|
/** |
|
* Matches when the observable sequence has an available value and projects the value. |
|
* |
|
* @param {Function} selector Selector that will be invoked for values in the source sequence. |
|
* @returns {Plan} Plan that produces the projected values, to be fed (with other plans) to the when operator. |
|
*/ |
|
observableProto.thenDo = function (selector) { |
|
return new Pattern([this]).thenDo(selector); |
|
}; |
|
|
|
/** |
|
* Joins together the results from several patterns. |
|
* |
|
* @param plans A series of plans (specified as an Array of as a series of arguments) created by use of the Then operator on patterns. |
|
* @returns {Observable} Observable sequence with the results form matching several patterns. |
|
*/ |
|
Observable.when = function () { |
|
var len = arguments.length, plans; |
|
if (Array.isArray(arguments[0])) { |
|
plans = arguments[0]; |
|
} else { |
|
plans = new Array(len); |
|
for(var i = 0; i < len; i++) { plans[i] = arguments[i]; } |
|
} |
|
return new AnonymousObservable(function (o) { |
|
var activePlans = [], |
|
externalSubscriptions = new Map(); |
|
var outObserver = observerCreate( |
|
function (x) { o.onNext(x); }, |
|
function (err) { |
|
externalSubscriptions.forEach(function (v) { v.onError(err); }); |
|
o.onError(err); |
|
}, |
|
function (x) { o.onCompleted(); } |
|
); |
|
try { |
|
for (var i = 0, len = plans.length; i < len; i++) { |
|
activePlans.push(plans[i].activate(externalSubscriptions, outObserver, function (activePlan) { |
|
var idx = activePlans.indexOf(activePlan); |
|
activePlans.splice(idx, 1); |
|
activePlans.length === 0 && o.onCompleted(); |
|
})); |
|
} |
|
} catch (e) { |
|
return observableThrow(e).subscribe(o); |
|
} |
|
var group = new CompositeDisposable(); |
|
externalSubscriptions.forEach(function (joinObserver) { |
|
joinObserver.subscribe(); |
|
group.add(joinObserver); |
|
}); |
|
|
|
return group; |
|
}); |
|
}; |
|
|
|
return Rx; |
|
}));
|
|
|