RxJS (5.0rc4): Pause and resume an interval timer

8
votes

I am using Rx to keep an animation clock. Every animation frame, it maps an interval tick to the new values for that tick.

Suppose I want to pause the animation. The most natural way would be to somehow suspend the clock rx and then resume it at a later time.

Unsubscribe then resubscribe isn't a natural fit, because this animation clock is a cold observable. I don't want to restart the animation when they resume. If I go for a workaround method, I will have to generate a new resume rx, vastly complicating all of the exposed APIs.

Backpressure methods don't seem promising:

pause doesn't work, because I want to resume where I left off, not jump forward. In other words I don't want to drop ticks while it's off.

pausableBuffered doesn't work, because on resume, it will drain all of the accumulated ticks as fast as it can.

Using some sort of a virtual time scheduler to completely stop time and then resume normal time might be possible(?)

I am on RxJS 5.0rc4, but I wouldn't know how to do this on RxJS 4, either. Any advice for either version would be appreciated.

2
rxjsrxjs5

2 Answers

9
votes

Use switchMap on a pauser stream to choose between the original source and Observable.never. If you don't want the timer to jump ahead, then manage it yourself (using the x variable in the below).

function pausableInterval(ms, pauser) {
    let x = 0;
    const source = IntervalObservable.create(ms);
 
    return pauser.switchMap(
        paused => paused ? 
        Observable.never() : 
        source.map(() => x++)
    );
}

The pauser stream should emit booleans.

Not tested.

5
votes

Using torazaburo's idea got me 95% of the way. The last step was that I needed that x closure to be unique for each subscription. Observable.create was what I needed to create a closure for each subscription.

pausableInterval(paused: Rx.Observable<boolean>): Rx.Observable<number> {
    return Rx.Observable.create((obs: Rx.Observer<number>) => {
        let i = 0;
        let ticker = Rx.Observable.interval(1000 / this.fps).map(() => i++)

        let p = paused.switchMap(paused => paused ? Rx.Observable.never() : ticker);
        return p.subscribe(val => obs.next(val), err => obs.error(err), () => obs.complete());
     });
 }