RXJava 忽略错误并继续链式答案

【问题标题】：RXJava Ignore Error and continue in chainRXJava 忽略错误并继续链式
【发布时间】：2017-11-17 11:22:09
【问题描述】：

我有一个可以引发错误的 RXJava 链。我希望链继续并忽略错误并到达订阅，仅针对某些特定错误。

例如在这段代码中

authDataManager
                    .getUserAccessToken(username, password)
                    .subscribeOn(Schedulers.io())
                    .doOnNext({
                        authDataManager.saveUserAccessToken(it)
                    })
                    .flatMap {
                        appDataManager.getStations()
                    }
                    .doOnNext({
                        appDataManager.persistStations(it)
                    })
                    .flatMap {
                        appDataManager.getDriverInformation()
                    }
                    .doOnNext({
                        appDataManager.persistDriverInformation(it)
                    })
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(
                            {
                                onLoginSuccess()
                            },
                            {
                                onLoginFailure(it)
                            }
                    )

如果appDataManager.getStations() 抛出错误，我还想继续并到达onLoginSuccess() 方法。

但是，如果 getUserAccessToken(username, password) 失败，则应调用 onLoginFailure。

我尝试在平面图之后和内部添加 onErrorResumeNext() 和 onExceptionResumeNext()，但如果我这样做，链就会退出并且不会继续并到达订阅

【问题讨论】：

你在处理 Observable 吗？如果是这样，您可以将 onLoginSuccess() 放在 onComplete 中，无论成功或错误，它都会被调用。
是的，它是一个 Observable。如果第一次调用（authDataManager.getUserAccessToken(username, password)）失败，我希望调用 onLoginFailure 的问题

标签： java android kotlin rx-java

【解决方案1】：

我在 Kotlin 中发现了这一点 here (stackoverflow link)

可以通过使用映射、过滤、可选和onErrorResumeNext 的组合来完成。我认为这与当前版本的 RxJava 一样优雅。

authDataManager
    .getUserAccessToken(username, password)
    .subscribeOn(Schedulers.io())
    .doOnNext({
        authDataManager.saveUserAccessToken(it)
    })
    .flatMap {
        appDataManager.getStations()
            .map(stations -> Single.just(Optional.of(stations))
            .onErrorResumeNext(error -> Single.just(Optional.empty()))                    
     }
     .filter(optional -> optional.isPresent())
     .map(stations -> optional.get())
     .doOnNext({
         appDataManager.persistStations(it)
     })
     .flatMap {
         appDataManager.getDriverInformation()
     }
     .doOnNext({
         appDataManager.persistDriverInformation(it)
     })
     .observeOn(AndroidSchedulers.mainThread())
     .subscribe(
         {
             onLoginSuccess()
         },
         {
             onLoginFailure(it)
         }
     )

【讨论】：

这将处理appDataManager.getStations() 流上的错误，但是，想象一下如果您的appDataManager.getDriverInformation 抛出异常 - 它会杀死整个流并调用onLoginFailure(it)。如果发生异常，OP 想跳过该项目。
您所描述的我的解决方案的问题正是 OP 要求的行为。

【解决方案2】：

我认为你可以使用onErrorResumeNext 运算符。

              authDataManager
                    .getUserAccessToken(username, password)
                    .subscribeOn(Schedulers.io())
                    .doOnNext({
                        authDataManager.saveUserAccessToken(it)
                    })
                    .flatMap {
                        appDataManager.getStations()
                             .onErrorResumeNext(Observable.empty())
                    }
                    .doOnNext({
                        appDataManager.persistStations(it)
                    })
                    .flatMap {
                        appDataManager.getDriverInformation()
                    }
                    .doOnNext({
                        appDataManager.persistDriverInformation(it)
                    })
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(
                            {
                                onLoginSuccess()
                            },
                            {
                                onLoginFailure(it)
                            }
                    )

【讨论】：

如果我尝试让链永远不会到达订阅。我认为问题在于Observable.empty()，因为如果我使用Observable.just(the return type)，它就可以工作。然而这很烦人，因为我必须创建一个 flatMap 的返回类型的实例
当然。这取决于您的业务逻辑。
问题是 Observable.empty() 没有调用 doOnNext 或 onNext.. 我希望我能做类似 Observable.just(null) 的事情
为什么 Observable.just 不适合你？您可以只返回预期对象的新实例。
当然可以，但是我的方法appDataManager.persistStations(it) 将被调用，我无法避免这种情况。如果我可以返回一个空值，我可以检查它是否为空而不调用它

【解决方案3】：

站在@savepopulation 的答案的肩膀上，并使用您拥有三个不同的观察者链的观察结果：

         authDataManager
                .getUserAccessToken(username, password)
                .subscribeOn(Schedulers.io())
                .doOnNext({
                    authDataManager.saveUserAccessToken(it)
                })
                .doOnNext(
                    appDataManager.getStations()
                         .onErrorResumeNext(Observable.empty())
                         .subscribe( appDataManager.persistStations(it) )
                )
                .doOnNext( appDataManager.getDriverInformation()
                             .subscribe( appDataManager.persistDriverInformation(it)
                )
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                        {
                            onLoginSuccess()
                        },
                        {
                            onLoginFailure(it)
                        }
                )

主观察者链流向完成，而两条子链完全不参与日志；它们只是通过获取用户访问令牌来触发。

如果您需要这些附属分支与主链异步运行，则必须添加 subscribeOn()/observeOn() 运算符。

【讨论】：