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(promotions -> { // used to chain

        // this assumes we'll always find currentPromotion
        // we'll see how're gonna handle when there's no currentPromotion
        // in the next post
        Promotion currentPromotion = null;
        for(Promotion p: promotions){
          if (p.getStartDate().isBeforeNow() && p.getEndDate().isAfterNow()) {
            currentPromotion = p;
            break;
          }
        }

        // fetchSales can continue running in the same thread fetchPromotions just used
        Observable<List<Sale>> sales = repository.fetchSales(currentPromotion.getId());
        Observable<List<Ad>> ads = repository.fetchAds(currentPromotion.getId())
          .subscribeOn(Schedulers.newThread()); // notice we call newThread for fetchAds

        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 (mView != null) {
          mView.onDataFetched(pair.first, pair.second);
        }
      }, throwable -> {
        // this is reached when fetchSales or fetchAds throws an error
        if (mView != null) mView.onDataFetchFailed();
      });

    disposeBag.add(disposable);
  }
  
  @Override
  public void onAttach(PromotionContract.View view) {
    this.view = view;
  }
  
  @Override
  public void onDetach() {
    disposeBag.clear();
    this.view = null;
  }

}