Support white-list and black-list commands and keys (#3755)

* Create csc package

* Create csc.util package

* Create a config interface for client-side caching

* Default isCacheable

* Config to WhiteList/BlackList commands and String keys

* Create csc test package(s)

* Test white-list/black-list commands and keys

* Merge fix

* Remove csc.util package

* Fix javadoc links

* Added ClientSideCacheable interface and removed ClientSideCacheConfig interface

* Format imports

* Re-create csc.util package

* Rename to allow/deny instead of white/black

src/main/java/redis/clients/jedis/Connection.java

@@ -18,6 +18,7 @@
 import redis.clients.jedis.args.ClientAttributeOption;
 import redis.clients.jedis.args.Rawable;
 import redis.clients.jedis.commands.ProtocolCommand;
+import redis.clients.jedis.csc.ClientSideCache;
 import redis.clients.jedis.exceptions.JedisConnectionException;
 import redis.clients.jedis.exceptions.JedisDataException;
 import redis.clients.jedis.exceptions.JedisException;

src/main/java/redis/clients/jedis/ConnectionFactory.java

@@ -6,7 +6,7 @@
 import org.apache.commons.pool2.impl.DefaultPooledObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
+import redis.clients.jedis.csc.ClientSideCache;
 import redis.clients.jedis.exceptions.JedisException;
 
 /**

src/main/java/redis/clients/jedis/ConnectionPool.java

@@ -2,6 +2,7 @@
 
 import org.apache.commons.pool2.PooledObjectFactory;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import redis.clients.jedis.csc.ClientSideCache;
 import redis.clients.jedis.util.Pool;
 
 public class ConnectionPool extends Pool<Connection> {

src/main/java/redis/clients/jedis/JedisCluster.java

@@ -7,6 +7,7 @@
 
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import redis.clients.jedis.providers.ClusterConnectionProvider;
+import redis.clients.jedis.csc.ClientSideCache;
 import redis.clients.jedis.util.JedisClusterCRC16;
 
 public class JedisCluster extends UnifiedJedis {

src/main/java/redis/clients/jedis/JedisClusterInfoCache.java

@@ -21,6 +21,7 @@
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import redis.clients.jedis.csc.ClientSideCache;
 import redis.clients.jedis.exceptions.JedisClusterOperationException;
 import redis.clients.jedis.exceptions.JedisException;
 import redis.clients.jedis.util.SafeEncoder;

src/main/java/redis/clients/jedis/JedisPooled.java

@@ -7,6 +7,7 @@
 
 import org.apache.commons.pool2.PooledObjectFactory;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import redis.clients.jedis.csc.ClientSideCache;
 import redis.clients.jedis.providers.PooledConnectionProvider;
 import redis.clients.jedis.util.JedisURIHelper;
 import redis.clients.jedis.util.Pool;

src/main/java/redis/clients/jedis/JedisSentineled.java

@@ -2,6 +2,7 @@
 
 import java.util.Set;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import redis.clients.jedis.csc.ClientSideCache;
 import redis.clients.jedis.providers.SentineledConnectionProvider;
 
 public class JedisSentineled extends UnifiedJedis {

src/main/java/redis/clients/jedis/Protocol.java

@@ -11,6 +11,7 @@
 import redis.clients.jedis.exceptions.*;
 import redis.clients.jedis.args.Rawable;
 import redis.clients.jedis.commands.ProtocolCommand;
+import redis.clients.jedis.csc.ClientSideCache;
 import redis.clients.jedis.util.KeyValue;
 import redis.clients.jedis.util.RedisInputStream;
 import redis.clients.jedis.util.RedisOutputStream;

src/main/java/redis/clients/jedis/UnifiedJedis.java

@@ -17,6 +17,7 @@
 import redis.clients.jedis.commands.SampleBinaryKeyedCommands;
 import redis.clients.jedis.commands.SampleKeyedCommands;
 import redis.clients.jedis.commands.RedisModuleCommands;
+import redis.clients.jedis.csc.ClientSideCache;
 import redis.clients.jedis.exceptions.JedisException;
 import redis.clients.jedis.executors.*;
 import redis.clients.jedis.gears.TFunctionListParams;
@@ -300,11 +301,11 @@ public void setBroadcastAndRoundRobinConfig(JedisBroadcastAndRoundRobinConfig co
   }
 
   private <T> T checkAndClientSideCacheCommand(CommandObject<T> command, Object... keys) {
-    if (clientSideCache == null) {
-      return executeCommand(command);
+    if (clientSideCache != null) {
+      return clientSideCache.get((cmd) -> executeCommand(cmd), command, keys);
     }
 
-    return clientSideCache.getValue((cmd) -> executeCommand(cmd), command, keys);
+    return executeCommand(command);
   }
 
   public String ping() {

src/main/java/redis/clients/jedis/ClientSideCache.java → src/main/java/redis/clients/jedis/csc/ClientSideCache.java

@@ -1,46 +1,51 @@
-package redis.clients.jedis;
+package redis.clients.jedis.csc;
 
 import java.nio.ByteBuffer;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
-
 import java.util.function.Function;
+import redis.clients.jedis.CommandObject;
 import redis.clients.jedis.util.SafeEncoder;
 
 /**
  * The class to manage the client-side caching. User can provide any of implementation of this class to the client
- * object; e.g. {@link redis.clients.jedis.util.CaffeineCSC CaffeineCSC} or
- * {@link redis.clients.jedis.util.GuavaCSC GuavaCSC} or a custom implementation of their own.
+ * object; e.g. {@link redis.clients.jedis.csc.util.CaffeineCSC CaffeineCSC} or
+ * {@link redis.clients.jedis.csc.util.GuavaCSC GuavaCSC} or a custom implementation of their own.
  */
 public abstract class ClientSideCache {
 
   protected static final int DEFAULT_MAXIMUM_SIZE = 10_000;
   protected static final int DEFAULT_EXPIRE_SECONDS = 100;
 
-  private final Map<ByteBuffer, Set<Long>> keyToCommandHashes;
+  private final Map<ByteBuffer, Set<Long>> keyToCommandHashes = new ConcurrentHashMap<>();
+  private final ClientSideCacheable cacheable;
 
   protected ClientSideCache() {
-    this.keyToCommandHashes = new ConcurrentHashMap<>();
+    this.cacheable = DefaultClientSideCacheable.INSTANCE;
+  }
+
+  protected ClientSideCache(ClientSideCacheable cacheable) {
+    this.cacheable = cacheable;
   }
 
-  protected abstract void invalidateAllCommandHashes();
+  protected abstract void invalidateAllHashes();
 
-  protected abstract void invalidateCommandHashes(Iterable<Long> hashes);
+  protected abstract void invalidateHashes(Iterable<Long> hashes);
 
-  protected abstract void put(long hash, Object value);
+  protected abstract void putValue(long hash, Object value);
 
-  protected abstract Object get(long hash);
+  protected abstract Object getValue(long hash);
 
-  protected abstract long getCommandHash(CommandObject command);
+  protected abstract long getHash(CommandObject command);
 
   public final void clear() {
     invalidateAllKeysAndCommandHashes();
   }
 
-  final void invalidate(List list) {
+  public final void invalidate(List list) {
     if (list == null) {
       invalidateAllKeysAndCommandHashes();
       return;
@@ -50,7 +55,7 @@ final void invalidate(List list) {
   }
 
   private void invalidateAllKeysAndCommandHashes() {
-    invalidateAllCommandHashes();
+    invalidateAllHashes();
     keyToCommandHashes.clear();
   }
 
@@ -63,23 +68,27 @@ private void invalidateKeyAndRespectiveCommandHashes(Object key) {
 
     Set<Long> hashes = keyToCommandHashes.get(mapKey);
     if (hashes != null) {
-      invalidateCommandHashes(hashes);
+      invalidateHashes(hashes);
       keyToCommandHashes.remove(mapKey);
     }
   }
 
-  final <T> T getValue(Function<CommandObject<T>, T> loader, CommandObject<T> command, Object... keys) {
+  public final <T> T get(Function<CommandObject<T>, T> loader, CommandObject<T> command, Object... keys) {
+
+    if (!cacheable.isCacheable(command.getArguments().getCommand(), keys)) {
+      return loader.apply(command);
+    }
 
-    final long hash = getCommandHash(command);
+    final long hash = getHash(command);
 
-    T value = (T) get(hash);
+    T value = (T) getValue(hash);
     if (value != null) {
       return value;
     }
 
     value = loader.apply(command);
     if (value != null) {
-      put(hash, value);
+      putValue(hash, value);
       for (Object key : keys) {
         ByteBuffer mapKey = makeKeyForKeyToCommandHashes(key);
         if (keyToCommandHashes.containsKey(mapKey)) {

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

@@ -0,0 +1,8 @@
+package redis.clients.jedis.csc;
+
+import redis.clients.jedis.commands.ProtocolCommand;
+
+public interface ClientSideCacheable {
+
+  boolean isCacheable(ProtocolCommand command, Object... keys);
+}

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

@@ -0,0 +1,15 @@
+package redis.clients.jedis.csc;
+
+import redis.clients.jedis.commands.ProtocolCommand;
+
+public class DefaultClientSideCacheable implements ClientSideCacheable {
+
+  public static final DefaultClientSideCacheable INSTANCE = new DefaultClientSideCacheable();
+
+  public DefaultClientSideCacheable() { }
+
+  @Override
+  public boolean isCacheable(ProtocolCommand command, Object... keys) {
+    return true;
+  }
+}

src/main/java/redis/clients/jedis/csc/util/AllowAndDenyListWithStringKeys.java

@@ -0,0 +1,36 @@
+package redis.clients.jedis.csc.util;
+
+import java.util.Set;
+import redis.clients.jedis.commands.ProtocolCommand;
+import redis.clients.jedis.csc.ClientSideCacheable;
+
+public class AllowAndDenyListWithStringKeys implements ClientSideCacheable {
+
+  private final Set<ProtocolCommand> allowCommands;
+  private final Set<ProtocolCommand> denyCommands;
+
+  private final Set<String> allowKeys;
+  private final Set<String> denyKeys;
+
+  public AllowAndDenyListWithStringKeys(Set<ProtocolCommand> allowCommands, Set<ProtocolCommand> denyCommands,
+      Set<String> allowKeys, Set<String> denyKeys) {
+    this.allowCommands = allowCommands;
+    this.denyCommands = denyCommands;
+    this.allowKeys = allowKeys;
+    this.denyKeys = denyKeys;
+  }
+
+  @Override
+  public boolean isCacheable(ProtocolCommand command, Object... keys) {
+    if (allowCommands != null && !allowCommands.contains(command)) return false;
+    if (denyCommands != null && denyCommands.contains(command)) return false;
+
+    for (Object key : keys) {
+      if (!(key instanceof String)) return false;
+      if (allowKeys != null && !allowKeys.contains((String) key)) return false;
+      if (denyKeys != null && denyKeys.contains((String) key)) return false;
+    }
+
+    return true;
+  }
+}

src/main/java/redis/clients/jedis/util/CaffeineCSC.java → src/main/java/redis/clients/jedis/csc/util/CaffeineCSC.java

@@ -1,54 +1,64 @@
-package redis.clients.jedis.util;
+package redis.clients.jedis.csc.util;
 
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
 import java.util.concurrent.TimeUnit;
 import net.openhft.hashing.LongHashFunction;
-import redis.clients.jedis.ClientSideCache;
+
 import redis.clients.jedis.CommandObject;
 import redis.clients.jedis.args.Rawable;
+import redis.clients.jedis.csc.ClientSideCache;
+import redis.clients.jedis.csc.ClientSideCacheable;
+import redis.clients.jedis.csc.DefaultClientSideCacheable;
 
 public class CaffeineCSC extends ClientSideCache {
 
   private static final LongHashFunction DEFAULT_HASH_FUNCTION = LongHashFunction.xx3();
 
   private final Cache<Long, Object> cache;
-  private final LongHashFunction function;
+  private final LongHashFunction hashFunction;
 
   public CaffeineCSC(Cache<Long, Object> caffeineCache, LongHashFunction hashFunction) {
+    super();
+    this.cache = caffeineCache;
+    this.hashFunction = hashFunction;
+  }
+
+  public CaffeineCSC(Cache<Long, Object> caffeineCache, LongHashFunction function, ClientSideCacheable cacheable) {
+    super(cacheable);
     this.cache = caffeineCache;
-    this.function = hashFunction;
+    this.hashFunction = function;
   }
 
   @Override
-  protected final void invalidateAllCommandHashes() {
+  protected final void invalidateAllHashes() {
     cache.invalidateAll();
   }
 
   @Override
-  protected void invalidateCommandHashes(Iterable<Long> hashes) {
+  protected void invalidateHashes(Iterable<Long> hashes) {
     cache.invalidateAll(hashes);
   }
 
   @Override
-  protected void put(long hash, Object value) {
+  protected void putValue(long hash, Object value) {
     cache.put(hash, value);
   }
 
   @Override
-  protected Object get(long hash) {
+  protected Object getValue(long hash) {
     return cache.getIfPresent(hash);
   }
 
   @Override
-  protected final long getCommandHash(CommandObject command) {
+  protected final long getHash(CommandObject command) {
     long[] nums = new long[command.getArguments().size() + 1];
     int idx = 0;
     for (Rawable raw : command.getArguments()) {
-      nums[idx++] = function.hashBytes(raw.getRaw());
+      nums[idx++] = hashFunction.hashBytes(raw.getRaw());
     }
-    nums[idx] = function.hashInt(command.getBuilder().hashCode());
-    return function.hashLongs(nums);
+    nums[idx] = hashFunction.hashInt(command.getBuilder().hashCode());
+    return hashFunction.hashLongs(nums);
   }
 
   public static Builder builder() {
@@ -63,6 +73,8 @@ public static class Builder {
 
     private LongHashFunction hashFunction = DEFAULT_HASH_FUNCTION;
 
+    private ClientSideCacheable cacheable = DefaultClientSideCacheable.INSTANCE;
+
     private Builder() { }
 
     public Builder maximumSize(int size) {
@@ -80,14 +92,19 @@ public Builder hashFunction(LongHashFunction function) {
       return this;
     }
 
+    public Builder cacheable(ClientSideCacheable cacheable) {
+      this.cacheable = cacheable;
+      return this;
+    }
+
     public CaffeineCSC build() {
       Caffeine cb = Caffeine.newBuilder();
 
       cb.maximumSize(maximumSize);
 
       cb.expireAfterWrite(expireTime, expireTimeUnit);
 
-      return new CaffeineCSC(cb.build(), hashFunction);
+      return new CaffeineCSC(cb.build(), hashFunction, cacheable);
     }
   }
 }