Convert CountDownTimer Cancel feature to RxJava

Question

I have an RxObservable that emits an item. I have an RxObserver that consumes the item, turns something on, waits 30 seconds, then turns that same thing off. If in the middle of that 30 second wait the Observable emits another item, the Observer needs to cancel the timer and restart the previous process. The solution was very simple to implement with a CountDownTimer. Here's a snippet of what I wrote:

.subscribe {
            ledTimer?.cancel()

            ledTimer = object : CountDownTimer(COUNTDOWN_MS, COUNTDOWN_INTERVAL_MS) {

                override fun onTick(timeRemaining: Long) {
                    Log.d(
                        TAG,
                        "%d seconds remaining from recent Pcu Event.".format(timeRemaining / COUNTDOWN_INTERVAL_MS)
                    )
                }

                override fun onFinish() {
                    Log.d(TAG, "Timer has completed.")
                    ledSubject.onNext(LedEvent.LedOff)
                }
            }
            ledSubject.onNext(it)
            ledTimer?.start()
        }

So as can you see I cancel the timer if it's currently running, create a new object and start it. I need to convert this to use Rx. The part I'm stuck on is the cancel feature of the timer. This is what I came up without the "interrupt" feature:

   ledStream
        .doOnNext{
            ledSubject.onNext(it)
        }.flatMap {
            Observable.timer(COUNTDOWN_S, TimeUnit.SECONDS)
        }.subscribe{
            ledSubject.onNext(LedEvent.LedOff)
        }

So it's very nice and concise but when subsequent items are emitted it doesn't reset. I've looked at operators like switchOnNext but just don't have a good enough grasp on Rx yet. If I could see a simple example with explanation that would be a great help.

Answer 1

You can just use debounce operator. Here is the example:

ledStream
        .doOnNext{
            ledSubject.onNext(it)
        }
        .debounce(COUNTDOWN_S, TimeUnit.SECONDS)
        .subscribe{
            ledSubject.onNext(LedEvent.LedOff)
        }