Complete write promise successfully on no-op writes #764

CommandHandler now completes a write promise if the write completes without actually writing a message. Previously, no-op writes did not complete and rendered an invalid state (e.g. queue size was not decremented).
redis · Apr 23, 2018 · d5bae76 · d5bae76
1 parent fb05f21
commit d5bae76
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 12 deletions.
diff --git a/src/main/java/io/lettuce/core/protocol/CommandHandler.java b/src/main/java/io/lettuce/core/protocol/CommandHandler.java
@@ -350,26 +350,23 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
         }
 
         if (msg instanceof Collection) {
-
-            Collection<RedisCommand<?, ?, ?>> batch = (Collection<RedisCommand<?, ?, ?>>) msg;
-
-            writeBatch(ctx, batch, promise);
+            writeBatch(ctx, (Collection<RedisCommand<?, ?, ?>>) msg, promise);
         }
     }
 
     private void writeSingleCommand(ChannelHandlerContext ctx, RedisCommand<?, ?, ?> command, ChannelPromise promise)
-            throws Exception {
+ {
 
         if (!isWriteable(command)) {
+            promise.trySuccess();
             return;
         }
 
         addToStack(command, promise);
         ctx.write(command, promise);
     }
 
-    private void writeBatch(ChannelHandlerContext ctx, Collection<RedisCommand<?, ?, ?>> batch, ChannelPromise promise)
-            throws Exception {
+    private void writeBatch(ChannelHandlerContext ctx, Collection<RedisCommand<?, ?, ?>> batch, ChannelPromise promise) {
 
         Collection<RedisCommand<?, ?, ?>> deduplicated = new LinkedHashSet<>(batch.size(), 1);
 
@@ -390,8 +387,7 @@ private void writeBatch(ChannelHandlerContext ctx, Collection<RedisCommand<?, ?,
                 redisCommand.completeExceptionally(e);
             }
 
-            promise.setFailure(e);
-            return;
+            throw e;
         }
 
         for (RedisCommand<?, ?, ?> command : deduplicated) {
@@ -400,6 +396,8 @@ private void writeBatch(ChannelHandlerContext ctx, Collection<RedisCommand<?, ?,
 
         if (!deduplicated.isEmpty()) {
             ctx.write(deduplicated, promise);
+        } else {
+            promise.trySuccess();
         }
     }
 
@@ -423,7 +421,6 @@ private void addToStack(RedisCommand<?, ?, ?> command, ChannelPromise promise) {
             }
         } catch (Exception e) {
             command.completeExceptionally(e);
-            promise.setFailure(e);
             throw e;
         }
     }

diff --git a/src/test/java/io/lettuce/core/ConnectionCommandTest.java b/src/test/java/io/lettuce/core/ConnectionCommandTest.java
@@ -88,11 +88,14 @@ public void authEmpty() throws Exception {
     public void authReconnect() throws Exception {
         new WithPasswordRequired() {
             @Override
-            public void run(RedisClient client) {
+            public void run(RedisClient client) throws InterruptedException {
+
                 RedisCommands<String, String> connection = client.connect().sync();
                 assertThat(connection.auth(passwd)).isEqualTo("OK");
                 assertThat(connection.set(key, value)).isEqualTo("OK");
                 connection.quit();
+
+                Thread.sleep(100);
                 assertThat(connection.get(key)).isEqualTo(value);
             }
         };

diff --git a/src/test/java/io/lettuce/core/protocol/CommandHandlerTest.java b/src/test/java/io/lettuce/core/protocol/CommandHandlerTest.java
@@ -250,14 +250,28 @@ public void isConnectedShouldReportFalseForCLOSED() throws Exception {
         assertThat(sut.isConnected()).isFalse();
     }
 
+    @Test
+    public void shouldNotWriteCancelledCommand() throws Exception {
+
+        command.cancel();
+        sut.write(context, command, promise);
+
+        verifyZeroInteractions(context);
+        assertThat(stack).isEmpty();
+
+        verify(promise).trySuccess();
+    }
+
     @Test
     public void shouldNotWriteCancelledCommands() throws Exception {
 
         command.cancel();
-        sut.write(context, command, null);
+        sut.write(context, Collections.singleton(command), promise);
 
         verifyZeroInteractions(context);
         assertThat(stack).isEmpty();
+
+        verify(promise).trySuccess();
     }
 
     @Test