Chain multiple network calls and database inserts with Room and RxJava in Android

Question

Let's say I have an android app with a single activity that contains a button. When I click the button I'd like to make several requests to a rest API that return JSON response. Then I parse the response to a java object and persist it with Room. For the http requests I implemented a Volley request queue as singleton. The requests are asynchronous and deliver their responses back to the UI thread. There I let Room persist the objects.

I send my http request like this:

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
    }
}

Since Room forces you to dispatch the queries to worker threads, I'm using RxJava to handle the task. So, for example my Insert method returns an ArrayList of the IDs of the inserted objects wrapped in a Single<ArrayList<Integer>>. Then I call the Insert method and subscribe to the result like this:

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

However, I want to chain multiple requests to the server and then get notified when all are complete and the DB insertions are ready in order to update the UI (e.g. display confirm message, disable the save button). I read numerous articles but nowhere I could find how to perform this apparently easy task. Basically what I want to achieve is:

// 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
    }
});

Then perform this request multiple times and add the responses to the collection of Single<> (or Maybe<> or Flowable<>) responses and subscribe not to every single stream but to the collection, because I only care that all the operations are complete. Chaining them by firing a request in the onRestSuccess of the previous one seems like a pretty awful solution. Do you know if there is a RxJava mechanism that allows this?

Is there any general approach/design pattern to handle this situation? I can think of numerous cases when you'd like to e.g. enable a button only after multiple requests have been performed and delivered results. How do you create and subscribe to such event in the context of RxJava? I haven't worked a lot with reactive data so any knowledge will be appreciated.

Give a try with Observable.zip (). It can handle multiple request simultaneously and notify when all finished — Md. Asaduzzaman

Christilyn Arjona Christilyn Arjona · Accepted Answer · 2019-12-10T10:19:21

You can wrap each request in a Single<Pair<ArrayList<Integer>, String>> to store each JSON responses per request. Then, execute them all together with 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()
);

Chain multiple network calls and database inserts with Room and RxJava in Android

1 Answers