[automatic-failover] Improve extensibility that is needed for automatic-failover feature

src/main/java/io/lettuce/core/CommandListenerWriter.java

@@ -45,7 +45,7 @@
  * @since 6.1
  */
 @SuppressWarnings("unchecked")
-public class CommandListenerWriter implements RedisChannelWriter {
+public class CommandListenerWriter implements RedisChannelWriter, Delegating<RedisChannelWriter> {
 
     private final RedisChannelWriter delegate;
 

src/main/java/io/lettuce/core/Delegating.java

@@ -0,0 +1,35 @@
+package io.lettuce.core;
+
+/**
+ * A delegating interface that allows access to the underlying delegate.
+ * 
+ * @param <T> the type of the delegate.
+ * 
+ * @author Ali Takavci
+ * @since 7.1
+ */
+public interface Delegating<T> {
+
+    /**
+     * The underlying delegate.
+     *
+     * @return never {@code null}.
+     */
+    T getDelegate();
+
+    /**
+     * Unwrap the underlying delegate, through the recursively wrapped {@link Delegating} interfaces if available.
+     *
+     * @return the unwrapped delegate.
+     */
+    default T unwrap() {
+        T delegate = getDelegate();
+        if (delegate instanceof Delegating) {
+            @SuppressWarnings("unchecked")
+            T unwrapped = ((Delegating<T>) delegate).unwrap();
+            return unwrapped;
+        }
+        return delegate;
+    }
+
+}

src/main/java/io/lettuce/core/RedisClient.java

@@ -275,7 +275,7 @@ private <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectStandalone
 
         logger.debug("Trying to get a Redis connection for: {}", redisURI);
 
-        DefaultEndpoint endpoint = new DefaultEndpoint(getOptions(), getResources());
+        DefaultEndpoint endpoint = createEndpoint();
         RedisChannelWriter writer = endpoint;
 
         if (CommandExpiryWriter.isSupported(getOptions())) {
@@ -408,7 +408,7 @@ private <K, V> ConnectionFuture<StatefulRedisPubSubConnection<K, V>> connectPubS
         assertNotNull(codec);
         checkValidRedisURI(redisURI);
 
-        PubSubEndpoint<K, V> endpoint = new PubSubEndpoint<>(getOptions(), getResources());
+        PubSubEndpoint<K, V> endpoint = createPubSubEndpoint();
         RedisChannelWriter writer = endpoint;
 
         if (CommandExpiryWriter.isSupported(getOptions())) {
@@ -575,7 +575,7 @@ private <K, V> ConnectionFuture<StatefulRedisSentinelConnection<K, V>> doConnect
         connectionBuilder.clientOptions(ClientOptions.copyOf(getOptions()));
         connectionBuilder.clientResources(getResources());
 
-        DefaultEndpoint endpoint = new DefaultEndpoint(getOptions(), getResources());
+        DefaultEndpoint endpoint = createEndpoint();
         RedisChannelWriter writer = endpoint;
 
         if (CommandExpiryWriter.isSupported(getOptions())) {
@@ -840,4 +840,22 @@ private void checkForRedisURI() {
         checkValidRedisURI(this.redisURI);
     }
 
+    /**
+     * Create a new {@link DefaultEndpoint}. Subclasses may override this method to change the default behavior.
+     *
+     * @return a new {@link DefaultEndpoint}.
+     */
+    protected DefaultEndpoint createEndpoint() {
+        return new DefaultEndpoint(getOptions(), getResources());
+    }
+
+    /**
+     * Create a new {@link PubSubEndpoint}. Subclasses may override this method to change the default behavior.
+     *
+     * @return a new {@link PubSubEndpoint}.
+     */
+    protected <K, V> PubSubEndpoint<K, V> createPubSubEndpoint() {
+        return new PubSubEndpoint<>(getOptions(), getResources());
+    }
+
 }

src/main/java/io/lettuce/core/cluster/NodeSelectionInvocationHandler.java

@@ -172,13 +172,13 @@ private Object doInvoke(Object[] args, Method targetMethod, StatefulRedisConnect
         return targetMethod.invoke(executionModel == ExecutionModel.REACTIVE ? it.reactive() : it.async(), argsToUse);
     }
 
-    @SuppressWarnings("unchecked")
+    @SuppressWarnings({ "unchecked", "rawtypes" })
     private Object getExecutions(Map<RedisClusterNode, Object> executions, long timeoutNs)
             throws ExecutionException, InterruptedException {
 
         if (executionModel == ExecutionModel.REACTIVE) {
             Map<RedisClusterNode, CompletionStage<? extends Publisher<?>>> reactiveExecutions = (Map) executions;
-            return new ReactiveExecutionsImpl<>(reactiveExecutions);
+            return new ReactiveExecutionsImpl(reactiveExecutions);
         }
 
         Map<RedisClusterNode, CompletionStage<?>> asyncExecutions = (Map) executions;

src/main/java/io/lettuce/core/protocol/CommandExpiryWriter.java

@@ -28,6 +28,7 @@
 import java.util.concurrent.TimeUnit;
 
 import io.lettuce.core.ClientOptions;
+import io.lettuce.core.Delegating;
 import io.lettuce.core.RedisChannelWriter;
 import io.lettuce.core.TimeoutOptions;
 import io.lettuce.core.internal.ExceptionFactory;
@@ -45,7 +46,7 @@
  * @since 5.1
  * @see io.lettuce.core.TimeoutOptions
  */
-public class CommandExpiryWriter implements RedisChannelWriter {
+public class CommandExpiryWriter implements RedisChannelWriter, Delegating<RedisChannelWriter> {
 
     private final RedisChannelWriter delegate;