private static Observable.Transformer<IndexInfo, IndexInfo> safeAbort(final long watchTimeout, final TimeUnit watchTimeUnit,
            final String indexName) {
        return new Observable.Transformer<IndexInfo, IndexInfo>() {
            @Override
            public Observable<IndexInfo> call(Observable<IndexInfo> source) {
                return source.timeout(watchTimeout, watchTimeUnit)
                .onErrorResumeNext(new Func1<Throwable, rx.Observable<IndexInfo>>() {
                    @Override
                    public rx.Observable<IndexInfo> call(Throwable t) {
                        if (t instanceof TimeoutException) {
                            if (indexName == null) {
                                INDEX_WATCH_LOG.debug("Watched indexes were not all online after the given {} {}", watchTimeout, watchTimeUnit);
                            } else {
                                INDEX_WATCH_LOG.debug("Index {} was not online after the given {} {}", indexName, watchTimeout, watchTimeUnit);
                            }
                            return Observable.empty();
                        }
                        //should not happen with the INDEX_WATCH_MAX_ATTEMPTS set close to Integer.MAX_VALUE, but in case it is later tuned down...
                        if (t instanceof CannotRetryException && t.getCause() instanceof IndexesNotReadyException) {
                            INDEX_WATCH_LOG.debug("{} after {} attempts", INDEX_WATCH_MAX_ATTEMPTS, t.getCause().getMessage());
                            return Observable.empty();
                        }
                        return rx.Observable.error(t);
                    }
                });
            }
        };
    }
private static Observable . Transformer < IndexInfo , IndexInfo > safeAbort ( final long watchTimeout , final TimeUnit watchTimeUnit , final String indexName ) { return new Observable . Transformer < IndexInfo , IndexInfo > ( ) { @ Override public Observable < IndexInfo > call ( Observable < IndexInfo > source ) { return source . timeout ( watchTimeout , watchTimeUnit ) . onErrorResumeNext ( new Func1 < Throwable , rx . Observable < IndexInfo > > ( ) { @ Override public rx . Observable < IndexInfo > call ( Throwable t ) { if ( t instanceof TimeoutException ) { if ( indexName == null ) { INDEX_WATCH_LOG . debug ( "Watched indexes were not all online after the given {} {}" , watchTimeout , watchTimeUnit ) ; } else { INDEX_WATCH_LOG . debug ( "Index {} was not online after the given {} {}" , indexName , watchTimeout , watchTimeUnit ) ; } return Observable . empty ( ) ; } if ( t instanceof CannotRetryException && t . getCause ( ) instanceof IndexesNotReadyException ) { INDEX_WATCH_LOG . debug ( "{} after {} attempts" , INDEX_WATCH_MAX_ATTEMPTS , t . getCause ( ) . getMessage ( ) ) ; return Observable . empty ( ) ; } return rx . Observable . error ( t ) ; } } ) ; } } ; }
A transformer (to be used with {@link Observable#compose(Observable.Transformer)}) that puts a timeout on an
{@link IndexInfo} Observable and will consider {@link IndexesNotReadyException} inside a
{@link CannotRetryException} and {@link TimeoutException} to be non-failing cases (resulting in an empty
observable), whereas all other errors are propagated as is.

@param watchTimeout the timeout to set.
@param watchTimeUnit the time unit for the timeout.
@param indexName null for a global timeout, or the name of the single index being watched.