test fix

Author: mhl-b

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

@@ -50,10 +50,11 @@ State getState() {
         return state;
     }
 
-    @Override
-    public void channelActive(ChannelHandlerContext ctx) throws Exception {
-        autoRead = GatedAutoRead.newGate(ctx.channel());
-        super.channelActive(ctx);
+    GatedAutoRead.Gate autoRead(ChannelHandlerContext ctx) {
+        if (autoRead == null) {
+            autoRead = GatedAutoRead.newGate(ctx.channel());
+        }
+        return autoRead;
     }
 
     @SuppressWarnings("fallthrough")
@@ -68,7 +69,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                 pending.add(ReferenceCountUtil.retain(httpObject));
                 requestStart(ctx);
                 assert state == QUEUEING_DATA;
-                assert autoRead.get() == false;
+                assert autoRead(ctx).get() == false;
                 break;
             case QUEUEING_DATA:
                 pending.add(ReferenceCountUtil.retain(httpObject));
@@ -90,7 +91,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
             case DROPPING_DATA_PERMANENTLY:
                 assert pending.isEmpty();
                 ReferenceCountUtil.release(httpObject); // consume without enqueuing
-                autoRead.set(false);
+                autoRead(ctx).set(false);
                 break;
         }
     }
@@ -113,7 +114,7 @@ private void requestStart(ChannelHandlerContext ctx) {
         }
 
         state = QUEUEING_DATA;
-        autoRead.set(false);
+        autoRead(ctx).set(false);
 
         if (httpRequest == null) {
             // this looks like a malformed request and will forward without validation
@@ -156,10 +157,10 @@ public void onFailure(Exception e) {
     private void forwardFullRequest(ChannelHandlerContext ctx) {
         Transports.assertDefaultThreadContext(threadContext);
         assert ctx.channel().eventLoop().inEventLoop();
-        assert autoRead.get() == false;
+        assert autoRead(ctx).get() == false;
         assert state == QUEUEING_DATA;
 
-        autoRead.set(true);
+        autoRead(ctx).set(true);
         boolean fullRequestForwarded = forwardData(ctx, pending);
 
         assert fullRequestForwarded || pending.isEmpty();
@@ -176,7 +177,7 @@ private void forwardFullRequest(ChannelHandlerContext ctx) {
     private void forwardRequestWithDecoderExceptionAndNoContent(ChannelHandlerContext ctx, Exception e) {
         Transports.assertDefaultThreadContext(threadContext);
         assert ctx.channel().eventLoop().inEventLoop();
-        assert autoRead.get() == false;
+        assert autoRead(ctx).get() == false;
         assert state == QUEUEING_DATA;
 
         HttpObject messageToForward = pending.getFirst();
@@ -187,7 +188,7 @@ private void forwardRequestWithDecoderExceptionAndNoContent(ChannelHandlerContex
         }
         messageToForward.setDecoderResult(DecoderResult.failure(e));
 
-        autoRead.set(true);
+        autoRead(ctx).set(true);
         ctx.fireChannelRead(messageToForward);
 
         assert fullRequestDropped || pending.isEmpty();

modules/transport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpHeaderValidatorTests.java

@@ -76,7 +76,6 @@ private void reset() {
         };
         netty4HttpHeaderValidator = new Netty4HttpHeaderValidator(validator, new ThreadContext(Settings.EMPTY));
         channel.pipeline().addLast(netty4HttpHeaderValidator);
-        channel.pipeline().fireChannelActive();
     }
 
     public void testValidationPausesAndResumesData() {

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/netty4/SecurityNetty4HttpServerTransportCloseNotifyTests.java

@@ -27,6 +27,7 @@
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 import io.netty.util.concurrent.Future;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
@@ -35,6 +36,7 @@
 import org.elasticsearch.http.HttpServerTransport;
 import org.elasticsearch.http.netty4.Netty4HttpServerTransport;
 import org.elasticsearch.rest.RestChannel;
+import org.elasticsearch.rest.RestContentAggregator;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
@@ -261,7 +263,23 @@ private static class QueuedDispatcher implements HttpServerTransport.Dispatcher
 
         @Override
         public void dispatchRequest(RestRequest request, RestChannel channel, ThreadContext threadContext) {
-            reqQueue.add(new ReqCtx(request, channel, threadContext));
+            if (request.isStreamedContent()) {
+                RestContentAggregator.aggregate(request, new ActionListener<>() {
+                    @Override
+                    public void onResponse(RestRequest restRequest) {
+                        reqQueue.add(new ReqCtx(request, channel, threadContext));
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        errQueue.add(new ErrCtx(channel, threadContext, e));
+
+                    }
+                });
+            } else {
+                reqQueue.add(new ReqCtx(request, channel, threadContext));
+            }
+
         }
 
         @Override