Having some issues testing a flow that emits infin...

# coroutines

[Håkon Pettersen]

Having some issues testing a flow that emits infinite messages:

Class:

class MessageManager(
    dispatcher: CoroutineDispatcher = <http://Dispatchers.IO|Dispatchers.IO>,
) {
    private val messageFlow = flow {
        while (true) {
            delay(60_000)  // Emit every minute
            try {
                val messages = service.getMessages().await()
                emit(messages)
            } catch (error: Exception) {
                Log.e("Messages", "Error fetching messages: $error")
            }
        }
    }.flowOn(dispatcher)
        .shareIn(scope, SharingStarted.Lazily, 1)

    val messages: StateFlow<List<Message>> = messageFlow.stateIn(
        CoroutineScope(Dispatchers.Main),
        SharingStarted.WhileSubscribed(),
        emptyList()
    )
}

Test:

@OptIn(ExperimentalCoroutinesApi::class)
@RunWith(RobolectricTestRunner::class)
class MessageManagerTest {

    private val testScheduler = TestCoroutineScheduler()
    private val application: Application = mockk(relaxed = true)
    private val testDispatcher = StandardTestDispatcher(testScheduler)
    private val mockMessages = listOf(Message("id1", "Test message 1", listOf("mock")))

    @Before
    fun setUp() {
        Dispatchers.setMain(testDispatcher)
    }

    @After
    fun tearDown() {
        Dispatchers.resetMain()
    }

    @Test
    fun `messages emits expected data`() = runTest {
        // Initialize BroadcastManager
        val messageManager = MessageManager(application, testDispatcher)

        // Test the messages flow with Turbine
        messageManager.messages.test {
            // Assert initial state
            assertTrue(awaitItem().isEmpty())

            // Advance time and assert next emission
            advanceTimeBy(60_000)

            assertEquals(mockMessages, awaitItem())

            cancelAndIgnoreRemainingEvents()
        }
    }
}

My expectation is that I should only receive one emitted list of messages before the remaning events gets cancelled, but that is not the case. Currently the messageFlow never stops emitting messages. Am I missing something obvious here?

[hfhbd]

Yes, you catch the CancellationException.

[Håkon Pettersen]

Like this in the messageFlow?

private val messageFlow = flow {
    while (true) {
        delay(60_000)  // Emit every minute
        try {
            val messages = service.getMessages().await()
            emit(messages)
        } catch (error: CancellationException) {
            Log.e("Messages", "Error fetching messages: $error")
        } catch (error: Exception) {
            Log.e("Messages", "Error fetching messages: $error")
        }
    }
}.flowOn(dispatcher)
    .shareIn(scope, SharingStarted.Lazily, 1)

[hfhbd]

Nope, you must rethrow the CE.

[phldavies]

Really you should avoid catching the cancellation and use while(isActive)

[Håkon Pettersen]

Ok, now throwing error:

private val messageFlow = flow {
        while (true) {
            delay(60_000)  // Emit every minute
            try {
                val messages = service.getMessages().await()
                emit(messages)
            } catch (error: CancellationException) {
                throw error
            } catch (error: Exception) {
                Log.e("Messages", "Error fetching messages: $error")
            }
        }
    }.flowOn(dispatcher)
        .shareIn(scope, SharingStarted.Lazily, 1)

From what I can tell the test is still not able to actually cancel anything, still runs until OOM.

[Håkon Pettersen]

I did some modifications. From what I can tell here it only collects the messages twice, as expected with a 120s advance. BUT the messages are still being emitted in the actual messageFlow until it eventually OOM.

@Test
    fun test() = runTest {
        manager = MessageManager(application, testDispatcher)

        val messages = mutableListOf<List<Message>>()

        val job = launch {
            manager.messages.collect {
                println("collecting message")
                messages.add(it)
            }
        }

        advanceTimeBy(120_000)

        job.cancel()
    }

[Håkon Pettersen]

Managed to cancel the messageFlow(hot flow) by explicitly cancelling the scope created in the Manager-class. In the actual usage of the class the messageFlow should never be cancelled, so I guess I will need the method to stop it just for the sake of testing.

[Dmitry Khalanskiy [JB]]

class MessageManager(
    dispatcher: CoroutineDispatcher = <http://Dispatchers.IO|Dispatchers.IO>,
    scope: CoroutineScope = CoroutineScope(Dispatchers.Main)
) {
    val messages: StateFlow<List<Message>> = messageFlow.stateIn(
        scope,
        SharingStarted.WhileSubscribed(),
        emptyList()
    )
}

Then, in the test,

val messageManager = MessageManager(testDispatcher, backgroundScope)

[Dmitry Khalanskiy [JB]]

Also, advanceTimeBy is heavily discouraged, delay(120.seconds) is the preferred solution.

[Håkon Pettersen]

I see, thanks. What are the main differences by using delay instead?

[Dmitry Khalanskiy [JB]]

• It's a suspend function, so it won't block the thread;
• When we introduce new features to the test framework, they may not interoperate with advanceTimeBy well, but are guaranteed to work with `delay`;
• One less thing to learn for anyone who reads the code;
More: https://github.com/Kotlin/kotlinx.coroutines/issues/3919