Example of controling execution parallerisms with kotlin flows

example.kt

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import java.util.concurrent.Executors
import kotlin.system.measureTimeMillis

@OptIn(FlowPreview::class)
fun main() = runBlocking {

    val urls = (1..1000).map { "https://example.com/todos/$it" }

    val ioThreadPool = Executors.newCachedThreadPool().asCoroutineDispatcher()

    for(mcr in (3..20).step(2)) {
        val timeMs = measureTimeMillis {
            val result = urls.asFlow()
                .flatMapMerge(mcr) { url ->
                    flowOf(url).flowOn(ioThreadPool)
                        .map {
                            queryRestEndpoint(it)
                        }
                }
                .toList()
            println("Visited ${result.size}")
        }
        println("Total time $timeMs for concurrency: $mcr")
    }
    ioThreadPool.close()

}

suspend fun queryRestEndpoint(url: String): String {
    delay(100L)
    return url
}