Hi everyone. I found an issue with broadcast channel, flows and tests:

@Test
    fun `BroadcastChannel keep a coroutine alive`() = runBlockingTest {
        val channel = BroadcastChannel<Int>(Channel.CONFLATED)
        val flow = channel.asFlow()
        launch {
            flow.collect { assert(true) }
            channel.send(1)
        }
    }

this test fails with`kotlinx.coroutines.test.UncompletedCoroutinesError: Test finished with active jobs: ["coroutine#2":StandaloneCoroutine{Active}@2a2d45ba]` it forces me to close the channel for the test to pass. But if i change the

flow

for a

receiveChannel

like this:

@Test
    fun `BroadcastChannel keep a coroutine alive`() = runBlockingTest {
        val channel = BroadcastChannel<Int>(Channel.CONFLATED)
        val receive = channel.openSubscription()
        launch {
            receive.consume { assert(true) }
            channel.send(1)
        }
    }

The test passes without problems

Those 2 code snippets do not have the same behavior : you need to change

consume

to

consumeEach

in the 2nd snippet. Then it should also suspend forever.

Correct. Then, how could I stop this from suspending forever?

Flow.collect

suspends until there are no more elements in the flow, calling your lamdba for each item, at the time an item is emitted. In your case you are producing a

Flow

from a

BroadcastChannel

with

asFlow

, therefore the coroutine calling

collect

will resume when the source

BroadcastChannel

will have no more elements (i.e., when it is closed or fails with an exception)

after fiddling with the code I've found out that for the suspension to break I need to cancel the job returned by launch or close the channel. I just need to make sure that all subscriptions are closed before the end of the test. With my actual code this is easy to achieve for testing purposes and it was already done by default in my production code.