Kotlin Flow Extensions

FlowExtensions.kt

@ExperimentalCoroutinesApi
fun <T> Flow<T>.bufferList(capacity: Int): Flow<List<T>> {
    require(capacity > 0) { "Buffer size should be positive, but was $capacity" }
    var buffer = listOf<T>()
    return transform { value ->
        if (buffer.size == capacity) emit(buffer)
        else buffer = buffer + value
    }
}

@InternalCoroutinesApi
@ExperimentalCoroutinesApi
suspend fun <T> Flow<T>.collectWhenAtLeast(
    count: Int,
    action: (List<T>) -> Unit
) {
    require(count >= 0) { "Buffer size should be non-negative, but was $count" }
    var buffer: List<T> = emptyList()
    var useBuffer = true
    collect { value ->
        when {
            !useBuffer -> action(listOf(value))
            buffer.size == count -> {
                action(buffer)
                useBuffer = false
            }
            else -> buffer = buffer + value
        }

    }
}