向可观测性添加条件(rx-java)

Question

总的来说，我对Rx和反应性编程还是比较陌生的。

这是我目前的密码。它现在所做的是：

1. 检查设备是否存在数据
2. 如果有，拿来
3. 如果不存在，请从服务器获取 可观测= Observable.concat( helper.loadFromDevice(params)，helper.loadFromServer(params)) .takeFirst(pastPaper -> pastPaper != null)；observable.subscribe(.)

我想要做的是：

1. 检查设备是否存在数据
2. 如果有，拿来
3. 如果不存在，请检查internet连接是否活动。
4. 如果internet连接是主动从服务器获取的

如何按照上述步骤添加互联网连接检查？(已经有了方法)。

任何帮助都是非常感谢的。谢谢

边问:我上面的代码可以改进吗？

Answer 1

您可以使用switchOnError，以便如果loadFromServer调用失败，它将被忽略。为了正常起见，您可能会记录来自服务器的任何错误，作为警告：

Observable<PastPaper> observable = Observable.concat(
    helper.loadFromDevice(params),
    helper.loadFromServer(params)
          .doOnError(e -> log.warn(e.getMessage()))
          .switchOnError(e -> Observable.empty())
    .takeFirst(pastPaper -> pastPaper != null);

否则，如果您想使用您的internet连接活动可观察(一个Observable<Boolean> )使用flatMap

Observable<PastPaper> observable = Observable.concat(
    helper.loadFromDevice(params),
    internetAvailable
        .filter(x -> x)
        .flatMap(x -> helper.loadFromServer(params)))
    .takeFirst(pastPaper -> pastPaper != null);

Answer 2

人工运行示例:从本地开始，如果本地为空，则切换到远程，如果网络不可用，则返回默认值

        Observable<Integer> local = Observable.fromCallable(() -> 1);
        Observable<Integer> remote = Observable.fromCallable(() -> 2);
        Observable<Boolean> hasInternet = Observable.fromCallable(() -> true);
        Integer noLocalNoInetVal = 42;

        Observable<Integer> localOrEmpty = local.filter(localVal -> localVal != null);
        Observable<Integer> remoteWithInet = hasInternet
                                               .filter(inetAvailable -> inetAvailable)
                                               .flatMap(__ -> remote);

        Observable<Integer> localOrRemote = localOrEmpty.switchIfEmpty(remoteWithInet)
                                                 .firstOrDefault(noLocalNoInetVal);

Answer 3

我认为在您的例子中，使用switchIfEmpty是最简单的方式

    Observable<PastPaper> observable = Observable.just(helper.loadFromDevice(params))
                  .switchIfEmpty(Observable.just(helper.loadFromServer(params)))
                  .subscribe(....);

这里还有关于如何工作https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/combining/ObservableSwitch.java的其他示例