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

Question

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

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

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

我写的一些代码：

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

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

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

就像是：

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

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

希望更清楚

Answer 1

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

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

如果 Room 返回的错误Flow，您可以通过以下方式处理：catch()

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

我同意你的观点，拥有一个State对象是一个很好的方法。在我看来，将物品ViewModel呈现给 是 的责任。该对象应该有一种暴露错误的方法。StateViewState

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

我发现让控件负责错误的State对象比从层中冒出的对象更容易。ViewModelService

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

正如您所提到的，通常的做法是使用 来Repository处理从多个数据源检索数据。在这种情况下，将Repository采用DAO和一个表示从网络获取数据的对象，我们将其称为Api。我假设您正在使用FirebaseFirestore，因此类和方法签名将如下所示：

class Api(private val firestore: FirebaseFirestore) {

fun getUsers() : Flow<List<UserApiPojo>

}

现在的问题是如何将基于回调的 API 转换为Flow. 幸运的是，我们可以用callbackFlow()它来实现。那么就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()看来concat()你必须想出运营商。我没有看到first()返回 ;的版本Flow。它是一个终端操作符（Rx 曾经有一个first()返回的版本Observable，Dan Lew 在这篇文章中使用它）。Flow.concat()似乎也不存在。的目标users是返回Flow发出任何源发出的第一个值的a Flows。另请注意，我将 DAO 用户和 Api 用户映射到一个公共User对象。

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

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

由于我们可以访问该内容，Repository因此ViewModel可以如下所示：

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

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