Does anyone know the best practice way to do a fla...

# coroutines

[Christophe Smet]

Does anyone know the best practice way to do a flatmapMerge on a Flow that runs every map in parallel ? It seems to be sequential by default.

[Zach Klippenstein (he/him) [MOD]]

It shouldn’t be sequential. flatMapMerge takes a concurrency parameter that defines how many upstream Flows may be collected simultaneously: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flat-map-merge.html

[Zach Klippenstein (he/him) [MOD]]

DEFAULT_CONCURRENCY is 16 by default, but may be overridden by a system property: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-d-e-f-a-u-l-t_-c-o-n-c-u-r-r-e-n-c-y.html

[Christophe Smet]

Then I don’t grasp what they mean with

This operator calls transform sequentially

My code maps a list of ip address to a rest result:

return generateAddressesInSubnet(preferredInterfaceAddress) //returns a list of inetAddresses
            .asFlow()
            .flatMapMerge() {
                val url = "http://${it.hostAddress}/status"
                try {
                    logger.debug { "Connecting to $url" }
                    val apiResponse = discoveryApiService.fetchStatus(url)
                    flowOf(LanDiscoveryResult(it, apiResponse))
                } catch (ex: Exception) {
                    emptyFlow<LanDiscoveryResult>()
                }
            }

These rest calls should run in parallel with that code right ?

[octylFractal]

it calls the transform sequentially, but collects the flows concurrently. you should put the work inside of the flow

[Christophe Smet]

Thanks @Zach Klippenstein (he/him) [MOD] & @octylFractal ! You guys are absolutely right.
It works flawlessly include the concurrency parameter.
My understanding of coroutines and flow seems to be lacking the fundamentals, I’ll check out the basics again first. If you have any favourite learning resources, feel free to send a link :)

[Zach Klippenstein (he/him) [MOD]]

This sort of mistake is pretty common and not specific to flow/kotlin. I’ve seen a lot of rxjava code that looks like this:

Observable.just(expensiveFunction())

When the author really wanted this:

Observable.fromCallable { expensiveFunction() }