将 RxJava observable 的错误转换为另一个 observable 并吞下成功答案

【问题标题】：Transform RxJava observable's error into another observable and swallow success将 RxJava observable 的错误转换为另一个 observable 并吞下成功
【发布时间】：2015-07-25 04:27:03
【问题描述】：

如何将 observable 的错误转换为另一个 observable？

我正在尝试实现以下方案中显示的算法：

我正在使用一个名为 Grokking RxJava 的教程开始学习，发现 flatMap 运算符可以将 Retrofit 返回的一个 Observable 转换为另一个，这允许我执行一系列服务器方法的调用。但在教程中，链接方法总是返回成功结果。 如果远程方法之一返回错误，我如何执行相同的链接？

目前我正在使用一种对我来说看起来很奇怪且不清楚的方法：

AppObservable.bindActivity(this, userService.checklogin(mPhone)
            .onErrorResumeNext(new Func1<Throwable, Observable<? extends Response>>() {
                @Override
                public Observable<? extends Response> call(Throwable throwable) {
                    return Observable.just(null);
                }
            }))
            .flatMap(new Func1<Response, Observable<Response>>() {
                @Override
                public Observable<Response> call(Response response) {
                    if (response == null) {
                        return AppObservable.bindActivity(RegistrationActivity.this, userService.register(
                                mPhone,
                                name.getText().toString(),
                                selectedSex,
                                selectedDateDay,
                                selectedDateMonth,
                                selectedDateYear,
                                Locale.getDefault().getLanguage(),
                                persistentUserInfoStore.getInstallationToken()
                        ));
                    }
                    phone.setError(getString(R.string.already_registered_phone));
                    progressDialog.dismiss();
                    return Observable.empty();
                }
            })
            .subscribe(new Action1<Response>() {
                @Override
                public void call(Response response) {
                    startConfirmationActivity();
                    progressDialog.dismiss();
                }
            }, new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    progressDialog.dismiss();
                    Toast.makeText(RegistrationActivity.this, "Error while register user", Toast.LENGTH_SHORT).show();
                }
            });

有没有更明确的方法来做到这一点。

附：我是 RxJava 的新手。

【问题讨论】：

为什么不从onErrorResumeNext函数调用userService.register？
@VladimirMironov 我不确定之后会发生什么。如果我理解正确，在 onErrorResumeNext 中传递 userService.register 以防可观察的 userService.checkLogin 发生错误后，其订阅者将重新订阅 userService.register 并消耗其排放量。如何区分 userService.register 和 userService.checkLogin 的成功发射？
pastebin.com/Cu32c8tc 之类的怎么样？
@VladimirMironov 它看起来又漂亮又清晰。我认为您应该将该解决方案发布为答案，我会接受它。
看看retrolambda github.com/evant/gradle-retrolambda，可以简化代码

标签： retrofit reactive-programming rx-java observable

【解决方案1】：

您应该从onErrorResumeNext 拨打userService.register
区分发生了何种错误的最简单方法是为每个错误引入一个单独的Exception 类。

这是它在代码中的样子：

userService.checkLogin(mPhone).flatMap(new Func1<Response, Observable<Response>() {
    @Override
    public Observable<? extends Response> call(final Response response) {
        // according to your diagram the whole Observable should fail if first call was successful
        return Observable.error(new FirstCallWasSuccessfulException(response));
    }
}).onErrorResumeNext(new Func1<Throwable, Observable<? extends Response>>() {
    @Override
    public Observable<? extends Response> call(final Throwable throwable) {
        if (throwable instanceof FirstCallWasSuccessfulException) {
            return Observable.error(throwable);
        }

        return userService.register(...);
    }
}).subscribe(new Subscriber<Response>() {
    @Override
    public void onError(final Throwable throwable) {
        if (throwable instanceof FirstCallWasSuccessfulException) {
            // foo
        } else {
            // bar
        }
    }

    @Override
    public void onNext(final Response response) {

    }

    @Override
    public void onCompleted() {

    }
});

【讨论】：