stwissel stwissel - 2 months ago 14
Java Question

Waiting for multiple observable to complete that return different number of elements

Scenario: I have a customerID string that is used to query multiple different backend systems: calendar, helpdesk, ERP, CRM etc. I want to compile a single report.
So I have roughly (psydocode):

Result myResult = new Result();
Observable<Cal> cal = Calbackend.get(customerid);
cal.subscribe(calentry -> myResult.addCal(calentry));

Observable<Erp> erp = ERPbackend.get(customerid);
erp.subscribe(erpentry -> myResult.addErp(erpentry));

Observable<Help> help = Helpbackend.get(customerid);
help.subscribe(helpentry -> myResult.addHelp(helpentry));

Observable<Crm> crm = CRMbackend.get(customerid);
crm.subscribe(crmentry -> myResult.addCrm(crmentry));

// Magic here?

return result;

The approach I was thinking of: using
to prevent the start and then additionally subscribe to
for each. Then I could ZIP the count elements since they only will emit a single item each (while the others will have different numbers of events). However that could lead to loss of data if the
is performing slower than the

The other option I was thinking of, is to set an array of boolean flags for each subscription and check in each completion (and error) event if all of them are done and do a callback or use blocking for that one.

I had a look here and here but that examples deal with constant numbers or data types.

Or is there a better / recommended way?


Operator toList can be used together with zip like this:

Observable<List<Cal>> cal = Calbackend.get(customerid).toList();
Observable<List<Erp>> erp = ERPbackend.get(customerid).toList();
Observable<List<Help>> help = Helpbackend.get(customerid).toList();
Observable<List<Crm>> crm = CRMbackend.get(customerid).toList();, erp, help, crm,
                new Func4<List<Cal>, List<Erp>, List<Help>, List<Crm>, Result>() {
                    public Result call(List<Cal> cals, List<Erp> erps, List<Help> helps, List<Crm> crms) {
                        Result myResult = new Result();
                        // add all cals, erps, helps and crms to result
                        return myResult;
                .subscribe(new Subscriber<Result>() {
                    public void onNext(Result result) {
                        // do something with the result


Explanation: As the name suggests, the toList operator creates a list of the items emitted by the source observable (the list is emitted just once, when the source observable completes) and zip is then used to combine the results of the observables.

Edit: In case of the possibility that those Observables can emit an error, you could use onErrorReturn to keep the normal flow going:

Observable<List<Cal>> cal = Calbackend.get(customerid)
            .onErrorReturn(new Func1<Throwable, Cal>() {
                public Cal call(Throwable throwable) {
                    // Return something in the error case
                    return null;