Possible event loss when pooled connection is closed before writing, close AsyncHttpClient#768

Author: Stephane Landelle

src/main/java/com/ning/http/client/providers/netty/handler/Processor.java

@@ -32,7 +32,6 @@
 import com.ning.http.client.providers.netty.future.NettyResponseFuture;
 import com.ning.http.client.providers.netty.future.StackTraceInspector;
 import com.ning.http.client.providers.netty.request.NettyRequestSender;
-import com.ning.http.util.AsyncHttpProviderUtils;
 
 import java.io.IOException;
 import java.nio.channels.ClosedChannelException;
@@ -128,12 +127,7 @@ public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws
                 return;
 
             protocol.onClose(future);
-
-            if (future.isDone())
-                channelManager.closeChannel(channel);
-
-            else if (!requestSender.retry(future))
-                requestSender.abort(channel, future, AsyncHttpProviderUtils.REMOTELY_CLOSED_EXCEPTION);
+            requestSender.handleUnexpectedClosedChannel(channel, future);
         }
     }
 

src/main/java/com/ning/http/client/providers/netty/request/NettyRequestSender.java

@@ -17,6 +17,7 @@
 import static com.ning.http.client.providers.netty.util.HttpUtils.useProxyConnect;
 import static com.ning.http.util.AsyncHttpProviderUtils.getDefaultPort;
 import static com.ning.http.util.AsyncHttpProviderUtils.requestTimeout;
+import static com.ning.http.util.AsyncHttpProviderUtils.REMOTELY_CLOSED_EXCEPTION;
 import static com.ning.http.util.ProxyUtils.avoidProxy;
 import static com.ning.http.util.ProxyUtils.getProxyServer;
 
@@ -207,28 +208,35 @@ private <T> ListenableFuture<T> sendRequestWithCachedChannel(Request request, Ur
         future.attachChannel(channel, false);
 
         LOGGER.debug("Using cached Channel {}\n for request \n{}\n", channel, future.getNettyRequest().getHttpRequest());
-        Channels.setAttribute(channel, future);
 
-        try {
-            writeRequest(future, channel);
-        } catch (Exception ex) {
-            // write request isn't supposed to throw Exceptions
-            LOGGER.debug("writeRequest failure", ex);
-            if (ex.getMessage() != null && ex.getMessage().contains("SSLEngine")) {
-                // FIXME what is this for? https://github.com/AsyncHttpClient/async-http-client/commit/a847c3d4523ccc09827743e15b17e6bab59c553b
-                // can such an exception happen as we write async?
-                LOGGER.debug("SSLEngine failure", ex);
-                future = null;
-            } else {
-                try {
-                    asyncHandler.onThrowable(ex);
-                } catch (Throwable t) {
-                    LOGGER.warn("doConnect.writeRequest()", t);
+        if (Channels.isChannelValid(channel)) {
+            Channels.setAttribute(channel, future);
+
+            try {
+                writeRequest(future, channel);
+            } catch (Exception ex) {
+                // write request isn't supposed to throw Exceptions
+                LOGGER.debug("writeRequest failure", ex);
+                if (ex.getMessage() != null && ex.getMessage().contains("SSLEngine")) {
+                    // FIXME what is this for? https://github.com/AsyncHttpClient/async-http-client/commit/a847c3d4523ccc09827743e15b17e6bab59c553b
+                    // can such an exception happen as we write async?
+                    LOGGER.debug("SSLEngine failure", ex);
+                    future = null;
+                } else {
+                    try {
+                        asyncHandler.onThrowable(ex);
+                    } catch (Throwable t) {
+                        LOGGER.warn("doConnect.writeRequest()", t);
+                    }
+                    IOException ioe = new IOException(ex.getMessage());
+                    ioe.initCause(ex);
+                    throw ioe;
                 }
-                IOException ioe = new IOException(ex.getMessage());
-                ioe.initCause(ex);
-                throw ioe;
             }
+        } else {
+            // bad luck, the channel was closed in-between
+            // there's a very good chance onClose was already notified but the future wasn't already registered
+            handleUnexpectedClosedChannel(channel, future);
         }
         return future;
     }
@@ -413,6 +421,14 @@ public void abort(Channel channel, NettyResponseFuture<?> future, Throwable t) {
         }
     }
 
+    public void handleUnexpectedClosedChannel(Channel channel, NettyResponseFuture<?> future) {
+        if (future.isDone())
+            channelManager.closeChannel(channel);
+
+        else if (!retry(future))
+            abort(channel, future, REMOTELY_CLOSED_EXCEPTION);
+    }
+
     public boolean retry(NettyResponseFuture<?> future) {
 
         if (isClosed())