code: 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);
                      }
                  });
              }
          };
      }
docs: 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.