code: public static <T> CloseableFlowableWithReset<T> cache(final Flowable<T> source,
              final long duration, final TimeUnit unit, final Scheduler scheduler) {
          final AtomicReference<CachedFlowable<T>> cacheRef = new AtomicReference<CachedFlowable<T>>();
          final AtomicReference<Optional<Scheduler.Worker>> workerRef = new AtomicReference<Optional<Scheduler.Worker>>(
                  Optional.<Scheduler.Worker>absent());
          CachedFlowable<T> cache = new CachedFlowable<T>(source);
          cacheRef.set(cache);
          Runnable closeAction = new Runnable() {
              @Override
              public void run() {
                  while (true) {
                      Optional<Scheduler.Worker> w = workerRef.get();
                      if (w == null) {
                          // we are finished
                          break;
                      } else {
                          if (workerRef.compareAndSet(w, null)) {
                              if (w.isPresent()) {
                                  w.get().dispose();
                              }
                              // we are finished
                              workerRef.set(null);
                              break;
                          }
                      }
                      // if not finished then try again
                  }
              }
          };
          Runnable resetAction = new Runnable() {
  
              @Override
              public void run() {
                  startScheduledResetAgain(duration, unit, scheduler, cacheRef, workerRef);
              }
          };
          return new CloseableFlowableWithReset<T>(cache, closeAction, resetAction);
      }
docs: Returns a cached {@link Flowable} like {@link Flowable#cache()} except that
  the cache may be reset by the user calling
  {@link CloseableFlowableWithReset#reset}.
  
  @param source
  the source observable
  @param duration
  duration till next reset
  @param unit
  units corresponding to the duration
  @param scheduler
  scheduler to use for scheduling reset.
  @param <T>
  generic type of source observable
  @return {@link CloseableFlowableWithReset} that should be closed once
  finished to prevent worker memory leak.