I have a (potentially large) list of async operati...

# coroutines

[Tian Tian098]

I have a (potentially large) list of async operations to perform not necessarily in order. What's the best way to run them?

I tried using the Flow.buffer function, which got a decent result

files.asFlow()
  .map { async { download(it) } }
  .buffer(CONCURRENCY)
  .collect { it.await() }

But this runs all the operations in order. In particular, if operation 1 takes really long, and the rest of the operations in the buffer are done, then I won't start any new operations until operation 1 is done.

Is there a better way?

[ephemient]

files.asFlow()
  .flatMapMerge(CONCURRENCY) {
    flow {
      download(it)
    }
  }
  .collect()

[Tian Tian098]

Thank you!

[debop]

consider chunked also

[renatomrcosta]

You could also potentially use this concept to create your own operator as needed: https://kt.academy/article/pattern_for_composing_flows

But if flatmapmerge fits the bill, better to use the available api

[efemoney]

Why do you need flows for this? 🤔 Seems strange you would not just do

val dispatcher = Dispatcher.IO[or Default or Whatever].limitedParallelism(CONCURRENCY)

files.map { 
  async(dispatcher) {
    download(it) 
  }
}.awaitAll()

as thats the direct equivalent of what you explained

(this is important because there is a proliferation of unnecessary flow operations in the wild which are modelled better as just simple routines)