Ignore errors in an Observable<Single<T>>

Viewed 348

I have an Observable<Single<T>> that I would like to transform that to a new Observable<T> where any Single<T> that fails is ignored.

Here is my attempt at an implementation:

public static <T> Observable<T> skipErrors(final Observable<Single<T>> xs) {
    Preconditions.checkNotNull(xs);
    return xs.flatMap(single -> single.map(Optional::of)
        .onErrorReturn(error -> Optional.empty())
        .flatMapObservable(optional ->
            optional.map(Observable::just).orElseGet(Observable::empty)));
}

Basically it wraps every success in an Optional and maps every failure to Optional.empty(). The optionals are then filtered in a flatMapObservable.

Is there a more idiomatic way to do this?


Unit-test:

final Observable<Single<String>> observable = skipErrors(Observable.just(
    Single.error(new Exception()),
    Single.error(new Exception()),
    Single.just("Hello"),
    Single.error(new Exception()),
    Single.just("world"),
    Single.error(new Exception())));

final ImmutableList<String> expected = ImmutableList.of("Hello", "world");
final ImmutableList<String> actual = observable.toList()
    .blockingGet()
    .stream()
    .collect(ImmutableList.toImmutableList());

assertEquals(expected, actual);
1 Answers
Related