在Android中使用Retrofit2和RxJava2實現嵌套的HTTP請求

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個flatMaps來完成此操作。您正在查找的觸發器是onComplete調用本身，因爲所有流都將正常終止或者將傳播到onError方法的異常終止。

從概念上講，你正在尋找的東西是這樣的：

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

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