Sha256: 5d2e74384a8678a1c0a0a063b515f9b81473be6999c50b463ac41ed6adb9e7ef
Contents?: true
Size: 1.37 KB
Versions: 14
Compression:
Stored size: 1.37 KB
Contents
function Plan(expression, selector) {
this.expression = expression;
this.selector = selector;
}
Plan.prototype.activate = function (externalSubscriptions, observer, deactivate) {
var self = this;
var joinObservers = [];
for (var i = 0, len = this.expression.patterns.length; i < len; i++) {
joinObservers.push(planCreateObserver(externalSubscriptions, this.expression.patterns[i], observer.onError.bind(observer)));
}
var activePlan = new ActivePlan(joinObservers, function () {
var result;
try {
result = self.selector.apply(self, arguments);
} catch (e) {
observer.onError(e);
return;
}
observer.onNext(result);
}, 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;
}
Version data entries
14 entries across 7 versions & 1 rubygems