Hi guys, I have a `List<Flow<T>>` what is the best way to keep collecting from each `Flo...

# coroutines

[Ruben Quadros]

Hi guys, I have a List<Flow<T>> what is the best way to keep collecting from each Flow<T>?

[Joffrey]

What do you mean by that? You mean you want to launch N corresponding coroutines to collect each of those flows concurrently?

[Ruben Quadros]

Can I do it in a single coroutine?

coroutineScope.launch {
    list.forEach {
        eachFlow.collect {
           //do something
        }
    }
}

[Joffrey]

This means the first flow has to complete before the next one starts. If this is what you want/need, then yes of course. But if those flows are infinite, nope

[Ruben Quadros]

If I do not want to wait and if another flow in the list emits something I want to perform some action. How can I achieve this?

[Joffrey]

ah I see, you want to merge the flows into a single steam of Ts and process elements as they come from all the flows, right?

[Ruben Quadros]

I want to collect them individually instead of merging T1, T2, T3 into T

[Javier]

list.forEach { eachFlow ->
    coroutineScope.launch {
        eachFlow.collect {
           //do something
        }
    }
}

[archecraft]

Maybe something like

list.forEach { flow ->
    flow.onEach {
        // do something
    }.launchIn(coroutineScope)
}

?

[Joffrey]

Yeah that's pretty much what I was suggesting the first message.

@Ruben Quadros are those flows of different types? I don't get why you don't want merge() here. It doesn't combine elements of different flows into one, it just collects them all in a single flow:

list.merge().collect {
    // do stuff on each element, which could come from any flow
}

This abstracts away the multiple coroutines, and doesn't even require a scope, unless you want to launch this multi-collection concurrently as well

[Ruben Quadros]

Yes merge makes sense will try it out and let you know

[Ruben Quadros]

merge waits for all flows to flatten the List<Flow<T>> right? I think what ArcheCraft suggested makes more sense for my usecase

[Joffrey]

merge() doesn't wait for anything, it returns immediately a Flow<T>. If you then collect this flow directly, collect will suspend until all flows are done.

If you want that to be asynchronous, you can of course run it in a coroutine:

coroutineScope.launch {
    list.merge().collect {
        // do stuff
    }
}

But yes the other 2 solutions above work fine too

[Joffrey]

Also if you are using a multi-threaded dispatcher, the option with merge will not perform the operations on the items in parallel, while the options launching N coroutines will.