Replacing Java Timer with Kotlin Coroutine Timer

[anggrayudi]

I often using this method to replace java.util.Timer in my projects:

inline fun startCoroutineTimer(delayMillis: Long = 0, repeatMillis: Long = 0, crossinline action: () -> Unit) = GlobalScope.launch {
    delay(delayMillis)
    if (repeatMillis > 0) {
        while (true) {
            action()
            delay(repeatMillis)
        }
    } else {
        action()
    }
}

java.util.Timer is buggy. On some Android devices, it causes the following crash, even though I re-init the object:

java.lang.IllegalStateException: Timer was canceled

With Kotlin Coroutines, it is possible for me to prevent this terrible error. I hope you add the method I wrote above.

[JakeWharton]

You probably want this to be a factory for a Flow that emits either Unit or a counter on each tick.

[anggrayudi]

@JakeWharton, I am sure that somebody else needs it as well. I hope this will be added to Coroutine. It is simple code, but powerful.

[JakeWharton]

I would be against it as is, if that wasn't clear. Having a Flow factory which is a timer, however, makes sense.

[anggrayudi]

@JakeWharton I am not sure whether this is related to Coroutine itself since it is an additional method to replace Java Timer. In Java Timer, I write the following code to start a repeated task every 1 second:

val timer = Timer()
timer.schedule(object : TimerTask() {
    override fun run() {
        // some actions
    }
}, 1, 1000) // Minimum delay is 1 millisecond. If lower, it throws IllegalArgumentException

The main backward from Java Timer is, we cannot start the repeated action immediately by setting the delay time to 0ms. Hence, the following approach with Coroutine is possible:

val timer: Job = startCoroutineTimer(delayMillis = 0, repeatMillis = 1000) {
    // some actions
}

Another backward from Java Timer is, it sometimes throws java.lang.IllegalStateException: Timer was canceled and causes app to crash. I tried to re-init the Timer and start a new one. But the error still appears. I don't know why. It affects less than 2% of my users.

I see Crashlytics reported this message. When I migrated to Coroutine timer, the error is gone. I hope you consider adding this method to Coroutine. Thanks in advance.

[napperley]

If users want a concurrency model agnostic version of this functionality then they should down vote the comment that was made earlier.

[napperley]

What Jake is proposing is to make this functionality only available if using the RX concurrency model via the Flow implementation. Doing so would be a extremely bad design decision that promotes tight coupling with the functionality only being available in Flow, even though this functionality isn't dependent on any concurrency model.

[Enrico2]

(edited, after a PR review)

Based on the original suggestion, took a stab at a more testable variation:

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.cancel
import kotlinx.coroutines.launch
import kotlinx.coroutines.time.delay
import java.time.Duration
import java.util.concurrent.atomic.AtomicBoolean

class KorosTimerTask internal constructor(
	name: String,
	private val delay: Duration = Duration.ZERO,
	private val repeat: Duration? = null,
	private val coroutineScope: CoroutineScope = GlobalScope,
	action: suspend () -> Unit
) {
	private val log = Logs.logger(this::class.java)
	private val keepRunning = AtomicBoolean(true)
	private var job: Job? = null
	private val tryAction = suspend {
		try {
			action()
		} catch (e: Throwable) {
			log.warn("$name timer action failed: $action")
		}
	}

	fun start() {
		job = coroutineScope.launch {
			delay(delay)
			if (repeat != null) {
				while (keepRunning.get()) {
					tryAction()
					delay(repeat)
				}
			} else {
				if (keepRunning.get()) {
					tryAction()
				}
			}
		}
	}

	/**
	 * Initiates an orderly shutdown, where if the timer task is currently running,
	 * we will let it finish, but not run it again.
	 * Invocation has no additional effect if already shut down.
	 */
	fun shutdown() {
		keepRunning.set(false)
	}

	/**
	 * Immediately stops the timer task, even if the job is currently running,
	 * by cancelling the underlying Koros Job.
	 */
	fun cancel() {
		shutdown()
		job?.cancel("cancel() called")
	}

	companion object {
		/**
		 * Runs the given `action` after the given `delay`,
		 * once the `action` completes, waits the `repeat` duration
		 * and runs again, until `shutdown` is called.
		 *
		 * if action() throws an exception, it will be swallowed and a warning will be logged.
		 */
		fun start(
			name: String,
			delay: Duration = Duration.ZERO,
			repeat: Duration? = null,
			coroutineScope: CoroutineScope = GlobalScope,
			action: suspend () -> Unit
		): KorosTimerTask =
			KorosTimerTask(name, delay, repeat, coroutineScope, action).also { it.start() }
	}
}

The test:

import io.kotlintest.milliseconds
import io.kotlintest.shouldBe
import io.kotlintest.specs.StringSpec
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.cancel
import kotlinx.coroutines.channels.Channel

class KorosTimerTaskTest : StringSpec(
	{
		"Basic operation" {
			val channel = Channel<Unit>()
			var times = 0
			lateinit var timer: KorosTimerTask
			timer = KorosTimerTask.start("test", repeat = 10.milliseconds, coroutineScope = this) {
				if (times == 4) {
					cancel("should not run a 5th time")
				} else if (times == 3) {
					// shutdown self to show the task finishes running.
					timer.shutdown()
				}

				channel.receive()
				times++
			}

			val loops = 4
			repeat(loops) {
				channel.send(Unit)
			}

			times.shouldBe(loops)
		}

		"cancel during task run" {
			val c1 = Channel<Unit>()
			val c2 = Channel<Unit>()
			val cancelled = Channel<Unit>()
			val timer = KorosTimerTask.start("test", repeat = 10.milliseconds, coroutineScope = this) {
				try {
					c1.receive()
					c2.receive()
				} catch (e: CancellationException) {
					cancelled.send(Unit)
				}
			}
			c1.send(Unit)
			timer.cancel()
			cancelled.receive()
		}

		"shutdown before start never runs task" {
			val task = KorosTimerTask("test", coroutineScope = this) {
				cancel("should never run")
			}
			task.shutdown()
			task.start()
		}
	}
)

[elizarov]

@anggrayudi Can you, please, elaborate a bit on how you use a function like that. You write in your message this example:

val timer: Job = startCoroutineTimer(delayMillis = 0, repeatMillis = 1000) {
    // some actions
}

Do what are those "some actions" that are inside? What do use it for? What is your use-case? Can you give a larger example of a piece of code that is using startCoroutineTimer?

[zach-klippenstein]

this functionality isn't dependent on any concurrency model.

Callbacks are just another concurrency model. It's trivial to convert between callbacks and Flows, but the latter already answers questions that aren't answered by the originally proposed solution.

How would you manage the lifecycle of this timer? How do you customize the dispatcher? The original proposal uses GlobalScope and provides no mechanism to stop/dispose of the timer, which means every timer that gets created will be leaked. It also doesn't provide a mechanism for controlling which dispatcher is used to run the timer, which can be inefficient because it forces consumers using the timer from other dispatchers to change threads just to run the delay (eg if I'm running a timer on a UI thread, I can just use the UI system's scheduling mechanism and not hop into the default pool and back).

The advantage of exposing this functionality using Flow is that the Flow API already has answers for those questions, and doesn't require defining a whole new API surface, with its own documentation, learning curve, etc.

I don't understand your aversion to using Flow as the primitive here. Flow is a lightweight abstraction on basic coroutine concepts, so it's useable from any concurrency model. It's built-in to the coroutines library, so there's no added dependency weight.

[Enrico2]

@zach-klippenstein see my suggestion above to address your (valid!) concerns.

[anggrayudi]

@elizarov I use the function to ping an IP address every 2 seconds. This app implement that function https://play.google.com/store/apps/details?id=com.anggrayudi.ping

[elizarov]

@anggrayudi Thanks. If this the only thing you are using it for in your application?

[anggrayudi]

@elizarov I am using it on another app too, https://play.google.com/store/apps/details?id=com.anggrayudi.watercat

This Coroutines timer is really helpful to report download speed to user.

[elizarov]

You are welcome to use your startCoroutineTimer solution, but we don't plan to a function like startCoroutineTimer to the core library, because it is easy to write, limited to this narrow use-case and cannot be easily composed with other operations.

Our plan is to provide ready-to-use tick-generators (~timers) for Kotlin Flow so that your particular case would be solved with code like this (note, that names are purely tentative here):

tickerFlow(1000)
    .onEach { /* some actions here */ }
    .launchIn(scope)

The advantage of using the flow here is that you can easily use other flow operators to control what happens when the action takes longer than ticker period, etc, without having to provide lots of configuration options to the tickerFlow function itself.

Please, use #540 as an umbrella issue for that work to post your use-cases.

[gmk57]

@anggrayudi Thanks for the idea and "reference" implementation! ;)

Here is my variant, almost as simple, but slightly more configurable (dispatcher from scope or context, cancellation via scope or job).

We use it for a lot of things on Android: periodic downloading or uploading data to server, periodic calculations, on-screen timers...

And I would prefer it to a Flow-based solution, because in many cases it is more convenient to calculate delay since previous run completion. Especially on Android, where in Doze mode multiple "time ticks" are queued and then all fire in quick succession. We learned this a hard way with RxJava Observable.interval() and had to add throttleFirst() everywhere to prevent e.g. dozens of simultaneous network requests. Coroutine-based solution just does not have this issue.

I also prefer to keep things as simple as possible, and "loop with delay" seems easier to grasp then "stream of time ticks". Flows are nice and definitely have their uses, but I don't like "everything is a stream" mantra. From my experience with RxJava, complex stream-based logic may become an undecipherable mess.