如何在Android应用程序中组合RxJava Single和Completable Retrofit调用

我目前的 Android应用程序使用Retrofit和RxJava来编排我的网络调用.

我已将我的HTTP GET建模为Single< Response< String>>和POST(s)为可完成.

我要求的电话顺序如下： –

顺序调用GET(1),GET(2),GET(3)

并行呼叫POST(1),POST(2)

POST(1)& POST(2)完成OK,调用GET(4).

我有部分解决方案.我编写了前三个GET的编号
然后是POST电话

我的代码类似于： –

Single.concat(getRequests())
                .subscribeOn(Schedulers.single())
                .doOnError(throwable -> Log.e(TAG,"Manage Totals Failed",throwable))
                .doFinally(new Action() {
                    @Override
                    public void run() throws Exception {
                        manageExecutions(combineExecutions());
                    }
                })
                .subscribe();

    /**
     * @return
     */
    private static Iterable<Single<Response<String>>> getRequests() {
        final API_CALL_GET[] apiCalls = API_CALL_GET.values();
        final List<Single<Response<String>>> requests = new ArrayList<>(apiCalls.length);

        for (final API_CALL_GET apiCall : apiCalls) {
            requests.add(apiCall.request());
        }

        return requests;
    }

public enum API_CALL_GET {

    GET_ONE {
        @Override
        public Single<Response<String>> request() {
            return RETRO_SERVICE
                .getOne(authToken,new HashMap<>())
                .doAfterSuccess(this::persistDataOne)
                .doOnError(error -> ever(error));
        }
    },GET_TWO {
        @Override
        public Single<Response<String>> request() {
            return RETRO_SERVICE
                .getTwo(authToken,new HashMap<>())
                .doAfterSuccess(this::persistDataTwo)
                .doOnError(error -> ever(error));
        }
    },GET_THREE {
        @Override
        public Single<Response<String>> request() {
            return RETRO_SERVICE
                .getThree(authToken,new HashMap<>())
                .doAfterSuccess(this::persistDataThree)
                .doOnError(error -> ever(error));
        }
    };

    public abstract Single<Response<String>> request();

}


    private static Action manageExecutions(final List<Completable> completables) {

        return new Action() {
            @Override
            public void run() throws Exception {
                Completable
                .concat(completables)
                .subscribeOn(Schedulers.io())
                .doOnError(throwable -> Log.e(TAG,throwable))
                .doOnComplete(new Action() {
                    @Override
                    public void run() throws Exception {
                        accumulateAmounts();
                    }
                })
                .subscribe();
            }
        };
    }


    /**
     * @return
     */
    private static List<Completable> combineExecutions() {
        final API_CALL_POST[] apiCalls = API_CALL_POST.values();
        final List<Completable> requests = new ArrayList<>(apiCalls.length);

        for (final API_CALL_POST apiCall : apiCalls) {
            requests.addAll(apiCall.requests());
        }

        return Lists.newArrayList(Iterables.unmodifiableIterable(requests));
    }

public enum API_CALL_POST {

    POST_ONE {
        @Override
        public List<Completable> requests() {
            return NetworkController.postRecommenderExecutions();
        }
    },POST_TWO {
        @Override
        public List<Completable> requests() {
            return NetworkController.postSavedSearcheExecutions();
        }
    };

    public abstract List<Completable> requests();

}


    public static List<Completable> postONE() {
        final List<Completable> completables = new ArrayList<>();

        final List<OneDO> oneDOS = fetchOnes();

        for (final OneDO oneDO : oneDOS) {
            completables.add(RETRO_SERVICE.runCompletableOnes(authToken,oneDO.getId())
                    .doOnError(new Consumer<Throwable>() {
                        @Override
                        public void accept(final Throwable throwable) throws Exception {
                            Log.e(TAG,"accept: ",throwable);
                        }
                    }));
        }

        return completables;
    }




    public static List<Completable> postTWO() {
        final List<Completable> completables = new ArrayList<>();

        final List<TwoDO> twoDOS = fetchTwos();

        for (final TwoDO twoDO : twoDOS) {
            completables.add(RETRO_SERVICE.runCompletableTwos(authToken,twoDO.getId())
                    .doOnError(new Consumer<Throwable>() {
                        @Override
                        public void accept(final Throwable throwable) throws Exception {
                            Log.e(TAG,throwable);
                        }
                    }));
        }

        return completables;
    }

我遇到的困难是正确地链接我的电话

例如我以为我能够开发出类似于这个伪代码的解决方案

Single.concat(GET_1 … GET_N).onComplete(POST_1 … POST_N).onComplete(GET_LAST)

但是我当前的部分解决方案只调用第一组GET,然后是POST,并且GET和POST调用不是“链接”

我看不到如何创建支持我的用例的调用链.

可以组合Single – >可完成 – >在链式电话中单身？

UPDATE

根据Daniil的回答,我最终得到了这个解决方案： –

Single.concat(getRequests())
                .subscribeOn(Schedulers.io())
                .doOnError(throwable -> Log.e(TAG,"accept[0000]: ",throwable))
                .ignoreElements()
                .andThen(Completable.merge(combineExecutions()))
                .doOnError(throwable -> Log.e(TAG,throwable))
                .doOnComplete(() -> Controller.accumulateTotals())
                .subscribe();

解决方法

在kotlin中,它看起来像这样：

fun generateGetRequests(): List<Single<Response<String>>> {
    return listOf(retrofit.firstGet(),retrofit.secondGet(),...,retrofit.lastGet())
}

fun generatePostRequests(): List<Completable> {
    return listOf(retrofit.firstPost(),retrofit.lastPost())
}

fun doSomethingWithResponses(responses: Array<Any>) {
    // Do Something,like save to db
}

fun runRequests() {
    Single.zip(generateGetRequests(),{ responses ->
        doSomethingWithResponses(responses)
    }).ignoreElements()
        .andThen(Completable.merge(generatePostRequests()))
        .subscribeOn(Schedulers.io())
        .subscribe()
}