对每个 DB 条目执行请求

Question

我正在尝试将使用 Retrofit 进行 API 请求的现有工作代码与 Android 中的 SQLite Room 集成。

val apiInterface = ApiInterface.create()
val requests = ArrayList<Observable<Business>>()
requests.addAll(DB.getAllBusinessEntry().map { // fun getAllBusinessEntry(): List<BusinessEntry>
    apiInterface.getBusiness(it.major, it.minor)
})
Observable.zip(requests)
{ objects ->
     val businesses = ArrayList<Business>()
     objects.forEach { businesses.add(it as Business) }
     return@zip businesses
}
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe {
    businessAdapter.updateElements(it)
}

BusinessEntry 是一个包含主键（major 和 minor）的数据类，允许执行 API 请求以获取业务详细信息（存储在 Business 中）。 DB.getAllBusinessEntry() 是一个函数，它使用 Android 文档中记录的 DB Helper 类来访问应用程序内部的 SQLite 数据库。

我想将此代码迁移到 Room 并依赖 RxJava 功能，但我不知道如何在 ApiInterface 调用中转换每个 BusinessEntry。

下面是BusinessDao接口

import androidx.room.Dao
import androidx.room.Query
import io.reactivex.rxjava3.core.Observable

@Dao
interface BusinessDao {
    @Query("SELECT * FROM Entry")
    fun getAll(): Observable<List<Entry>>
}

我正在使用 Room 2.3.0-alpha02

---

编辑：根据 cmets 的建议，我已经实现了这一点，但仍然无法正常工作，因为在最后的 subscribe 调用 businessList 是 List<Observable<Business>!>! 类型，而它应该是 @ 987654334@

swipeRefreshLayout?.isRefreshing = true
val apiInterface = ApiInterface.create()

businessDB.businessDao().getAll()
    .flatMapSingle { entries ->
        Observable.fromIterable(entries)
            .map { apiInterface.getBusiness(it.major, it.minor) }
            .toList()
    }
    .map { it.filter { true } }
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { businessList ->
        swipeRefreshLayout?.isRefreshing = false
        businessAdapter.updateElements(businessList) // <-- type error here
    }

Answer 1

我不是 RxJava 专家，但我可以建议尝试一下：

roomDb.businessDao().getAll()
    .flatMap { entryList -> Observable.fromIterable(entryList) }
    .flatMap { entry -> apiInterface.getBusiness(entry.major, entry.minor) }
    .toList()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { businessList -> businessAdapter.updateElements(businessList) }

其中roomDb - 是 Room 数据库对象的实例，roomDb.businessDao() 为您提供实现 BusinessDao 接口的自动生成类的实例（当然，如果您已将方法 businessDao() 添加到 Room 数据库类） .您可以阅读documentation 了解更多信息。

更新

上面使用的toList() 运算符存在问题，因为 Room 的 Observable 创建了一种无限流，它等待不断更新的数据并且它永远不会完成。在this answer 找到的解决方案。

businessDB.businessDao().getAll()
    .flatMapSingle { entries ->
        Observable.fromIterable(entries)
            .flatMap { apiInterface.getBusiness(it.major, it.minor) }
            .toList()
    }
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { businessList ->
        swipeRefreshLayout?.isRefreshing = false
        businessAdapter.updateElements(businessList)
    }