Kotlin Flow：带有回调对象延迟初始化程序的 callbackFlow

问题描述

我想在我的 Android 项目中使用 Kotlin Flow 来使用响应式范例。我有一个基于外部回调的 API，所以我的选择是在我的 callbackFlow 类中使用 Repository。

我已经在没有帮助的情况下深入阅读了一些适当的文档：

• callbackFlow documentation
• Callbacks and Kotlin Flows by Roman Elizarov

我想要达到的目标：

目前我的 Repository 类看起来像这样（简化代码）：

lateinit var callback: ApiCallback

fun someFlow() = callbackFlow<SomeModel> {
    callback = object : ApiCallback {
        override fun someApiMethod() {
            offer(SomeModel())
        }
    }

    awaitClose { Log.d("Suspending flow until methods aren't invoked") }
}

suspend fun someUnfortunateCallbackDependentCall() {
    externalapiclient.externalMethod(callback)
}

当调用 someUnfortunateCallbackDependentCall 比收集 someFlow() 更快时会出现问题。 现在为了避免 UninitializedPropertyAccessException，我在调用 someUnfortunateCallbackDependentCall 之前在我的协程中添加了一些延迟，但这对我来说是一种黑客/代码气味。

我的第一个想法是使用 by lazy 而不是 lateinit var 因为这是我想要的 - 回调对象的延迟初始化。但是，我无法完全对其进行编码。我想从 someApiMethod 发出/提供/发送一些数据以形成数据流，但在 callbackFlow 之外将需要 ProducerScope 在其中。另一方面，someUnfortunateCallbackDependentCall 根本不是基于 Kotlin Flow（最多可以使用 Coroutines API 暂停）。

可以吗？也许使用其他一些 Kotlin 代表？任何帮助将不胜感激。

解决方法

从技术上回答你的问题，你当然可以懒惰地或使用lateinit初始化一个回调，但你不能这样做并且同时共享协程范围（一个用于流程，一个用于挂起函数） -您需要自己构建某种同步。

下面我对您要实现的目标做了一些假设，也许它们并不适合您，但希望能提供一些改进方法。

由于它是您正在创建的存储库，因此我首先假设您希望存储 SomeModel 并允许应用程序的其余部分观察对它的更改。如果是这样，最简单的方法是使用 MutableStateFlow 属性而不是 callbackFlow：

interface Repository {
    val state: Flow<SomeModel>
    suspend fun reload()
}

class RepositoryImpl(private val service: ApiService) : Repository {

    override val state = MutableStateFlow(SomeModel())

    override suspend fun reload() {
        return suspendCoroutine { continuation ->
            service.callBackend(object : ApiCallback {
                override fun someApiMethod(data: SomeModel) {
                    state.value = data
                    if (continuation.context.isActive)
                        continuation.resume(Unit)
                }
            })
        }
    }
}

interface ApiCallback {
    fun someApiMethod(data: SomeModel)
}

data class SomeModel(val data: String = "")

interface ApiService {
    fun callBackend(callback: ApiCallback)
}

此解决方案的缺点是您必须调用 reload() 才能实际调用后端，仅收集 Flow 是不够的。

myrepository.state.collect {} 
myrepository.reload()

另一种解决方案，同样取决于您要实现的具体目标，是提供两种调用后端的方法：

interface Repository {
    fun someFlow(): Flow<SomeModel>
    suspend fun reload(): SomeModel
}

class RepositoryImpl(private val service: ApiService) : Repository {

    override fun someFlow() = callbackFlow<SomeModel> {
        service.callBackend(object : ApiCallback {
            override fun someApiMethod(data: SomeModel) {
                offer(data)
            }
        })
        awaitClose {
            Log.d("TAG","Callback Flow is closed")
        }
    }

    override suspend fun reload(): SomeModel {
        return suspendCoroutine<SomeModel> { continuation ->
            service.callBackend(object : ApiCallback {
                override fun someApiMethod(data: SomeModel) {
                    if (continuation.context.isActive)
                        continuation.resume(data)
                }
            })
        }
    }
}

interface ApiCallback {
    fun someApiMethod(data: SomeModel)
}

data class SomeModel(val data: String = "")

interface ApiService {
    fun callBackend(callback: ApiCallback)
}

现在您可以调用 reload() 或 someFlow() 来检索 SomeModel() 并且 Repository 没有“状态”。

请注意，reload() 函数只是 callbackFlow 想法的“协程”版本。