fix: let WebSocketClient share a borrowed client's event loop

changes.md

@@ -125,10 +125,13 @@ Changes log
           answers pings, pings idle sessions, and supports the "websocket.maxSessions",
           "websocket.maxMessageSize", "websocket.maxFrameSize" and "websocket.compression"
           parameters. Server and client support "permessage-deflate" compression (RFC 7692),
-          enabled by default.
+          enabled by default. `WebSocketClient.setClient` shares the I/O threads of a borrowed,
+          started Netty HTTP `Client` instead of creating its own.
         - Added the `NettyConnectionUpgrade` engine hook: a Restlet returning a "101 Switching
           Protocols" response with this response attribute takes over the HTTP/1.1 connection
           (`NettyUpgradeContext`), which the WebSocket extension relies on.
+          `NettyClientHelper.getEventLoopGroup()` lets extensions share the event loop of the Netty
+          HTTP client connector.
         - Added `Response.getProtocol()` and `setProtocol()`: the protocol, with its version,
           used by the server to reply (for example HTTP/1.1, HTTP/2.0 or HTTP/3.0), set by the HTTP
           client connector. When unknown, the protocol of the request is returned.
@@ -160,6 +163,9 @@ Changes log
         - `CookieAuthenticator` no longer lets the next restlet overwrite the redirection sent after
           a login. Issue #554.
         - CorsService and CorsFilter now add `Vary: Origin` to every response when the CORS answer depends on the request origin (origin allow-list or credentials allowed), including requests without an `Origin` header, so shared caches no longer serve the response of one origin to another. A static `*` without credentials still sends no `Vary`, as recommended by the Fetch standard. Issue #1611.
+        - `Client.setContext()` now stores the connector helper in the "org.restlet.engine.helper"
+          context attribute, as the constructors do, so clients added with
+          `Component.getClients().add(Protocol)` expose it too.
     - Misc
         - Fixed JDK HTTP client accessibility in core and WebSocket tests in Eclipse and VS Code,
           without adding a production module dependency.

org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketClient.java

@@ -8,6 +8,7 @@
  */
 package org.restlet.ext.websocket;
 
+import io.netty.channel.EventLoopGroup;
 import java.io.IOException;
 import java.net.URI;
 import java.time.Duration;
@@ -17,15 +18,21 @@
 import java.util.Objects;
 import java.util.logging.Logger;
 import javax.net.ssl.SSLContext;
+import org.restlet.Client;
+import org.restlet.engine.netty.NettyClientHelper;
 import org.restlet.ext.websocket.internal.NettyWebSocketConnector;
 
 /**
  * Opens WebSocket client sessions ("ws" and "wss" URIs) over HTTP/1.1 upgrades. Each session uses
  * its own connection.
  *
  * <p>Configure the client before starting it. {@link #connect(String)} starts a stopped client,
- * like {@link org.restlet.Restlet#handle(org.restlet.Request)}. Stop or close the client to release
- * its I/O threads; it can be started again afterwards.
+ * like {@link org.restlet.Restlet#handle(org.restlet.Request)}. Stop or close the client to close
+ * its sessions; it can be started again afterwards.
+ *
+ * <p>By default, the client creates its own I/O threads, released when it stops. To share the I/O
+ * threads of the application instead, set a borrowed HTTP {@link Client} with {@link
+ * #setClient(Client)}.
  */
 public final class WebSocketClient implements AutoCloseable {
 
@@ -43,6 +50,8 @@ public final class WebSocketClient implements AutoCloseable {
 
     private boolean hostnameVerification = true;
 
+    private Client client;
+
     private NettyWebSocketConnector connector;
 
     /** Constructor using the default settings. */
@@ -191,14 +200,40 @@ public synchronized void setHostnameVerification(boolean hostnameVerification) {
     }
 
     /**
-     * Starts the client, creating its I/O threads. Does nothing if already started.
+     * Returns the borrowed HTTP client whose I/O threads are shared.
+     *
+     * @return The borrowed client, or null if the WebSocket client creates its own I/O threads.
+     */
+    public synchronized Client getClient() {
+        return client;
+    }
+
+    /**
+     * Sets a borrowed HTTP client whose I/O threads are shared, instead of creating new ones. The
+     * client must use the Netty HTTP connector of the core, have a context, and be started before
+     * this WebSocket client. Its owner stops it, after this WebSocket client: stopping it closes
+     * the sessions. Only the I/O threads are shared; the other settings, including TLS, are those
+     * of this WebSocket client.
+     *
+     * @param client The borrowed client, or null to create new I/O threads.
+     */
+    public synchronized void setClient(Client client) {
+        requireStopped();
+        this.client = client;
+    }
+
+    /**
+     * Starts the client, creating its I/O threads unless a borrowed client is set. Does nothing if
+     * already started.
      *
      * @throws IOException If the client cannot be started.
+     * @throws IllegalStateException If the borrowed client is not a started Netty HTTP client.
      */
     public synchronized void start() throws IOException {
         if (connector == null) {
             connector =
                     new NettyWebSocketConnector(
+                            borrowedEventLoopGroup(),
                             sslContext,
                             connectTimeout.toMillis(),
                             idleTimeout.toMillis(),
@@ -219,7 +254,10 @@ public synchronized boolean isStarted() {
         return connector != null;
     }
 
-    /** Stops the client, releasing its I/O threads. Does nothing if already stopped. */
+    /**
+     * Stops the client, closing its sessions and releasing its own I/O threads, if any. Does
+     * nothing if already stopped.
+     */
     public synchronized void stop() {
         if (connector != null) {
             connector.close();
@@ -266,6 +304,31 @@ public void close() {
         stop();
     }
 
+    /**
+     * Returns the event loop group of the borrowed client.
+     *
+     * @return The event loop group, or null if no client is borrowed.
+     */
+    private EventLoopGroup borrowedEventLoopGroup() {
+        if (client == null) {
+            return null;
+        }
+        Object helper =
+                (client.getContext() == null)
+                        ? null
+                        : client.getContext().getAttributes().get("org.restlet.engine.helper");
+        if (!(helper instanceof NettyClientHelper nettyHelper)) {
+            throw new IllegalStateException(
+                    "The borrowed client must have a context and use the Netty HTTP connector");
+        }
+        EventLoopGroup group = nettyHelper.getEventLoopGroup();
+        if (group == null || group.isShuttingDown()) {
+            throw new IllegalStateException(
+                    "Start the borrowed client before the WebSocket client");
+        }
+        return group;
+    }
+
     private void requireStopped() {
         if (connector != null) {
             throw new IllegalStateException(

org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnector.java

@@ -18,6 +18,8 @@
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.MultiThreadIoEventLoopGroup;
 import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.channel.nio.NioIoHandler;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
@@ -42,6 +44,7 @@
 import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GlobalEventExecutor;
 import io.netty.util.concurrent.Promise;
 import java.io.IOException;
 import java.io.InterruptedIOException;
@@ -66,6 +69,16 @@ public final class NettyWebSocketConnector implements AutoCloseable {
 
     private final EventLoopGroup group;
 
+    /** Whether the event loop group was created by this connector, which then shuts it down. */
+    private final boolean ownedGroup;
+
+    /**
+     * The open connections, closed with the connector. The group stays closed, closing the late
+     * connections of concurrent connect() calls, which a borrowed event loop would keep open.
+     */
+    private final ChannelGroup channels =
+            new DefaultChannelGroup("restlet-websocket-client", GlobalEventExecutor.INSTANCE, true);
+
     private final SslContext sslContext;
 
     private final long connectTimeoutMs;
@@ -85,6 +98,8 @@ public final class NettyWebSocketConnector implements AutoCloseable {
     /**
      * Constructor.
      *
+     * @param sharedGroup The event loop group to share, which this connector never shuts down, or
+     *     null to create one.
      * @param sslContext The JDK SSL context used for "wss" URIs, or null to use the default one.
      * @param connectTimeoutMs The connect and handshake timeout in milliseconds, 0 for none.
      * @param idleTimeoutMs The idle time after which the connection is pinged, then closed, 0 for
@@ -97,6 +112,7 @@ public final class NettyWebSocketConnector implements AutoCloseable {
      * @throws IOException If the SSL context cannot be created.
      */
     public NettyWebSocketConnector(
+            EventLoopGroup sharedGroup,
             SSLContext sslContext,
             long connectTimeoutMs,
             long idleTimeoutMs,
@@ -128,7 +144,11 @@ public NettyWebSocketConnector(
         this.compression = compression;
         this.hostnameVerification = hostnameVerification;
         this.logger = logger;
-        this.group = new MultiThreadIoEventLoopGroup(0, NioIoHandler.newFactory());
+        this.ownedGroup = (sharedGroup == null);
+        this.group =
+                ownedGroup
+                        ? new MultiThreadIoEventLoopGroup(0, NioIoHandler.newFactory())
+                        : sharedGroup;
     }
 
     /**
@@ -158,6 +178,11 @@ public WebSocketSession connect(URI uri, List<String> subprotocols, Map<String,
         int defaultPort = secure ? 443 : 80;
         int port = (uri.getPort() > 0) ? uri.getPort() : defaultPort;
 
+        if (group.isShuttingDown()) {
+            throw new IOException(
+                    "The shared event loop is shut down: the borrowed client was stopped");
+        }
+
         HttpHeaders extraHeaders = new DefaultHttpHeaders();
         headers.forEach(extraHeaders::add);
         WebSocketClientHandshaker handshaker =
@@ -179,6 +204,7 @@ public WebSocketSession connect(URI uri, List<String> subprotocols, Map<String,
                                 new ChannelInitializer<SocketChannel>() {
                                     @Override
                                     protected void initChannel(SocketChannel ch) {
+                                        channels.add(ch);
                                         initPipeline(ch, secure, host, port, handshaker, ready);
                                     }
                                 });
@@ -273,7 +299,11 @@ private void initPipeline(
 
     @Override
     public void close() {
-        group.shutdownGracefully(0, 2, TimeUnit.SECONDS).awaitUninterruptibly(5, TimeUnit.SECONDS);
+        channels.close().awaitUninterruptibly(5, TimeUnit.SECONDS);
+        if (ownedGroup) {
+            group.shutdownGracefully(0, 2, TimeUnit.SECONDS)
+                    .awaitUninterruptibly(5, TimeUnit.SECONDS);
+        }
     }
 
     /** Sends the handshake request and installs the session once the response is accepted. */