When an error occurs during a streaming response, call HttpServerResponse.reset()

Fixes: quarkusio#50336

Author: geoand

extensions/resteasy-reactive/rest-servlet/runtime/src/main/java/io/quarkus/resteasy/reactive/server/servlet/runtime/ServletRequestContext.java

@@ -724,4 +724,9 @@ public void handle(Void event) {
         });
         return this;
     }
+
+    @Override
+    public void reset() {
+        context.response().reset();
+    }
 }

independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java

@@ -246,6 +246,9 @@ protected void handleException(ResteasyReactiveRequestContext requestContext, Th
             // it will appear to be an SSE value, which is incorrect, so we should only log it and close the connection
             if (requestContext.serverResponse().headWritten()) {
                 log.error("Exception in SSE server handling, impossible to send it to client", t);
+                // HTTP chunked encoding sends an indeterminate number of chunks, but it has to end with an end chunk of zero size to indicate successful transmission.
+                // reset() will cause this last chunk to not be sent, even though every other chunk was sent, and so clients can detect the error
+                requestContext.serverResponse().reset();
             } else {
                 // we can go through the abort chain
                 requestContext.resume(t, true);

independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/spi/ServerHttpResponse.java

@@ -50,4 +50,6 @@ public interface ServerHttpResponse extends StreamingResponse<ServerHttpResponse
     boolean isWriteQueueFull();
 
     ServerHttpResponse addDrainHandler(Runnable onDrain);
+
+    void reset();
 }

independent-projects/resteasy-reactive/server/vertx/src/main/java/org/jboss/resteasy/reactive/server/vertx/VertxResteasyReactiveRequestContext.java

@@ -531,6 +531,11 @@ public void handle(Void event) {
         return this;
     }
 
+    @Override
+    public void reset() {
+        response.reset();
+    }
+
     @Override
     public boolean isWriteQueueFull() {
         return response.writeQueueFull();