java - 使用 RxJava 同时从本地和远程获取数据

Question

所以我是 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)
                }
        )
}

感谢您的宝贵时间。

Answer 1

您需要使用 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。