Kotlin Flow Offline Caching

user_16967713 0
Viewed 474

I am new with kotlin flow and I am working about this document. Kotlin Flows. In this code every five seconds datasource fetch data from api and emits it.

This is my example datasource class.

I am getting data and emitting it.

class RemoteDataSourceImpl @Inject constructor(
private val api:CryptoApi
): RemoteDataSource {


override suspend fun cryptoList(): Flow<List<CryptoCoinDto>> {
    return flow {
        while (true){
            
            val data = api.getCoinList()
            emit(data)
            delay(5000L)
        }
    }
   }
}

This is my example repository.

I am mapping data and saving it room database. I want to get data from room database and emit it because of single source of truth principle but I still have to return dataSource because if I open new flow{} I can't reach datasource's data. Of course I can fix the problem by using List instead of Flow<List> inside of RemoteDataSource class. But I want to understand this example. How can I apply here single source of truth.

class CoinRepositoryImpl @Inject constructor(
private val dataSource:RemoteDataSource,
private val dao: CryptoDao
):CoinRepository {

override fun getDataList(): Flow<List<CryptoCoin>> {

     dataSource.cryptoList().map { dtoList ->
        val entityList = dtoList.map { dto ->
            dto.toCryptoEntity()
        }
        dao.insertAll(entityList)
    }
    return dataSource.cryptoList().map {
        it.map { it.toCryptoCoin() }
    }

}
kotlin-flow kotlin repository android
1 Answers

This is actually more complicated than it seems. Flows were designed to support back-pressure which means that they usually only produce items on demand, when being consumed. They are passive, instead of pushing items, items are pulled from the flow.

(Disclaimer: this is all true for cold flows, not for hot flows. But cryptoList() is a cold flow.)

It was designed this way to greatly simplify cases when the consumer is slower than producer or nobody is consuming items at all. Then producer just stops producing and everything is fine.

In your case there are two consumers, so this is again more complicated. You need to decide what should happen if one consumer is slower than the other. For example, what should happen if nobody collects data from getDataList()? There are multiple options, each requires a little different approach:

  1. Stop consuming the source flow and therefore stop updating the database.
  2. Update the database all the time and queue items if nobody is collecting from getDataList(). What if there are more and more items in the queue?
  3. Update the database all the time and discard items if nobody is collecting from getDataList().

Ad.1.

It can be done by using onEach():

return dataSource.cryptoList().onEach {
    // update db
}.map {
    it.map { it.toCryptoCoin() }
}

In this solution updating the database is a "side effect" of consuming the getDataList() flow.

Ad.2. and Ad.3.

In this case we can't passively wait until someone asks us for an item. We need to actively consume items from the source flow and push them to the downstream flow. So we need a hot flow: SharedFlow. Also, because we remain the active side in this case, we have to launch a coroutine that will do this in the background. So we need a CoroutineScope.

Solution depends on your specific needs: do you need a queue or not, what should happen if queue exceeded the size limit, etc., but it will be similar to:

return dataSource.cryptoList().onEach {
    // update db
}.map {
    it.map { it.toCryptoCoin() }
}.shareIn(scope, SharingStarted.Eagerly)

You can also read about buffer() and MutableSharedFlow - they could be useful to you.

user_4488750
Related