PromotionPresenter.java

public class PromotionPresenter implements PromotionContract.Presenter {

  PromotionContract.View view;
  PromotionRepository repository;

  private CompositeDisposable disposeBag;

  @Inject
  public PromotionPresenter(PromotionRepository repository){
    this.repository = repository;
    disposeBag = new CompositeDisposable();
  }

  @Override
  public void fetchData() {
    Disposable disposable = repository.fetchPromotions()
        .flatMap(optional -> { // used to chain

          // this is not an error. this only mean we don't have any promotion
          if (optional.isEmpty())
            return Observable.just(Pair.create(Optional.<List<Sale>>empty(),
                Optional.<List<Ad>>empty()));

          List<Promotion> promotions = optional.get();
          Promotion currentPromotion = null;
          for(Promotion p: promotions){
            if (p.getStartDate().isBeforeNow() && p.getEndDate().isAfterNow()) {
              currentPromotion = p;
              break;
            }
          }

          // this is not an error. this only mean we don't have currentPromotion
          if (currentPromotion == null)
            return Observable.just(Pair.create(Optional.<List<Sale>>empty(),
                Optional.<List<Ad>>empty()));

          // fetchSales can continue running in the same thread fetchPromotions just used
          Observable<Optional<List<Sale>>> sales = repository.fetchSales(currentPromotion.getId());
          Observable<Optional<List<Ad>>> ads = repository.fetchAds(currentPromotion.getId())
              .subscribeOn(Schedulers.newThread()) // notice we call newThread for fetchAds
              .onErrorResumeNext(Observable.just(Optional.empty())); // we still gonna show the sale list even if we have some error with fetching ads

          return sales.zipWith(ads, Pair::create); // zip two observables to run two retrofit calls in parallel
        })
        .subscribeOn(Schedulers.io()) // RxJava default thread
        .observeOn(AndroidSchedulers.mainThread()) // observe on main thread
        .subscribe(pair -> {
          // this is reached only when fetchSales and fetchAds are done
          if (view != null) {
            view.onDataFetched(pair.first.getNullable(), pair.second.getNullable());
          }
        }, throwable -> {
          // this is reached when fetchSales
          // we don't reach here when fetchAds throws an error because of the `onErrorResumeNext` we used
          if (view != null) view.onDataFetchFailed();
        });

    disposeBag.add(disposable);
  }

  @Override
  public void onAttach(PromotionContract.View view) {
    this.view = view;
  }

  @Override
  public void onDetach() {
    disposeBag.clear();
    this.view = null;
  }

}