requested reads in validator

Author: mhl-b

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

@@ -29,6 +29,7 @@ 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;
@@ -47,6 +48,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                 } else if (msg instanceof HttpContent content) {
                     if (droppingContent) {
                         content.release();
+                        ctx.read();
                     } else {
                         assert validatingRequest == false : "unexpected content before validation completed";
                         ctx.fireChannelRead(content);
@@ -60,7 +62,10 @@ 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 == false) {
+        if (validatingRequest) {
+            readRequested = true;
+        } else {
+            readRequested = false;
             ctx.read();
         }
     }
@@ -108,6 +113,10 @@ void handleValidationResult(ChannelHandlerContext ctx, HttpRequest request, @Nul
             }
             validatingRequest = false;
             ctx.fireChannelRead(request);
+            if (readRequested) {
+                readRequested = false;
+                ctx.channel().eventLoop().execute(ctx::read);
+            }
         });
     }
 

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

@@ -146,9 +146,6 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
                     }
                 }
                 handlePipelinedRequest(ctx, netty4HttpRequest);
-                if (request instanceof FullHttpRequest) {
-                    ctx.read();
-                }
             } else {
                 assert msg instanceof HttpContent : "expect HttpContent got " + msg;
                 assert currentRequestStream != null : "current stream must exists before handling http content";
@@ -158,6 +155,11 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
                 }
             }
         } finally {
+            if (msg instanceof HttpRequest httpRequest) {
+                if (httpRequest instanceof FullHttpRequest || httpRequest.decoderResult().isFailure()) {
+                    ctx.channel().eventLoop().execute(ctx::read);
+                }
+            }
             activityTracker.stopActivity();
         }
     }