How to emit data from an asycnhronous callback using Kotlin Flow?

Question

I'm starting to learn Kotlin Flow and Coroutines but I do not know how to make the code below works. What am I doing wrong?

interface MessagesListener {
    fun onNewMessageReceived(message: String)
}

fun messages(): Flow<String> = flow {

    val messagesListener = object : MessagesListener {
        override fun onNewMessageReceived(message: String) {

            // The line below generates the error 'Suspension functions can be called only within coroutine body'

            emit(message)
        }
    }

    val messagesPublisher = MessagesPublisher(messagesListener)
    messagesPublisher.connect()
}

Answer 1

I believe you should be able to use callbackFlow ....something like:

fun messages(): Flow<String> = callbackFlow {

    val messagesListener = object : MessagesListener {
        override fun onNewMessageReceived(message: String) {
            trySend(message)
        }
    }

    val messagesPublisher = MessagesPublisher(messagesListener)
    messagesPublisher.connect()
}

Answer 2

What you are trying to achieve is not possible because emit is a suspend function.

However, you can use callbackFlow which is designed for such cases to convert Listeners/Callbacks into Coroutine's Flows.

fun messages() = callbackFlow<String> {
    val messagesListener = object : MessagesListener {
        override fun onNewMessageReceived(message: String) {
            offer(message)
        }
    }
    val messagesPublisher = MessagesPublisher(messagesListener)
    messagesPublisher.connect()
}