takeUntil implementation

ApiService.java

  @GET(ApiConstants.GET_QUESTIONS_URL) Observable<RequestResponse> getQuestions();

ineteractor.java

   public void performGetElQuestions(String query, QuestionsRequestServerCallback callback) {
   
 getFreshNetworkData()//
        .publish(network ->//
            Observable.merge(network,//
                getCachedDiskData().takeUntil(network)))
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new DisposableObserver<Question>() {
          @Override
          public void onComplete() {
            callback.onQuestionsReady(mQuestionsList);
          }
          @Override
          public void onError(Throwable e) {
            callback.onQuestionsFailed();
          }

          @Override
          public void onNext(Question question) {
            // mQuestionsList is an arraylist
            mQuestionsList.add(question);
          }
        });
        
        
        
        private Observable<Question> getCachedDiskData() {
    List<Question> list = new ArrayList<>();
    //get cached data from SQLite or disk

    return Observable.fromIterable(list)//
        .doOnSubscribe((data) -> new Handler(Looper.getMainLooper())//
            .post(() -> Timber.d("(disk) cache subscribed")))//
        .doOnComplete(() -> new Handler(Looper.getMainLooper())//
            .post(() -> Timber.d("(disk) cache completed")));
  }
  
  
  
  private Observable<Question> getFreshNetworkData() {
        return apiService.getQuestions()
              .flatMap(Observable::fromIterable)
              .doOnSubscribe((data) -> new Handler(Looper.getMainLooper())//
                    .post(() -> adapterSubscriptionInfo.add("(network) subscribed")))//
              .doOnComplete(() -> new Handler(Looper.getMainLooper())//
                    .post(() -> adapterSubscriptionInfo.add("(network) completed")));
    }

RequestResponse.java

Public class RequestResponse {

    @SerializedName("questions")
    ArrayList<Question> questions;

    public ArrayList<Question> getQuestions() {
        return questions;
    }
}

1. what exactly is the problem you're facing?
2. you should probably return the observable directly and use that in the interactor vs having a callback

Hey kaushik, I am using MVP in my app so I need the callback to connect with the presenter .
As you can see the freshnetworkethod returns a < question > while the API response is RequestResponse so how to connect both ?
Thanks again