可观察的条件链

Question

我想通过多个 REST API 异步检索数据。我在带有 rxJava 扩展的 Android 上使用 Retrofit，即我通过订阅 Observable 来执行任何 GET 请求。

正如我所说，我有多个源 API，所以当第一个源没有产生所需的结果时，我想尝试下一个，如果同样失败，请再次尝试下一个，依此类推，直到所有源都已完成查询或找到结果。

我正在努力将这种方法转化为对 Observables 的正确使用，因为我不知道哪些运营商可以实现这种行为，而且还需要遵守一些限制：

• 找到结果后，不应查询剩余的 API（如果有）
• 其他组件依赖于查询的结果，我希望它们在启动请求的时候得到一个Observable，这样这个Observable就可以通知它们请求完成了
• 我需要保留对上述 Observable 的引用，因为在完成之前可能会多次发出相同的请求，在这种情况下，我只会在第一次需要它时启动它，后续请求只会获得通知的 Observable当请求完成时

我一开始只使用一个 API 进行查询，并使用以下 API 进行依赖组件的请求和后续通知：

private Observable<String> loadData(int jobId) {

    final ConnectableObservable<String> result = Async
            .fromCallable(() -> getResult(jobId))
            .publish();

    getRestRequest()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                    dataHolder -> {
                        if (dataHolder.getData() != null && !dataHolder.getData().isEmpty()) {
                            saveData(dataHolder.getData());
                        } else {
                            markNotFound(dataHolder);
                        }
                    },
                    error -> currentJobs.remove(jobId),
                    () -> {
                        currentJobs.remove(jobId);
                        result.connect();
                    });

    return result;
}

此代码仅在第一次请求时调用，返回的 Observable 结果将保存在 currentJobs 中，后续请求只会获取 Observable 而不会再次触发请求。

非常感谢任何帮助。

Answer 1

假设您有一组在您每次订阅时重新连接的可观察对象：

List<Observable<Result>> suppliers = ...

那么你只需要做合乎逻辑的事情：

Observable<Result> results = Observable
          .from(suppliers)
          .concatMap(supplier -> supplier)
          .takeFirst(result -> isAcceptable(result))
          .cache()

Answer 2

使用.onErrorResumeNext，并假设每个服务可观察对象可能返回 0 或 1 个元素，如果没有发出任何元素，则使用 first 发出错误：

Observable<T> a, b, c;
...
a.first().onErrorResumeNext(t -> b.first())
 .onErrorResumeNext(t -> c.first())
 .onErrorResumeNext(t -> d.first())
 ...