Don't close the connection when returning Multi

Fixes #18445
quarkusio · Jul 8, 2021 · 5f120f8 · 5f120f8
1 parent 5bd6f46
commit 5f120f8
Show file tree

Hide file tree

Showing 2 changed files with 42 additions and 4 deletions.
diff --git a/...loyment/src/test/java/io/quarkus/resteasy/reactive/server/test/stream/StreamTestCase.java b/...loyment/src/test/java/io/quarkus/resteasy/reactive/server/test/stream/StreamTestCase.java
@@ -1,12 +1,14 @@
 package io.quarkus.resteasy.reactive.server.test.stream;
 
 import java.net.URI;
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.ws.rs.client.Client;
 import javax.ws.rs.client.ClientBuilder;
@@ -28,6 +30,13 @@
 import io.restassured.RestAssured;
 import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.subscription.Cancellable;
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpClient;
+import io.vertx.core.http.HttpClientRequest;
+import io.vertx.core.http.HttpConnection;
+import io.vertx.core.http.HttpMethod;
 
 @DisabledOnOs(OS.WINDOWS)
 public class StreamTestCase {
@@ -40,6 +49,39 @@ public class StreamTestCase {
             .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
                     .addClasses(StreamResource.class));
 
+    @Test
+    public void testStreamingDoesNotCloseConnection() throws Exception {
+        Vertx v = Vertx.vertx();
+        try {
+            AtomicInteger count = new AtomicInteger();
+            HttpClient client = v.createHttpClient().connectionHandler(new Handler<HttpConnection>() {
+                @Override
+                public void handle(HttpConnection event) {
+                    count.incrementAndGet();
+                }
+            });
+            HttpClientRequest req = client.request(HttpMethod.GET, RestAssured.port, "localhost", "/stream/text/stream")
+                    .toCompletionStage().toCompletableFuture().get();
+            req.end();
+            Buffer body = req.connect().toCompletionStage().toCompletableFuture().get().body().toCompletionStage()
+                    .toCompletableFuture().get();
+            Assertions.assertEquals("foobar", body.toString(StandardCharsets.US_ASCII));
+
+            req = client.request(HttpMethod.GET, RestAssured.port, "localhost", "/stream/text/stream")
+                    .toCompletionStage().toCompletableFuture().get();
+            req.end();
+            body = req.connect().toCompletionStage().toCompletableFuture().get().body().toCompletionStage()
+                    .toCompletableFuture().get();
+            Assertions.assertEquals("foobar", body.toString(StandardCharsets.US_ASCII));
+
+            //should have only connected once for both requests
+            Assertions.assertEquals(1, count.get());
+
+        } finally {
+            v.close().toCompletionStage().toCompletableFuture().get();
+        }
+    }
+
     @Test
     public void testStreaming() throws Exception {
         RestAssured.get("/stream/text/stream")

diff --git a/...e/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java b/...e/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java
@@ -135,10 +135,6 @@ public void onComplete() {
             // no need to cancel on complete
             // FIXME: are we interested in async completion?
             requestContext.serverResponse().end();
-            // so, if I don't also close the connection, the client isn't notified that the request is over
-            // I guess that's true of chunked responses, but not clear why I need to close the connection
-            // because it means it can't get reused, right?
-            requestContext.serverRequest().closeConnection();
             requestContext.close();
         }