If I run `myFlow.collect { /* some work */ }` , is that work being done concurrently? If so, on how ...

# getting-started

[Tianyu Zhu]

If I run myFlow.collect { /* some work */ } , is that work being done concurrently? If so, on how many threads? How do I control it?

[MR3Y]

collect() is suspend fun ,so, it will suspend the caller thread until some work is done, if you want to concurrently collect some work you might be interested in launchIn(scope: CoroutineScope) instead of collect()

[Joffrey]

The lambda passed to collect is called sequentially on each element of the flow in the current coroutine. If you don't launch coroutines inside the collect lambda, then the whole thing will be sequential

[Joffrey]

@MR3Y using launchIn will still not process the flow in multiple coroutines - only one will be launched

[Tianyu Zhu]

@Joffrey Can you show me an example of launching coroutines inside a collet lambda?

[Tianyu Zhu]

I was able to do something like

<http://Dispatchers.IO|Dispatchers.IO> {
  myFlow
    .map { async { /* some work */ } }
    .buffer()
    .map { it.await() }
    .collect { println(it) }
}

[MR3Y]

@Joffrey This is true, a new one will be launched but I think there is misunderstanding here, If I understand his question correctly, he is asking if collect {} is being executed on another thread (or a pool of threads) concurrently and the the caller thread would continue to execute the next code block, but collect {} will just suspend the current thread until the work is done that is why I suggested launchIn that takes a scope parameter in which he can configure the scope to collect the flow in a different pool of threads and even control the number of threads in that pool. I hope you get my POV and feel free to correct me if I'm wrong.

[Joffrey]

@MR3Y yep what you said is correct. I just think the OP was asking about processing the elements of the flow concurrently, not about processing the flow concurrently with some other work (the new code from the op seems to confirm that)