Guard initialization against premature disconnect #608

Properly propagate connect/disconnect events to prevent a race in which the disconnect is faster than setting up future chaining for event propagation.
redis · Oct 13, 2017 · b209f28 · b209f28
1 parent 42da4f8
commit b209f28
Show file tree

Hide file tree

Showing 5 changed files with 14 additions and 2 deletions.
diff --git a/src/main/java/com/lambdaworks/redis/AbstractRedisClient.java b/src/main/java/com/lambdaworks/redis/AbstractRedisClient.java
@@ -293,6 +293,7 @@ protected <K, V, T extends RedisChannelHandler<K, V>> ConnectionFuture<T> initia
         redisBootstrap.handler(initializer);
 
         clientResources.nettyCustomizer().afterBootstrapInitialized(redisBootstrap);
+        CompletableFuture<Boolean> initFuture = initializer.channelInitialized();
         ChannelFuture connectFuture = redisBootstrap.connect(redisAddress);
 
         connectFuture.addListener(future -> {
@@ -305,7 +306,6 @@ protected <K, V, T extends RedisChannelHandler<K, V>> ConnectionFuture<T> initia
                 return;
             }
 
-            CompletableFuture<Boolean> initFuture = initializer.channelInitialized();
             initFuture.whenComplete((success, throwable) -> {
 
                 if (throwable == null) {

diff --git a/src/main/java/com/lambdaworks/redis/PlainChannelInitializer.java b/src/main/java/com/lambdaworks/redis/PlainChannelInitializer.java
@@ -79,6 +79,11 @@ public CompletableFuture<Boolean> channelInitialized() {
                 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 
                     clientResources.eventBus().publish(new DisconnectedEvent(local(ctx), remote(ctx)));
+
+                    if (!initializedFuture.isDone()) {
+                        initializedFuture.completeExceptionally(new RedisConnectionException("Connection closed prematurely"));
+                    }
+
                     initializedFuture = new CompletableFuture<>();
                     pingCommand = null;
                     super.channelInactive(ctx);

diff --git a/src/main/java/com/lambdaworks/redis/SslConnectionBuilder.java b/src/main/java/com/lambdaworks/redis/SslConnectionBuilder.java
@@ -178,6 +178,12 @@ public CompletableFuture<Boolean> channelInitialized() {
 
                     @Override
                     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+
+                        if (!initializedFuture.isDone()) {
+                            initializedFuture.completeExceptionally(new RedisConnectionException(
+                                    "Connection closed prematurely"));
+                        }
+
                         initializedFuture = new CompletableFuture<>();
                         pingCommand = null;
                         super.channelInactive(ctx);

diff --git a/src/main/java/com/lambdaworks/redis/protocol/CommandHandler.java b/src/main/java/com/lambdaworks/redis/protocol/CommandHandler.java
@@ -101,7 +101,7 @@ public class CommandHandler<K, V> extends ChannelDuplexHandler implements RedisC
 
     private Thread exclusiveLockOwner;
     private RedisChannelHandler<K, V> redisChannelHandler;
-    private Throwable connectionError;
+    private volatile Throwable connectionError;
     private String logPrefix;
     private boolean autoFlushCommands = true;
 

diff --git a/src/test/java/com/lambdaworks/redis/support/CdiTest.java b/src/test/java/com/lambdaworks/redis/support/CdiTest.java
@@ -20,6 +20,7 @@
 
 import javax.enterprise.inject.Produces;
 
+import com.lambdaworks.TestClientResources;
 import org.apache.webbeans.cditest.CdiTestContainer;
 import org.apache.webbeans.cditest.CdiTestContainerLoader;
 import org.junit.AfterClass;