Pause / Resume a timer/delay in RX

Question

I'm trying to pause / resume a delayed operation using rx-Java, and surprisingly I can't find any details on how to do that.

Obviously, I know how to do it by creating a specific Timer thread and keeping track of the time, but I'm looking for a more elegant and reactive way.

I have three different observables, playDetected, one for pauseDetected and one for stopDetected. I want to emit something after a certain delay of PLAY, but pause when my pause observable emits, and resume when I get another PLAY

What I have so far: (it's written in kotlin but Java, pseudo-code or any language will do for an answer)

val playSubscription = playDetected
            .delay(DELAY, SECONDS, schedulers.computation)
            .subscribe { emitFinalEvent(it) }

stopDetected.subscribe { playSubscription.unsubscribe() }

My delay works, and when I detect a STOP, it successfully removes the delay so that the next PLAY can start it again. But how to pause and resume when pauseDetected emits something???

I suppose that after resume you would like to start in the point when the delay stopped, not from beginning? I mean your delay is e.g. 10 seconds, assume that 6 seconds already passed and then pauseEvent occured. Once the playEvent will be emitted delay should be 4 seconds now (since initial delay was 10 seconds and 6 seconds pass? — Rzodkiewka, Apr 27 '16 at 10:04
Yes indeed, so that's why I think delay might not work for that. I need to somehow keep the state of the timer. I am working on another solution using interval(1 second) for ticks, which seems to work. I'll post the answer here after more testing — Guillaume, Apr 27 '16 at 21:47
Is this relevant to your question?: http://stackoverflow.com/questions/35782767/how-can-an-observable-be-paused-without-loosing-the-items-emitted/35805100#35805100 — yurgis, Apr 28 '16 at 00:27
Thanks @yurgis, it is actually, even though it's not exactly the same — Guillaume, Apr 28 '16 at 10:50

score 1 · Accepted Answer · answered Apr 28 '16 at 10:53

Here is how I ended up doing it:

playDetected
            .doOnNext {
                if (trackIsDifferent(it)) resetTimer()
                trackPlaying.set(it.track)
            }
            .switchMap { state ->
                interval(1, SECONDS, schedulers.computation)
                        .doOnNext { currentTimer.incrementAndGet() }
                        .takeUntil(merge(pauseDetected, stopDetected.doOnNext { resetTimer() }))
                        .filter { currentTimer.get() == DELAY }
                        .map { state }
            }.subscribe { emitFinalEvent(it)) }

with:

private val trackPlaying = AtomicReference<Track>()
private val currentTimer = AtomicLong()

private fun resetTimer() {
    currentTimer.set(0)
}

private fun trackIsDifferent(payload: StateWithTrack) = payload.track != trackPlaying.get()

score 0 · Answer 2 · answered Jul 24 '17 at 06:13

Some time ago, I was also looking for kind of RX "timer" solutions, but non of them met my expectations. So there you can find my own solution:

AtomicLong elapsedTime = new AtomicLong();
AtomicBoolean resumed = new AtomicBoolean();
AtomicBoolean stopped = new AtomicBoolean();

public Flowable<Long> startTimer() { //Create and starts timper
    resumed.set(true);
    stopped.set(false);
    return Flowable.interval(1, TimeUnit.SECONDS)
            .takeWhile(tick -> !stopped.get())
            .filter(tick -> resumed.get())
            .map(tick -> elapsedTime.addAndGet(1000));
}

public void pauseTimer() {
    resumed.set(false);
}

public void resumeTimer() {
    resumed.set(true);
}

public void stopTimer() {
    stopped.set(true);
}

public void addToTimer(int seconds) {
    elapsedTime.addAndGet(seconds * 1000);
}

In case of delay, just overload interval function with delay.

Pause / Resume a timer/delay in RX

2 Answers2