combineLatest(Iterable<Flow>) operator

[Thomas-Vos]

Please add a combineLatest operator which accepts a list of Flows. Here is an example of my use case:

fun <T> combineLatest(flows: Iterable<Flow<T>>): Flow<List<T>> = TODO()

// A list of connected devices.
val devices: List<Device> = getDevices()

// A list of Flows which receive the state of the device every time it changes.
val flows: List<Flow<DeviceState>> = devices.map { device ->
    device.observeState() // Flow<DeviceState>
}

// Combine the latest state of the devices into a single Flow.
val flow: Flow<List<DeviceState>> = combineLatest(flows)


// Somewhere else in the code
flow.collect { states ->
    for (state in states) {
        // TODO: do something with the states.
    }
}

[qwwdfsad]

What do you expect this method to return, Flow<Array<Any?>>?

[zach-klippenstein]

For consistency with the other overloads of this operator, it should probably take a function parameter that accepts the combined array or list and returns an arbitrary type. This is also how RxJava's combineLatest(Iterable) operator works.

[Thomas-Vos]

I agree with @zach-klippenstein, it should take a function parameter instead. Something like this would be better:

fun <T, R> combineLatest(
    flows: Iterable<Flow<T>>, 
    transform: suspend (Array<T>) -> R
): Flow<R>

[qwwdfsad]

I see no real reasons not to do it.
Previously, we have a discussion of whether we should support flow.combineLatest(other) and combineLatest(flow, other) and decided to provide only the first one for the sake of discoverability in IDEA (also, varargs are more convenient for the former API shape)

I let this issue to stay for a while (to see if there are people interested in it) and fix it.

[PaulWoitaschek]

I find flow.combineLatest(other) only useful for that single overload. If there are more than 2 sources I do strongly prefer a factory method.

For example if we take this one:

kotlinx.coroutines/kotlinx-coroutines-core/common/src/flow/operators/Zip.kt

Lines 120 to 134 in f77533c

	public inline fun <T1, T2, T3, T4, T5, R> Flow<T1>.combineLatest(
	other: Flow<T2>,
	other2: Flow<T3>,
	other3: Flow<T4>,
	other4: Flow<T5>,
	crossinline transform: suspend (T1, T2, T3, T4, T5) -> R
	): Flow<R> = (this as Flow<*>).combineLatest(other, other2, other3, other4) { args: Array<*> ->
	transform(
	args[0] as T1,
	args[1] as T2,
	args[2] as T3,
	args[3] as T4,
	args[4] as T5
	)
	}

This makes no sense for me. It's not that I want to combine source A with the latest of B,C,D,E. Instead all these sources are from the user perspective equally important.
If we look at this signature, it implies as if the receiver flow is different from the flows passed as arguments.

[ZakTaccardi]

Just to add on, I find the first example below easier to read than the second.

combineLatest(
  flow1,
  flow2,
  flow3
) { .. }

flow1.combineLatest(
  flow2,
  flow3
) { .. }

flow1 has no additional significance over flow2, so I find seeing them written in parallel easier to read because they emit in parallel