Цепь двух модифицированных наблюдаемых ж/RxJava

Question

Я хочу выполнить 2 сетевых вызова один за другим. Оба сетевых вызова возвращаются Observable. Второй вызов использует данные из успешного результата первого вызова, метод в успешном результате второго вызова использует данные от и успешный результат первого и второго вызова. Также я должен иметь возможность обрабатывать как onError «события» по-разному. Как я могу достичь этого избежать обратного вызова ад, как в примере ниже:

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

Я знаю о молнии, но я хочу, чтобы избежать создания «объединителя класс».

Update 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) и т.д.). Im новое для программирования rx и lambdas - пожалуйста, скажите мне, в чем проблема. В любом случае код выглядит намного чище.

Update 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

В дополнении к ответу Энтони R., есть перегрузка flatMap которая принимает func2 и пары основных и сглаженных значения для вас. Кроме того, обратите внимание на onErrorXXX и onExceptionXXX операторов для манипулирования ошибок и приковать их с первым и вторым наблюдаемыми характеристиками

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

Answer 2

Вы заглянули в flatMap()? Если ваше отвращение к нему (или zip()) - это необходимость сделать ненужный класс просто для хранения двух объектов, 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 
      } 
     });