Do not retry completed commands through RetryListener #767

redis · Apr 23, 2018 · 273ca69 · 273ca69
1 parent faa6cc0
commit 273ca69
Show file tree

Hide file tree

Showing 2 changed files with 56 additions and 10 deletions.
diff --git a/src/main/java/com/lambdaworks/redis/protocol/CommandHandler.java b/src/main/java/com/lambdaworks/redis/protocol/CommandHandler.java
@@ -1265,7 +1265,7 @@ protected void complete(Throwable t) {
         }
     }
 
-    private class AtMostOnceWriteListener extends ListenerSupport implements ChannelFutureListener {
+    class AtMostOnceWriteListener extends ListenerSupport implements ChannelFutureListener {
 
         AtMostOnceWriteListener(RedisCommand<K, V, ?> sentCommand) {
             super(sentCommand);
@@ -1276,7 +1276,7 @@ private class AtMostOnceWriteListener extends ListenerSupport implements Channel
         }
 
         @Override
-        public void operationComplete(ChannelFuture future) throws Exception {
+        public void operationComplete(ChannelFuture future) {
 
             dequeue();
 
@@ -1289,7 +1289,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
     /**
      * A generic future listener which retries unsuccessful writes.
      */
-    private class RetryListener extends ListenerSupport implements GenericFutureListener<Future<Void>> {
+    class RetryListener extends ListenerSupport implements GenericFutureListener<Future<Void>> {
 
         RetryListener(RedisCommand<K, V, ?> sentCommand) {
             super(sentCommand);
@@ -1301,7 +1301,7 @@ private class RetryListener extends ListenerSupport implements GenericFutureList
 
         @SuppressWarnings("unchecked")
         @Override
-        public void operationComplete(Future<Void> future) throws Exception {
+        public void operationComplete(Future<Void> future) {
 
             Throwable cause = future.cause();
 
@@ -1314,12 +1314,7 @@ public void operationComplete(Future<Void> future) throws Exception {
                     return;
                 }
 
-                Channel channel = CommandHandler.this.channel;
-                if (channel != null) {
-                    channel.eventLoop().submit(this::requeueCommands);
-                } else {
-                    CommandHandler.this.clientResources.eventExecutorGroup().submit(this::requeueCommands);
-                }
+                potentiallyRequeueCommands(channel, sentCommand, sentCommands);
             }
 
             if (!success && !(cause instanceof ClosedChannelException)) {
@@ -1335,7 +1330,45 @@ public void operationComplete(Future<Void> future) throws Exception {
             }
         }
 
+        /**
+         * Requeue command/commands
+         *
+         * @param channel
+         * @param sentCommand
+         * @param sentCommands
+         */
+        private void potentiallyRequeueCommands(Channel channel, RedisCommand<?, ?, ?> sentCommand,
+                Collection<? extends RedisCommand<?, ?, ?>> sentCommands) {
+
+            if (sentCommand != null && sentCommand.isDone()) {
+                return;
+            }
+
+            if (sentCommands != null) {
+
+                boolean foundToSend = false;
+
+                for (RedisCommand<?, ?, ?> command : sentCommands) {
+                    if (!command.isDone()) {
+                        foundToSend = true;
+                        break;
+                    }
+                }
+
+                if (!foundToSend) {
+                    return;
+                }
+            }
+
+            if (channel != null) {
+                channel.eventLoop().submit(this::requeueCommands);
+            } else {
+                CommandHandler.this.clientResources.eventExecutorGroup().submit(this::requeueCommands);
+            }
+        }
+
         private void requeueCommands() {
+
             if (sentCommand != null) {
                 try {
                     write(sentCommand);

diff --git a/src/test/java/com/lambdaworks/redis/protocol/CommandHandlerTest.java b/src/test/java/com/lambdaworks/redis/protocol/CommandHandlerTest.java
@@ -627,6 +627,19 @@ public void shouldRebuildHugeQueue() throws Exception {
         assertThat(disconnectedBuffer).isEmpty();
     }
 
+    @Test
+    public void retryListenerDoesNotRetryCompletedCommands() {
+
+        CommandHandler.RetryListener listener = sut.new RetryListener(command);
+
+        command.complete();
+        when(promise.cause()).thenReturn(new Exception());
+
+        listener.operationComplete(promise);
+
+        verify(channel, never()).writeAndFlush(command);
+    }
+
     @Test
     public void testMTCConcurrentWriteThenReset() throws Throwable {
         TestFramework.runOnce(new MTCConcurrentWriteThenReset(clientResources, command));