如何使用 Kotlin Flow 从异步回调中发出数据？

Question

我开始学习 Kotlin Flow 和 Coroutines，但我不知道如何使下面的代码起作用。我究竟做错了什么？

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

我相信你应该能够使用callbackFlow......类似的东西：

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

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

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

Answer 2

你想要实现的目标是不可能的，因为它emit是一个suspend函数。

但是，您可以使用专为此类情况设计的callbackFlow将侦听器/回调转换为协程的流。

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