Repository: quarkusio/quarkus
Issue: Mutiny Dropped Exception on SSE Cancel

### Describe the bug

Undesirable `ERROR` message appears in logs, e.g:

`2022-08-25 11:30:51,013 ERROR [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-2) Mutiny had to drop the following exception: io.vertx.core.http.HttpClosedException: Connection was closed`

### Expected behavior

Would prefer not to have these appearing in Production logs.  Can they be avoided or suppressed?

### Actual behavior

_No response_

### How to Reproduce?

https://github.com/jimbogithub/mutiny-sse

Run the server and the client.  Observe client logs.

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

_No response_

### Output of `java -version`

_No response_

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

_No response_

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

_No response_

### Additional information

Mutiny SSE otherwise looks much better than legacy SSE (https://github.com/quarkusio/quarkus/issues/23997).

Changed files:
independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java

diff --git a/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java b/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java
index 0b17f299d25..6ae2a510969 100644
--- a/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java
+++ b/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java
@@ -59,6 +59,26 @@ public MultiRequest(MultiEmitter<? super R> emitter) {
             });
         }
 
+        void emit(R item) {
+            if (!isCancelled()) {
+                emitter.emit(item);
+            }
+        }
+
+        void fail(Throwable t) {
+            if (!isCancelled()) {
+                emitter.fail(t);
+                cancel();
+            }
+        }
+
+        void complete() {
+            if (!isCancelled()) {
+                emitter.complete();
+                cancel();
+            }
+        }
+
         public boolean isCancelled() {
             return onCancel.get() == CLEARED;
         }
@@ -130,20 +150,20 @@ private <R> void registerForSse(MultiRequest<? super R> multiRequest,
         // For now we don't want multi to reconnect
         SseEventSourceImpl sseSource = new SseEventSourceImpl(invocationBuilder.getTarget(),
                 invocationBuilder, Integer.MAX_VALUE, TimeUnit.SECONDS);
-        // FIXME: deal with cancellation
+
+        multiRequest.onCancel(() -> {
+            sseSource.close();
+        });
         sseSource.register(event -> {
             // DO NOT pass the response mime type because it's SSE: let the event pick between the X-SSE-Content-Type header or
             // the content-type SSE field
-            multiRequest.emitter.emit((R) event.readData(responseType));
+            multiRequest.emit(event.readData(responseType));
         }, error -> {
-            multiRequest.emitter.fail(error);
+            multiRequest.fail(error);
         }, () -> {
-            multiRequest.emitter.complete();
+            multiRequest.complete();
         });
         // watch for user cancelling
-        multiRequest.onCancel(() -> {
-            sseSource.close();
-        });
         sseSource.registerAfterRequest(vertxResponse);
     }