RxJava Zip with priorities

2
votes

Here's the problem. I have a number of async operations whose result gets aggregated into a single one that is further processed. However, not all the operations are equal and the error handling is different based on which operation fails.

To elaborate, let's say we have operations A, B and C. If A fails, we need to end processing but if B or C fail, we continue processing the others as normal.

Currently, we achieve this using count down latches and a lot of state management taking close to a hundred lines of code. I would like to move this to an RxJava based implementation. My first thought was trying the Observable.zip operator but that treats all the observables as equals and that is not true in this case. My other idea was chaining the calls, which works, but that means all the operations won't start at the same time resulting in greater overall time.

Can anyone guide me on how to achieve this?

1
rx-javareactive-programming

1 Answers

4
votes

Use .onErrorResumeNext:

Observable<T> a, b, c;
Observable.zip(
    a,
    b.onErrorResumeNext(t -> Observable.just(null)),
    c.onErrorResumeNext(t -> Observable.just(null)),
   (x, y, z) -> <your aggregation>)
...

Representing the errored observables with null is up to you. You could also use Optional:

Observable.zip(
    a,
    b.map(x -> Optional.of(x))
     .onErrorResumeNext(t -> Observable.just(Optional.empty())),
    c.map(x -> Optional.of(x))
     .onErrorResumeNext(t -> Observable.just(Optional.empty())),
   (x, y, z) -> <your aggregation>)

If b and c correspond to external service calls you might also want to ignore them if they take too long by replacing b with b.timeout(5, TimeUnit.SECONDS). Concise eh!