344 lines
11 KiB
JavaScript
344 lines
11 KiB
JavaScript
// 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;
|
|
}));
|