Kotlin flatMapMerge inconsistent when combining same flow multiple times

Issue

I have an app that flatMapMerges a shared flow A to create a flow B. I then combine Flow A with Flow B to get a final result that I collectLatest. Sometimes it outputs all the results. However, for other times, it skips some of the output. Here's a simplified example:

val flowA = MutableSharedFlow<Int>(replay = Int.MAX_VALUE)
val flowB = flowA.flatMapMerge {
    // In the real app, this flow is created in a complex way
    flow {
        emit(it + 2)
        emit(it + 3)
        emit(it + 4)
        emit(it + 5)
        emit(it + 6)
        emit(it + 7)
        emit(it + 8)
        emit(it + 9)
    }
}
flowA.emit(1)
flowA.combine(flowB, ::Pair).collect { (a, b) ->
    // I need access to the value of flowA for calculations here
    Log.d(TAG, "result A: $a, B: $b")
}

1st run:

result A: 1, B: 3
result A: 1, B: 5
result A: 1, B: 7
result A: 1, B: 9
result A: 1, B: 10

2nd run:

result A: 1, B: 3
result A: 1, B: 5
result A: 1, B: 7
result A: 1, B: 8
result A: 1, B: 10

Expected output:

result A: 1, B: 3
result A: 1, B: 4
result A: 1, B: 5
result A: 1, B: 6
result A: 1, B: 7
result A: 1, B: 8
result A: 1, B: 9
result A: 1, B: 10

I also tried adjusting the replay, but it has no effect. I also tried collectLatest instead of collect, but also no change.

Should I be using something other than flatMapMerge? I tried flatMapLatest, but that only outputs the first and the last result. Should the MutableSharedFlow be something else?

How do I get all the results to output every time?

Solution

The problem here is with combine function

as of official documentation, combine function:

Returns a Flow whose values are generated with transform function by combining the most recently emitted values by each flow.

If transform function of combine() has some heavy calculation inside it, it is possible that combine() doesn't emit all results.

So for obtaining all results, the below strategy works fine.

val flowA = MutableSharedFlow<Int>()
        .onStart { emit(1) }

    val flowB = flowA.flatMapMerge {
        flow {
            for (i in 2..9) {
                emit(Pair(it, i))
            }
        }
    }

    runBlocking {
        flowB.collect { (a, b) ->
            println("result A: $a, B: $b")

        }
    }

Output:

result A: 1, B: 3
result A: 1, B: 4
result A: 1, B: 5
result A: 1, B: 6
result A: 1, B: 7
result A: 1, B: 8
result A: 1, B: 9
result A: 1, B: 10

Answered By - Ariya