Change Protocol method to directly use a NettyResponseFuture

Author: Stephane Landelle

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

@@ -474,9 +474,9 @@ public void handle(final Channel channel, final NettyResponseFuture<?> future, f
         }
     }
 
-    public void onError(Channel channel, Throwable e) {
+    public void onError(NettyResponseFuture<?> future, Throwable e) {
     }
 
-    public void onClose(Channel channel) {
+    public void onClose(NettyResponseFuture<?> future) {
     }
 }

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

@@ -133,7 +133,7 @@ public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws
                     && requestSender.applyIoExceptionFiltersAndReplayRequest(future, CHANNEL_CLOSED_EXCEPTION, channel))
                 return;
 
-            protocol.onClose(channel);
+            protocol.onClose(future);
 
             if (future == null || future.isDone())
                 channelManager.closeChannel(channel);
@@ -193,12 +193,11 @@ public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws
             try {
                 LOGGER.debug("Was unable to recover Future: {}", future);
                 requestSender.abort(future, cause);
+                protocol.onError(future, e.getCause());
             } catch (Throwable t) {
                 LOGGER.error(t.getMessage(), t);
             }
 
-        protocol.onError(channel, e.getCause());
-
         channelManager.closeChannel(channel);
         ctx.sendUpstream(e);
     }

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

@@ -87,9 +87,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 e);
+    public abstract void onError(NettyResponseFuture<?> future, Throwable e);
 
-    public abstract void onClose(Channel channel);
+    public abstract void onClose(NettyResponseFuture<?> future);
 
     protected boolean exitAfterHandlingRedirect(//
             Channel channel,//

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

@@ -33,7 +33,6 @@
 import com.ning.http.client.HttpResponseHeaders;
 import com.ning.http.client.HttpResponseStatus;
 import com.ning.http.client.Request;
-import com.ning.http.client.providers.netty.DiscardEvent;
 import com.ning.http.client.providers.netty.NettyAsyncHttpProviderConfig;
 import com.ning.http.client.providers.netty.channel.ChannelManager;
 import com.ning.http.client.providers.netty.channel.Channels;
@@ -177,16 +176,11 @@ public void setContent(ChannelBuffer content) {
     }
 
     @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) future.getAsyncHandler();
 
             NettyWebSocket webSocket = NettyWebSocket.class.cast(h.onCompleted());
             if (webSocket != null) {
@@ -199,20 +193,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) future.getAsyncHandler();
             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);