将多个网络调用和数据库插入与 Android 中的 Room 和 RxJava 链接在一起

Question

假设我有一个安卓应用程序，其中包含一个包含按钮的活动。当我单击该按钮时，我想向返回 JSON 响应的 rest API 发出多个请求。然后，我将响应解析为 java 对象，并使用 Room 将其持久化。对于 http 请求，我将 Volley 请求队列实现为单例。 请求是异步的，并将其响应传递回 UI 线程。在那里，我让 Room 保留对象。

我像这样发送我的 http 请求：

RestService.requestSomeData(context, objectId, new ResponseListener() {
@Override
public void onRestSuccess(String response) {    
// parse response JSON
// call the insert method
}
@Override
public void onRestError(int code, String errorMessage) {
// handle error
}
}

由于 Room 强制您将查询调度到工作线程，因此我使用 RxJava 来处理该任务。因此，例如，我的 Insert 方法返回一个 ArrayList 的插入对象的 ID 包装在Single<ArrayList<Integer>>中。然后我调用 Insert 方法并订阅结果，如下所示：

myDisposable = MyDatabase.getInstance().myDao()
.insert(myObject)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(idList -> {
Log.d(TAG, "IDs inserted: " + idList.toString());
}, Throwable::printStackTrace);

但是，我想将多个请求链接到服务器，然后在所有请求完成并且数据库插入准备就绪时收到通知以更新 UI(例如显示确认消息，禁用保存按钮(。我阅读了许多文章，但没有任何地方可以找到如何执行这项看似简单的任务。基本上我想要实现的是：

// a some sort of container for all the observables I get from the database insertions
private Object aPoolOfObservables;
RestService.requestSomeData(context, objectId, new ResponseListener() {
@Override
public void onRestSuccess(String response) {    
// parse response JSON
aPoolOfObservables.add(MyDatabase.getInstance().myDao()
.insert(myObject)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()));
}
}
// repeat this n-times more
...
aPoolOfObservables.subscribe(new Listener() {
@Override
public void onComplete() {
// update UI
}
});

然后多次执行此请求并将响应添加到Single<>(或Maybe<>或Flowable<>(响应的集合中，并且不是订阅每个流，而是订阅集合，因为我只关心所有操作都已完成。通过在前一个请求的onRestSuccess中触发请求来链接它们似乎是一个非常糟糕的解决方案。 你知道是否有允许这样做的RxJava机制吗？

是否有任何通用方法/设计模式来处理这种情况？我可以想到很多情况，例如，您希望仅在执行多个请求并交付结果后启用按钮。如何在 RxJava 的上下文中创建和订阅此类事件？我没有在反应式数据上做过很多工作，所以任何知识都会不胜感激。

Answer 1

您可以将每个请求包装在Single<Pair<ArrayList<Integer>, String>>中，以存储每个请求的每个 JSON 响应。然后，将它们与Single.zip(...)一起执行

private CompositeDisposable disposables;
private ArrayList<Single<Pair<ArrayList<Integer>, String>>> singles;
RestService.requestSomeData(context, objectId, new ResponseListener() {
@Override
public void onRestSuccess(String response) {    
// parse response JSON
// kotlin syntax
singles.add(
MyDatabase.getInstance().myDao().insert(myObject)
.flatMap { ids: ArrayList<String> -> 
// transform single to include JSON response
return@flatMap Single.just(Pair(ids, response))
}
);
}
}
// kotlin syntax
disposables.add(
// execute all singles
Single.zip(singles) {}.subscribe()
);