Change Protocol method to directly use a NettyResponseFuture

Author: Stephane Landelle

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/HttpProtocol.java

@@ -463,10 +463,10 @@ public void handle(final Channel channel, final NettyResponseFuture<?> future, f
     }
 
     @Override
-    public void onError(Channel channel, Throwable error) {
+    public void onError(NettyResponseFuture<?> future, Throwable error) {
     }
 
     @Override
-    public void onClose(Channel channel) {
+    public void onClose(NettyResponseFuture<?> future) {
     }
-}
+}

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/Processor.java

@@ -113,7 +113,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
             if (!config.getIOExceptionFilters().isEmpty() && requestSender.applyIoExceptionFiltersAndReplayRequest(future, CHANNEL_CLOSED_EXCEPTION, channel))
                 return;
 
-            protocol.onClose(channel);
+            protocol.onClose(future);
 
             if (future == null || future.isDone())
                 channelManager.closeChannel(channel);
@@ -174,12 +174,11 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Excep
             try {
                 LOGGER.debug("Was unable to recover Future: {}", future);
                 requestSender.abort(future, cause);
+                protocol.onError(future, e);
             } catch (Throwable t) {
                 LOGGER.error(t.getMessage(), t);
             }
 
-        protocol.onError(channel, e);
-
         channelManager.closeChannel(channel);
         // FIXME not really sure
         // ctx.fireChannelRead(e);

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/Protocol.java

@@ -86,9 +86,9 @@ public Protocol(ChannelManager channelManager, AsyncHttpClientConfig config, Net
 
     public abstract void handle(Channel channel, NettyResponseFuture<?> future, Object message) throws Exception;
 
-    public abstract void onError(Channel channel, Throwable error);
+    public abstract void onError(NettyResponseFuture<?> future, Throwable error);
 
-    public abstract void onClose(Channel channel);
+    public abstract void onClose(NettyResponseFuture<?> future);
 
     protected boolean exitAfterHandlingRedirect(//
             Channel channel,//

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/WebSocketProtocol.java

@@ -35,7 +35,6 @@
 import org.asynchttpclient.HttpResponseHeaders;
 import org.asynchttpclient.HttpResponseStatus;
 import org.asynchttpclient.Request;
-import org.asynchttpclient.providers.netty.DiscardEvent;
 import org.asynchttpclient.providers.netty.NettyAsyncHttpProviderConfig;
 import org.asynchttpclient.providers.netty.channel.ChannelManager;
 import org.asynchttpclient.providers.netty.channel.Channels;
@@ -169,16 +168,11 @@ public void handle(Channel channel, NettyResponseFuture<?> future, Object e) thr
     }
 
     @Override
-    public void onError(Channel channel, Throwable e) {
-        try {
-            Object attribute = Channels.getAttribute(channel);
-            logger.warn("onError {}", e);
-            if (!(attribute instanceof NettyResponseFuture)) {
-                return;
-            }
+    public void onError(NettyResponseFuture<?> future, Throwable e) {
+        logger.warn("onError {}", e);
 
-            NettyResponseFuture<?> nettyResponse = (NettyResponseFuture<?>) attribute;
-            WebSocketUpgradeHandler h = WebSocketUpgradeHandler.class.cast(nettyResponse.getAsyncHandler());
+        try {
+            WebSocketUpgradeHandler h = WebSocketUpgradeHandler.class.cast(future);
 
             NettyWebSocket webSocket = NettyWebSocket.class.cast(h.onCompleted());
             if (webSocket != null) {
@@ -191,21 +185,15 @@ public void onError(Channel channel, Throwable e) {
     }
 
     @Override
-    public void onClose(Channel channel) {
+    public void onClose(NettyResponseFuture<?> future) {
         logger.trace("onClose {}");
-        Object attribute = Channels.getAttribute(channel);
-        if (!(attribute instanceof NettyResponseFuture))
-            return;
 
         try {
-            NettyResponseFuture<?> nettyResponse = NettyResponseFuture.class.cast(attribute);
-            WebSocketUpgradeHandler h = WebSocketUpgradeHandler.class.cast(nettyResponse.getAsyncHandler());
+            WebSocketUpgradeHandler h = WebSocketUpgradeHandler.class.cast(future);
             NettyWebSocket webSocket = NettyWebSocket.class.cast(h.onCompleted());
 
-            // FIXME How could this test not succeed, we just checked above that
-            // attribute is a NettyResponseFuture????
             logger.trace("Connection was closed abnormally (that is, with no close frame being sent).");
-            if (attribute != DiscardEvent.INSTANCE && webSocket != null)
+            if (webSocket != null)
                 webSocket.close(1006, "Connection was closed abnormally (that is, with no close frame being sent).");
         } catch (Throwable t) {
             logger.error("onError", t);