Add TTL support for CSC

src/main/java/redis/clients/jedis/csc/AbstractCache.java

@@ -23,15 +23,17 @@ public abstract class AbstractCache implements Cache {
   private Cacheable cacheable;
   private final Map<ByteBuffer, Set<CacheKey<?>>> redisKeysToCacheKeys = new ConcurrentHashMap<>();
   private final int maximumSize;
-  private ReentrantLock lock = new ReentrantLock();
+  private final long ttl;
+  ReentrantLock lock = new ReentrantLock();
   private volatile CacheStats stats = new CacheStats();
 
-  protected AbstractCache(int maximumSize) {
-    this(maximumSize, DefaultCacheable.INSTANCE);
+  protected AbstractCache(int maximumSize, long ttl) {
+    this(maximumSize, ttl, DefaultCacheable.INSTANCE);
   }
 
-  protected AbstractCache(int maximumSize, Cacheable cacheable) {
+  protected AbstractCache(int maximumSize, long ttl, Cacheable cacheable) {
     this.maximumSize = maximumSize;
+    this.ttl = ttl;
     this.cacheable = cacheable;
   }
 
@@ -42,6 +44,11 @@ public int getMaxSize() {
     return maximumSize;
   }
 
+  @Override
+  public long getTtl() {
+    return ttl;
+  };
+
   @Override
   public abstract int getSize();
 

src/main/java/redis/clients/jedis/csc/Cache.java

@@ -18,6 +18,11 @@ public interface Cache {
      */
     int getSize();
 
+    /**
+     * @return The ttl of the cache
+     */
+    long getTtl();
+
     /**
      * @return All the entries within the cache
      */
@@ -110,4 +115,9 @@ public interface Cache {
      * @return The compatibility of cache against different Redis versions
      */
     boolean compatibilityMode();
+
+    /**
+     * Releases any system resources associated with it
+     */
+    void close();
 }

src/main/java/redis/clients/jedis/csc/CacheConfig.java

@@ -3,6 +3,7 @@
 public class CacheConfig {
 
     private int maxSize;
+    private long ttl;
     private Cacheable cacheable;
     private EvictionPolicy evictionPolicy;
     private Class cacheClass;
@@ -11,6 +12,10 @@ public int getMaxSize() {
         return maxSize;
     }
 
+    public long getTtl() {
+        return ttl;
+    }
+
     public Cacheable getCacheable() {
         return cacheable;
     }
@@ -28,7 +33,12 @@ public static Builder builder() {
 
     public static class Builder {
         private final int DEFAULT_MAX_SIZE = 10000;
+        /**
+         * 30000ms
+         */
+        private final long DEFAULT_TTL = 30000;
         private int maxSize = DEFAULT_MAX_SIZE;
+        private long ttl = DEFAULT_TTL;
         private Cacheable cacheable = DefaultCacheable.INSTANCE;
         private EvictionPolicy evictionPolicy;
         private Class cacheClass;
@@ -38,6 +48,11 @@ public Builder maxSize(int maxSize) {
             return this;
         }
 
+        public Builder ttl(long ttl) {
+            this.ttl = ttl;
+            return this;
+        }
+
         public Builder evictionPolicy(EvictionPolicy policy) {
             this.evictionPolicy = policy;
             return this;
@@ -56,6 +71,7 @@ public Builder cacheClass(Class cacheClass) {
         public CacheConfig build() {
             CacheConfig cacheConfig = new CacheConfig();
             cacheConfig.maxSize = this.maxSize;
+            cacheConfig.ttl = this.ttl;
             cacheConfig.cacheable = this.cacheable;
             cacheConfig.evictionPolicy = this.evictionPolicy;
             cacheConfig.cacheClass = this.cacheClass;

src/main/java/redis/clients/jedis/csc/CacheConnection.java

@@ -90,7 +90,7 @@ public <T> T executeCommand(final CommandObject<T> commandObject) {
     // CACHE MISS !!
     cache.getStats().miss();
     T value = super.executeCommand(commandObject);
-    cacheEntry = new CacheEntry<>(cacheKey, value, this);
+    cacheEntry = new CacheEntry<>(cacheKey, value, this, cache.getTtl());
     cache.set(cacheKey, cacheEntry);
     // this line actually provides a deep copy of cached object instance 
     value = cacheEntry.getValue();

src/main/java/redis/clients/jedis/csc/CacheEntry.java

@@ -14,11 +14,13 @@ public class CacheEntry<T> {
   private final CacheKey<T> cacheKey;
   private final WeakReference<CacheConnection> connection;
   private final byte[] bytes;
+  private long expireTime;
 
-  public CacheEntry(CacheKey<T> cacheKey, T value, CacheConnection connection) {
+  public CacheEntry(CacheKey<T> cacheKey, T value, CacheConnection connection, long ttl) {
     this.cacheKey = cacheKey;
     this.connection = new WeakReference<>(connection);
     this.bytes = toBytes(value);
+    this.expireTime = System.currentTimeMillis() + ttl;
   }
 
   public CacheKey<T> getCacheKey() {
@@ -53,4 +55,8 @@ private T toObject(byte[] data) {
       throw new JedisCacheException("Failed to deserialize object", e);
     }
   }
+
+  public boolean isExpired() {
+    return System.currentTimeMillis() > expireTime;
+  }
 }

src/main/java/redis/clients/jedis/csc/CacheFactory.java

@@ -13,7 +13,7 @@ public static Cache getCache(CacheConfig config) {
             if (config.getCacheable() == null) {
                 throw new JedisCacheException("Cacheable is required to create the default cache!");
             }
-            return new DefaultCache(config.getMaxSize(), config.getCacheable(), getEvictionPolicy(config));
+            return new DefaultCache(config.getMaxSize(), config.getTtl(), config.getCacheable(), getEvictionPolicy(config));
         }
         return instantiateCustomCache(config);
     }

src/main/java/redis/clients/jedis/csc/DefaultCache.java

@@ -2,34 +2,47 @@
 
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 public class DefaultCache extends AbstractCache {
 
     protected final Map<CacheKey, CacheEntry> cache;
     private final EvictionPolicy evictionPolicy;
+    private final ScheduledExecutorService scheduler;
 
-    protected DefaultCache(int maximumSize) {
-        this(maximumSize, new HashMap<CacheKey, CacheEntry>());
+    protected DefaultCache(int maximumSize, long ttl) {
+        this(maximumSize, ttl, new HashMap<CacheKey, CacheEntry>());
     }
 
-    protected DefaultCache(int maximumSize, Map<CacheKey, CacheEntry> map) {
-        this(maximumSize, map, DefaultCacheable.INSTANCE, new LRUEviction(maximumSize));
+    protected DefaultCache(int maximumSize, long ttl, Map<CacheKey, CacheEntry> map) {
+        this(maximumSize, ttl, map, DefaultCacheable.INSTANCE, new LRUEviction(maximumSize));
     }
 
-    protected DefaultCache(int maximumSize, Cacheable cacheable) {
-        this(maximumSize, new HashMap<CacheKey, CacheEntry>(), cacheable, new LRUEviction(maximumSize));
+    protected DefaultCache(int maximumSize, long ttl, Cacheable cacheable) {
+        this(maximumSize, ttl, new HashMap<CacheKey, CacheEntry>(), cacheable, new LRUEviction(maximumSize));
     }
 
-    protected DefaultCache(int maximumSize, Cacheable cacheable, EvictionPolicy evictionPolicy) {
-        this(maximumSize, new HashMap<CacheKey, CacheEntry>(), cacheable, evictionPolicy);
+    protected DefaultCache(int maximumSize, long ttl, Cacheable cacheable, EvictionPolicy evictionPolicy) {
+        this(maximumSize, ttl, new HashMap<CacheKey, CacheEntry>(), cacheable, evictionPolicy);
     }
 
-    protected DefaultCache(int maximumSize, Map<CacheKey, CacheEntry> map, Cacheable cacheable, EvictionPolicy evictionPolicy) {
-        super(maximumSize, cacheable);
+    protected DefaultCache(int maximumSize, long ttl, Map<CacheKey, CacheEntry> map, Cacheable cacheable, EvictionPolicy evictionPolicy) {
+        super(maximumSize, ttl, cacheable);
         this.cache = map;
         this.evictionPolicy = evictionPolicy;
         this.evictionPolicy.setCache(this);
+        this.scheduler = Executors.newSingleThreadScheduledExecutor();
+        // Periodically clear expired cache every 2 seconds
+        this.scheduler.scheduleAtFixedRate(new Runnable() {
+            @Override
+            public void run() {
+                cleanupExpiredEntries();
+            }
+        }, ttl, 2000, TimeUnit.MILLISECONDS);
     }
 
     @Override
@@ -47,9 +60,18 @@ public EvictionPolicy getEvictionPolicy() {
         return this.evictionPolicy;
     }
 
+    @Override
+    public void close() {
+        this.scheduler.shutdown();
+    }
+
     @Override
     public CacheEntry getFromStore(CacheKey key) {
-        return cache.get(key);
+        CacheEntry entry = cache.get(key);
+        if (entry != null && !entry.isExpired()) {
+            return entry;
+        }
+        return null;
     }
 
     @Override
@@ -72,4 +94,20 @@ protected boolean containsKeyInStore(CacheKey cacheKey) {
         return cache.containsKey(cacheKey);
     }
 
+    private void cleanupExpiredEntries() {
+        this.lock.lock();
+        try {
+            Iterator<Map.Entry<CacheKey, CacheEntry>> iterator = cache.entrySet().iterator();
+            while (iterator.hasNext()) {
+                Map.Entry<CacheKey, CacheEntry> entry = iterator.next();
+                if (entry.getValue().isExpired()) {
+                    iterator.remove();
+                    evictionPolicy.reset(entry.getKey());
+                }
+            }
+        } finally {
+            lock.unlock();
+        }
+    }
+
 }

src/test/java/redis/clients/jedis/csc/ClientSideCacheFunctionalityTest.java

@@ -60,7 +60,7 @@ public void lruEvictionTest() {
     }
 
     Map<CacheKey, CacheEntry> map = new LinkedHashMap<>(count);
-    Cache cache = new DefaultCache(count, map);
+    Cache cache = new DefaultCache(count, 5000, map);
     try (JedisPooled jedis = new JedisPooled(hnp, clientConfig.get(), cache)) {
 
       // Retrieve the 100 keys in the same order
@@ -565,4 +565,27 @@ public void testCacheFactory() throws InterruptedException {
       assertEquals(1, stats.getMissCount());
     }
   }
+
+  @Test
+  public void testCacheEntryExpiredWithUnifiedJedis() throws InterruptedException {
+    Cache cache = new TestCache();
+    UnifiedJedis client = new UnifiedJedis(hnp, clientConfig.get(), cache);
+
+    try {
+      // "foo" is cached
+      client.set("foo", "bar");
+      client.get("foo"); // read from the server
+      Assert.assertEquals("bar", client.get("foo")); // cache hit
+      Assert.assertEquals(1, cache.getSize());
+
+      Thread.sleep(10000); // sleep 10000ms
+
+      Assert.assertEquals(0, cache.getSize()); // The cache "foo bar" has expired
+
+      client.get("foo"); // cache miss, read from the server
+      Assert.assertEquals(1, cache.getSize());
+    } finally {
+      client.close();
+    }
+  }
 }

src/test/java/redis/clients/jedis/csc/TestCache.java

@@ -10,19 +10,19 @@ public TestCache() {
   }
 
   public TestCache(Map<CacheKey, CacheEntry> map) {
-    super(10000, map);
+    super(10000, 5000, map);
   }
 
   public TestCache(Map<CacheKey, CacheEntry> map, Cacheable cacheable) {
-    super(10000, map, cacheable, new LRUEviction(10000));
+    super(10000, 5000, map, cacheable, new LRUEviction(10000));
   }
 
   public TestCache(int maximumSize, EvictionPolicy evictionPolicy ) {
-    super(maximumSize, new HashMap<CacheKey, CacheEntry>(), DefaultCacheable.INSTANCE, evictionPolicy);
+    super(maximumSize, 5000, new HashMap<CacheKey, CacheEntry>(), DefaultCacheable.INSTANCE, evictionPolicy);
   }
 
   public TestCache(int maximumSize, EvictionPolicy evictionPolicy, Cacheable cacheable ) {
-    super(maximumSize, new HashMap<CacheKey, CacheEntry>(), cacheable, evictionPolicy);
+    super(maximumSize, 5000, new HashMap<CacheKey, CacheEntry>(), cacheable, evictionPolicy);
   }
 
 }