Missing emissions when using combine operator

[utwyko]

I'm getting an unexpected result when checking flow emissions after using the combine operator. Using .test{} results in less emissions than using .collect{}. I expect .test{} to result in the same # of emissions.

If this is indeed considered a bug and not a misuse from my side, I'd be happy to try and contribute a fix.

Reproduction test (based on the combine operator docs):

@Test
fun combineTurbine() = runBlockingTest {
    combine(flow1(), flow2()) { i, s -> i.toString() + s }.collect {
        println("Collect: $it") // Will print "1a 2a 2b 2c"
    }

    combine(flow1(), flow2()) { i, s -> i.toString() + s }.test {
        println("Turbine test: ${expectItem()}") // Will print 2b
        println("Turbine test: ${expectItem()}") // Will print 2c
    }
}

private fun flow1() = flowOf(1, 2)
private fun flow2() = flowOf("a", "b", "c")

Output:

Collect: 1a
Collect: 2a
Collect: 2b
Collect: 2c
Turbine test: 2b
Turbine test: 2c

Running:

• Turbine version 0.4.0
• Kotlin version 1.4.30
• Coroutines version 1.4.1

[andrey-bolduzev]

I think this issue is related to Kotlin/kotlinx.coroutines#2082 and similar.

Running your example locally with coroutines 1.4.2, I'm getting yet another output variant:

Collect: 1a
Collect: 2b
Collect: 2c
Turbine test: 1a
Turbine test: 2b

The combine operator is broken as far as tests are concerned I personally believe.

[JakeWharton]

Specifically, this comment seems to highlight the problem.

I try really, really hard to never use runBlockingTest because it has some wacky behaviors associated with it.

[utwyko]

Thanks for the responses and the link to the coroutines issue. There is indeed a difference between running it in runBlocking vs runBlockingTest, although the output in Turbine is the same in both cases. On my machine, the output is the same for coroutines 1.4.1 and 1.4.2 (could not test 1.4.3 due to #290). I will add extra information to the coroutines ticket and close this one for now.

For completeness, this is the code for the test comparing runBlocking with runBlockingTest using collect() and Turbine.

@Test
fun combineTest() {
    runBlocking {
        combine(flow1(), flow2()) { i, s -> i.toString() + s }.collect {
            println("collect + runBlocking: $it") // Will print "1a 2b 2c"
        }
    }

    runBlockingTest {
        combine(flow1(), flow2()) { i, s -> i.toString() + s }.collect {
            println("collection + runBlockingTest: $it") // Will print "1a 2a 2b 2c"
        }
    }

    runBlocking {
        combine(flow1(), flow2()) { i, s -> i.toString() + s }.test {
            println("Turbine + runBlocking: ${expectItem()}") // Will print 2b
            println("Turbine + runBlocking: ${expectItem()}") // Will print 2c
            expectComplete()
        }
    }

    runBlockingTest {
        combine(flow1(), flow2()) { i, s -> i.toString() + s }.test {
            println("Turbine + runBlockingTest: ${expectItem()}") // Will print 2b
            println("Turbine + runBlockingTest: ${expectItem()}") // Will print 2c
            expectComplete()
        }
    }
}

Output:

collect + runBlocking: 1a
collect + runBlocking: 2b
collect + runBlocking: 2c

collection + runBlockingTest: 1a
collection + runBlockingTest: 2a
collection + runBlockingTest: 2b
collection + runBlockingTest: 2c

Turbine + runBlocking: 2b
Turbine + runBlocking: 2c

Turbine + runBlockingTest: 2b
Turbine + runBlockingTest: 2c