Trigger retry in WriteListener on SSLException, close #1645

Motivation:

There's a possibility that we have a write failure because we're trying to write on a pooled channel and the remote peer has closed the SSLEngine but we haven't process the connection close yet.

In this case, we don't trigger retry, and don't forcefully close the channel.

Modifications:

* Trigger retry on write failure on SSLException.
* Always close channel on write failure

Result:

Retry triggered on SSLEngine crash.
No more pooled connection in stalled state.

Author: slandelle

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

@@ -16,11 +16,13 @@
 import io.netty.channel.Channel;
 import org.asynchttpclient.handler.ProgressAsyncHandler;
 import org.asynchttpclient.netty.NettyResponseFuture;
+import org.asynchttpclient.netty.channel.ChannelState;
 import org.asynchttpclient.netty.channel.Channels;
 import org.asynchttpclient.netty.future.StackTraceInspector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.net.ssl.SSLException;
 import java.nio.channels.ClosedChannelException;
 
 public abstract class WriteListener {
@@ -36,27 +38,27 @@ public abstract class WriteListener {
     this.notifyHeaders = notifyHeaders;
   }
 
-  private boolean abortOnThrowable(Channel channel, Throwable cause) {
-    if (cause != null) {
-      if (cause instanceof IllegalStateException || cause instanceof ClosedChannelException || StackTraceInspector.recoverOnReadOrWriteException(cause)) {
-        LOGGER.debug(cause.getMessage(), cause);
-        Channels.silentlyCloseChannel(channel);
+  private void abortOnThrowable(Channel channel, Throwable cause) {
+    if (future.getChannelState() == ChannelState.POOLED
+      && (cause instanceof IllegalStateException
+      || cause instanceof ClosedChannelException
+      || cause instanceof SSLException
+      || StackTraceInspector.recoverOnReadOrWriteException(cause))) {
+      LOGGER.debug("Write exception on pooled channel, letting retry trigger", cause);
 
-      } else {
-        future.abort(cause);
-      }
-      return true;
+    } else {
+      future.abort(cause);
     }
-
-    return false;
+    Channels.silentlyCloseChannel(channel);
   }
 
   void operationComplete(Channel channel, Throwable cause) {
     future.touch();
 
-    // The write operation failed. If the channel was cached, it means it got asynchronously closed.
+    // The write operation failed. If the channel was pooled, it means it got asynchronously closed.
     // Let's retry a second time.
-    if (abortOnThrowable(channel, cause)) {
+    if (cause != null) {
+      abortOnThrowable(channel, cause);
       return;
     }