Ignore Errors in Observable <Single <T>>

I have Observable<Single<T>>

one that I would like to convert to a new Observable<T>

one where any Single<T>

that fails is ignored.

Here is my attempt at 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 completes each success in Optional

and matches each failure Optional.empty()

. The options are then filtered into 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);