HTTP/1.1 connection state might be corrupted when processing pipelined requests - closes #3082

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

@@ -132,11 +132,12 @@ public class Http1xServerConnection extends Http1xConnectionBase<ServerWebSocket
         if (METRICS_ENABLED) {
           req.reportRequestBegin();
         }
+        req.handleBegin();
       }
       ContextInternal ctx = req.context;
       ContextInternal prev = ctx.beginDispatch();
       try {
-        req.handleBegin(requestHandler);
+        requestHandler.handle(req);
       } catch (Throwable t) {
         ctx.reportException(t);
       } finally {
@@ -203,9 +204,10 @@ public class Http1xServerConnection extends Http1xConnectionBase<ServerWebSocket
 
   private void handleNext(HttpServerRequestImpl next) {
     responseInProgress = next;
-    getContext().runOnContext(v -> {
+    next.handleBegin();
+    context.runOnContext(v -> {
       next.resume();
-      next.handleBegin(requestHandler);
+      requestHandler.handle(next);
     });
   }
 

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

@@ -149,12 +149,11 @@ public class HttpServerRequestImpl implements HttpServerRequest {
     }
   }
 
-  void handleBegin(Handler<HttpServerRequest> handler) {
+  void handleBegin() {
     response = new HttpServerResponseImpl((VertxInternal) conn.vertx(), context, conn, request, metric);
     if (conn.handle100ContinueAutomatically) {
       check100();
     }
-    handler.handle(this);
   }
 
   /**

src/test/java/io/vertx/core/http/Http1xTest.java

@@ -22,7 +22,6 @@ import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.*;
-import io.vertx.core.net.impl.ConnectionBase;
 import io.vertx.core.parsetools.RecordParser;
 import io.vertx.core.streams.WriteStream;
 import io.vertx.test.core.Repeat;
@@ -1177,7 +1176,6 @@ public class Http1xTest extends HttpTest {
     await();
   }
 
-  @Repeat(times = 10)
   @Test
   public void testCloseServerConnectionWithPendingMessages() throws Exception {
     int n = 5;
@@ -1246,6 +1244,86 @@ public class Http1xTest extends HttpTest {
     await();
   }
 
+  /**
+   * A test that stress HTTP server pipe-lining.
+   */
+  @Repeat(times = 100)
+  @Test
+  public void testPipelineStress() throws Exception {
+
+    // A client that will aggressively pipeline HTTP requests and close the connection abruptly after one second
+    class Client {
+      private final NetSocket so;
+      private StringBuilder received;
+      private int curr;
+      private int count;
+      private boolean closed;
+      Client(NetSocket so) {
+        this.so = so;
+      }
+      private void close() {
+        closed = true;
+      }
+      private void receiveChunk(Buffer chunk) {
+        received.append(chunk);
+        int c;
+        while ((c = received.indexOf("\r\n\r\n", curr)) != -1) {
+          curr = c + 4;
+          count++;
+        }
+        if (count == 16 && !closed) {
+          send();
+        }
+      }
+      private void send() {
+        received = new StringBuilder();
+        curr = 0;
+        count = 0;
+        for (int i = 0;i < 16;i++) {
+          so.write(Buffer.buffer("" +
+            "GET / HTTP/1.1\r\n" +
+            "content-length:0\r\n" +
+            "\r\n"));
+        }
+      }
+      void run() {
+        so.handler(this::receiveChunk);
+        so.closeHandler(v -> {
+          close();
+          complete();
+        });
+        send();
+        vertx.setTimer(1000, id -> {
+          so.close();
+        });
+      }
+    }
+
+    // We want to be aware of uncaught exceptions and fail the test when it happens
+    vertx.exceptionHandler(err -> {
+      fail(err);
+    });
+
+    server.requestHandler(req -> {
+      // Use runOnContext to allow pipelined requests to pile up in the server
+      // when we send a response right away it's nearly like no pipe-lining is occurring
+      Vertx.currentContext().runOnContext(v -> {
+        req.response().end("Hello World");
+      });
+    });
+    startServer(testAddress);
+    NetClient tcpClient = vertx.createNetClient(new NetClientOptions().setSoLinger(0));
+    int numConn = 32;
+    waitFor(numConn);
+    for (int i = 0;i < numConn;i++) {
+      tcpClient.connect(testAddress, onSuccess(so -> {
+        Client client = new Client(so);
+        client.run();
+      }));
+    }
+    await();
+  }
+
   @Test
   public void testPipeliningPauseRequest() throws Exception {
     int n = 10;