TimeChunkedFlow.kt.diff

suspend fun collect(collector: FlowCollector<List<T>>) = coroutineScope<Unit> {
+   val mutex = Mutex()
    // 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)
+           mutex.withLock {
                // 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()) {
+                   return@launch
+               }
+               collector.emit(values.toList())
+               values.clear()
+           }
        }
    }

    // Collect the upstream flow and add the items to the above `values` list
    upstream.collect {
+       mutex.withLock {
+           values.add(it)
+       }
    }
    
    isFlowCompleted = true
}