Hey guys how do run flows concurrently then collect the outp kotlinlang #coroutines

@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")
    }

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

07/15/2020, 1:31 AM

provide a parameter to

flatMapMerge

for a concurrency level other than 1

Luis Munoz

07/15/2020, 1:32 AM

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

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

octylFractal

07/15/2020, 1:33 AM

oh, I see

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

scope

then

for some reason I recalled

flatMapMerge

as defaulting to 1

Luis Munoz

07/15/2020, 1:34 AM

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

octylFractal

07/15/2020, 1:34 AM

what you probably want is to do

flow { ... }.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

Luis Munoz

07/15/2020, 1:34 AM

but doesn't work with thread sleep

octylFractal

07/15/2020, 1:34 AM

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

Luis Munoz

07/15/2020, 1:35 AM

thank you that works

julian

07/15/2020, 1:58 AM

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

Luis Munoz

07/15/2020, 1:59 AM

@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

07/15/2020, 2:02 AM

Thanks @Luis Munoz!

Zach Klippenstein (he/him) [MOD]

07/15/2020, 3:42 PM

Any reason you're using

sleep

instead of

delay

? Unless you're actually trying to simulate IO that is almost always wrong.

Luis Munoz

07/21/2020, 1:00 AM

exactly wanted to simulate a blocking IO call

6 Views

#coroutines Join Slack