intercept readComplete

Author: mhl-b

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpHeaderValidator.java

@@ -29,7 +29,6 @@ public class Netty4HttpHeaderValidator extends ChannelDuplexHandler {
     private final ThreadContext threadContext;
     private boolean droppingContent;
     private boolean validatingRequest;
-    private boolean readRequested;
 
     public Netty4HttpHeaderValidator(HttpValidator validator, ThreadContext threadContext) {
         this.validator = validator;
@@ -62,14 +61,18 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
     public void read(ChannelHandlerContext ctx) throws Exception {
         // until validation is completed we can ignore read calls,
         // once validation is finished HttpRequest will be fired and downstream can read from there
-        if (validatingRequest) {
-            readRequested = true;
-        } else {
-            readRequested = false;
+        if (validatingRequest == false) {
             ctx.read();
         }
     }
 
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        if (validatingRequest == false) {
+            ctx.fireChannelReadComplete();
+        }
+    }
+
     void validate(ChannelHandlerContext ctx, HttpRequest request) {
         assert Transports.assertDefaultThreadContext(threadContext);
         droppingContent = false;
@@ -113,10 +116,7 @@ void handleValidationResult(ChannelHandlerContext ctx, HttpRequest request, @Nul
             }
             validatingRequest = false;
             ctx.fireChannelRead(request);
-            if (readRequested) {
-                readRequested = false;
-                ctx.channel().eventLoop().execute(ctx::read);
-            }
+            ctx.fireChannelReadComplete();
         });
     }
 

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java

@@ -366,6 +366,8 @@ protected HttpMessage createMessage(String[] initialLine) throws Exception {
             decoder.setCumulator(ByteToMessageDecoder.COMPOSITE_CUMULATOR);
             ch.pipeline().addLast("decoder", decoder); // parses the HTTP bytes request into HTTP message pieces
             ch.pipeline().addLast(new FlowControlHandler());
+            ch.config().setAutoRead(false);
+            ch.read();
             if (httpValidator != null) {
                 // runs a validation function on the first HTTP message piece which contains all the headers
                 // if validation passes, the pieces of that particular request are forwarded, otherwise they are discarded
@@ -423,8 +425,6 @@ protected Result beginEncode(HttpResponse httpResponse, String acceptEncoding) t
                     new Netty4HttpPipeliningHandler(transport.pipeliningMaxEvents, transport, threadWatchdogActivityTracker)
                 );
             transport.serverAcceptedChannel(nettyHttpChannel);
-            ch.config().setAutoRead(false);
-            ch.read();
         }
 
         @Override