链接多个调用 RxJava、Room 和 Retrofit

Posted 2023-04-18

【中文标题】链接多个调用 RxJava、Room 和 Retrofit

【英文标题】：Chaining multiple calls RxJava, Room and Retrofit

【发布时间】：2018-06-16 01:01:29

【问题描述】：

我正在使用带有 Room 和 Retrofit 的 RxJava/Kotlin。我确定我刚开始学习 RxJava 时不会写东西。场景是我调用检查数据库中是否有收藏记录并将它们放入列表中，从 API 获取数据并将其插入数据库，使用以前的收藏列表更新数据库并获取所有记录，现在更新，列表。我在我的片段中得到结果，但每次得到它时，就好像我少了 1 个最喜欢的项目，直到我没有最喜欢的项目。

Repository

fun getKafaniFromApi(): Observable<List<Kafana>> 
    return apiService.getKafani().toObservable().doOnNext 
        insertKafaniInDb(it)
    


fun getKafaniFromDb(): Observable<List<Kafana>> 
    return kafanaDao.getKafani().toObservable()


fun insertKafaniInDb(kafani: List<Kafana>) 
    Observable.fromCallable  kafanaDao.insertAll(kafani) 
            .subscribeOn(Schedulers.io())
            .subscribe 
                Timber.d("Inserted $kafani.size kafani from API in DB...")
            


fun getFavoriteKafani(): Single<List<Kafana>> 
    return kafanaDao.getFavoriteKafani()


fun setKafanaFavorite(kafana: Kafana, isFavorite: Int) 
    return kafanaDao.setFavourite(kafana.name, isFavorite)


fun updateFavoriteKafana(kafana: Kafana) 
    return kafanaDao.updateFavoriteKafana(kafana)

在我的viewmodel

fun get(): Observable<List<Kafana>> 
    return kafanaRepository.getFavoriteKafani()
            .toObservable()
            .doOnNext  kafaniList = it 
            .flatMap  kafanaRepository.getKafaniFromApi() 
            .doOnNext  kafaniList?.forEach  kafanaRepository.updateFavoriteKafana(it)  
            .flatMap  kafanaRepository.getKafaniFromDb() 
            .subscribeOn(Schedulers.io())
            .observeOn(androidSchedulers.mainThread())

我实际上会在我的片段中得到这个列表，但是，正如我所说，它总是会少一个，直到没有。

【参考方案1】：

首先尽量不要依赖副作用，这会让事情变得不可预测..例如这个函数

fun insertKafaniInDb(kafani: List<Kafana>) 
    Observable.fromCallable  kafanaDao.insertAll(kafani) 
            .subscribeOn(Schedulers.io())
            .subscribe 
                Timber.d("Inserted $kafani.size kafani from API in DB...")
            

它的返回类型是Unit，最好将它包含在流中，这是通过转换为Completable来完成的，所以它会是这样的

  fun insertKafaniInDb(kafani: List<Kafana>) 
    return  Observable.fromAction  kafanaDao.insertAll(kafani) 
        .subscribeOn(Schedulers.io())
        .doOnComplete  Timber.d("Inserted $kafani.size kafani from API in DB...") 
  

其他返回 Unit（java 中为 void）的函数应以相同的方式转换为 completabel。所以现在我将尝试在不使用副作用的情况下重写您的逻辑。并解释每个步骤。

fun getUpdatedData(): Single<MutableList<String>>? 
    return kafanaRepository.getFavoriteKafani()
        .toObservable()
        .flatMap  Observable.fromIterable(it)  //to iterate on favorite items
        .flatMap  localItem ->
          kafanaRepository.getKafaniFromApi()
              .flatMap  Observable.fromIterable(it)  //to iterate api items
              .filter  localItem == it  //search for the favorite item in Api response
              .flatMap 
                //we update this item then we pass it after update
                kafanaRepository.updateFavoriteKafana(it)
                    .andThen(Observable.just(it))
              
              .defaultIfEmpty(localItem) //if it's not found, then no update needed we take this it.

        .toList() // we collect the updated and non updated local items to list
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
  

希望这可以帮助你。 最后的话.. Rx 是关于整理您的想法并以功能方式插入您的逻辑.. 尽量避免使用onNext() 来更新全局变量，仅将其用于日志记录和非业务逻辑。