使用 RxJava 链接两个改造 Observable

Question

我想依次执行2个网络调用。两个网络调用 return 可观察。第二次调用使用来自第一次调用成功结果的数据，第二次调用成功结果中的方法使用来自第一次和第二次调用的 both 成功结果的数据。此外，我应该能够以不同的方式处理 both onError "events"。我怎样才能像下面的例子那样避免回调地狱：

       API().auth(email, password)
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<AuthResponse>() {
                @Override
                public void call(final AuthResponse authResponse) {
                    API().getUser(authResponse.getAccessToken())
                            .subscribe(new Action1<List<User>>() {
                                @Override
                                public void call(List<User> users) {
                                    doSomething(authResponse, users);
                                }
                            }, new Action1<Throwable>() {
                                @Override
                                public void call(Throwable throwable) {
                                    onErrorGetUser();
                                }
                            });
                }
            }, new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    onErrorAuth();
                }
            });

我知道 zip，但我想避免创建 "Combiner class".

更新 1. 试图实现 akarnokd 的回答：

         API()
            .auth(email, password)
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .flatMap(authResponse -> API()
                    .getUser(authResponse.getAccessToken())
                    .doOnError(throwable -> {
                        getView().setError(processFail(throwable));
                    }), ((authResponse, users) -> {
                // Ensure returned user is the which was authenticated
                if (authResponse.getUserId().equals(users.get(0).getId())) {
                    SessionManager.getInstance().initSession(email, password, authResponse.getAccessToken(), users.get(0));
                    getView().toNews();
                } else {
                    getView().setError(R.string.something_went_wrong);
                }
            }));

但是在 flatMap 方法内部，编译器说它无法解析 authResponse 和用户（authResponse.getAccessToken()、users.get(0) 等）的方法。我是 rx 编程和 lambdas 的新手 - 请告诉我问题是什么。不管怎样，代码现在看起来更干净了。

更新 2.

API()
            .auth(email, password)
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(throwable -> getView().setError(processFail(throwable)))
            .flatMap((AuthResponse authResponse) -> API()
                    .getUser(authResponse.getAccessToken())
                    .doOnError(throwable -> getView().setError(processFail(throwable))), ((AuthResponse authResponse, List<User> users) -> {
                            // Ensure returned user is the which was authenticated
                            if (authResponse.getUserId().equals(users.get(0).getId())) {
                                SessionManager.getInstance().initSession(email, password, authResponse.getAccessToken(), users.get(0));
                                getView().toNews();
                            }
                            return Observable.just(this);
            }));

这样做了，但现在我的网络调用根本没有执行。

Answer 1

你看过 flatMap() 了吗？如果您对它（或 zip()）的厌恶是需要制作一个不必要的 class 来容纳两个对象，那么 android.util.Pair 可能是一个答案。不过，我不确定如何准确地获得您正在寻找的错误处理。

       API().auth(email, password)
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
        .flatMap(new Func1<AuthResponse, Observable<List<User>>>() {
          @Override
          public Observable<List<User>> call(AuthResponse authResponse) {
            return API().getUser(authResponse.getAccessToken());
          }
        }, new Func2<AuthResponse, List<User>, Pair<AuthResponse, List<User>>>() {
          @Override
          public Pair<AuthResponse, List<User>> call(AuthResponse authResponse, List<User> users) {
            return new Pair<>(authResponse, users);
          }
        }).subscribe(new Action1<Pair<AuthResponse, List<User>>>() {
          @Override
          public void call(Pair<AuthResponse, List<User>> pair) {
            doSomething(pair.first, pair.second);
          }
        }, new Action1<Throwable>() {
          @Override
          public void call(Throwable throwable) {
            // not sure how to tell which one threw the error
          }
        });

Answer 2

除了 Anthony R. 的回答之外，还有一个 flatMap 重载，它采用 Func2 并为您配对您的主要值和扁平化值。此外，查看用于错误操作的 onErrorXXX 和 onExceptionXXX 运算符，并将它们与您的第一个和第二个 Observables

链接起来

first.onErrorReturn(1)
.flatMap(v -> service(v).onErrorReturn(2), (a, b) -> a + b);

使用 RxJava 链接两个改造 Observable

Chain two retrofit observables w/ RxJava

java

android

reactive-programming

rx-java

retrofit