How to handle database call errors using Flows

Question

Usually I'm returning from my dao suspend function:

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

And handle the call within the repository:

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)

}

With generic fun:

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

The problem is that I want return fun getAllData(): Flow<List<Data>> instead of fun getAllData(): List<Data> In order to get immediate updates, But if I'm returning Flow from the Dao, I can't handle the call with safe call and catch errors. I thought about collecting the data, but if i'm collecting the data the call already done without error handling

Basically I need the cache result return CacheResult<Data> and not CacheResult<Flow<Data>>

How can I solve the problem And make a generic safeDatabaseCall while returning Flow from Dao?

Answer 1

So if I understand correctly you just want to handle the query and return of information safely in a flow. My only question is around the types. I can sorta assume Data DataState and CacheResult are not the same types so I use a "magic" function that converts the intermediary values to the correct one. You will need to adjust accordingly

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

You shouldn't need flowOn with a dispatcher here since the work inside this flow doesn't require thread dispatching to Dispatcher.IO. The code we are putting in our flow, is purely exception handling and invoking a function. The only place that seems to require any manual dispatch changing is, safeDatabaseCall(). I am not familiar with this function but if it does exist and takes a dispatcher for the result of actualing making the db calls on an IO thread, then all should be good without flowOn. Otherwise you will be switching dispatchers from original dispatcher -> IO and then to IO again. It's not much but the extra no-op context switch doesn't add anything other than confusion later on.

The flow itself traps any upstream issues and you then make them part of the resulting flow