具有 Room 和状态处理功能的 Kotlin 协程流

我正在尝试新的协程流程，我的目标是创建一个简单的存储库，可以从 Web api 获取数据并将其保存到数据库，还可以从数据库返回流程。

我使用 room 和 firebase 作为 Web api，现在一切看起来都非常简单，直到我尝试将来自 api 的错误传递到 ui。

由于我从数据库获取的流仅包含数据而没有状态，因此通过将其与 Web api 结果结合起来为其提供状态（如加载、内容、错误）的正确方法是什么？

我写的一些代码：

The DAO:

@Query("SELECT * FROM users")
fun getUsers(): Flow<List<UserPojo>>

存储库：

val users: Flow<List<UserPojo>> = userDao.getUsers()

API 调用：

override fun downloadUsers(filters: UserListFilters, onResult: (result: FailableWrapper<MutableList<UserApiPojo>>) -> Unit) {
    val data = Gson().toJson(filters)

    functions.getHttpsCallable("users").call(data).addOnSuccessListener {
        try {
            val type = object : TypeToken<List<UserApiPojo>>() {}.type
            val users = Gson().fromJson<List<UserApiPojo>>(it.data.toString(), type)
            onResult.invoke(FailableWrapper(users.toMutableList(), null))
        } catch (e: java.lang.Exception) {
            onResult.invoke(FailableWrapper(null, "Error parsing data"))
        }
    }.addOnFailureListener {
        onResult(FailableWrapper(null, it.localizedMessage))
    }
}

我希望问题足够清楚 谢谢您的帮助

Edit:由于问题不清楚，我会尽力澄清。我的问题是，使用房间发出的默认流，您只有数据，因此如果我要订阅该流，我只会收到数据（例如，在这种情况下，我只会收到用户列表）。我需要实现的是某种方式来通知应用程序的状态，例如加载或错误。目前我能想到的唯一方法是包含状态的“响应”对象，但我似乎找不到实现它的方法。

就像是：

fun getUsers(): Flow<Lce<List<UserPojo>>>{
    emit(Loading())
    downloadFromApi()
    if(downloadSuccessful)
        return flowFromDatabase
    else
        emit(Error(throwable))
}

但我遇到的明显问题是来自数据库的流的类型Flow<List<UserPojo>>，我不知道如何通过编辑流程的状态来“丰富它”，而不会丢失数据库的订阅，也不会在每次更新数据库时运行新的网络调用（通过在地图转换中进行）。

希望更清楚

我相信这更多的是一个架构问题，但让我先尝试回答您的一些问题。

我的问题是，使用房间发出的默认流量，您只能 数据，所以如果我订阅该流，我只会收到 数据

如果有错误Flow由 Room 返回，您可以通过以下方式处理catch() https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/catch.html

我需要实现的是某种方式来通知应用程序的状态， 例如加载或错误。

我同意你的观点，有一个State对象是一个很好的方法。在我心目中，它是ViewModel的责任是呈现State反对View. This State对象应该有一种方法来暴露错误。

目前我能想到的唯一方法是一个“响应”对象 包含状态，但我似乎找不到实现它的方法。

我发现更容易拥有State反对ViewModel控件对错误负责，而不是从冒泡的对象Service layer.

现在，解决了这些问题，让我尝试为您的问题提出一个特定的“解决方案”。

正如您提到的，通常的做法是Repository处理从多个数据源检索数据。在这种情况下，Repository会采取DAO以及一个代表从网络获取数据的对象，我们称之为Api。我假设你正在使用FirebaseFirestore，所以类和方法签名看起来像这样：

class Api(private val firestore: FirebaseFirestore) {

fun getUsers() : Flow<List<UserApiPojo>

}

现在的问题是如何将基于回调的 API 变成Flow。幸运的是，我们可以使用callbackFlow() https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/callback-flow.html为了这。然后Api变成：

class Api(private val firestore: FirebaseFirestore) {

fun getUsers() : Flow<List<UserApiPojo> = callbackFlow {

val data = Gson().toJson(filters)

functions.getHttpsCallable("users").call(data).addOnSuccessListener {
    try {
        val type = object : TypeToken<List<UserApiPojo>>() {}.type
        val users = Gson().fromJson<List<UserApiPojo>>(it.data.toString(), type)
        offer(users.toMutableList())
    } catch (e: java.lang.Exception) {
       cancel(CancellationException("API Error", e))
    }
}.addOnFailureListener {
    cancel(CancellationException("Failure", e))
    }
  }
}

如你看到的，callbackFlow允许我们在出现问题时取消流程并让下游人员处理错误。

移动到Repository我们现在想做类似的事情：

val users: Flow<List<User>> = Flow.concat(userDao.getUsers().toUsers(), api.getUsers().toUsers()).first()

这里有一些注意事项。first() and concat()看来你必须想出运营商。我没看到版本first()返回一个Flow;它是一个终端操作员（Rx 曾经有一个版本first()返回一个Observable，丹·卢 (Dan Lew) 将其用于this https://blog.danlew.net/2015/06/22/loading-data-from-multiple-sources-with-rxjava/ post). Flow.concat()似乎也不存在。目标是users是返回一个Flow发出任何源发出的第一个值Flows。另请注意，我将 DAO 用户和 Api 用户映射到一个公共的User object.

我们现在可以谈论ViewModel。正如我之前所说，ViewModel应该有可以容纳的东西State. This State应代表数据、错误和加载状态。可以实现的一种方法是使用数据类。

data class State(val users: List<User>, val loading: Boolean, val serverError: Boolean)

由于我们可以访问Repository the ViewModel可以看起来像：

val state = repo.users.map {users -> State(users, false, false)}.catch {emit(State(emptyList(), false, true)}

请记住，这是一个粗略的解释，为您指明方向，有很多方法可以完成状态管理，这绝不是完整的实现。将 API 调用转变为Flow， 例如。