Have a mutex for handling handleUnexpectedClosedChannel, close AsyncHttpClient#1265

Author: slandelle

client/src/main/java/org/asynchttpclient/netty/channel/Channels.java

@@ -13,6 +13,8 @@
  */
 package org.asynchttpclient.netty.channel;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelId;
 import io.netty.channel.DefaultChannelId;
@@ -29,6 +31,7 @@ public class Channels {
 
     private static final AttributeKey<Object> DEFAULT_ATTRIBUTE = AttributeKey.valueOf("default");
     private static final AttributeKey<ChannelId> CHANNEL_ID_ATTRIBUTE = AttributeKey.valueOf("channelId");
+    private static final AttributeKey<AtomicBoolean> INACTIVE_TOKEN_ATTRIBUTE = AttributeKey.valueOf("inactiveToken");
 
     public static Object getAttribute(Channel channel) {
         Attribute<Object> attr = channel.attr(DEFAULT_ATTRIBUTE);
@@ -46,6 +49,14 @@ public static void setDiscard(Channel channel) {
     public static boolean isChannelValid(Channel channel) {
         return channel != null && channel.isActive();
     }
+    
+    public static void setInactiveToken(Channel channel) {
+        channel.attr(INACTIVE_TOKEN_ATTRIBUTE).set(new AtomicBoolean(true));
+    }
+    
+    public static boolean getInactiveToken(Channel channel) {
+        return channel != null && channel.attr(INACTIVE_TOKEN_ATTRIBUTE).get().getAndSet(false);
+    }
 
     public static ChannelId getChannelId(Channel channel) {
         Attribute<ChannelId> attr = channel.attr(CHANNEL_ID_ATTRIBUTE);

client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java

@@ -95,6 +95,8 @@ private void writeRequest(Channel channel) {
 
     public void onSuccess(Channel channel, InetSocketAddress remoteAddress) {
 
+        Channels.setInactiveToken(channel);
+
         TimeoutsHolder timeoutsHolder = future.getTimeoutsHolder();
 
         if (futureIsAlreadyCancelled(channel)) {

client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java

@@ -405,12 +405,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 (future.incrementRetryAndCheck() && retry(future)) {
-            future.pendingException = null;
-        } else {
-            abort(channel, future, future.pendingException != null ? future.pendingException : RemotelyClosedException.INSTANCE);
+        if (Channels.getInactiveToken(channel)) {
+            if (future.isDone()) {
+                channelManager.closeChannel(channel);
+            } else if (future.incrementRetryAndCheck() && retry(future)) {
+                future.pendingException = null;
+            } else {
+                abort(channel, future, future.pendingException != null ? future.pendingException : RemotelyClosedException.INSTANCE);
+            }
         }
     }