make caching configured via options, not an explicit wrapper

Author: craig-day

src/main/asciidoc/index.adoc

@@ -199,14 +199,34 @@ To create a client with client-side caching, one would do:
 {@link examples.RedisExamples#clientCaching1}
 ----
 
-The default implementation will use a simple Least-Recently-Used (LFU) cache backed by a `LinkedHashMap`.
-You can also provide your own implementation of the cache:
+A specific interface, `CachingRedis`, is exposed for a caching client that allows an invalidation handler to be attached or a flush command to be issued.
+The invalidation handler will be invoked with all the keys being invalidated whenever a message is received on the invalidations connection.
+
+To attach an invalidations handler:
 
 [source,$lang]
 ----
 {@link examples.RedisExamples#clientCaching2}
 ----
 
+To manually flush the client's cache store:
+
+[source,$lang]
+----
+{@link examples.RedisExamples#clientCaching3}
+----
+
+The client comes with a default cache store out of the box, but you can write your own if you prefer.
+
+The implementations are expected to follow the `ServiceLoader` conventions and all stores that are available at runtime from the classpath will be exposed.
+When more than 1 implementation is available the first one that can be instantiated and configured with success becomes the default.
+If none is available, then the default is a simple Least-Recently-Used (LRU) cache backed by a `LinkedHashMap`.
+
+[source,$lang]
+----
+{@link examples.RedisExamples#clientCaching4}
+----
+
 NOTE: The cache is not a write-through cache. A value will not be stored in the client-side cache until the value is fetched from Redis for the first time.
 To avoid write-then-read race conditions within the same batch, read commands that are part of a batch will not check the cache first.
 Additionally, the current implementation does not support the `OPTIN` or `NOLOOP` options.

src/main/java/examples/RedisExamples.java

@@ -1,12 +1,20 @@
 package examples;
 
+import io.vertx.codegen.annotations.Nullable;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.tracing.TracingPolicy;
 import io.vertx.redis.client.*;
 
+import io.vertx.redis.client.impl.CachingRedis;
+import io.vertx.redis.client.impl.CachingRedisClient;
+import io.vertx.redis.client.impl.RedisClient;
+import io.vertx.redis.client.impl.cache.CacheKey;
+import io.vertx.redis.client.spi.RedisClientCache;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -248,8 +256,12 @@ public void tracing1(RedisOptions options) {
   }
 
   public void clientCaching1(Vertx vertx) {
-    CachingRedis
-      .create(vertx)
+    Redis.createClient(
+      vertx,
+      new RedisOptions()
+        .setCacheEnabled(true)
+        .setCacheMaxSize(256)
+        .setCacheMaxAge(60_000))
       .connect()
       .onSuccess(conn -> {
         // get the value for a key, returning from a local in-memory cache if
@@ -259,9 +271,59 @@ public void clientCaching1(Vertx vertx) {
       });
   }
 
-  public void clientCaching2(Vertx vertx, RedisClientCache customCache) {
-    CachingRedis
-      .create(vertx, customCache)
+  public void clientCaching2(Redis redis) {
+    CachingRedis cachingClient = (CachingRedis) redis;
+
+    cachingClient.invalidationHandler(keys -> {
+      // something...
+    });
+  }
+
+  public void clientCaching3(Redis redis) {
+    CachingRedis cachingClient = (CachingRedis) redis;
+
+    cachingClient.flush().onSuccess(ignored -> {
+      // Success!
+    });
+  }
+
+  public void clientCaching4(Vertx vertx, RedisClientCache customCache) {
+
+    // Register this class in META-INF/services/io.vertx.redis.client.spi.RedisClientCache
+    class CustomCache implements RedisClientCache {
+
+      private final Map<CacheKey, Response> store = new HashMap<>();
+
+      @Override
+      public @Nullable Response get(CacheKey key) {
+        return store.get(key);
+      }
+
+      @Override
+      public void put(CacheKey key, Response value) {
+        store.put(key, value);
+      }
+
+      @Override
+      public void delete(CacheKey key) {
+        store.remove(key);
+      }
+
+      @Override
+      public void flush() {
+        store.clear();
+      }
+
+      @Override
+      public void close() {
+        // Nothing to do here
+      }
+    }
+
+    Redis.createClient(
+      vertx,
+      new RedisOptions()
+        .setCacheEnabled(true))
       .connect()
       .onSuccess(conn -> {
         // get the value for a key, returning from the custom cache if

src/main/java/io/vertx/redis/client/CachingRedis.java

@@ -19,6 +19,7 @@
 import io.vertx.core.json.JsonObject;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
@@ -36,6 +37,7 @@ public class CachingRedisOptions {
   public static final long DEFAULT_MAX_AGE = 3_600_000; // 1 hour in milliseconds
   public static final TimeUnit DEFAULT_MAX_AGE_UNIT = TimeUnit.MILLISECONDS;
 
+  private boolean enabled;
   private int maxCacheSize;
   private long maxAge;
   private TimeUnit maxAgeUnit;
@@ -46,6 +48,7 @@ public class CachingRedisOptions {
    * Creates a default configuration.
    */
   public CachingRedisOptions() {
+    this.enabled = false;
     this.maxCacheSize = DEFAULT_MAX_SIZE;
     this.maxAge = DEFAULT_MAX_AGE;
     this.maxAgeUnit = DEFAULT_MAX_AGE_UNIT;
@@ -59,6 +62,7 @@ public CachingRedisOptions() {
    * @param other the options to clone
    */
   public CachingRedisOptions(CachingRedisOptions other) {
+    this.enabled = other.enabled;
     this.maxCacheSize = other.maxCacheSize;
     this.maxAge = other.maxAge;
     this.maxAgeUnit = other.maxAgeUnit;
@@ -76,6 +80,24 @@ public CachingRedisOptions(JsonObject json) {
     CachingRedisOptionsConverter.fromJson(json, this);
   }
 
+  /**
+   * @return true if caching is enabled
+   */
+  public boolean getEnabled() {
+    return enabled;
+  }
+
+  /**
+   * Set if caching is enabled.
+   *
+   * @param enabled if caching is enabled
+   * @return fluent self
+   */
+  public CachingRedisOptions setEnabled(boolean enabled) {
+    this.enabled = enabled;
+    return this;
+  }
+
   /**
    * Get the max cache size.
    *
@@ -174,7 +196,7 @@ public List<String> getPrefixes() {
    * @param prefixes the prefix list
    * @return fluent self
    */
-  public CachingRedisOptions setPrefixes(List<String> prefixes) {
+  public CachingRedisOptions setPrefixes(Collection<String> prefixes) {
     this.prefixes.clear();
     this.prefixes.addAll(prefixes);
     return this;
@@ -193,7 +215,7 @@ public CachingRedisOptions setPrefix(String prefix) {
   }
 
   /**
-   * Add a prefix teh server should send invalidation messages for.
+   * Add a prefix the server should send invalidation messages for.
    *
    * @param prefix the prefix to add
    * @return fluent self

src/main/java/io/vertx/redis/client/CachingRedisOptions.java

@@ -19,6 +19,7 @@
 import io.vertx.codegen.annotations.VertxGen;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
+import io.vertx.redis.client.impl.CachingRedisClient;
 import io.vertx.redis.client.impl.RedisClient;
 import io.vertx.redis.client.impl.RedisClusterClient;
 import io.vertx.redis.client.impl.RedisReplicationClient;
@@ -62,18 +63,30 @@ static Redis createClient(Vertx vertx, String connectionString) {
    * @return the client
    */
   static Redis createClient(Vertx vertx, RedisOptions options) {
+    final Redis clientImpl;
+
     switch (options.getType()) {
       case STANDALONE:
-        return new RedisClient(vertx, options.getNetClientOptions(), options.getPoolOptions(), new RedisStandaloneConnectOptions(options), options.getTracingPolicy());
+        clientImpl = new RedisClient(vertx, options.getNetClientOptions(), options.getPoolOptions(), new RedisStandaloneConnectOptions(options), options.getTracingPolicy());
+        break;
       case SENTINEL:
-        return new RedisSentinelClient(vertx, options.getNetClientOptions(), options.getPoolOptions(), new RedisSentinelConnectOptions(options), options.getTracingPolicy());
+        clientImpl = new RedisSentinelClient(vertx, options.getNetClientOptions(), options.getPoolOptions(), new RedisSentinelConnectOptions(options), options.getTracingPolicy());
+        break;
       case CLUSTER:
-        return new RedisClusterClient(vertx, options.getNetClientOptions(), options.getPoolOptions(), new RedisClusterConnectOptions(options), options.getTracingPolicy());
+        clientImpl = new RedisClusterClient(vertx, options.getNetClientOptions(), options.getPoolOptions(), new RedisClusterConnectOptions(options), options.getTracingPolicy());
+        break;
       case REPLICATION:
-        return new RedisReplicationClient(vertx, options.getNetClientOptions(), options.getPoolOptions(), new RedisClusterConnectOptions(options), options.getTracingPolicy());
+        clientImpl = new RedisReplicationClient(vertx, options.getNetClientOptions(), options.getPoolOptions(), new RedisClusterConnectOptions(options), options.getTracingPolicy());
+        break;
       default:
         throw new IllegalStateException("Unknown Redis Client type: " + options.getType());
     }
+
+    if (options.getCacheEnabled()) {
+      return new CachingRedisClient(vertx, clientImpl, options.getCachingOptions());
+    }
+
+    return clientImpl;
   }
 
   /**