android - 用于切换方法的 RxJava Operator

Question

我是 Android 项目上的 Rxjava 新手，这里是我的代码

class RadioListRepositoryImpl(private val apiServices: ApiServices, private val chlDao: ChannelDao) : RadioListRepository {

private val results: MutableList<DataResponse>

init {
    results = ArrayList<DataResponse>()
}

override fun getData(): Observable<DataResponse> {
    return dataFromMemory().switchIfEmpty(dataFromNetwork())
}

override fun dataFromMemory(): Observable<DataResponse> {

    val cacheDateExp = DateTime().minusHours(6)

    if(chlDao.isCacheExpired(cacheDateExp).isNotEmpty()){

        Logger.d("Get data from cache SQLITE")

        val chList: MutableList<DataResponse> = ArrayList()

        val cache = chlDao.loadAll()
        repeat(cache.size){ i ->
            val ch = DataResponse()
            ch.channelId = cache[i].channelId
            ch.channelTitle = cache[i].title

            chList.add(ch)
        }

        return Observable.from(chList)
    }else{
        chlDao.deleteAll()

        return Observable.empty<DataResponse>()
    }
}

override fun dataFromNetwork(): Observable<DataResponse> {
    val dttime = DateTime()
    return apiServices.getChannelList()
            .concatMap {
                dataListResponseModel -> Observable.from(dataListResponseModel.radio)
            }
            .doOnNext {
                channelDataResponse -> results.add(channelDataResponse)
            }
            .doOnNext { channelDataResponse ->
                Logger.d("Put data to cache")
                val c: ChannelEntitiy = ChannelEntitiy()
                c.channelId = channelDataResponse.channelId
                c.title = channelDataResponse.channelTitle
                chlDao.insert(c)
            }
}

我的其他类访问方法 getData() 并且我希望如果内存 (sqlite) 中的数据为空，则从网络中获取数据。

但我想要的是，如果内存中的数据为空，则从网络中获取数据插入到内存中，然后 getData() 方法返回 dataFromMemory()

我可以使用另一个 Rx 运算符来处理它以简化我的代码吗？

Answer 1

当您想从多个来源获取数据时，concat 和first 应该适合您。

// Our sources (left as an exercise for the reader)
Observable<Data> memory = ...;  
Observable<Data> disk = ...;  
Observable<Data> network = ...;

// Retrieve the first source with data
Observable<Data> source = Observable  
  .concat(memory, disk, network)
  .first();

concat() 接受多个 Observables 并连接它们的序列。 first() 仅发出序列中的第一项。因此，如果您使用 concat().first()，它会检索多个源发出的第一个项目。

此模式的关键是 concat() 仅在需要时订阅每个子 Observable。如果缓存了数据，则不会对较慢的源进行不必要的查询，因为 first() 会提前停止序列。换句话说，如果内存返回一个结果，那么我们就不会费心去磁盘或网络了。反之，如果内存和磁盘都没有数据，就会重新发起网络请求。

请注意，concat() 中源 Observables 的顺序很重要，因为它会逐一检查它们。

然后，如果您想为每个源保存数据，只需使用 doOnNext() 稍微更改您的源即可

Observable<Data> networkWithSave = network.doOnNext(data -> {  
  saveToDisk(data);
  cacheInMemory(data);
});

Observable<Data> diskWithCache = disk.doOnNext(data -> {  
  cacheInMemory(data);
});