I'm trying this with something like ```inline fun <T, R> Flow<T>.concurrentMap( con...

# coroutines

[Shreyas Patil]

I'm trying this with something like

inline fun <T, R> Flow<T>.concurrentMap(
    concurrency: Int,
    @BuilderInference crossinline block: suspend (T) -> R
) = flow<R> {
    val scope = CoroutineScope(Executors.newFixedThreadPool(concurrency).asCoroutineDispatcher())
    val deferredResults = mutableListOf<Deferred<R>>()
    collect { value ->
        deferredResults.add(scope.async { block(value) })
    }
    deferredResults.awaitAll().forEach { newValue -> emit(newValue) }
    scope.cancel()
}

[Sam]

I think you could try flatMapMerge for this

[Sam]

There's a long thread on the GitHub issue: https://github.com/Kotlin/kotlinx.coroutines/issues/1147

[Shreyas Patil]

Cool, let me check

[Shreyas Patil]

flatMapMerge it is

inline fun <T, R> Flow<T>.concurrentMap(
    concurrency: Int,
    @BuilderInference crossinline block: suspend (T) -> R
) = flatMapMerge(concurrency) { value -> flow { emit(block(value)) } }

[Sam]

👌🐕