kotlin coroutines: how to avoid multiple calls of map() if we have 2 or more collectors

Question

There is code sample:

fun main() {
    runBlocking {
        val sourceFlow = MutableSharedFlow<Int>()

        val mappedFlow = sourceFlow
            .map {
                println("INSIDE MAP")
                return@map it.toString()
            }


        launch {
            mappedFlow.collect {
                println(it)
            }
        }

        launch {
            mappedFlow.collect {
                println(it)
            }
        }

        launch {
            delay(1000)
            sourceFlow.emit(1)
        }
    }
}

I have one sourceFlow with mapping and 2 collectors. this program will print:

INSIDE MAP
1
INSIDE MAP
1

In real case I have sourceFlow with some mapping from database, and it is heavy operation. I want to avoid double-mapping for the same value.

For first attemption I created this fun:

suspend fun <R> Flow<R>.flatten(scope: CoroutineScope): Flow<R> {
    val flow = MutableSharedFlow<R>()
    scope.launch {
        collect {
            flow.emit(it)
        }
    }
    return flow
}

To use it like this:

val mappedFlow = sourceFlow
    .map {
        println("INSIDE MAP")
        return@map it.toString()
    }.flatten(this)

Does Kotlin coroutines some other way to implement it? Does anybody see any problems with offered solution?

Answer 1

You have recreated the behavior of the base library's shareIn flow operator function with started set to SharingStarted.Eagerly and replay set to 0, and the result upcast to a generic Flow. Also, your function is needlessly marked suspend. Like any other flow operator, there's no reason it should need to suspend. The name "flatten" could be confusing since the behavior is nothing like the List.flatten, Flow.flattenConcat, and Flow.flattenMerge functions.