Collect a Flow in parallel

Flow.collectInParallel.kt

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*

suspend fun <T> Flow<T>.collectInParallel(
    concurrency: Int,
    collector: suspend (T) -> Unit
) = coroutineScope {
    val channel: Channel<T> = Channel()
    repeat(concurrency) { launch { for (e in channel) collector(e) } }
    collect { channel.send(it) }
    channel.close()
}

suspend fun <T> Flow<T>.collectInParallel(
    collectors: List<suspend (T) -> Unit>
) = coroutineScope {
    val channel: Channel<T> = Channel()
    collectors.forEach { collector -> launch { for (e in channel) collector(e) } }
    collect { channel.send(it) }
    channel.close()
}