CODEWITHSUNDEEP
javarx-javareactive-programmingjersey-clientreactivex
Adam
Adam

Reputation: 44959

How can I call multiple observable calls asynchronously, yet synchronously do some computation before and after those calls?

I'm new to RxJava. I have a few Jersey RxJava clients that return Observables. I need to make one call to get some data, that data becomes the input to my next 3 calls. I want those calls to be made in parallel. Finally, I want to do a calculation once all the calls have completed that requires all the data. Here is how it looks:

interface Service {
  Observable<ResultA> callServiceA(InitialData input);
  Observable<ResultB> callServiceB(ResultA resultA);
  Observable<ResultC> callServiceC(ResultA resultA);
  Observable<ResultD> callServiceD(ResultA resultA);
  FinalResult simpleCalculation(ResultA a, ResultB b, ResultC c, ResultD d);
}

class MyClass{

   @Autowired
   ExecutorService myExecutorService;

   Observable<FinalResult> myMethod(InitialData initialData){
   /* Make call to ServiceA, get the results, then make calls to services B, C, and D in parallel (on different threads), finally perform simpleCalculation, and emit the result */
   }
}

Upvotes: 0

Views: 1572

Answers (2)

kjones
kjones

Reputation: 5823

flatMap() and zip() are your friends in this situation.

Observable<FinalResult> myMethod(InitialData initialData) {
    return service
            .callServiceA(initialData)
            .flatMap(resultA -> Observable.zip(
                    service.callServiceB(resultA),
                    service.callServiceC(resultA),
                    service.callServiceD(resultA),
                    (resultB, resultC, resultD) -> 
                      service.simpleCalculation(resultA, resultB, resultC, resultD))
            );
}

Using the return observable will look like this:

Subscription subscription =
        myMethod(new InitialData())
                .subscribe(finalResult -> {
                            // FinalResult will end up here.
                        },
                        throwable -> {
                            // Handle all errors here.
                        });

Upvotes: 4

Graeme
Graeme

Reputation: 1117

You can use flatMap to do a synchronous call, then use zip or merge to send out multiple calls, then flatMap that again when complete.

Upvotes: 0

Related Questions