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?

Do you expect more than one emission from these Observables? If yes, you could use the `toList` operator, that compiles all the emitted items to a list and emits it, when the source Observable completes. Then you could just zip all those 4 Observables emitting lists together and compose your Result. — ehehhh, Oct 22 '16 at 12:48
Yes. Each observable has an unknown number of emissions. The numbers are not related. The toList operator does return a list that keeps the insertion order? Turn your comment into an answer for acceptance — stwissel, Oct 22 '16 at 17:06

ehehhh · Accepted Answer · 2016-10-23T07:27:06.380

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

Thx. Now I need to evaluate the .toList() behaviour in error cases. Like I get 5 elements and then an error. Will I get no list or the good ones - but that's best put into a new question. — stwissel, Oct 23 '16 at 02:50

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

1 Answers1