rxjava2 - Cadena de dos reequipables observables con RxJava

rxjava single example

Question

Quiero ejecutar 2 llamadas de red una tras otra. Ambas llamadas de red devuelven Observable. La segunda llamada usa datos del resultado exitoso de la primera llamada, el método en el resultado exitoso de la segunda llamada usa datos tanto del resultado exitoso de la primera llamada como de la segunda. También debería poder manejar ambos "eventos" onError de manera diferente. ¿Cómo puedo lograr esto evitando el infierno de devolución de llamada como en el siguiente ejemplo:

       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();
                }
            });

Sé sobre zip, pero quiero evitar crear "clase Combiner".

Actualización 1. Intenté implementar la respuesta de 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);
                }
            }));

Sin embargo, el compilador del método flatMap dice que no puede resolver los métodos de authResponse y los usuarios ( authResponse.getAccessToken() , users.get(0) etc.). Soy nuevo en la programación de rx y lambdas, por favor, dígame cuál es el problema. De todos modos el código se ve mucho más limpio ahora.

Actualización 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);
            }));

Lo he hecho así, pero ahora mis llamadas de red no se ejecutan en absoluto.

Answer 1

¿Has mirado en flatMap ()? Si su aversión a ella (o zip ()) es la necesidad de crear una clase innecesaria solo para contener dos objetos, android.util.Pair podría ser una respuesta. Sin embargo, no estoy seguro de cómo obtener exactamente el manejo de errores que estás buscando.

       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

Además de la respuesta de Anthony R., hay una sobrecarga de mapa plano que toma un Func2 y empareja sus valores primarios y planos para usted. Además, observe los operadores onErrorXXX y onExceptionXXX para la manipulación de errores, y encadéntelos con su primer y segundo observables

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