ChunkedThrottled.kt

import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlin.time.Duration

fun <T> Flow<T>.chunkedThrottled(interval: Duration, bufferSize: Int = 1024): Flow<List<T>> = channelFlow {
    val mutex = Mutex()
    val buffer = ArrayList<T>(bufferSize)

    coroutineScope {
        launch {
            while (isActive) {
                mutex.withLock {
                    if (buffer.isNotEmpty()) {
                        send(buffer.toList())
                        buffer.clear()
                    }
                }

                delay(interval)
            }
        }

        collect {
            mutex.withLock {
                buffer.add(it)

                if (buffer.size == bufferSize) {
                    send(buffer.toList())
                    buffer.clear()
                }
            }
        }
    }
}