Rxjava operator for list of observables and stops concatenating on first completed stream

user_9678048 0
Viewed 95

I have a list of observables and most likely some of them will raise an error whilst subscribing. I'd like to concat these observables to one stream & ignore all raised errors. In case of error, it should start subscription to new stream from list and continue until some stream completes. In this case, it should NOT subscribe to other streams and complete.

For example a have a list of three observables.

  • subscribe to first -> it raised an error, so I subscribe to second
  • subscribe to second -> it completes without any error, so emit onCompleted
  • subscription to third stream should never happen as second stream completed successfully

Marble:

--o---o----x--------------------
-------------o-------o-----|----
--------------------------------
  
--o---o------o-------o-----|----

I tried to use onerrorresumenext, but don't know, how to connect it with concat operator.

rx-java
1 Answers

You need concatMap for sure, capture any potential errors in intermediate sources, share it between operators, then reintroduce the error(s) at the very end.

List<Observable<Integer>> sources = new ArrayList<>();

for (int i = 0; i < 10; i++) {
    final int j = i;
    if (i == 5) {
        sources.add(Observable.just(i)
                .doOnSubscribe(d -> System.out.println("Subscribed: " + j)));
    } else {
        sources.add(Observable.<Integer>error(new Exception("" + i))
                .doOnSubscribe(d -> System.out.println("Subscribed: " + j))
        );
    }
}
// -------------------------------------------------------

Observable.defer(() -> {
    AtomicReference<Throwable> lastError = new AtomicReference<>();
    AtomicBoolean lastSuccessful = new AtomicReference<>();
    return Observable.fromIterable(sources)
                     .takeWhile(t -> !lastSuccessful.get())
                     .concatMap(source -> {
                         return source
                                .doOnComplete(() -> lastSuccessful.set(true))
                                .doOnError(e -> lastError.set(e))
                                .onErrorComplete();
                     })
                     .concatWith(Observable.defer(() -> {
                         if (lastSuccessful.get()) {
                             return Observable.empty();
                         }
                         return Observable.error(lastError.get());
                     }));

})
.test()
.assertResult(5);
user_611580
Related