java - 在 Android 中使用 Room 和 RxJava 链接多个网络调用和数据库插入

Question

假设我有一个 Android 应用程序，其中有一个包含按钮的 Activity。当我单击该按钮时，我想向返回 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 强制您将查询分派(dispatch)给工作线程，因此我使用 RxJava 来处理该任务。因此，例如，我的 Insert 方法返回一个 ArrayList，其中包含包装在 Single<ArrayList<Integer>> 中的插入对象的 ID。 。然后我调用 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 上下文中创建和订阅此类事件？我没有太多使用 react 性数据的工作，因此任何知识都将不胜感激。

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()
);