This repository was archived by the owner on Apr 20, 2018. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 2.1k
/
Copy pathpausable.js
58 lines (48 loc) · 1.75 KB
/
pausable.js
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
var PausableObservable = (function (__super__) {
inherits(PausableObservable, __super__);
function PausableObservable(source, pauser) {
this.source = source;
this.controller = new Subject();
this.paused = true;
if (pauser && pauser.subscribe) {
this.pauser = this.controller.merge(pauser);
} else {
this.pauser = this.controller;
}
__super__.call(this);
}
PausableObservable.prototype._subscribe = function (o) {
var conn = this.source.publish(),
subscription = conn.subscribe(o),
connection = disposableEmpty;
var pausable = this.pauser.startWith(!this.paused).distinctUntilChanged().subscribe(function (b) {
if (b) {
connection = conn.connect();
} else {
connection.dispose();
connection = disposableEmpty;
}
});
return new NAryDisposable([subscription, connection, pausable]);
};
PausableObservable.prototype.pause = function () {
this.paused = true;
this.controller.onNext(false);
};
PausableObservable.prototype.resume = function () {
this.paused = false;
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);
};