对每个DB条目执行请求

Question

我正在尝试将现有的工作代码与Android中的SQLite Room集成在一起，该代码使用针对API请求的改装。

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是包含允许执行API请求以获取业务详细信息(存储在Business中(的主键(major和minor(的数据类。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>>
}

我使用的是2.3.0-alpha02 房间

---

EDIT：根据评论的建议，我已经实现了这一点，但仍然不起作用，因为在最终的subscribe调用中，businessList的类型是List<Observable<Business>!>!，而它应该是List<Business>

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数据库类中(。您可以在文档中阅读更多关于这方面的信息。

更新

上面使用的toList()运算符有一个问题，因为Room的Observable创建了一个无限流，等待数据的不断更新，但它永远不会完成。在这个答案中找到了解决方案。

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