Handle Netty 4's HttpObject.getDecoderResult().cause(), close AsyncHttpClient#1072

Author: slandelle

client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java

@@ -18,6 +18,7 @@
 import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpObject;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponse;
 import io.netty.handler.codec.http.LastHttpContent;
@@ -68,7 +69,7 @@ private boolean updateBodyAndInterrupt(NettyResponseFuture<?> future, AsyncHandl
         return interrupt;
     }
 
-    private boolean exitAfterHandler(Channel channel, NettyResponseFuture<?> future, HttpResponse response, AsyncHandler<?> handler, NettyResponseStatus status,
+    private void notifyHandler(Channel channel, NettyResponseFuture<?> future, HttpResponse response, AsyncHandler<?> handler, NettyResponseStatus status,
             HttpRequest httpRequest, HttpResponseHeaders responseHeaders) throws IOException, Exception {
 
         boolean exit = exitAfterHandlingStatus(channel, future, response, handler, status, httpRequest) || //
@@ -77,8 +78,6 @@ private boolean exitAfterHandler(Channel channel, NettyResponseFuture<?> future,
 
         if (exit)
             finishUpdate(future, channel, HttpHeaders.isTransferEncodingChunked(httpRequest) || HttpHeaders.isTransferEncodingChunked(response));
-
-        return exit;
     }
 
     private boolean exitAfterHandlingStatus(//
@@ -118,7 +117,7 @@ private boolean exitAfterHandlingReactiveStreams(//
         return false;
     }
 
-    private boolean handleHttpResponse(final HttpResponse response, final Channel channel, final NettyResponseFuture<?> future, AsyncHandler<?> handler) throws Exception {
+    private void handleHttpResponse(final HttpResponse response, final Channel channel, final NettyResponseFuture<?> future, AsyncHandler<?> handler) throws Exception {
 
         HttpRequest httpRequest = future.getNettyRequest().getHttpRequest();
         logger.debug("\n\nRequest {}\n\nResponse {}\n", httpRequest, response);
@@ -128,8 +127,9 @@ private boolean handleHttpResponse(final HttpResponse response, final Channel ch
         NettyResponseStatus status = new NettyResponseStatus(future.getUri(), config, response, channel);
         HttpResponseHeaders responseHeaders = new HttpResponseHeaders(response.headers());
 
-        return interceptors.intercept(channel, future, handler, response, status, responseHeaders)
-                || exitAfterHandler(channel, future, response, handler, status, httpRequest, responseHeaders);
+        if (!interceptors.exitAfterIntercept(channel, future, handler, response, status, responseHeaders)) {
+            notifyHandler(channel, future, response, handler, status, httpRequest, responseHeaders);
+        }
     }
 
     private void handleChunk(HttpContent chunk,//
@@ -173,9 +173,17 @@ public void handleRead(final Channel channel, final NettyResponseFuture<?> futur
 
         AsyncHandler<?> handler = future.getAsyncHandler();
         try {
-            if (e instanceof HttpResponse) {
-                if (handleHttpResponse((HttpResponse) e, channel, future, handler))
+            if (e instanceof HttpObject) {
+                HttpObject object = (HttpObject) e;
+                Throwable t = object.getDecoderResult().cause();
+                if (t != null) {
+                    readFailed(channel, future, t);
                     return;
+                }
+            }
+            
+            if (e instanceof HttpResponse) {
+                handleHttpResponse((HttpResponse) e, channel, future, handler);
 
             } else if (e instanceof HttpContent) {
                 handleChunk((HttpContent) e, channel, future, handler);
@@ -189,16 +197,20 @@ public void handleRead(final Channel channel, final NettyResponseFuture<?> futur
                 return;
             }
 
-            try {
-                requestSender.abort(channel, future, t);
-            } catch (Exception abortException) {
-                logger.debug("Abort failed", abortException);
-            } finally {
-                finishUpdate(future, channel, false);
-            }
+            readFailed(channel, future, t);
             throw t;
         }
     }
+    
+    private void readFailed(Channel channel, NettyResponseFuture<?> future, Throwable t) throws Exception {
+        try {
+            requestSender.abort(channel, future, t);
+        } catch (Exception abortException) {
+            logger.debug("Abort failed", abortException);
+        } finally {
+            finishUpdate(future, channel, false);
+        }
+    }
 
     @Override
     public void handleException(NettyResponseFuture<?> future, Throwable error) {

client/src/main/java/org/asynchttpclient/netty/handler/WebSocketHandler.java

@@ -140,11 +140,10 @@ public void handleRead(Channel channel, NettyResponseFuture<?> future, Object e)
             HttpResponseStatus status = new NettyResponseStatus(future.getUri(), config, response, channel);
             HttpResponseHeaders responseHeaders = new HttpResponseHeaders(response.headers());
 
-            if (interceptors.intercept(channel, future, handler, response, status, responseHeaders)) {
-                return;
+            if (!interceptors.exitAfterIntercept(channel, future, handler, response, status, responseHeaders)) {
+                Channels.setAttribute(channel, new UpgradeCallback(future, channel, response, handler, status, responseHeaders));
             }
 
-            Channels.setAttribute(channel, new UpgradeCallback(future, channel, response, handler, status, responseHeaders));
 
         } else if (e instanceof WebSocketFrame) {
 

client/src/main/java/org/asynchttpclient/netty/handler/intercept/Interceptors.java

@@ -55,7 +55,7 @@ public Interceptors(//
         hasResponseFilters = !config.getResponseFilters().isEmpty();
     }
 
-    public boolean intercept(//
+    public boolean exitAfterIntercept(//
             Channel channel,//
             NettyResponseFuture<?> future,//
             AsyncHandler<?> handler,//