所以我是 RxJava 的初学者,但这是我想要完成的:
MainViewModel 与存储库对话。 Repository 同时具有 LocalDataStore(与数据库对话)和 RemoteDataStore(Retrofit)两者都是接口(interface) DataStore 的不同实现。
我想要实现的是从返回 Observable 的 Repository 调用 fetchData 但是:
- 它首先从 RemoteDataStore 中获取它
- 在获取每一个东西(onNext())之后,它将它插入到数据库中
- 如果失败,它会从 LocalDataStore 返回结果。
但是,我不知道如何实现这个逻辑。订阅发生在 ViewModel 的一端,但我无法真正将可观察到的从存储库端(?)更改为 LocalDataStore。将数据更新到数据库中也会返回一个 Observable(准确地说是 Single),它需要订阅才能正常工作。
谁能给我解释一下,或者给我指明一个好的方向?
我的代码(存储库注释中的问题):
远程数据存储
override fun getData(): Observable<SomeData> = api
.getData(token)
.flatMapIterable { x -> x }
本地数据存储
override fun saveData(data: SomeData): Single<SomeData> {
return database.upsert(data)
}
存储库
fun getData(): Observable<SomeData> {
return
remoteDataStore.getData()
.doOnError {
localDataStore.getData() //? this will invoke but nothing happens because I'm not subscribed to it
}
.doOnNext {
saveData(it) //The same as before, nothing will happen
}
}
View 模型
override fun fetchData() {
repository.getData()
.observeOn(androidScheduler)
.subscribeOn(threadScheduler)
.subscribe(
{ data: SomeData ->
dataList.add(data)
},
{ throwable: Throwable? ->
handleError(throwable)
},
{
//send data to view
},
{ disposable: Disposable ->
compositeDisposable.add(disposable)
}
)
}
感谢您的宝贵时间。
最佳答案
您需要使用 onErrorResumeNext 之一方法。我还建议将您的流类型从 Observable
更改为 Single
,因为您的数据性质看起来像一次获取数据或抛出错误。这只是一个很好的 API 设计。
在您的特定情况下,我将以这种方式实现存储库:
class RepositoryImpl @Inject constructor(private val localRepository: Repository, private val remoteRepository: Repository) : Repository {
override fun getData(): Single<Data> = remoteRepository.getData()
.onErrorResumeNext { throwable ->
if (throwable is IOException) {
return localRepository.getData()
}
return Single.error(throwable)
}
}
你可能会问为什么只捕获IOException
?我通常只处理这个异常,以免错过任何重要但不重要的网络错误。如果您将捕获您可能错过的所有异常,例如 NullPointerException
。
关于java - 使用 RxJava 同时从本地和远程获取数据,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/45902362/