Clean up finishUpdate

Motivation:

As ABORT now closes the channel and never tries to offer the channel to
the pool, we no longer need an `expectOtherChunks` parameter.

Modifications:

* remove `expectOtherChunks` parameter
* reverse keepAlive in to close

Result:

Removed dead path in code

Author: slandelle

client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java

@@ -226,15 +226,13 @@ private boolean isHandledByReactiveStreams(ChannelHandlerContext ctx) {
         return Channels.getAttribute(ctx.channel()) instanceof StreamedResponsePublisher;
     }
 
-    protected void finishUpdate(NettyResponseFuture<?> future, Channel channel, boolean keepAlive, boolean expectOtherChunks) throws IOException {
+    protected void finishUpdate(NettyResponseFuture<?> future, Channel channel, boolean close) throws IOException {
         future.cancelTimeouts();
 
-        if (!keepAlive) {
+        if (close) {
             channelManager.closeChannel(channel);
-        } else if (expectOtherChunks) {
-            channelManager.drainChannelAndOffer(channel, future);
         } else {
-            channelManager.tryToOfferChannelToPool(channel, future.getAsyncHandler(), keepAlive, future.getPartitionKey());
+            channelManager.tryToOfferChannelToPool(channel, future.getAsyncHandler(), true, future.getPartitionKey());
         }
 
         try {

client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java

@@ -87,7 +87,7 @@ private void handleHttpResponse(final HttpResponse response, final Channel chann
                     abortAfterHandlingReactiveStreams(channel, future, handler);
 
             if (abort) {
-                finishUpdate(future, channel, false, true);
+                finishUpdate(future, channel, true);
             }
         }
     }
@@ -116,8 +116,8 @@ private void handleChunk(HttpContent chunk,//
         }
 
         if (abort || last) {
-            boolean keepAlive = !abort && future.isKeepAlive();
-            finishUpdate(future, channel, keepAlive, !last);
+            boolean close = abort || !future.isKeepAlive();
+            finishUpdate(future, channel, close);
         }
     }
 
@@ -168,7 +168,7 @@ private void readFailed(Channel channel, NettyResponseFuture<?> future, Throwabl
         } catch (Exception abortException) {
             logger.debug("Abort failed", abortException);
         } finally {
-            finishUpdate(future, channel, false, false);
+            finishUpdate(future, channel, true);
         }
     }
 

client/src/main/java/org/asynchttpclient/netty/handler/WebSocketHandler.java

@@ -87,7 +87,7 @@ private void abort(Channel channel, NettyResponseFuture<?> future, WebSocketUpgr
         try {
             handler.onThrowable(new IOException("Invalid Status code=" + status.getStatusCode() + " text=" + status.getStatusText()));
         } finally {
-            finishUpdate(future, channel, false, false);
+            finishUpdate(future, channel, true);
         }
     }
 

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

@@ -527,6 +527,7 @@ public boolean applyIoExceptionFiltersAndReplayRequest(NettyResponseFuture<?> fu
         }
 
         if (fc.replayRequest() && future.incrementRetryAndCheck() && future.isReplayPossible()) {
+            future.setKeepAlive(false);
             replayRequest(future, fc, channel);
             replayed = true;
         }