RxJava2 在 Observable 中有远程数据覆盖本地数据

Question

目前我在存储库类中有一个方法，它从本地缓存和远程 API 中获取数据。

public Observable<List<Items>> getItemsForUser(String userId {
    return Observable.concatArrayEager(
            getUserItemsLocal(userId), // returns Observable<List<Items>>
            getUserItemsRemote(userId) // returns Observable<List<Items>>
    );
}

目前，该方法首先获取本地数据（可能已过时）并返回，然后使用来自远程 API 的新数据进行更新。

我想将实现更改为使用Observable.merge，这样如果远程 API 请求首先完成，该数据就会首先显示。但是，如果我只使用Observable.merge，我担心本地数据库请求可能会返回陈旧数据，这将覆盖来自远程的新数据。

基本上，我想要这样的东西：

public Observable<List<ShoutContent>> getItemsForUser(String userId, ErrorCallback errorCallback) {
    return Observable.merge(
            getUserItemsRemote(userId),
            getUserItemsLocal(userId)
                .useOnlyIfFirstResponse()
}

因此，如果远程 API 请求首先完成，则该响应是唯一返回的响应。但是如果本地请求首先完成，我想返回它，然后在完成后返回远程请求。 RxJava 有这样的内置功能吗？

编辑：我想补充一点，getUserItemsRemote 在 Observable 发出时会更新本地数据库，但我认为我不能确保在本地请求完成之前更新数据库，这留下了可能性本地请求将响应陈旧数据。

Answer 1

您可以使用takeUntil 运算符。

takeUntil 返回一个 Observable，它发出源 Observable 发出的项目，直到第二个 ObservableSource 发出一个项目。

在您的情况下，一旦远程 Observable 发出，您需要停止观察本地 observable。代码如下所示。

public Observable<String> getUserItemsLocal() {
    return Observable.just("Local db response")
            .delay(5, TimeUnit.SECONDS);  // assume local db takes 5 seconds to emit
}


public Observable<String> getUserItemsRemote() {
    return Observable.just("Remote Data")
            .delay(1, TimeUnit.SECONDS); // remote data comes quicker, in 1 second
}

您的存储库代码如下

Observable<String> remoteResponse = getUserItemsRemote();
getUserItemsLocal().takeUntil(remoteResponse)
            .mergeWith(remoteResponse)
            .subscribe(new Consumer<String>() {
                @Override
                public void accept(String s) throws Exception {
                    Log.d(TAG, "result: " + s);
                }
            });