在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 
    })