Always process a message in Http1xServerConnection and don't put it b…

…ack on the queue, instead when the HttpRequest ends and the corresponding response is still in progress then we pause the connection to avoid processing pipelined requests concurrently

src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java

@@ -139,8 +139,10 @@ synchronized void resume() {
   }
 
   synchronized void handleMessage(Object msg) {
-    if (queueing || !processMessage(msg)) {
+    if (queueing) {
       enqueue(msg);
+    } else {
+      processMessage(msg);
     }
   }
 
@@ -180,7 +182,9 @@ synchronized void responseComplete() {
       }
     }
     pendingResponse = null;
-    checkNextTick();
+    if (currentRequest == null && paused) {
+      resume();
+    }
   }
 
   synchronized void requestHandlers(HttpHandlers handlers) {
@@ -427,15 +431,12 @@ private void handleError(HttpObject obj) {
     }
   }
 
-  private boolean processMessage(Object msg) {
+  private void processMessage(Object msg) {
     if (msg instanceof HttpRequest) {
-      if (pendingResponse != null) {
-        return false;
-      }
       HttpRequest request = (HttpRequest) msg;
       if (request.decoderResult().isFailure()) {
         handleError(request);
-        return false;
+        return;
       }
       if (options.isHandle100ContinueAutomatically() && HttpUtil.is100ContinueExpected(request)) {
         write100Continue();
@@ -457,7 +458,6 @@ private boolean processMessage(Object msg) {
     } else {
       handleOther(msg);
     }
-    return true;
   }
 
   private void handleContent(Object msg) {
@@ -484,6 +484,9 @@ private void handleLastHttpContent() {
       bytesRead = 0;
     }
     currentRequest = null;
+    if (pendingResponse != null) {
+      pause();
+    }
   }
 
   private void handleOther(Object msg) {
@@ -506,15 +509,14 @@ private void checkNextTick() {
             // The only place we poll the pending queue, so we are sure that pending.size() > 0
             // since we got there because queueing was true
             Object msg = pending.poll();
-            if (processMessage(msg)) {
-              if (pending.isEmpty()) {
-                unsetQueueing();
-              } else {
-                checkNextTick();
-              }
-            } else {
-              pending.addFirst(msg);
+            if (pending.isEmpty()) {
+              // paused == false && pending.size() == 0 => queueing = false
+              unsetQueueing();
             }
+            // Process message, it might pause the connection
+            processMessage(msg);
+            // Check next tick in case we still have pending messages and the connection is not paused
+            checkNextTick();
           }
         }
       });