TimeChunkedFlow.kt.diff

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

    // 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)
            
+           // If the upstream flow has been completed and there are no values
+           // pending to emit in the collector, just break this loop.
+           if (isFlowCompleted && values.isEmpty()) {
+               break
+           }

            collector.emit(values.toList())
            values.clear()
        }
    }

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

    // If we reach here it means the upstream flow has been completed and won't 
    // produce any values anymore. So set the flag as flow is completed so that
    // child coroutine will break its loop
+   isFlowCompleted = true
}