Have one common way of closing channels: silentlyCloseChannel

Author: Stephane Landelle

providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/ChannelManager.java

@@ -309,13 +309,9 @@ public void close() {
     public void closeChannel(Channel channel) {
 
         LOGGER.debug("Closing Channel {} ", channel);
-        try {
-            removeAll(channel);
-            Channels.setDiscard(channel);
-            channel.close();
-        } catch (Throwable t) {
-            LOGGER.debug("Error closing a connection", t);
-        }
+        removeAll(channel);
+        Channels.setDiscard(channel);
+        Channels.silentlyCloseChannel(channel);
         openChannels.remove(channel);
     }
 

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

@@ -18,9 +18,13 @@
 import io.netty.util.AttributeKey;
 
 import org.asynchttpclient.providers.netty.DiscardEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class Channels {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(Channels.class);
+
     private static final AttributeKey<Object> DEFAULT_ATTRIBUTE = AttributeKey.valueOf("default");
 
     public static Object getAttribute(Channel channel) {
@@ -37,6 +41,15 @@ public static void setDiscard(Channel channel) {
     }
 
     public static boolean isChannelValid(Channel channel) {
-        return channel != null && channel.isOpen() && channel.isActive();
+        return channel != null && channel.isActive();
+    }
+
+    public static void silentlyCloseChannel(Channel channel) {
+        try {
+            if (channel != null && channel.isActive())
+                channel.close();
+        } catch (Throwable t) {
+            LOGGER.debug("Failed to close channel", t);
+        }
     }
 }

providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/CleanupChannelGroup.java

@@ -89,7 +89,7 @@ public boolean add(Channel channel) {
         try {
             if (this.closed.get()) {
                 // Immediately close channel, as close() was already called.
-                channel.close();
+                Channels.silentlyCloseChannel(channel);
                 return false;
             }
 

providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/pool/DefaultChannelPool.java

@@ -22,7 +22,6 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
@@ -315,13 +314,9 @@ public void destroy() {
     }
 
     private void close(Channel channel) {
-        try {
-            // FIXME pity to have to do this here
-            Channels.setDiscard(channel);
-            channel2Creation.remove(channel);
-            channel.close();
-        } catch (Throwable t) {
-            // noop
-        }
+        // FIXME pity to have to do this here
+        Channels.setDiscard(channel);
+        channel2Creation.remove(channel);
+        Channels.silentlyCloseChannel(channel);
     }
-}
+}

providers/netty/src/main/java/org/asynchttpclient/providers/netty/future/NettyResponseFuture.java

@@ -132,12 +132,9 @@ public boolean cancel(boolean force) {
         if (isCancelled.getAndSet(true))
             return false;
 
-        try {
-            Channels.setDiscard(channel);
-            channel.close();
-        } catch (Throwable t) {
-            // Ignore
-        }
+        Channels.setDiscard(channel);
+        Channels.silentlyCloseChannel(channel);
+
         if (!onThrowableCalled.getAndSet(true)) {
             try {
                 asyncHandler.onThrowable(new CancellationException());

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

@@ -76,11 +76,8 @@ public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exce
             protocol.handle(channel, future, msg);
 
         } else if (attribute != DiscardEvent.INSTANCE) {
-            try {
-                LOGGER.trace("Closing an orphan channel {}", channel);
-                channel.close();
-            } catch (Throwable t) {
-            }
+            LOGGER.trace("Closing an orphan channel {}", channel);
+            Channels.silentlyCloseChannel(channel);
         }
     }
 
@@ -146,14 +143,9 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Excep
 
                     // FIXME why drop the original exception and throw a new one?
                     if (!config.getIOExceptionFilters().isEmpty()) {
-                        if (!requestSender.applyIoExceptionFiltersAndReplayRequest(future, CHANNEL_CLOSED_EXCEPTION, channel)) {
+                        if (!requestSender.applyIoExceptionFiltersAndReplayRequest(future, CHANNEL_CLOSED_EXCEPTION, channel))
                             // Close the channel so the recovering can occurs.
-                            try {
-                                channel.close();
-                            } catch (Throwable t) {
-                                // Swallow.
-                            }
-                        }
+                            Channels.silentlyCloseChannel(channel);
                         return;
                     }
                 }
@@ -181,6 +173,6 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Excep
         channelManager.closeChannel(channel);
         // FIXME not really sure
         // ctx.fireChannelRead(e);
-        ctx.close();
+        Channels.silentlyCloseChannel(channel);
     }
 }

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

@@ -313,11 +313,7 @@ public <T> void writeRequest(NettyResponseFuture<T> future, Channel channel) {
                 } catch (Throwable cause) {
                     // FIXME why not notify?
                     LOGGER.debug(cause.getMessage(), cause);
-                    try {
-                        channel.close();
-                    } catch (RuntimeException ex) {
-                        LOGGER.debug(ex.getMessage(), ex);
-                    }
+                    Channels.silentlyCloseChannel(channel);
                     return;
                 }
             }
@@ -326,11 +322,7 @@ public <T> void writeRequest(NettyResponseFuture<T> future, Channel channel) {
                 nettyRequest.getBody().write(channel, future, config);
 
         } catch (Throwable ioe) {
-            try {
-                channel.close();
-            } catch (RuntimeException ex) {
-                LOGGER.debug(ex.getMessage(), ex);
-            }
+            Channels.silentlyCloseChannel(channel);
         }
 
         scheduleTimeouts(future);

providers/netty/src/main/java/org/asynchttpclient/providers/netty/request/ProgressListener.java

@@ -13,21 +13,22 @@
  */
 package org.asynchttpclient.providers.netty.request;
 
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelProgressiveFuture;
+import io.netty.channel.ChannelProgressiveFutureListener;
+
+import java.nio.channels.ClosedChannelException;
+
 import org.asynchttpclient.AsyncHandler;
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.ProgressAsyncHandler;
 import org.asynchttpclient.Realm;
+import org.asynchttpclient.providers.netty.channel.Channels;
 import org.asynchttpclient.providers.netty.future.NettyResponseFuture;
 import org.asynchttpclient.providers.netty.future.StackTraceInspector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelProgressiveFuture;
-import io.netty.channel.ChannelProgressiveFutureListener;
-
-import java.nio.channels.ClosedChannelException;
-
 public class ProgressListener implements ChannelProgressiveFutureListener {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ProgressListener.class);
@@ -54,28 +55,17 @@ public ProgressListener(AsyncHttpClientConfig config,//
     private boolean abortOnThrowable(Throwable cause, Channel channel) {
 
         if (cause != null && future.getState() != NettyResponseFuture.STATE.NEW) {
-
             if (cause instanceof IllegalStateException) {
                 LOGGER.debug(cause.getMessage(), cause);
-                try {
-                    channel.close();
-                } catch (RuntimeException ex) {
-                    LOGGER.debug(ex.getMessage(), ex);
-                }
-            } else if (cause instanceof ClosedChannelException || StackTraceInspector.abortOnReadOrWriteException(cause)) {
-
-                if (LOGGER.isDebugEnabled())
-                    LOGGER.debug(cause.getMessage(), cause);
+                Channels.silentlyCloseChannel(channel);
 
-                try {
-                    channel.close();
-                } catch (RuntimeException ex) {
-                    LOGGER.debug(ex.getMessage(), ex);
-                }
+            } else if (cause instanceof ClosedChannelException || StackTraceInspector.abortOnReadOrWriteException(cause)) {
+                LOGGER.debug(cause.getMessage(), cause);
+                Channels.silentlyCloseChannel(channel);
+                
             } else {
                 future.abort(cause);
             }
-
             return true;
         }
 
@@ -84,10 +74,8 @@ private boolean abortOnThrowable(Throwable cause, Channel channel) {
 
     @Override
     public void operationComplete(ChannelProgressiveFuture cf) {
-        // The write operation failed. If the channel was cached, it means it
-        // got asynchronously closed.
+        // The write operation failed. If the channel was cached, it means it got asynchronously closed.
         // Let's retry a second time.
-
         if (!abortOnThrowable(cf.cause(), cf.channel())) {
 
             future.touch();