WebSockets Next: error handlers part 4

- use error handlers to process Mutiny Multi failures
- update docs

docs/src/main/asciidoc/websockets-next-reference.adoc

@@ -193,6 +193,7 @@ A WebSocket endpoint comprises the following components:
 * At most one `@OnBinaryMessage` method: Handles the binary messages the connected client sends.
 * At most one `@OnOpen` method: Invoked when a client connects to the WebSocket.
 * At most one `@OnClose` method: Executed upon the client disconnecting from the WebSocket.
+* Any number of `@OnError` methods: Invoked when an error occurs; that is when an endpoint callback throws a runtime error, or when a conversion errors occurs, or when a returned `io.smallrye.mutiny.Uni`/`io.smallrye.mutiny.Multi` receives a failure. 
 
 Only some endpoints need to include all methods.
 However, it must contain at least `@On[Text|Binary]Message` or `@OnOpen`.
@@ -222,23 +223,26 @@ Here are the rules governing execution:
 * Methods annotated with `@RunOnVirtualThread` are considered blocking and should execute on a virtual thread.
 * Blocking methods must execute on a worker thread if not annotated with `@RunOnVirtualThread`.
 * When `@RunOnVirtualThread` is employed, each invocation spawns a new virtual thread.
-* Methods returning `CompletionStage` and `Uni` are considered non-blocking
-* Methods returning `Multi` are considered non-blocking and must be subscribed to, except if they return their own `Multi`.
+* Methods returning `CompletionStage`, `Uni` and `Multi` are considered non-blocking.
 * Methods returning `void` or plain objects are considered blocking.
 
-=== Parameters
+=== Method Parameters
 
-These methods can accept parameters in two formats:
+The method must accept exactly one message parameter:
 
 * The message object (of any type).
 * A `Multi<X>` with X as the message type.
-* Any other parameters should be flagged as errors.
 
+However, it may also accept the following parameters:
+
+ * `WebSocketConnection`
+ * `HandshakeRequest`
+ * `String` parameters annotated with `@PathParam`
+
 The message object represents the data sent and can be accessed as either raw content (`String`, `JsonObject`, `JsonArray`, `Buffer` or `byte[]`) or deserialized high-level objects, which is the recommended approach.
 
 When receiving a `Multi`, the method is invoked once per connection, and the provided `Multi` receives the items transmitted by this connection.
 The method must subscribe to the `Multi` to receive these items (or return a Multi).
-Cancelling this subscription closes the associated connection.
 
 === Allowed Returned Types
 
@@ -324,7 +328,7 @@ When a method is intended to produce a message written to the client, it can emi
 Emitting `null` signifies no response to be sent to the client, allowing for skipping a response when needed.
 
 === JsonObject and JsonArray
-Vert.x `JSONObject` and `JSONArray` instances bypass the serialization and deserialization mechanisms.
+Vert.x `JsonObject` and `JsonArray` instances bypass the serialization and deserialization mechanisms.
 Messages are sent as text messages.
 
 === Broadcasting
@@ -341,6 +345,8 @@ String emitToAll(String message) {
 
 The same principle applies to methods returning instances of `Multi` or `Uni`.
 
+NOTE: If you need to select the connected clients that should receive the message, you can use `WebSocketConnection.broadcast().filter().sendText()`.
+
 == OnOpen and OnClose methods
 
 The WebSocket endpoint can also be notified when a client connects or disconnects.
@@ -369,8 +375,11 @@ These methods have access to the _session-scoped_ `WebSocketConnection` bean.
 
 === Parameters
 
-Methods annotated with `@OnOpen` and `@OnClose` do not accept any parameters.
-If such methods declare parameters, they will be flagged as errors and reported at build time.
+Methods annotated with `@OnOpen` and `@OnClose` may accept the following parameters:
+
+ * `WebSocketConnection`
+ * `HandshakeRequest`
+ * `String` parameters annotated with `@PathParam`
 
 === Allowed Returned Types
 
@@ -425,6 +434,25 @@ String onOpen() {
 }
 ----
 
+== Error Handling
+
+The WebSocket endpoint can also be notified when an error occurs.
+A WebSocket endpoint method annotated with `@io.quarkus.websockets.next.OnError` is invoked when an endpoint callback throws a runtime error, or when a conversion errors occurs, 
+or when a returned `io.smallrye.mutiny.Uni`/`io.smallrye.mutiny.Multi` receives a failure.
+
+The method must accept exactly one "error" parameter, i.e. a parameter that is assignable from `java.lang.Throwable`.
+The method may also accept the following parameters:
+
+* `WebSocketConnection`
+* `HandshakeRequest`
+* `String` parameters annotated with `@PathParam`
+
+An endpoint may declare multiple methods annotated with `@io.quarkus.websockets.next.OnError`.
+However, each method must declare a different error parameter.
+The method that declares a most-specific supertype of the actual exception is selected.
+
+NOTE: The `@io.quarkus.websockets.next.OnError` annotation can be also used to declare a global error handler, i.e. a method that is not declared on a WebSocket endpoint. Such a method may not accept `@PathParam` paremeters. Error handlers declared on an endpoint take precedence over the global error handlers.
+
 == Access to the WebSocketConnection
 
 The `io.quarkus.websockets.next.WebSocketConnection` object represents the WebSocket connection.

extensions/websockets-next/server/deployment/src/main/java/io/quarkus/websockets/next/deployment/WebSocketServerProcessor.java

@@ -463,7 +463,7 @@ private String generateEndpoint(WebSocketEndpointBuildItem endpoint,
                     MethodDescriptor.ofMethod(WebSocketEndpointBase.class, "beanInstance", Object.class, String.class),
                     doOnOpen.getThis(), doOnOpen.load(endpoint.bean.getIdentifier()));
             // Call the business method
-            TryBlock tryBlock = onErrorTryBlock(doOnOpen);
+            TryBlock tryBlock = onErrorTryBlock(doOnOpen, doOnOpen.getThis());
             ResultHandle[] args = callback.generateArguments(tryBlock.getThis(), tryBlock, transformedAnnotations, index);
             ResultHandle ret = tryBlock.invokeVirtualMethod(MethodDescriptor.of(callback.method), beanInstance, args);
             encodeAndReturnResult(tryBlock.getThis(), tryBlock, callback, globalErrorHandlers, endpoint, ret);
@@ -488,7 +488,7 @@ private String generateEndpoint(WebSocketEndpointBuildItem endpoint,
                     MethodDescriptor.ofMethod(WebSocketEndpointBase.class, "beanInstance", Object.class, String.class),
                     doOnClose.getThis(), doOnClose.load(endpoint.bean.getIdentifier()));
             // Call the business method
-            TryBlock tryBlock = onErrorTryBlock(doOnClose);
+            TryBlock tryBlock = onErrorTryBlock(doOnClose, doOnClose.getThis());
             ResultHandle[] args = callback.generateArguments(tryBlock.getThis(), tryBlock, transformedAnnotations, index);
             ResultHandle ret = tryBlock.invokeVirtualMethod(MethodDescriptor.of(callback.method), beanInstance, args);
             encodeAndReturnResult(tryBlock.getThis(), tryBlock, callback, globalErrorHandlers, endpoint, ret);
@@ -632,7 +632,7 @@ private void generateOnMessage(ClassCreator endpointCreator, WebSocketEndpointBu
         MethodCreator doOnMessage = endpointCreator.getMethodCreator("doOn" + messageType + "Message", Uni.class,
                 methodParameterType);
 
-        TryBlock tryBlock = onErrorTryBlock(doOnMessage);
+        TryBlock tryBlock = onErrorTryBlock(doOnMessage, doOnMessage.getThis());
         // Foo foo = beanInstance("foo");
         ResultHandle beanInstance = tryBlock.invokeVirtualMethod(
                 MethodDescriptor.ofMethod(WebSocketEndpointBase.class, "beanInstance", Object.class, String.class),
@@ -673,13 +673,13 @@ private TryBlock uniFailureTryBlock(BytecodeCreator method) {
         return tryBlock;
     }
 
-    private TryBlock onErrorTryBlock(BytecodeCreator method) {
+    private TryBlock onErrorTryBlock(BytecodeCreator method, ResultHandle endpointThis) {
         TryBlock tryBlock = method.tryBlock();
         CatchBlockCreator catchBlock = tryBlock.addCatch(Throwable.class);
         // return doOnError(t);
         catchBlock.returnValue(catchBlock.invokeVirtualMethod(
                 MethodDescriptor.ofMethod(WebSocketEndpointBase.class, "doOnError", Uni.class, Throwable.class),
-                catchBlock.getThis(), catchBlock.getCaughtException()));
+                endpointThis, catchBlock.getCaughtException()));
         return tryBlock;
     }
 
@@ -810,23 +810,28 @@ private ResultHandle encodeMessage(ResultHandle endpointThis, BytecodeCreator me
                     return uniOnFailureDoOnError(endpointThis, method, callback, uniChain, endpoint, globalErrorHandlers);
                 }
             } else if (callback.isReturnTypeMulti()) {
-                // return multiBinary(multi, broadcast, m -> {
-                //    Buffer buffer = encodeBuffer(m);
-                //    return sendBinary(buffer,broadcast);
-                //});
+                //    try {
+                //      Buffer buffer = encodeBuffer(m);
+                //      return sendBinary(buffer,broadcast);
+                //    } catch(Throwable t) {
+                //      return doOnError(t);
+                //    }
                 FunctionCreator fun = method.createFunction(Function.class);
                 BytecodeCreator funBytecode = fun.getBytecode();
-                ResultHandle buffer = encodeBuffer(funBytecode, callback.returnType().asParameterizedType().arguments().get(0),
-                        funBytecode.getMethodParam(0), endpointThis, callback);
-                funBytecode.returnValue(funBytecode.invokeVirtualMethod(
+                // This checkcast should not be necessary but we need to use the endpoint in the function bytecode
+                // otherwise gizmo does not access the endpoint reference correcly
+                ResultHandle endpointBase = funBytecode.checkCast(endpointThis, WebSocketEndpointBase.class);
+                TryBlock tryBlock = onErrorTryBlock(fun.getBytecode(), endpointBase);
+                ResultHandle buffer = encodeBuffer(tryBlock, callback.returnType().asParameterizedType().arguments().get(0),
+                        tryBlock.getMethodParam(0), endpointThis, callback);
+                tryBlock.returnValue(tryBlock.invokeVirtualMethod(
                         MethodDescriptor.ofMethod(WebSocketEndpointBase.class,
                                 "sendBinary", Uni.class, Buffer.class, boolean.class),
                         endpointThis, buffer,
-                        funBytecode.load(callback.broadcast())));
+                        tryBlock.load(callback.broadcast())));
                 return method.invokeVirtualMethod(MethodDescriptor.ofMethod(WebSocketEndpointBase.class,
-                        "multiBinary", Uni.class, Multi.class, boolean.class, Function.class), endpointThis,
+                        "multiBinary", Uni.class, Multi.class, Function.class), endpointThis,
                         value,
-                        method.load(callback.broadcast()),
                         fun.getInstance());
             } else {
                 // return sendBinary(buffer,broadcast);
@@ -865,22 +870,29 @@ private ResultHandle encodeMessage(ResultHandle endpointThis, BytecodeCreator me
                 }
             } else if (callback.isReturnTypeMulti()) {
                 // return multiText(multi, broadcast, m -> {
-                //    String text = encodeText(m);
-                //    return sendText(buffer,broadcast);
+                //    try {
+                //      String text = encodeText(m);
+                //      return sendText(buffer,broadcast);
+                //    } catch(Throwable t) {
+                //      return doOnError(t);
+                //    }
                 //});
                 FunctionCreator fun = method.createFunction(Function.class);
                 BytecodeCreator funBytecode = fun.getBytecode();
-                ResultHandle text = encodeText(funBytecode, callback.returnType().asParameterizedType().arguments().get(0),
-                        funBytecode.getMethodParam(0), endpointThis, callback);
-                funBytecode.returnValue(funBytecode.invokeVirtualMethod(
+                // This checkcast should not be necessary but we need to use the endpoint in the function bytecode
+                // otherwise gizmo does not access the endpoint reference correcly
+                ResultHandle endpointBase = funBytecode.checkCast(endpointThis, WebSocketEndpointBase.class);
+                TryBlock tryBlock = onErrorTryBlock(fun.getBytecode(), endpointBase);
+                ResultHandle text = encodeText(tryBlock, callback.returnType().asParameterizedType().arguments().get(0),
+                        tryBlock.getMethodParam(0), endpointThis, callback);
+                tryBlock.returnValue(tryBlock.invokeVirtualMethod(
                         MethodDescriptor.ofMethod(WebSocketEndpointBase.class,
                                 "sendText", Uni.class, String.class, boolean.class),
                         endpointThis, text,
-                        funBytecode.load(callback.broadcast())));
+                        tryBlock.load(callback.broadcast())));
                 return method.invokeVirtualMethod(MethodDescriptor.ofMethod(WebSocketEndpointBase.class,
-                        "multiText", Uni.class, Multi.class, boolean.class, Function.class), endpointThis,
+                        "multiText", Uni.class, Multi.class, Function.class), endpointThis,
                         value,
-                        method.load(callback.broadcast()),
                         fun.getInstance());
             } else {
                 // return sendText(text,broadcast);

extensions/websockets-next/server/deployment/src/test/java/io/quarkus/websockets/next/test/args/OnClosePathParamConnectionArgumentTest.java

@@ -0,0 +1,68 @@
+package io.quarkus.websockets.next.test.args;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import jakarta.inject.Inject;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.quarkus.websockets.next.OnClose;
+import io.quarkus.websockets.next.OnOpen;
+import io.quarkus.websockets.next.PathParam;
+import io.quarkus.websockets.next.WebSocket;
+import io.quarkus.websockets.next.WebSocketConnection;
+import io.quarkus.websockets.next.test.utils.WSClient;
+import io.vertx.core.Vertx;
+import io.vertx.core.http.WebSocketConnectOptions;
+
+public class OnClosePathParamConnectionArgumentTest {
+
+    @RegisterExtension
+    public static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> {
+                root.addClasses(MontyEcho.class, WSClient.class);
+            });
+
+    @Inject
+    Vertx vertx;
+
+    @TestHTTPResource("echo/monty/and/foo")
+    URI testUri;
+
+    @Test
+    void testArguments() throws InterruptedException {
+        String header = "fool";
+        WSClient client = WSClient.create(vertx).connect(new WebSocketConnectOptions().addHeader("X-Test", header), testUri);
+        client.disconnect();
+        assertTrue(MontyEcho.CLOSED_LATCH.await(5, TimeUnit.SECONDS));
+        assertEquals("foo:monty:fool", MontyEcho.CLOSED_MESSAGE.get());
+    }
+
+    @WebSocket(path = "/echo/{grail}/and/{life}")
+    public static class MontyEcho {
+
+        static final CountDownLatch CLOSED_LATCH = new CountDownLatch(1);
+        static final AtomicReference<String> CLOSED_MESSAGE = new AtomicReference<>();
+
+        @OnOpen
+        void open() {
+        }
+
+        @OnClose
+        void close(@PathParam String life, @PathParam String grail, WebSocketConnection connection) {
+            CLOSED_MESSAGE.set(life + ":" + grail + ":" + connection.handshakeRequest().header("X-Test"));
+            CLOSED_LATCH.countDown();
+        }
+
+    }
+
+}

extensions/websockets-next/server/deployment/src/test/java/io/quarkus/websockets/next/test/args/OnOpenPathParamConnectionArgumentTest.java

@@ -0,0 +1,54 @@
+package io.quarkus.websockets.next.test.args;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.net.URI;
+
+import jakarta.inject.Inject;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.quarkus.websockets.next.OnOpen;
+import io.quarkus.websockets.next.PathParam;
+import io.quarkus.websockets.next.WebSocket;
+import io.quarkus.websockets.next.WebSocketConnection;
+import io.quarkus.websockets.next.test.utils.WSClient;
+import io.vertx.core.Vertx;
+import io.vertx.core.http.WebSocketConnectOptions;
+
+public class OnOpenPathParamConnectionArgumentTest {
+
+    @RegisterExtension
+    public static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> {
+                root.addClasses(MontyEcho.class, WSClient.class);
+            });
+
+    @Inject
+    Vertx vertx;
+
+    @TestHTTPResource("echo/monty/and/foo")
+    URI testUri;
+
+    @Test
+    void testArguments() {
+        String header = "fool";
+        WSClient client = WSClient.create(vertx).connect(new WebSocketConnectOptions().addHeader("X-Test", header), testUri);
+        client.waitForMessages(1);
+        assertEquals("foo:monty:fool", client.getMessages().get(0).toString());
+    }
+
+    @WebSocket(path = "/echo/{grail}/and/{life}")
+    public static class MontyEcho {
+
+        @OnOpen
+        String process(@PathParam String life, @PathParam String grail, WebSocketConnection connection) {
+            return life + ":" + grail + ":" + connection.handshakeRequest().header("X-Test");
+        }
+
+    }
+
+}