WebSocket Next Server-Side Streaming cancelation

[andlinger]

Describe the bug

If a server-side streaming connection is closed by the client, the subscription of Multi should also be cancelled.

@WebSocket(path = "/v1/products")
@RequestScoped
public class ProductSocket {

    @OnOpen
    public Multi<Product> onOpen() {
        return Multi.createFrom()
                .ticks()
                .every(Duration.ofSeconds(2))
                .map(aLong -> new Product(String.valueOf(aLong)));
    }
}

Expected behavior

The returned Multi<Product> should be cancelled if the client closes the WebSocket connection.

Actual behavior

The Mutli<Product> subscription stays active and produces following exceptions every tick:

2024-06-06 13:20:07,851 ERROR [io.qua.web.nex.run.WebSocketEndpointBase] (executor-thread-1) Unable to send text message from Multi: WebSocket connection [id=13bedb47-618f-4631-9a5a-164807c53572, path=/v1/products]: io.vertx.core.impl.NoStackTraceThrowable: WebSocket is closed

How to Reproduce?

Created a reproducer: https://github.com/andlinger/quarkus-reproducer/tree/quarkus-41025

[geoand]

cc @mkouba