如何使用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()
}

- Androider2

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()
}

- John O'Reilly

---

有没有不使用实验性API的方法？ :) - Androider2

Answer 2

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

但是，你可以使用callbackFlow 来解决这种情况，它专门将监听器/回调转化为协程流。

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

- Giorgos Neokleous

---

offer()已被弃用。 - Ananthakrishnan K R