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

Question

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 defer() to prevent the start and then additionally subscribe to count() 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 myResult.add is performing slower than the count().

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?

Answer 1

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();
Observable.zip(cal, erp, help, crm,
                new Func4<List<Cal>, List<Erp>, List<Help>, List<Crm>, Result>() {
                    @Override
                    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>() {
                    @Override
                    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>() {
                @Override
                public Cal call(Throwable throwable) {
                    // Return something in the error case
                    return null;
                }
            })
            .toList();