How to implement timer with Kotlin coroutines

A very pragmatic approach with Kotlin Flows could be:

// Create the timer flow
val timer = (0..Int.MAX_VALUE)
    .asSequence()
    .asFlow()
    .onEach { delay(1_000) } // specify delay

// Consume it
timer.collect { 
    println("bling: ${it}")
}

Be aware that this solution times out at Int.MAX_VALUE (2,147,483,647), so it might not be the right solution for very short interval values, or extreme long running timer values. Consider Long.MAX_VALUE in that case (see comments).

another possible solution as a reusable kotlin extension of CoroutineScope

fun CoroutineScope.launchPeriodicAsync(
    repeatMillis: Long,
    action: () -> Unit
) = this.async {
    if (repeatMillis > 0) {
        while (isActive) {
            action()
            delay(repeatMillis)
        }
    } else {
        action()
    }
}

and then usage as:

var job = CoroutineScope(Dispatchers.IO).launchPeriodicAsync(100) {
  //...
}

and then to interrupt it:

job.cancel()

another note: we consider here that action is non-blocking and does not take time.

You can create a countdown timer like this

GlobalScope.launch(Dispatchers.Main) {
            val totalSeconds = TimeUnit.MINUTES.toSeconds(2)
            val tickSeconds = 1
            for (second in totalSeconds downTo tickSeconds) {
                val time = String.format("%02d:%02d",
                    TimeUnit.SECONDS.toMinutes(second),
                    second - TimeUnit.MINUTES.toSeconds(TimeUnit.SECONDS.toMinutes(second))
                )
                timerTextView?.text = time
                delay(1000)
            }
            timerTextView?.text = "Done!"
        }

Here's a possible solution using Kotlin Flow

fun tickFlow(millis: Long) = callbackFlow<Int> {
    val timer = Timer()
    var time = 0
    timer.scheduleAtFixedRate(
        object : TimerTask() {
            override fun run() {
                try { offer(time) } catch (e: Exception) {}
                time += 1
            }
        },
        0,
        millis)
    awaitClose {
        timer.cancel()
    }
}

Usage

val job = CoroutineScope(Dispatchers.Main).launch {
   tickFlow(125L).collect {
      print(it)
   }
}

...

job.cancel()

Edit: Joffrey has edited his solution with a better approach.

Old :

Joffrey's solution works for me but I ran into a problem with the for loop.

I have to cancel my ticker in the for loop like this :

            val ticker = ticker(500, 0)
            for (event in ticker) {
                if (...) {
                    ticker.cancel()
                } else {
                    ...
                    }
                }
            }

But ticker.cancel() was throwing a cancellationException because the for loop kept going after this.

I had to use a while loop to check if the channel was not closed to not get this exception.

                val ticker = ticker(500, 0)
                while (!ticker.isClosedForReceive && ticker.iterator().hasNext()) {
                    if (...) {
                        ticker.cancel()
                    } else {
                        ...
                        }
                    }
                }

Timer with START, PAUSE and STOP functions.

Usage:

val timer = Timer(millisInFuture = 10_000L, runAtStart = false)
timer.start()

Timer class:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.asStateFlow

enum class PlayerMode {
    PLAYING,
    PAUSED,
    STOPPED
}

class Timer(
    val millisInFuture: Long,
    val countDownInterval: Long = 1000L,
    runAtStart: Boolean = false,
    val onFinish: (() -> Unit)? = null,
    val onTick: ((Long) -> Unit)? = null
) {
    private var job: Job = Job()
    private val _tick = MutableStateFlow(0L)
    val tick = _tick.asStateFlow()
    private val _playerMode = MutableStateFlow(PlayerMode.STOPPED)
    val playerMode = _playerMode.asStateFlow()

    private val scope = CoroutineScope(Dispatchers.Default)

    init {
        if (runAtStart) start()
    }

    fun start() {
        if (_tick.value == 0L) _tick.value = millisInFuture
        job.cancel()
        job = scope.launch(Dispatchers.IO) {
            _playerMode.value = PlayerMode.PLAYING
            while (isActive) {
                if (_tick.value <= 0) {
                    job.cancel()
                    onFinish?.invoke()
                    _playerMode.value = PlayerMode.STOPPED
                    return@launch
                }
                delay(timeMillis = countDownInterval)
                _tick.value -= countDownInterval
                onTick?.invoke(this@Timer._tick.value)
            }
        }
    }

    fun pause() {
        job.cancel()
        _playerMode.value = PlayerMode.PAUSED
    }

    fun stop() {
        job.cancel()
        _tick.value = 0
        _playerMode.value = PlayerMode.STOPPED
    }
}

I took inspiration from here.

Here is Flow version of Observable.intervalRange(1, 5, 0, 1, TimeUnit.SECONDS) based on Joffrey's answer:

fun tickerFlow(start: Long,
               count: Long,
               initialDelayMs: Long,
               periodMs: Long) = flow<Long> {
    delay(initialDelayMs)

    var counter = start
    while (counter <= count) {
        emit(counter)
        counter += 1

        delay(periodMs)
    }
}

//...

tickerFlow(1, 5, 0, 1_000L)

Made a copy of Observable.intervalRange(0, 90, 0, 1, TimeUnit.SECONDS) ( will emit item in 90 sec each 1 sec ):

fun intervalRange(start: Long, count: Long, initialDelay: Long = 0, period: Long, unit: TimeUnit): Flow<Long> {
        return flow<Long> {
            require(count >= 0) { "count >= 0 required but it was $count" }
            require(initialDelay >= 0) { "initialDelay >= 0 required but it was $initialDelay" }
            require(period > 0) { "period > 0 required but it was $period" }

            val end = start + (count - 1)
            require(!(start > 0 && end < 0)) { "Overflow! start + count is bigger than Long.MAX_VALUE" }

            if (initialDelay > 0) {
                delay(unit.toMillis(initialDelay))
            }

            var counter = start
            while (counter <= count) {
                emit(counter)
                counter += 1

                delay(unit.toMillis(period))
            }
        }
    }

Usage:

lifecycleScope.launch {
intervalRange(0, 90, 0, 1, TimeUnit.SECONDS)
                .onEach {
                    Log.d(TAG, "intervalRange: ${90 - it}")
                }
                .lastOrNull()
}

Used this recently to chunk values based on a timer and max buffer size.

private object Tick

@Suppress("UNCHECKED_CAST")
fun <T : Any> Flow<T>.chunked(size: Int, initialDelay: Long, delay: Long): Flow<List<T>> = flow {
    if (size <= 0) throw IllegalArgumentException("invalid chunk size $size - expected > 0")
    val chunkedList = mutableListOf<T>()
    if (delay > 0L) {
        merge(this@chunked, timerFlow(initialDelay, delay, Tick))
    } else {
        this@chunked
    }
        .collect {
            when (it) {
                is Tick -> {
                    if (chunkedList.isNotEmpty()) {
                        emit(chunkedList.toList())
                        chunkedList.clear()
                    }
                }
                else -> {
                    chunkedList.add(it as T)
                    if (chunkedList.size >= size) {
                        emit(chunkedList.toList())
                        chunkedList.clear()
                    }
                }
            }
        }
    if (chunkedList.isNotEmpty()) {
        emit(chunkedList.toList())
    }
}

fun <T> timerFlow(initialDelay: Long, delay: Long, o: T) = flow {
    if (delay <= 0) throw IllegalArgumentException("invalid delay $delay - expected > 0")
    if (initialDelay > 0) delay(initialDelay)
    while (currentCoroutineContext().isActive) {
        emit(o)
        delay(delay)
    }
}

You could do something like this...

class JitterTimer {
    fun scheduleAtRandom(
        initialDelay : Duration = 2.seconds,
        base: Duration = 3.seconds,
        interval: Duration = 0.5.seconds
    ) = flow {
        delay(initialDelay)
        while (true) {
            emit(Unit)
            val jitter = randomizeJitter(
                baseDelay = base.inWholeMilliseconds,
                interval = interval.inWholeMilliseconds
            )
            delay(jitter)
        }
    }.cancellable()
}

//Usage

timerJob?.cancel()
JitterTimer()
 .scheduleAtRandom(10.seconds, 3.seconds)
 .onEach {
    //Do some work
  }
 .flowOn(dispatcher)
 .catch { it.remoteLog("Recovery", "Attempts") }
 .launchIn(scope)
 .apply { timerJob = this }

It's not using Kotlin coroutines, but if your use case is simple enough you can always just use something like a fixedRateTimer or timer (docs here) which resolve to JVM native Timer.

I was using RxJava's interval for a relatively simple scenario and when I switched to using Timers I saw significant performance and memory improvements.

You can also run your code on the main thread on Android by using View.post() or it's mutliple variants.

The only real annoyance is you'll need to keep track of the old time's state yourself instead of relying on RxJava to do it for you.

But this will always be much faster (important if you're doing performance critical stuff like UI animations etc) and will not have the memory overhead of RxJava's Flowables.

Here's the question's code using a fixedRateTimer:

var currentTime: LocalDateTime = LocalDateTime.now()

fixedRateTimer(period = 5000L) {
    val newTime = LocalDateTime.now()
    if (currentTime.minute != newTime.minute) {
        post { // post the below code to the UI thread to update UI stuff
            setDateTime(newTime)
        }
        currentTime = newTime
    }
}

enter image description here

enter code here
private val updateLiveShowTicker = flow {
    while (true) {
        emit(Unit)
        delay(1000L * UPDATE_PROGRAM_INFO_INTERVAL_SECONDS)
    }
}

private val updateShowProgressTicker = flow {
    while (true) {
        emit(Unit)
        delay(1000L * UPDATE_SHOW_PROGRESS_INTERVAL_SECONDS)
    }
}

private val liveShow = updateLiveShowTicker
    .combine(channelId) { _, channelId -> programInfoRepository.getShow(channelId) }
    .catch { emit(LiveShow(application.getString(R.string.activity_channel_detail_info_error))) }
    .shareIn(viewModelScope, SharingStarted.WhileSubscribed(), replay = 1)
    .distinctUntilChanged()

My solution,You can now use the Flow API to create your own ticker flow: