timedBuffer

TimedBuffer.kt

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

private class Buffer<T> {
    private val list = arrayListOf<T>()

    @Synchronized
    fun add(value: T) { list.add(value) }

    @Synchronized
    fun pollIfNotEmpty() =
        list.takeIf { it.isNotEmpty() }?.toList()?.also { list.clear() }
}

fun <T> Flow<T>.timedBuffer(millis: Long): Flow<List<T>> = channelFlow {
    val buffer = Buffer<T>()
    suspend fun flush() { buffer.pollIfNotEmpty()?.let { send(it) } }
    val job = launch {
        while (isActive) {
            delay(millis)
            flush()
        }
    }
    collect { buffer.add(it) }
    job.cancelAndJoin()
    flush()
}