Реализация вложенных HTTP-запросов в Android с помощью Retrofit2 и RxJava2

Question

В рамках моей деятельности я запускаю HTTP-запрос с Retrofit2. При успехе я получаю массив элементов. На каждом элементе я запускаю второй запрос, который снова реагирует с массивом элементов на успех. Мне нужно событие для запуска, как только все вложенные запросы будут завершены. Как я могу достичь этого, не считая успехов каждого вложенного запроса и сравнивая его с размером первого массива? Вот мой Activity:

public class ABCActivity extends AppCompatActivity { 

    private static final String TAG = "StackoverflowActivity"; 

    private Disposable mDmRequestSubscription; 
    private StopRequestApiInteractor mStopRequestInteractor; 
    private DmRequestApi mDmRequestApiInteractor; 

    private Position mPosition; 

    private List<ItdOdvAssignedStop> mStopsList = new ArrayList<>(); 
    private List<List<ItdDeparture>> mAllDepartures; 
    private List<Place> mStopsInRange = new ArrayList<>(); 
    private List<Place> stopsInRange; 

    @Override 
    protected void onCreate(Bundle savedInstanceState) { 
     super.onCreate(savedInstanceState); 
     requestStops(); 
    } 

    public void requestStops() { 
     mStopRequestInteractor 
       .getStopFinderRespond(mPosition) 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .flatMap(s -> Observable.just(s.getStopFinder().getItdOdvAssignedStops())) 
       .subscribe(this::onSuccess, this::onError); 
    } 

    private void onError(Throwable throwable) { 
     throwable.printStackTrace(); 
    } 

    private void onSuccess(ItdOdvAssignedStop[] response) { 
     mStopsList.clear(); 
     mStopsList.addAll(Arrays.asList(response)); 

     for (ItdOdvAssignedStop stop : mStopsList) { 
      mDmRequestSubscription = mDmRequestApiInteractor 
        .getStopDmRespond(stop.getStopID()) 
        .subscribeOn(Schedulers.io()) 
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe(this::onDmRequestSuccess, this::onDmRequestError); 
     } 
    } 

    private void onDmRequestError(Throwable throwable) { 
     Log.e(TAG, "DM Request failed"); 
    } 

    private void onDmRequestSuccess(ResponseBody entry) { 
     Log.d(TAG, "DM Request success"); 
     mStopsInRange = getStopsInRange(); 
     update(); 
    } 

    public List<Place> getStopsInRange() { 
     //Some things done here... 
     return stopsInRange; 
    } 

    private void update() { 
     //Shall be called only once as soon as all onDmRequestSuccess() were fired 
    } 
} 

Моя активность не работоспособной вне моего проекта. Я просто хочу, чтобы чистое решение вызывало update() только один раз, как только отвечают на все запросы. И не каждый раз, когда отвечает вложенный запрос.

Как это реализовать, используя onNext(), onCompleted() и onError() методы Subscriber? Итак, как преобразовать мой код в нечто вроде следующего?

Observable.create(new Observable.OnSubscribe<String>() { 
    @Override 
    public void call(Subscriber<? super String> subscriber) { 
     subscriber.onNext("Test"); 
     subscriber.onError(null); 
    } 
}).observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(new Observer<String>() { 
      @Override 
      public void onCompleted() { 
       System.out.println("onCompleted"); 
      } 

      @Override 
      public void onError(Throwable e) { 
       System.out.println("onError"); 
      } 

      @Override 
      public void onNext(String s) { 
       System.out.println("onNext"); 
      } 
     }); 

Answer 1

Вы можете выполнить это, используя 2 плоские карты. Триггер, который вы ищете, это сам вызов onComplete, так как все потоки либо прекратятся нормально, либо с исключением, которое будет распространено на метод onError.

Концептуально, вы ищете что-то вроде этого:

getMainRequest() 
    .flatMap(itemList -> Observable.from(itemList)) 
    .flatMap(item -> secondRequest(item)) 
    .subscribe(result -> { 

    }, error -> { 
     // some stream encountered an error 
    },() -> { 
     // all requests have finished 
    })