Use a thread-safe Jedis pool

Author: jkodumal

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

@@ -1,5 +1,6 @@
 package com.launchdarkly.client;
 
+import java.io.Closeable;
 import java.util.Map;
 
 /**
@@ -13,7 +14,7 @@
  * of features based on update messages that may be received out-of-order.
  *
  */
-public interface FeatureStore {
+public interface FeatureStore extends Closeable {
   /**
    *
    * Returns the {@link com.launchdarkly.client.FeatureRep} to which the specified key is mapped, or
@@ -71,4 +72,5 @@ public interface FeatureStore {
    * @return true if this store has been initialized
    */
   boolean initialized();
+
 }

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

@@ -1,5 +1,6 @@
 package com.launchdarkly.client;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -141,4 +142,14 @@ public void upsert(String key, FeatureRep<?> feature) {
   public boolean initialized() {
     return initialized;
   }
+
+  /**
+   * Does nothing; this class does not have any resources to release
+   * @throws IOException
+   */
+  @Override
+  public void close() throws IOException
+  {
+    return;
+  }
 }

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

@@ -7,8 +7,11 @@
 import com.google.gson.reflect.TypeToken;
 import org.apache.http.util.EntityUtils;
 import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.JedisPoolConfig;
 import redis.clients.jedis.Transaction;
 
+import java.io.IOException;
 import java.lang.reflect.Type;
 import java.net.URI;
 import java.util.HashMap;
@@ -22,7 +25,7 @@
  */
 public class RedisFeatureStore implements FeatureStore {
   private static final String DEFAULT_PREFIX = "launchdarkly";
-  private final Jedis jedis;
+  private final JedisPool pool;
   private LoadingCache<String, FeatureRep<?>> cache;
   private LoadingCache<String, Boolean> initCache;
   private String prefix;
@@ -36,7 +39,7 @@ public class RedisFeatureStore implements FeatureStore {
    * @param cacheTimeSecs an optional timeout for the in-memory cache. If set to 0, no in-memory caching will be performed
    */
   public RedisFeatureStore(String host, int port, String prefix, long cacheTimeSecs) {
-    jedis = new Jedis(host, port);
+    pool = new JedisPool(new JedisPoolConfig(), host, port);
     setPrefix(prefix);
     createCache(cacheTimeSecs);
     createInitCache(cacheTimeSecs);
@@ -50,7 +53,7 @@ public RedisFeatureStore(String host, int port, String prefix, long cacheTimeSec
    * @param cacheTimeSecs an optional timeout for the in-memory cache. If set to 0, no in-memory caching will be performed
    */
   public RedisFeatureStore(URI uri, String prefix, long cacheTimeSecs) {
-    jedis = new Jedis(uri);
+    pool = new JedisPool(new JedisPoolConfig(), uri);
     setPrefix(prefix);
     createCache(cacheTimeSecs);
     createInitCache(cacheTimeSecs);
@@ -61,7 +64,7 @@ public RedisFeatureStore(URI uri, String prefix, long cacheTimeSecs) {
    *
    */
   public RedisFeatureStore() {
-    jedis = new Jedis("localhost");
+    pool = new JedisPool(new JedisPoolConfig(), "localhost");
     this.prefix = DEFAULT_PREFIX;
   }
 
@@ -128,7 +131,7 @@ public FeatureRep<?> get(String key) {
    */
   @Override
   public Map<String, FeatureRep<?>> all() {
-    Map<String,String> featuresJson = jedis.hgetAll(featuresKey());
+    Map<String,String> featuresJson = jedis().hgetAll(featuresKey());
     Map<String, FeatureRep<?>> result = new HashMap<String, FeatureRep<?>>();
     Gson gson = new Gson();
 
@@ -149,6 +152,7 @@ public Map<String, FeatureRep<?>> all() {
    */
   @Override
   public void init(Map<String, FeatureRep<?>> features) {
+    Jedis jedis = jedis();
     Gson gson = new Gson();
     Transaction t = jedis.multi();
 
@@ -172,6 +176,7 @@ public void init(Map<String, FeatureRep<?>> features) {
    */
   @Override
   public void delete(String key, int version) {
+    Jedis jedis = jedis();
     try {
       Gson gson = new Gson();
       jedis.watch(featuresKey());
@@ -206,6 +211,7 @@ public void delete(String key, int version) {
    */
   @Override
   public void upsert(String key, FeatureRep<?> feature) {
+    Jedis jedis = jedis();
     try {
       Gson gson = new Gson();
       jedis.watch(featuresKey());
@@ -245,18 +251,28 @@ public boolean initialized() {
     return getInit();
   }
 
+  /**
+   * Releases all resources associated with the store. The store must no longer be used once closed.
+   * @throws IOException
+   */
+  public void close() throws IOException
+  {
+    pool.destroy();
+  }
+
+
 
   private String featuresKey() {
     return prefix + ":features";
   }
 
   private Boolean getInit() {
-    return jedis.exists(featuresKey());
+    return jedis().exists(featuresKey());
   }
 
   private FeatureRep<?> getRedis(String key) {
     Gson gson = new Gson();
-    String featureJson = jedis.hget(featuresKey(), key);
+    String featureJson = jedis().hget(featuresKey(), key);
 
     if (featureJson == null) {
       return null;
@@ -267,4 +283,8 @@ private FeatureRep<?> getRedis(String key) {
 
     return f.deleted ? null : f;
   }
+
+  private final Jedis jedis() {
+    return pool.getResource();
+  }
 }

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

@@ -101,6 +101,9 @@ public void close() throws IOException {
     if (es != null) {
       es.close();
     }
+    if (store != null) {
+      store.close();
+    }
   }
 
   boolean initialized() {