TimeChunkedFlow.kt

suspend fun collect(collector: FlowCollector<List<T>>) = coroutineScope<Unit> {
    // For storing un-emitted values
    val values = mutableListOf<T>()

    // Continue looping after intervals `duration` and emit the items in the collector
    // and clear the existing items from the `values`.
    launch {
        while (true) {
            delay(duration)
            
            collector.emit(values.toList())
            values.clear()
        }
    }

    // Collect the upstream flow and add the items to the above `values` list
    upstream.collect {
        values.add(it)
    }
}