I want to combine several flows into a single one with a transformation, but some of my flows may not emit anything right away. The

combine

method only emits from the resultant flow after all the source flows have emitted something, but I need a result as soon as there is one available. I am fine transforming nulls or empty lists as default values for flows that haven't produced. Is there anything out there that can combine several flows but emit after the first source flow emits? I've written the following

inline fun <reified T, R> combineIterative(
    flows: Iterable<Flow<T>>,
    crossinline transform: suspend (Array<T?>) -> R
) = channelFlow {
  val latest = arrayOfNulls<T>(flows.count())
  flows.forEachIndexed { i, flow ->
    launch {
      flow.collect {
        latest[i] = it
        send(transform(latest))
      }
    }
  }
}

It emulates the behavior of

merge

but tracks the latest values and emits the whole list rather than just the most recent output. Are there any major problems you can see with my approach? Am I missing some super simple helper that does exactly what I'm looking for?