Kotlin flow: Weird behavior, code executed 3 times while it is supposed to be executed only once

Question

I start recently studying a project which uses kotlin flow + coroutine and I found a weird thing, I write some code to reproduce the issue, here is the code

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.channels.*

sealed class ViewIntent {
    class Initial : ViewIntent()
    class Refresh : ViewIntent()
    class Click: ViewIntent()
}

class ViewState

@kotlinx.coroutines.ExperimentalCoroutinesApi
fun Flow<ViewIntent>.toEvent() : Flow<Int> {
    return merge(
       filterIsInstance<ViewIntent.Initial>().map{1},
       filterIsInstance<ViewIntent.Refresh>().map{2},
       filterIsInstance<ViewIntent.Click>().map{3},   
    )
}

@kotlinx.coroutines.ExperimentalCoroutinesApi
fun main() = runBlocking {
    val intentFlow = listOf(ViewIntent.Initial()).asFlow()
    intentFlow.onEach { println(it) }
    .toEvent()
    .collect()

}

the output of the code is as follow:

ViewIntent$Initial@2ff5659e
ViewIntent$Initial@2ff5659e
ViewIntent$Initial@2ff5659e

so things that confused me is that why ViewIntent$Initial@2ff5659e is shown 3 times? if .toEvent() removed, there are only one ViewIntent$Initial@2ff5659e is shown.

Answer 1

Because Flow are cold streams. Every time you call filterIsInstance, upstream will emits ViewIntent.Initial().

In future, SharedFlow will be added into library, see pull request, we can write like this:

@kotlinx.coroutines.ExperimentalCoroutinesApi
fun Flow<ViewIntent>.toEvent(coroutineScope: CoroutineScope) : Flow<Int> {
    val shared = this.shareIn(
            coroutineScope,
            replay = 0,
            started = SharingStarted.WhileSubscribed()
    )

    return merge(
       shared.filterIsInstance<ViewIntent.Initial>().map{1},
       shared.filterIsInstance<ViewIntent.Refresh>().map{2},
       shared.filterIsInstance<ViewIntent.Click>().map{3},   
    )
}