Flow message not delivered in unit test

Issue

I have a consumer that reads messages off MutableSharedFlow (which acts as an EventBus in my application). I am trying to write a unit test to show that passing a message into the Flow triggers my Listener.

This is my Flow definition:

class MessageBus {

    private val _messages = MutableSharedFlow<Message>()
    val messages = _messages.asSharedFlow()

    suspend fun send(message: Message) {
        _messages.emit(message)
    }
}

Here is the Listener:

class Listener(private val messageBus: MessageBus) {

    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())

    init {
        scope.launch {
            messageBus.messages.collectLatest { message ->
                when (message) {
                    is CustomMessage -> handleCustomMessage(message)
                }
            }
        }
    }

And finally here is my unit test:

class CommandTest {
    @Test
    fun `should process CustomMessage`(): Unit = runBlocking {
        val messageBus = MessageBus()
        val listener = Listener(messageBus)
        messageBus.send(CustomMessage("test command"))
        //argumentCaptor...verify[removed for brevity]
    }
}

Unfortunately the above code does not trigger the break point in my Listener (breakpoint on line init is triggered, but a message is never received and no breakpoints triggered in the collectLatest block).

I even tried adding a Thread.sleep(5_000) before the verify statement but the result is the same. Am I missing something obvious with how coroutines work?

Edit: if it matters this is not an Android project. Simply Kotlin + Ktor

Solution

I imagine that since the code is in the init block in the Listener once you initialize val listener = Listener(messageBus, this) in the test it reads all messages and at this point you have none then in the next line you emit a message messageBus.send(CustomMessage("test command")) but your launch block should have finished by then. You can emit the message first or place your launch in an loop or in a different method that can be called after you emit the message

Answered By - Alain Plana