Repository: quarkusio/quarkus
Issue: Quarkus Resteasy Reactive does not respect keep-alive when using Multi<>s

## Describe the bug

Quarkus Resteasy Reactive kills connections even if Connection: keep-alive is specified and acknowledged by the server.

The problem only occurs when returning Multi<>s, not when returning Uni<>s or synchronous results.

A simple method such as

```
    @GET
    public Multi<String> async(){
        return Multi.createFrom().item("async");
    }

```
and then calling

`curl http://localhost:8080 http://localhost:8080 -v -H "Connection: keep-alive`

reproduces the problem.

### Expected behavior
Quarkus Resteasy Reactive should behave like non-reactive and keep the connection alive.

### Actual behavior
The connection is dead after the first request.

## To Reproduce

Small project based on the Quarkus Reactive quick-start.

https://github.com/rasmusfaber/quarkus-resteasy-reactive-connection-drop

Steps to reproduce the behavior:
1. Run the included tests and notice that `testHelloEndpointAsyncMulti` fails.
Or
1. Build and start the application
2. Run `curl http://localhost:8081/hello/async http://localhost:8081/hello/async -v -H "Connection: keep-alive"`
3. Notice that the second connection fails and curl needs to create a new connection (compare with `curl http://localhost:8081/hello/sync http://localhost:8081/hello/sync -v -H "Connection: keep-alive"`)

## Environment (please complete the following information):

### Output of `uname -a` or `ver`
Linux xxx 5.4.0-73-generic #82~18.04.1-Ubuntu SMP Fri Apr 16 15:10:02 UTC 2021 x86_64 x86_64 x86_64 GNU/Linux

### Output of `java -version`
openjdk version "15.0.2" 2021-01-19
OpenJDK Runtime Environment AdoptOpenJDK (build 15.0.2+7)
OpenJDK 64-Bit Server VM AdoptOpenJDK (build 15.0.2+7, mixed mode, sharing)

### Quarkus version or git rev
2.0.0.Final (also 1.13.3.Final)

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)
apache-maven-3.6.3


Changed files:
extensions/resteasy-reactive/quarkus-resteasy-reactive/deployment/src/test/java/io/quarkus/resteasy/reactive/server/test/stream/StreamTestCase.java
independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java

diff --git a/extensions/resteasy-reactive/quarkus-resteasy-reactive/deployment/src/test/java/io/quarkus/resteasy/reactive/server/test/stream/StreamTestCase.java b/extensions/resteasy-reactive/quarkus-resteasy-reactive/deployment/src/test/java/io/quarkus/resteasy/reactive/server/test/stream/StreamTestCase.java
index a4a20751170..ca3de29b2cb 100644
--- a/extensions/resteasy-reactive/quarkus-resteasy-reactive/deployment/src/test/java/io/quarkus/resteasy/reactive/server/test/stream/StreamTestCase.java
+++ b/extensions/resteasy-reactive/quarkus-resteasy-reactive/deployment/src/test/java/io/quarkus/resteasy/reactive/server/test/stream/StreamTestCase.java
@@ -1,9 +1,11 @@
 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.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -28,6 +30,12 @@
 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.http.HttpClient;
+import io.vertx.core.http.HttpClientOptions;
+import io.vertx.core.http.HttpClientRequest;
+import io.vertx.core.http.HttpMethod;
 
 @DisabledOnOs(OS.WINDOWS)
 public class StreamTestCase {
@@ -40,6 +48,56 @@ public class StreamTestCase {
             .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
                     .addClasses(StreamResource.class));
 
+    @Test
+    public void testStreamingDoesNotCloseConnection() throws Exception {
+        Vertx v = Vertx.vertx();
+        try {
+            final CompletableFuture<Object> latch = new CompletableFuture<>();
+            HttpClient client = v
+                    .createHttpClient(
+                            new HttpClientOptions().setKeepAlive(true).setIdleTimeout(10).setIdleTimeoutUnit(TimeUnit.SECONDS));
+            sendRequest(latch, client, () -> sendRequest(latch, client, () -> latch.complete(null)));
+
+            //should not have been closed
+            latch.get();
+
+        } finally {
+            v.close().toCompletionStage().toCompletableFuture().get();
+        }
+    }
+
+    private void sendRequest(CompletableFuture<Object> latch, HttpClient client, Runnable runnable) {
+        Handler<Throwable> failure = latch::completeExceptionally;
+        client.request(HttpMethod.GET, RestAssured.port, "localhost", "/stream/text/stream")
+                .onFailure(failure)
+                .onSuccess(new Handler<HttpClientRequest>() {
+                    @Override
+                    public void handle(HttpClientRequest event) {
+                        event.end();
+                        event.connect().onFailure(failure)
+                                .onSuccess(response -> {
+                                    response.request().connection().closeHandler(new Handler<Void>() {
+                                        @Override
+                                        public void handle(Void event) {
+                                            latch.completeExceptionally(new Throwable("Connection was closed"));
+                                        }
+                                    });
+                                    response.body().onFailure(failure)
+                                            .onSuccess(buffer -> {
+                                                try {
+                                                    Assertions.assertEquals("foobar",
+                                                            buffer.toString(StandardCharsets.US_ASCII));
+                                                } catch (Throwable t) {
+                                                    latch.completeExceptionally(t);
+                                                }
+                                                runnable.run();
+                                            });
+                                });
+
+                    }
+                });
+    }
+
     @Test
     public void testStreaming() throws Exception {
         RestAssured.get("/stream/text/stream")
diff --git a/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java b/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java
index 0b30d2e2714..edb1625f17b 100644
--- a/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java
+++ b/independent-projects/resteasy-reactive/server/runtime/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();
         }