Sha256: 83cf4cf9fd4a0b815d0d47710e75887e29e1ee304b9c7391eb83434013061bf5
Contents?: true
Size: 1.71 KB
Versions: 14
Compression:
Stored size: 1.71 KB
Contents
var PausableObservable = (function (__super__) {
inherits(PausableObservable, __super__);
function subscribe(observer) {
var conn = this.source.publish(),
subscription = conn.subscribe(observer),
connection = disposableEmpty;
var pausable = this.pauser.distinctUntilChanged().subscribe(function (b) {
if (b) {
connection = conn.connect();
} else {
connection.dispose();
connection = disposableEmpty;
}
});
return new CompositeDisposable(subscription, connection, pausable);
}
function PausableObservable(source, pauser) {
this.source = source;
this.controller = new Subject();
if (pauser && pauser.subscribe) {
this.pauser = this.controller.merge(pauser);
} else {
this.pauser = this.controller;
}
__super__.call(this, subscribe, source);
}
PausableObservable.prototype.pause = function () {
this.controller.onNext(false);
};
PausableObservable.prototype.resume = function () {
this.controller.onNext(true);
};
return PausableObservable;
}(Observable));
/**
* Pauses the underlying observable sequence based upon the observable sequence which yields true/false.
* @example
* var pauser = new Rx.Subject();
* var source = Rx.Observable.interval(100).pausable(pauser);
* @param {Observable} pauser The observable sequence used to pause the underlying sequence.
* @returns {Observable} The observable sequence which is paused based upon the pauser.
*/
observableProto.pausable = function (pauser) {
return new PausableObservable(this, pauser);
};
Version data entries
14 entries across 7 versions & 1 rubygems