ObservingCache - scheduled task executor that returns a result I wrote a utility class called ObservingCache, which receives a task in the form of Supplier&lt;T&gt; and an interval, computes the task every [interval] millis and allows its result to be queried via the getItems() method.
I'm using for making sure that outside resources (such as configuration files or DB tables) are always in sync with the app and can be modified without having to restart the application.

The code is as follows:


public class ObservingCache&lt;T&gt; {
  private static final int DEFAULT_CACHE_REFRESH_INTERVAL = 10 * 60 * 1000; // 10 minutes
  private static final int DEFAULT_THREAD_POOL_SIZE = 10;

  private static volatile ScheduledExecutorService executor;

  protected T items;

  public ObservingCache(Supplier&lt;? extends T&gt; syncFunc) {
      this(syncFunc, DEFAULT_CACHE_REFRESH_INTERVAL, true);
  }

  public ObservingCache(Supplier&lt;? extends T&gt; syncFunc, boolean firstRunBlocking) {
      this(syncFunc, DEFAULT_CACHE_REFRESH_INTERVAL, firstRunBlocking);
  }

  public ObservingCache(Supplier&lt;? extends T&gt; syncFunc, int intervalMillis) {
      this(syncFunc, intervalMillis, true);
  }

  public ObservingCache(Supplier&lt;? extends T&gt; syncFunc, int intervalMillis, boolean firstRunBlocking) {
      initExecutor();
      Runnable task = () -&gt; {
          T result = syncFunc.get();
          if (result != null) {
              items = result;
          }
      };
      if (firstRunBlocking) {         
          task.run(); // First run is blocking (saves a lot of trouble later).
      }
      executor.scheduleAtFixedRate(task, firstRunBlocking ? intervalMillis : 0, intervalMillis, TimeUnit.MILLISECONDS);
  }

  private void initExecutor() {
      if (executor == null || executor.isShutdown()) {
          synchronized (this) {
              if (executor == null || executor.isShutdown()) {
                  executor = Executors.newScheduledThreadPool(DEFAULT_THREAD_POOL_SIZE);
              }
          }
      }
  }

  public T getItems() {
      return items;
  }
}



I'd really like to hear your opinion about this piece of code, especially regarding the subject of potential memory leaks caused by either a bug of mine or any known issue with Java 8's ScheduledExecutorService class.
Class is open to modification

ObservingCache should be closed for modification, this is the OCP principle. If it's not, its behaviour can be altered (by inheritance for example) and can lead to an unpredictable behaviour which make it rather difficult to test and maintain.

How to achieve that ? Mark ObservableCache final.

Executor never shutdowned

Once started, the executor can never get shutdown, which causes the task to be run indefinitly.

How to achieve that ? Create a method:

/**
 * Once called, the cached value won't be updated anymore.
 */
public void stopObserving() {
    executor.shutdownNow();
}


firstRunBlocking

This parameter is the thing that bothers me the most in your class, and also the one who gave me the most trouble when trying to remove it.

Why does it bothers me ? Everytime you have a boolean in a method (or constructor), it's a sign of poor design because the class/method should do 2 things now (one for each boolean's value), therefore violating the single responsibility principle. It makes your code hardest to test and to maintain because of the conditionnal logic flow.

If I understand correctly, the purpose of this flag is to avoid client code retrieving a null when calling getItem because the cache has not been updated at least once. I would rather resolve this problem by blocking in getItem as long as the cache has not been computed once. In Java, one have the CountDownLatch who is a thread-safe class able to fulfill this responsibility by blocking only until the cache is computed the first time.

I came up with the following wrapper to achieve that:

public class BlockingHolder&lt;T&gt; {
    private T value;
    private final CountDownLatch barrier = new CountDownLatch(1);

    public T get() {
        try {
            barrier.await();
            return value;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public void set(T value) {
        this.value = value;
        barrier.countDown();
    }
}


Final solution

All these remarks put together, I came up with the following solution. Note that BlockingHolder is an inner class because it doesn't have to be known in the outside world.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

public final class ObservingCache&lt;T&gt; {
    private final BlockingHolder&lt;T&gt; holder;
    private final ScheduledExecutorService executor;

    /**
     * The cache will be refreshed every 10 minutes
     */
    public ObservingCache(Supplier&lt;? extends T&gt; syncFunc) {
        this(syncFunc, 10 * 60 * 1000);
    }

    public ObservingCache(Supplier&lt;? extends T&gt; syncFunc, int refreshIntervalMillis) {
        this.holder = new BlockingHolder&lt;&gt;();
        this.executor = Executors.newScheduledThreadPool(1);
        executor.scheduleAtFixedRate(() -&gt; holder.set(syncFunc.get()), 0, refreshIntervalMillis, TimeUnit.MILLISECONDS);
    }

    /**
     * Blocks until the cached value has been computed at least once
     */
    public T getItem() {
        return holder.get();
    }

    /**
     * Once called, the cached value won't be updated anymore.
     */
    public void stopObserving() {
        executor.shutdownNow();
    }

    private static class BlockingHolder&lt;T&gt; {
        private T value;
        private final CountDownLatch barrier = new CountDownLatch(1);

        public T get() {
            try {
                barrier.await();
                return value;
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        public void set(T value) {
            this.value = value;
            barrier.countDown();
        }
    }
}




Update: taking into account firstRunBlocking

This is how I would fully replace a boolean into a business class. It looks like overkill but in reality it's not. Each class has a single responsibility, is testable and maintainable.

ObservingCache is now abstract because the "blocking" behaviour can't be defined here but only in suclasses.

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

public abstract class ObservingCache&lt;T&gt; {
    /**
     * 10 minutes
     */
    public static final int DEFAULT_CACHE_REFRESH_INTERVAL = 10 * 60 * 1000;

    private final ScheduledExecutorService executor;

    public ObservingCache(Supplier&lt;? extends T&gt; syncFunc, int refreshIntervalMillis) {
        this.executor = Executors.newScheduledThreadPool(1);
        executor.scheduleAtFixedRate(() -&gt; setItem(syncFunc.get()), 0, refreshIntervalMillis, TimeUnit.MILLISECONDS);
    }

    public abstract T getItem();

    protected abstract void setItem(T value);

    /**
     * Once called, the cached value won't be updated anymore.
     */
    public final void stopObserving() {
        executor.shutdownNow();
    }
}


BlockingCache contains only notion of blocking until the value is not computed. How the value is computed is totally abstracted here.

import java.util.concurrent.CountDownLatch;
import java.util.function.Supplier;

public final class BlockingCache&lt;T&gt; extends ObservingCache&lt;T&gt; {
    private final BlockingHolder&lt;T&gt; holder;

    public BlockingCache(Supplier&lt;? extends T&gt; syncFunc, int refreshIntervalMillis) {
        super(syncFunc, refreshIntervalMillis);
        this.holder = new BlockingHolder&lt;&gt;();
    }

    /**
     * Blocks until the cached value has been computed at least once
     */
    @Override
    public T getItem() {
        return holder.get();
    }

    @Override
    protected void setItem(T value) {
        holder.set(value);
    }

    private static class BlockingHolder&lt;T&gt; {
        private T value;
        private final CountDownLatch barrier = new CountDownLatch(1);

        public T get() {
            try {
                barrier.await();
                return value;
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        public void set(T value) {
            this.value = value;
            barrier.countDown();
        }
    }
}


And finally, the cache that doesn't block but can return null (that I personnaly would never use):

import java.util.function.Supplier;

public final class NonBlockingCache&lt;T&gt; extends ObservingCache&lt;T&gt; {
    private volatile T value;

    public NonBlockingCache(Supplier&lt;? extends T&gt; syncFunc, int refreshIntervalMillis) {
        super(syncFunc, refreshIntervalMillis);
    }

    /**
     * Never blocks. Returns null if the value has not been computed !
     * Otherwise, returns the cached value.
     */
    @Override
    public T getItem() {
        return value;
    }

    @Override
    protected void setItem(T value) {
        this.value = value;
    }
}