Repository: quarkusio/quarkus
Issue: Server Sent Events delaying 200 response

### Describe the bug

I am on 2.6.1.Final. I have an endpoint that is an SSE endpoint. The status code is only returned after the first element is produced.

This was not the case in 1.x and can lead to timeouts on clients.

### Expected behavior

I would expect a 200 to be returned immediately (provided everything worked as expected).

### Actual behavior

The status code, headers, etc. is not returned until the moment the first item is produced.

### How to Reproduce?

[getting-started.zip](https://github.com/quarkusio/quarkus/files/7837816/getting-started.zip)


### Output of `uname -a` or `ver`

Linux jose-buguroo 5.11.0-41-generic #45-Ubuntu SMP Fri Nov 5 11:37:01 UTC 2021 x86_64 x86_64 x86_64 GNU/Linux

### Output of `java -version`

openjdk version "14.0.2" 2020-07-14 OpenJDK Runtime Environment (build 14.0.2+12-46) OpenJDK 64-Bit Server VM (build 14.0.2+12-46, mixed mode, sharing)

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

2.6.1.Final

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

Apache Maven 3.6.3 (cecedd343002696d0abb50b32b541b8a6ba2883f) Maven home: /home/jose/.m2/wrapper/dists/apache-maven-3.6.3-bin/1iopthnavndlasol9gbrbg6bf2/apache-maven-3.6.3 Java version: 14.0.2, vendor: Oracle Corporation, runtime: /home/jose/.sdkman/candidates/java/14.0.2-open Default locale: en_US, platform encoding: UTF-8 OS name: "linux", version: "5.11.0-41-generic", arch: "amd64", family: "unix"

### Additional information

From what I see `SseUtil.send` is producing the response the first time an element is produced. Therefore, until the first element is generated, no call to it happens and no response is flushed. I see this could have to do with `PublisherResponseHandler` suspending the context. I wonder if an option could be to have that class handle SSE slightly different and generate a status code.

I see that adding:
```
requestContext.serverResponse().setStatusCode(200)
requestContext.serverResponse().setChunked(true)
requestContext.serverResponse().write(new byte[]{})
```

to PublisherResponseHandler works as expected (i.e. we get the status code, headers, etc. and then the events get produced as they come). Any reason you can think of not to do that?

Changed files:
independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/jaxrs/SseEventSinkImpl.java
independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/SseUtil.java
independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java

diff --git a/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/SseUtil.java b/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/SseUtil.java
index 963d71a82fe..07564a0982b 100644
--- a/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/SseUtil.java
+++ b/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/SseUtil.java
@@ -151,8 +151,6 @@ public static void setHeaders(ResteasyReactiveRequestContext context, ServerHttp
 
     public static void setHeaders(ResteasyReactiveRequestContext context, ServerHttpResponse response,
             List<PublisherResponseHandler.StreamingResponseCustomizer> customizers) {
-        // FIXME: spec says we should flush the headers when first message is sent or when the resource method returns, whichever
-        // happens first
         if (!response.headWritten()) {
             response.setStatusCode(Response.Status.OK.getStatusCode());
             response.setResponseHeader(HttpHeaders.CONTENT_TYPE, MediaType.SERVER_SENT_EVENTS);
@@ -164,6 +162,7 @@ public static void setHeaders(ResteasyReactiveRequestContext context, ServerHttp
                 customizers.get(i).customize(response);
             }
             // FIXME: other headers?
+
         }
     }
 }
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 43ddb8b91cf..64dc218adb1 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
@@ -1,10 +1,14 @@
 package org.jboss.resteasy.reactive.server.handlers;
 
+import static org.jboss.resteasy.reactive.server.jaxrs.SseEventSinkImpl.EMPTY_BUFFER;
+
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 import javax.ws.rs.core.MediaType;
 import org.jboss.logging.Logger;
 import org.jboss.resteasy.reactive.common.util.RestMediaType;
@@ -43,9 +47,9 @@ private static class SseMultiSubscriber extends AbstractMultiSubscriber {
         @Override
         public void onNext(Object item) {
             OutboundSseEventImpl event = new OutboundSseEventImpl.BuilderImpl().data(item).build();
-            SseUtil.send(requestContext, event, customizers).handle(new BiFunction<Object, Throwable, Object>() {
+            SseUtil.send(requestContext, event, customizers).whenComplete(new BiConsumer<Object, Throwable>() {
                 @Override
-                public Object apply(Object v, Throwable t) {
+                public void accept(Object v, Throwable t) {
                     if (t != null) {
                         // need to cancel because the exception didn't come from the Multi
                         subscription.cancel();
@@ -54,7 +58,6 @@ public Object apply(Object v, Throwable t) {
                         // send in the next item
                         subscription.request(1);
                     }
-                    return null;
                 }
             });
         }
@@ -250,11 +253,11 @@ public void handle(ResteasyReactiveRequestContext requestContext) throws Excepti
             requestContext.setResponseContentType(mediaType);
             // this is the non-async return type
             requestContext.setGenericReturnType(requestContext.getTarget().getReturnType());
-            // we have several possibilities here, but in all we suspend
-            requestContext.suspend();
+
             if (mediaType.isCompatible(MediaType.SERVER_SENT_EVENTS_TYPE)) {
                 handleSse(requestContext, result);
             } else {
+                requestContext.suspend();
                 boolean json = mediaType.toString().contains(JSON);
                 if (requiresChunkedStream(mediaType)) {
                     handleChunkedStreaming(requestContext, result, json);
@@ -279,7 +282,18 @@ private void handleStreaming(ResteasyReactiveRequestContext requestContext, Publ
     }
 
     private void handleSse(ResteasyReactiveRequestContext requestContext, Publisher<?> result) {
-        result.subscribe(new SseMultiSubscriber(requestContext, streamingResponseCustomizers));
+        SseUtil.setHeaders(requestContext, requestContext.serverResponse(), streamingResponseCustomizers);
+        requestContext.suspend();
+        requestContext.serverResponse().write(EMPTY_BUFFER, new Consumer<Throwable>() {
+            @Override
+            public void accept(Throwable throwable) {
+                if (throwable == null) {
+                    result.subscribe(new SseMultiSubscriber(requestContext, streamingResponseCustomizers));
+                } else {
+                    requestContext.resume(throwable);
+                }
+            }
+        });
     }
 
     public interface StreamingResponseCustomizer {
diff --git a/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/jaxrs/SseEventSinkImpl.java b/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/jaxrs/SseEventSinkImpl.java
index 6cdf099eab4..48735ac388c 100644
--- a/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/jaxrs/SseEventSinkImpl.java
+++ b/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/jaxrs/SseEventSinkImpl.java
@@ -11,7 +11,7 @@
 
 public class SseEventSinkImpl implements SseEventSink {
 
-    private static final byte[] EMPTY_BUFFER = new byte[0];
+    public static final byte[] EMPTY_BUFFER = new byte[0];
     private ResteasyReactiveRequestContext context;
     private SseBroadcasterImpl broadcaster;
     private boolean closed;