Android RxJava Observable.interval() doesn't stop emitting items

Question

I have a static field

 private static Subscription timer;

and two static methods:

public static void setTimer() {
    timer = Observable.interval(2, TimeUnit.SECONDS, Schedulers.computation())
            .doOnNext(tick -> update(tick))
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe();
}

public static void removeTimer() {
    if (timer != null && !timer.isUnsubscribed()) {
        timer.unsubscribe();
        timer = null;
    }
}

Guessing after unsubsription Observable have to stop emitting items. However it doesn't work. If function updatePrices is

private static void update(long tick) {
    Log.d(TAG, "tick");
}

Logs continue to be printed after calling removeTimer().

So the question is how to stop emitting items in my observable correctly?

Solved

The issue was in double calling of setTimer().

However I still have a question. Can anybody explain why is the old copy of timer still continues to emit items after the second call of setTimer()?

Cerussite

I'd tried your code and I don't see any logs printed after removeTimer is called. Could you verify that you are not calling the setTimer again from somewhere after the removeTimer call? – Backsight

Thank you for your reply! I found an issue. Actually the problem was in double calling of setTimer() before removeTimer(). However this behavior looks a bit magically for me, because I guessed that the second call of setTimer() should just replace the old observable + subscriber with the new one. Could you explain it? – Cerussite

Each setTimer call creates a new subscription(timer) thereby replacing the old one. When you call removeTimer it is applied to the last subscription, thereby leaking others. You can avoid that by checking for null before creating a new one. if(timer == null || timer.isUnsubscribed()){ timer=...} – Backsight

Answer 1

Maybe too late, But may it help others!

Observable.interval(TICK_MILLIS, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread())
                .map(v -> v * TICK_MILLIS) // use to map value to onNext func,
                .takeUntil(v -> v > MAX_TICKS) // Stop Timer here.
                .take() // You can use take to stop timer after N count
                .subscribe(this::onNext, Log::d ,this::onComplete);

Antennule