Merge pull request #49 from ivorp2/ivorp/redis-lazy-refresh

RedisFeatureStore config and functionality additions

Author: jkodumal

src/main/java/com/launchdarkly/client/RedisFeatureStore.java

@@ -2,7 +2,11 @@
 
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
+import com.google.common.cache.CacheStats;
 import com.google.common.cache.LoadingCache;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
 import redis.clients.jedis.Jedis;
@@ -15,6 +19,9 @@
 import java.net.URI;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -24,11 +31,13 @@
  */
 public class RedisFeatureStore implements FeatureStore {
   private static final String DEFAULT_PREFIX = "launchdarkly";
+  private static final String INIT_KEY = "$initialized$";
+  private static final String CACHE_REFRESH_THREAD_POOL_NAME_FORMAT = "RedisFeatureStore-cache-refresher-pool-%d";
   private final JedisPool pool;
   private LoadingCache<String, FeatureRep<?>> cache;
   private LoadingCache<String, Boolean> initCache;
   private String prefix;
-  private static final String INIT_KEY = "$initialized$";
+  private ListeningExecutorService executorService;
 
   /**
    * Creates a new store instance that connects to Redis with the provided host, port, prefix, and cache timeout. Uses a default
@@ -38,7 +47,9 @@ public class RedisFeatureStore implements FeatureStore {
    * @param port the port for the Redis connection
    * @param prefix a namespace prefix for all keys stored in Redis
    * @param cacheTimeSecs an optional timeout for the in-memory cache. If set to 0, no in-memory caching will be performed
+   * @deprecated as of 1.1. Please use the {@link RedisFeatureStoreBuilder#build()} for a more flexible way of constructing a {@link RedisFeatureStore}.
    */
+  @Deprecated
   public RedisFeatureStore(String host, int port, String prefix, long cacheTimeSecs) {
     this(host, port, prefix, cacheTimeSecs, getPoolConfig());
   }
@@ -50,20 +61,24 @@ public RedisFeatureStore(String host, int port, String prefix, long cacheTimeSec
    * @param uri the URI for the Redis connection
    * @param prefix a namespace prefix for all keys stored in Redis
    * @param cacheTimeSecs an optional timeout for the in-memory cache. If set to 0, no in-memory caching will be performed
+   * @deprecated as of 1.1. Please use the {@link RedisFeatureStoreBuilder#build()} for a more flexible way of constructing a {@link RedisFeatureStore}.
    */
+  @Deprecated
   public RedisFeatureStore(URI uri, String prefix, long cacheTimeSecs) {
     this(uri, prefix, cacheTimeSecs, getPoolConfig());
   }
 
   /**
-   * Creates a new store instance that connects to Redis with the provided URI, prefix, cache timeout, and connection pool settings.
+   * Creates a new store instance that connects to Redis with the provided host, port, prefix, cache timeout, and connection pool settings.
    *
    * @param host the host for the Redis connection
    * @param port the port for the Redis connection
    * @param prefix a namespace prefix for all keys stored in Redis
    * @param cacheTimeSecs an optional timeout for the in-memory cache. If set to 0, no in-memory caching will be performed
    * @param poolConfig an optional pool config for the Jedis connection pool
+   * @deprecated as of 1.1. Please use the {@link RedisFeatureStoreBuilder#build()} for a more flexible way of constructing a {@link RedisFeatureStore}.
    */
+  @Deprecated
   public RedisFeatureStore(String host, int port, String prefix, long cacheTimeSecs, JedisPoolConfig poolConfig) {
     pool = new JedisPool(poolConfig, host, port);
     setPrefix(prefix);
@@ -78,14 +93,34 @@ public RedisFeatureStore(String host, int port, String prefix, long cacheTimeSec
    * @param prefix a namespace prefix for all keys stored in Redis
    * @param cacheTimeSecs an optional timeout for the in-memory cache. If set to 0, no in-memory caching will be performed
    * @param poolConfig an optional pool config for the Jedis connection pool
+   * @deprecated as of 1.1. Please use the {@link RedisFeatureStoreBuilder#build()} for a more flexible way of constructing a {@link RedisFeatureStore}.
    */
+  @Deprecated
   public RedisFeatureStore(URI uri, String prefix, long cacheTimeSecs, JedisPoolConfig poolConfig) {
     pool = new JedisPool(poolConfig, uri);
     setPrefix(prefix);
     createCache(cacheTimeSecs);
     createInitCache(cacheTimeSecs);
   }
 
+  /**
+   * Creates a new store instance that connects to Redis based on the provided {@link RedisFeatureStoreBuilder}.
+   *
+   * See the {@link RedisFeatureStoreBuilder} for information on available configuration options and what they do.
+   *
+   * @param builder the configured builder to construct the store with.
+   */
+  protected RedisFeatureStore(RedisFeatureStoreBuilder builder) {
+    if (builder.poolConfig == null) {
+      this.pool = new JedisPool(getPoolConfig(), builder.uri, builder.connectTimeout, builder.socketTimeout);
+    } else {
+      this.pool = new JedisPool(builder.poolConfig, builder.uri, builder.connectTimeout, builder.socketTimeout);
+    }
+    setPrefix(builder.prefix);
+    createCache(builder.cacheTimeSecs, builder.refreshStaleValues, builder.asyncRefresh);
+    createInitCache(builder.cacheTimeSecs);
+  }
+
   /**
    * Creates a new store instance that connects to Redis with a default connection (localhost port 6379) and no in-memory cache.
    *
@@ -95,7 +130,6 @@ public RedisFeatureStore() {
     this.prefix = DEFAULT_PREFIX;
   }
 
-
   private void setPrefix(String prefix) {
     if (prefix == null || prefix.isEmpty()) {
       this.prefix = DEFAULT_PREFIX;
@@ -105,21 +139,56 @@ private void setPrefix(String prefix) {
   }
 
   private void createCache(long cacheTimeSecs) {
+    createCache(cacheTimeSecs, false, false);
+  }
+
+  private void createCache(long cacheTimeSecs, boolean refreshStaleValues, boolean asyncRefresh) {
     if (cacheTimeSecs > 0) {
-      cache = CacheBuilder.newBuilder().expireAfterWrite(cacheTimeSecs, TimeUnit.SECONDS).build(new CacheLoader<String, FeatureRep<?>>() {
+      if (refreshStaleValues) {
+        createRefreshCache(cacheTimeSecs, asyncRefresh);
+      } else {
+        createExpiringCache(cacheTimeSecs);
+      }
+    }
+  }
 
-        @Override
-        public FeatureRep<?> load(String key) throws Exception {
-          return getRedis(key);
-        }
-      });
+  private CacheLoader<String, FeatureRep<?>> createDefaultCacheLoader() {
+    return new CacheLoader<String, FeatureRep<?>>() {
+      @Override
+      public FeatureRep<?> load(String key) throws Exception {
+        return getRedis(key);
+      }
+    };
+  }
+
+  /**
+   * Configures the instance to use a "refresh after write" cache. This will not automatically evict stale values, allowing them to be returned if failures
+   * occur when updating them. Optionally set the cache to refresh values asynchronously, which always returns the previously cached value immediately.
+   * @param cacheTimeSecs the length of time in seconds, after a {@link FeatureRep} value is created that it should be refreshed.
+   * @param asyncRefresh makes the refresh asynchronous or not.
+   */
+  private void createRefreshCache(long cacheTimeSecs, boolean asyncRefresh) {
+    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat(CACHE_REFRESH_THREAD_POOL_NAME_FORMAT).setDaemon(true).build();
+    ExecutorService parentExecutor = Executors.newSingleThreadExecutor(threadFactory);
+    executorService = MoreExecutors.listeningDecorator(parentExecutor);
+    CacheLoader<String, FeatureRep<?>> cacheLoader = createDefaultCacheLoader();
+    if (asyncRefresh) {
+      cacheLoader = CacheLoader.asyncReloading(cacheLoader, executorService);
     }
+    cache = CacheBuilder.newBuilder().refreshAfterWrite(cacheTimeSecs, TimeUnit.SECONDS).build(cacheLoader);
+  }
+
+  /**
+   * Configures the instance to use an "expire after write" cache. This will evict stale values and block while loading the latest from Redis.
+   * @param cacheTimeSecs the length of time in seconds, after a {@link FeatureRep} value is created that it should be automatically removed.
+   */
+  private void createExpiringCache(long cacheTimeSecs) {
+    cache = CacheBuilder.newBuilder().expireAfterWrite(cacheTimeSecs, TimeUnit.SECONDS).build(createDefaultCacheLoader());
   }
 
   private void createInitCache(long cacheTimeSecs) {
     if (cacheTimeSecs > 0) {
       initCache = CacheBuilder.newBuilder().expireAfterWrite(cacheTimeSecs, TimeUnit.SECONDS).build(new CacheLoader<String, Boolean>() {
-
         @Override
         public Boolean load(String key) throws Exception {
           return getInit();
@@ -129,7 +198,6 @@ public Boolean load(String key) throws Exception {
   }
 
   /**
-   *
    * Returns the {@link com.launchdarkly.client.FeatureRep} to which the specified key is mapped, or
    * null if the key is not associated or the associated {@link com.launchdarkly.client.FeatureRep} has
    * been deleted.
@@ -148,12 +216,10 @@ public FeatureRep<?> get(String key) {
     }
   }
 
-
   /**
    * Returns a {@link java.util.Map} of all associated features. This implementation does not take advantage
    * of the in-memory cache, so fetching all features will involve a fetch from Redis.
    *
-   *
    * @return a map of all associated features.
    */
   @Override
@@ -171,8 +237,8 @@ public Map<String, FeatureRep<?>> all() {
       }
       return result;
     }
-
   }
+
   /**
    * Initializes (or re-initializes) the store with the specified set of features. Any existing entries
    * will be removed.
@@ -195,9 +261,7 @@ public void init(Map<String, FeatureRep<?>> features) {
     }
   }
 
-
   /**
-   *
    * Deletes the feature associated with the specified key, if it exists and its version
    * is less than or equal to the specified version.
    *
@@ -278,10 +342,26 @@ public boolean initialized() {
    */
   public void close() throws IOException
   {
-    pool.destroy();
+    try {
+      if (executorService != null) {
+        executorService.shutdownNow();
+      }
+    } finally {
+      pool.destroy();
+    }
   }
 
-
+  /**
+   * Return the underlying Guava cache stats object.
+   *
+   * @return the cache statistics object.
+   */
+  public CacheStats getCacheStats() {
+    if (cache != null) {
+      return cache.stats();
+    }
+    return null;
+  }
 
   private String featuresKey() {
     return prefix + ":features";