连锁两个改造的可观察性w/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(); 
       } 
      }); 

我知道的拉链，但我想避免打造“合班”。

更新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编程和lambda - 请告诉我有什么问题。无论如何，代码看起来更清洁。

更新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

除了Anthony 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 
      } 
     });