如何在Android应用程序中组合RxJava Single & Compleable Retrofit调用

Question

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

我已经将我的HTTP GET(s(建模为Single<Response<String>>，并将POST建模为Completable。

我需要的调用顺序如下：-

顺序调用 GET(1(、GET(2(、GET(3(

并行调用 POST(1(， POST(2(

当 POST(1( 和 POST(2( 都完成 OK 时，调用 GET(4(。

我有一个部分解决方案。我已经为前三个 GET 编写了调用代码 后跟开机自检呼叫

我的代码类似于这样：-

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, "Manage Totals Failed", 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, "accept: ", throwable);
}
}));
}
return completables;
}

我遇到的困难是正确链接我的呼叫

例如，我认为我将能够开发类似于此伪代码的解决方案

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

但是我当前的部分解决方案仅调用第一组 GET，然后是 POST，并且 GET 和 POST 调用没有"链接">

我看不出如何创建支持我的用例的调用链。

是否可以在链式调用中组合Single->Completable->Single？

更新

根据丹尼尔的回答，我最终得到了这个解决方案：-

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

Answer 1

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

Answer 2

可以通过将它们转换为共享的反应式类型(例如Observable(并连接，或者通过flatMapX和andThen使用一种延续方法来链接不同的类型：

someSingle
.flatMapCompletable(result1 -> {
sideEffect(result1);
return someCompletable;
})
.andThen(Single.defer(() -> {
sideEffectAfterCompletable();
return someOtherSingle;
}))
...