Merge pull request #4077 from eclipse-vertx/http-client-tunnel-fixes-4.1

Http client tunnel fixes 4.1
eclipse-vertx · Aug 23, 2021 · 8c92bfe · 8c92bfe
2 parents 4d408d2 + 7ad3e50
commit 8c92bfe
Show file tree

Hide file tree

Showing 2 changed files with 45 additions and 6 deletions.
diff --git a/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java
@@ -350,6 +350,7 @@ private static class StreamImpl extends Stream implements HttpClientStream {
     private final InboundBuffer<Object> queue;
     private boolean reset;
     private boolean writable;
+    private boolean closed;
     private HttpRequestHead request;
     private Handler<HttpResponseHead> headHandler;
     private Handler<Buffer> chunkHandler;
@@ -585,9 +586,7 @@ void handleChunk(Buffer buff) {
 
     void handleEnd(LastHttpContent trailer) {
       queue.write(new HeadersAdaptor(trailer.trailingHeaders()));
-      if (closeHandler != null) {
-        closeHandler.handle(null);
-      }
+      tryClose();
     }
 
     void handleException(Throwable cause) {
@@ -599,8 +598,18 @@ void handleException(Throwable cause) {
     @Override
     void handleClosed() {
       handleException(CLOSED_EXCEPTION);
-      if (closeHandler != null) {
-        closeHandler.handle(null);
+      tryClose();
+    }
+
+    /**
+     * Attempt to close the stream.
+     */
+    private void tryClose() {
+      if (!closed) {
+        closed = true;
+        if (closeHandler != null) {
+          closeHandler.handle(null);
+        }
       }
     }
   }
@@ -744,9 +753,11 @@ private void handleResponseBegin(Stream stream, HttpResponseHead response) {
         if ((request.method == HttpMethod.CONNECT &&
              response.statusCode == 200) || (
              request.method == HttpMethod.GET &&
-             request.headers.contains("connection", "Upgrade", false) &&
+             request.headers != null && request.headers.contains("connection", "Upgrade", false) &&
              response.statusCode == 101)) {
           removeChannelHandlers();
+        } else {
+          isConnect = false;
         }
       }
     }

diff --git a/src/test/java/io/vertx/core/http/HttpTest.java b/src/test/java/io/vertx/core/http/HttpTest.java
@@ -5367,6 +5367,34 @@ public void testHttpInvalidConnectResponseChunked() {
     await();
   }
 
+  @Test
+  public void testUpgradeTunnelNoSwitch() throws Exception {
+    server.requestHandler(req -> {
+      HttpServerResponse resp = req.response();
+      resp.setChunked(true);
+      resp.write("chunk-1")
+        .compose(v -> resp.write("chunk-2"))
+        .compose(v -> resp.end("chunk-3"));
+    });
+
+    startServer();
+
+    client.request(HttpMethod.GET, DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, "/", onSuccess(req -> {
+      req.connect(onSuccess(resp -> {
+        assertEquals(200, resp.statusCode());
+        List<String> chunks = new ArrayList<>();
+        resp.handler(chunk -> {
+          chunks.add(chunk.toString());
+        });
+        resp.endHandler(v -> {
+          assertEquals(Arrays.asList("chunk-1", "chunk-2", "chunk-3"), chunks);
+          testComplete();
+        });
+      }));
+    }));
+    await();
+  }
+
   @Test
   public void testEndFromAnotherThread() throws Exception {
     waitFor(2);