Polishing #1382

Move events from models.events to event.command. Move CommandListener configuration from ClientResources to AbstractRedisClient.addListener. Consider changed writer chain in RedisChannelHandler.setTimeout. Add tests for standalone and Redis Cluster operations. Remove generics from command events to simplify interaction and listener implementation.

Replace generic ExecutionException with ExceptionFactory usage.

Original pull request: #1424.

Author: mp911de

src/main/asciidoc/new-features.adoc

@@ -5,9 +5,9 @@
 == What's new in Lettuce 6.1
 
 * Kotlin Coroutines support for `SCAN`/`HSCAN`/`SSCAN`/`ZSCAN` through `ScanFlow`.
+* Command Listener API through `RedisClient.addListener(CommandListener)`.
 
 [[new-features.6-0-0]]
-
 == What's new in Lettuce 6.0
 
 * Support for RESP3 usage with Redis 6 along with RESP2/RESP3 handshake and protocol version discovery.

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

@@ -22,20 +22,35 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.*;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import reactor.core.publisher.Mono;
 import io.lettuce.core.Transports.NativeTransports;
-import io.lettuce.core.internal.*;
+import io.lettuce.core.event.command.CommandListener;
+import io.lettuce.core.internal.AsyncCloseable;
+import io.lettuce.core.internal.Exceptions;
+import io.lettuce.core.internal.Futures;
+import io.lettuce.core.internal.LettuceAssert;
+import io.lettuce.core.internal.LettuceStrings;
 import io.lettuce.core.protocol.ConnectionWatchdog;
 import io.lettuce.core.protocol.RedisHandshakeHandler;
 import io.lettuce.core.resource.ClientResources;
 import io.lettuce.core.resource.DefaultClientResources;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBufAllocator;
-import io.netty.channel.*;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
 import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
@@ -79,6 +94,8 @@ public abstract class AbstractRedisClient {
 
     private final ClientResources clientResources;
 
+    private final List<CommandListener> commandListeners = new ArrayList<>();
+
     private final Map<Class<? extends EventLoopGroup>, EventLoopGroup> eventLoopGroups = new ConcurrentHashMap<>(2);
 
     private final boolean sharedResources;
@@ -208,6 +225,33 @@ public void removeListener(RedisConnectionStateListener listener) {
         connectionEvents.removeListener(listener);
     }
 
+    /**
+     * Add a listener for Redis Command events. The listener is notified on each command start/success/failure.
+     *
+     * @param listener must not be {@code null}
+     * @since 6.1
+     */
+    public void addListener(CommandListener listener) {
+        LettuceAssert.notNull(listener, "CommandListener must not be null");
+        commandListeners.add(listener);
+    }
+
+    /**
+     * Removes a listener.
+     *
+     * @param listener must not be {@code null}
+     * @since 6.1
+     */
+    public void removeListener(CommandListener listener) {
+
+        LettuceAssert.notNull(listener, "CommandListener must not be null");
+        commandListeners.remove(listener);
+    }
+
+    protected List<CommandListener> getCommandListeners() {
+        return commandListeners;
+    }
+
     /**
      * Populate connection builder with necessary resources.
      *