Sha256: 3580ca412be48599eb2738155bd4d39ca72b862afb938f4527b2b28ceac1c031
Contents?: true
Size: 1.79 KB
Versions: 14
Compression:
Stored size: 1.79 KB
Contents
var ScheduledObserver = Rx.internals.ScheduledObserver = (function (__super__) {
inherits(ScheduledObserver, __super__);
function ScheduledObserver(scheduler, observer) {
__super__.call(this);
this.scheduler = scheduler;
this.observer = observer;
this.isAcquired = false;
this.hasFaulted = false;
this.queue = [];
this.disposable = new SerialDisposable();
}
ScheduledObserver.prototype.next = function (value) {
var self = this;
this.queue.push(function () { self.observer.onNext(value); });
};
ScheduledObserver.prototype.error = function (e) {
var self = this;
this.queue.push(function () { self.observer.onError(e); });
};
ScheduledObserver.prototype.completed = function () {
var self = this;
this.queue.push(function () { self.observer.onCompleted(); });
};
ScheduledObserver.prototype.ensureActive = function () {
var isOwner = false, parent = this;
if (!this.hasFaulted && this.queue.length > 0) {
isOwner = !this.isAcquired;
this.isAcquired = true;
}
if (isOwner) {
this.disposable.setDisposable(this.scheduler.scheduleRecursive(function (self) {
var work;
if (parent.queue.length > 0) {
work = parent.queue.shift();
} else {
parent.isAcquired = false;
return;
}
try {
work();
} catch (ex) {
parent.queue = [];
parent.hasFaulted = true;
throw ex;
}
self();
}));
}
};
ScheduledObserver.prototype.dispose = function () {
__super__.prototype.dispose.call(this);
this.disposable.dispose();
};
return ScheduledObserver;
}(AbstractObserver));
Version data entries
14 entries across 7 versions & 1 rubygems