android - 如何使用 Flows 处理数据库调用错误

Question

通常我从我的 dao 挂起函数返回:

@Dao
interface DataDao {
    @Query("SELECT * FROM data")
    fun getAllData(): List<Data>
}

并处理存储库中的调用:

class DataRepository(
    private val dataDao: DataDao
) {
    fun getAllData(): Flow<DataState> = flow {
        val cacheResult = safeDatabaseCall(dispatcher = Dispatchers.IO) { dataDao.getAllData() }

    //handle cacheResult, convert to DataState, emit DataState values

    }.flowOn(Dispatchers.IO)

}
具有一般的乐趣:

suspend fun <T> safeDatabaseCall(
    dispatcher: CoroutineDispatcher,
    cacheCall: suspend () -> T?
): CacheResult<T?> {
    return withContext(dispatcher) {
        try {
            withTimeout(10000L) {
                CacheResult.Success(cacheCall.invoke())
            }
        } catch (t: Throwable) {
            when (t) {
                is TimeoutCancellationException -> {
                    CacheResult.Error("Timeout error")
                }
                else -> {
                    CacheResult.Error("Unknown error")
                }
            }
        }
    }
}

问题是我想返回 fun getAllData(): Flow<List<Data>>而不是 fun getAllData(): List<Data>为了获得即时更新，但是如果我从 Dao 返回 Flow，我无法使用安全调用处理调用并捕获错误。
我考虑过收集数据，但如果我正在收集数据，那么调用已经完成而没有错误处理
基本上我需要缓存结果返回 CacheResult<Data>而不是 CacheResult<Flow<Data>>如何解决问题并在从 Dao 返回 Flow 时进行通用 safeDatabaseCall？

Answer 1

因此，如果我理解正确，您只想在流程中安全地处理查询和信息返回。我唯一的问题是关于类型。我可以假设 Data DataState和 CacheResult不是相同的类型，所以我使用“魔术”函数将中间值转换为正确的值。您将需要进行相应的调整

class DataRepository(
  private val dataDao: DataDao
) {

  fun getAllData(): Flow<DataState> = flow {
    val result = safeDatabaseCall(dispatcher = Dispatchers.IO) { 
      dataDao.getAllData() 
    }
    // Emit the result
    emit(result)
  }.catch { t : Throwable ->
      // Do our transformation like before
      val result = when (t) {
        is TimeoutCancellationException -> {
          CacheResult.Error("Timeout error")
        }
        else -> {
          CacheResult.Error("Unknown error")
        }
      }
      // And because catch is actually extending a FlowCollector
      // We can emit the result in the stream
      emit(result)
  }.map { cacheResult ->
      convertToDataOrDataState(cacheResult)
  } 

你不应该需要 flowOn这里有一个调度程序，因为这个流程中的工作不需要线程调度
至 Dispatcher.IO .我们放入流程中的代码纯粹是异常处理和调用函数。唯一需要手动更改调度的地方是 safeDatabaseCall() .我不熟悉这个函数，但是如果它确实存在并且需要一个调度程序来实现在 IO 线程上进行数据库调用的结果，那么没有 flowOn 一切都应该是好的。 .否则，您将从原始调度程序 -> IO 切换调度程序，然后再次切换到 IO。这并不多，但是额外的无操作上下文切换除了以后的困惑之外没有增加任何东西。
流程本身会捕获任何上游问题，然后您将它们作为结果流程的一部分