What RxJava operator to use to chain observables o

问题:

I'm developing a whole application in RxJava/Android, trying to make the things the most Rx-way possible.

I think I'm achieving what I wanted, but now I've encountered an issue that I'm sure that exists a better way to do it. It consists of:

1. Get a Boolean from an Observable
2. A: If it's true, you are done, return true. B: If it's false, make a request (call it firstRequest) and receive an Observable.
3. A: The same, if it's true, you are done, return true. B: If it's false, then make another request (secondRequest).
4. A: If it returns true, then again you are done. If it's false, launch processFailed();

Right now I've implemented it in a very naive way (and sub-optimal):

public void startProcess(){
  dataRepository.getStatus()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(status -> {
      if(status){
        processCompleted();
      } else {
        makeFirstRequest();
      }
    });
}

private void makeFirstRequest(){
  dataRepository.firstRequest()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(status -> {
      if(status){
        processCompleted();
      } else {
        makeSecondRequest();
      }
    });
}

private void makeSecondRequest(){
  dataRepository.firstRequest()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(status -> {
      if(status){
        processCompleted();
      } else {
        processFailed();
      }
    });
}

I would like to know how to combine those operations in a way that made more sense (I peeked at combine, but I think it's not the intended use case for this), and subscribing three times it's not right, I think.

I thought about flatMapping (as it makes sense returning an Observable), but the nested chain of operations (flatMap inside flatMap inside flatMap) made me think that there has to be a simpler way of achieving the same result with some operators.

Thanks in advance.

回答1:

flatMap to the rescue:

  dataRepository.getStatus()
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .flatMap(status -> status ? Observable.empty() : dataRepository.firstRequest())
  .flatMap(status -> status ? Observable.empty() : dataRepository.secondRequest())
  .defaultIfEmpty(false)
  .subscribe(status -> {
    if(status){
      processCompleted();
    } else {
      processFailed();
    }
  });

You can also replace Observable.empty with Observable.just(status).