使用 Retrofit2 和 RxJava2 在 Android 中实现嵌套的 HTTP 请求

Question

在我的活动中，我正在使用Retrofit2 发出 HTTP 请求。成功后，我会收到一组元素。在每个元素上，我都会触发第二个请求，该请求再次响应成功时的元素数组。我需要一个事件在所有嵌套请求完成后立即触发。如何在不计算每个嵌套请求的成功并将其与第一个数组的大小进行比较的情况下实现这一点？这是我的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()。而且不是每次响应嵌套请求时。

如何使用Subscriber 的onNext()、onCompleted() 和onError() 方法来实现它？那么如何将我的代码转换成如下的样子呢？

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 个 flatMap 来完成此操作。您正在寻找的触发器是 onComplete 调用本身，因为所有流都将正常终止或出现异常，该异常将传播到 onError 方法中。

从概念上讲，您正在寻找这样的东西：

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

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