Fast flow combination

FlowCombineLatest.kt

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


private val undefined = Any()

internal fun <T> Iterable<Flow<T>>.combineLatest(): Flow<List<T>> =
    combineLatest { it }


internal fun <T, R> Iterable<Flow<T>>.combineLatest(transform: suspend (value: List<T>) -> R): Flow<R> = flow {
    class Update(@JvmField val index: Int, @JvmField val value: T)

    val flows = toList()
    var remainingCount = flows.size
    var values: MutableList<Any?> = MutableList(remainingCount) { undefined }

    channelFlow {
        coroutineScope {
            flows.forEachIndexed { index, flow ->
                launch { flow.collectLatest { send(Update(index, it)) } }
            }
        }
    }.collect { update ->
        val index = update.index
        val value = update.value

        when {
            remainingCount == 0 -> values = values.toMutableList()
            values[index] === undefined -> remainingCount -= 1
        }

        values[index] = value

        if (remainingCount == 0)
            @Suppress("UNCHECKED_CAST")
            emit(transform(values as List<T>))
    }
}