I ended up with:

findUserIds()
  .map { async(<http://Dispatchers.IO|Dispatchers.IO>) { getUserInfo(it) } }
  .buffer(10)
  .map { it.await() }

Which seems to do the trick pretty nicely. Any better suggestions?

I don't think this looks right... is there a better way to make this reusable?

suspend fun <T> Flow<T>.mapParallel(concurrency: Int, transform: suspend CoroutineScope.() -> T): Flow<T> = coroutineScope {
	map { async(<http://Dispatchers.IO|Dispatchers.IO>, block = transform) }
		.buffer(concurrency)
		.map { it.await() }
}

?

if you want to limit the concurrency properly, you should remove

buffer

in favor of adding a

Semaphore

around the block (see

ChannelFlowMerge#mergeImpl

internal implementation)

look weird at the first look 🙂

This is what I came up with: https://pl.kotl.in/q_G6Vq0Y5 Uses

flow { coroutineScope {

to establish a new context to run the async tasks in, limits concurrency by using semaphore before submitting a task, then buffers all of the

Deferred

objects (to ensure they are submitted, since flows are cold by default) up to

UNLIMITED

but in practice never larger than

concurrency

. Then finally joins in order to produce the result. In my testing it seemed to be about 200ms slower than it should be, but I chalk that up to setting up the threads + other coroutine machinery. It's a constant startup cost, and in the long run is amortized to zero.