Hey guys how do run flows concurrently then collec...

# coroutines

[Luis Munoz]

Hey guys how do run flows concurrently then collect the output

@Test
    fun `experimental`() {
        runBlocking {
            val scope = CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>) 
            val songs = listOf("a", "b", "c", "d", "e")
            songs.asFlow().flatMapMerge {
                requestFlow("${Thread.currentThread().name} $it check")
            }
            .collect {
                println("$it")
            }
        }
    }

    fun requestFlow(i: String): Flow<String> = flow {
        emit("$i: First")
        Thread.sleep(1000)
        emit("$i: Second")
    }

[Luis Munoz]

main @coroutine#2 a check: First
main @coroutine#2 a check: Second
main @coroutine#2 b check: First
main @coroutine#2 b check: Second
main @coroutine#2 c check: First
main @coroutine#2 c check: Second
main @coroutine#2 d check: First
main @coroutine#2 d check: Second
main @coroutine#2 e check: First
main @coroutine#2 e check: Second

[octylFractal]

provide a parameter to flatMapMerge for a concurrency level other than 1

[Luis Munoz]

songs.asFlow().flatMapMerge(16)  makes no difference it already defaults to 16

[Luis Munoz]

public val DEFAULT_CONCURRENCY = systemProp(DEFAULT_CONCURRENCY_PROPERTY_NAME,
    16, 1, Int.MAX_VALUE
)

[octylFractal]

oh, I see

[octylFractal]

I presume the issue is that you're not using that scope then

[octylFractal]

for some reason I recalled flatMapMerge as defaulting to 1

[Luis Munoz]

when I do delay(1000) it seems to do what I want

[octylFractal]

what you probably want is to do flow { ... }.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

[Luis Munoz]

but doesn't work with thread sleep

[octylFractal]

otherwise you're executing in `runBlocking`'s dispatcher

[Luis Munoz]

thank you that works

[julian]

@Luis Munoz Would you mind posting the final working version of your code?

[Luis Munoz]

@Test
fun `experimental`() {
    runBlocking {
        val songs = listOf("a", "b", "c", "d", "e")
        songs.asFlow().flatMapMerge(16) {
            requestFlow(" $it check").flowOn(<http://Dispatchers.IO|Dispatchers.IO>)
        }.collect {
            println("$it")
        }
    }
}

fun requestFlow(i: String): Flow<String> = flow {
    emit("${Thread.currentThread().name} $i: First")
    Thread.sleep(1000)
    emit("${Thread.currentThread().name} $i: Second")
}

[julian]

Thanks @Luis Munoz!

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

Any reason you're using sleep instead of delay? Unless you're actually trying to simulate IO that is almost always wrong.

[Luis Munoz]

exactly wanted to simulate a blocking IO call