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. + * + *
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(
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnector.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnector.java
index 90a12de5f4..6654906cd6 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnector.java
+++ b/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
+ * The content of this file is subject to the terms of the Apache 2.0 open
+ * source license available at https://www.opensource.org/licenses/apache-2.0
+ *
+ * Restlet is a registered trademark of QlikTech International AB.
+ */
+package org.restlet.ext.websocket;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Logger;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.restlet.Client;
+import org.restlet.Context;
+import org.restlet.Request;
+import org.restlet.Response;
+import org.restlet.Server;
+import org.restlet.data.Protocol;
+import org.restlet.engine.Engine;
+import org.restlet.engine.connector.HttpClientHelper;
+import org.restlet.engine.connector.HttpServerHelper;
+import org.restlet.engine.netty.NettyClientHelper;
+import org.restlet.ext.websocket.internal.NettyWebSocketConnector;
+
+/** Tests WebSocket clients sharing the I/O threads of a borrowed HTTP client. */
+class WebSocketSharedEventLoopTestCase {
+
+ private Server server;
+
+ private Client http;
+
+ private String uri;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ Engine.clearThreadLocalVariables();
+ Engine.register(false);
+ Engine.getInstance().getRegisteredServers().add(new HttpServerHelper(null));
+ Engine.getInstance().getRegisteredClients().add(new HttpClientHelper(null));
+ server =
+ new Server(
+ new Context(),
+ Protocol.HTTP,
+ 0,
+ new WebSocketUpgrader() {
+ @Override
+ protected WebSocketHandler accept(Request request, Response response) {
+ return session -> {
+ WebSocketMessage message;
+ while ((message = session.receive()) != null) {
+ session.send(message);
+ }
+ };
+ }
+ });
+ server.start();
+ uri = "ws://localhost:" + server.getActualPort() + "/";
+ http = new Client(new Context(), Protocol.HTTP);
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ http.stop();
+ server.stop();
+ Engine.clearThreadLocalVariables();
+ Engine.register();
+ }
+
+ private NettyClientHelper helper(Client client) {
+ return (NettyClientHelper)
+ client.getContext().getAttributes().get("org.restlet.engine.helper");
+ }
+
+ @Test
+ void stoppingKeepsTheBorrowedEventLoop() throws Exception {
+ http.start();
+ WebSocketClient client = new WebSocketClient();
+ client.setClient(http);
+ assertSame(http, client.getClient());
+
+ WebSocketSession session = client.connect(uri);
+ session.sendText("shared");
+ assertEquals("shared", session.receive().getText());
+
+ client.stop();
+ assertNull(session.receive());
+ assertFalse(session.isOpen());
+ assertFalse(helper(http).getEventLoopGroup().isShuttingDown());
+
+ // The borrowed client still serves new WebSocket clients.
+ try (WebSocketClient other = new WebSocketClient()) {
+ other.setClient(http);
+ WebSocketSession again = other.connect(uri);
+ again.sendText("again");
+ assertEquals("again", again.receive().getText());
+ }
+ }
+
+ @Test
+ void stoppingTheBorrowedClientClosesTheSessions() throws Exception {
+ http.start();
+ try (WebSocketClient client = new WebSocketClient()) {
+ client.setClient(http);
+ WebSocketSession session = client.connect(uri);
+
+ http.stop();
+ assertNull(session.receive());
+ assertThrows(IOException.class, () -> client.connect(uri));
+ }
+ }
+
+ @Test
+ void closedConnectorClosesLateConnections() throws Exception {
+ http.start();
+ NettyWebSocketConnector connector =
+ new NettyWebSocketConnector(
+ helper(http).getEventLoopGroup(),
+ null,
+ 5000,
+ 0,
+ 65536,
+ 65536,
+ false,
+ true,
+ Logger.getLogger(WebSocketSharedEventLoopTestCase.class.getName()));
+ connector.close();
+
+ // As a connect() call racing with stop() would do after the connections were closed.
+ assertThrows(
+ IOException.class, () -> connector.connect(URI.create(uri), List.of(), Map.of()));
+ assertFalse(helper(http).getEventLoopGroup().isShuttingDown());
+ }
+
+ @Test
+ void acceptsAClientWhoseContextIsSetLater() throws Exception {
+ // As done by Component.getClients().add(Protocol.HTTP).
+ Client late = new Client(Protocol.HTTP);
+ late.setContext(new Context());
+ late.start();
+ try (WebSocketClient client = new WebSocketClient()) {
+ client.setClient(late);
+ WebSocketSession session = client.connect(uri);
+ session.sendText("late");
+ assertEquals("late", session.receive().getText());
+ } finally {
+ late.stop();
+ }
+ }
+
+ @Test
+ void rejectsUnsuitableClients() throws Exception {
+ WebSocketClient client = new WebSocketClient();
+
+ client.setClient(http);
+ assertThrows(IllegalStateException.class, client::start);
+
+ client.setClient(new Client(Protocol.HTTP));
+ assertThrows(IllegalStateException.class, client::start);
+
+ client.setClient(new Client(new Context(), Protocol.FTP));
+ assertThrows(IllegalStateException.class, client::start);
+
+ http.start();
+ client.setClient(http);
+ client.start();
+ assertThrows(IllegalStateException.class, () -> client.setClient(null));
+ client.stop();
+ }
+}
diff --git a/org.restlet/src/main/java/org/restlet/Client.java b/org.restlet/src/main/java/org/restlet/Client.java
index 79d008356b..5b84376c5a 100644
--- a/org.restlet/src/main/java/org/restlet/Client.java
+++ b/org.restlet/src/main/java/org/restlet/Client.java
@@ -142,6 +142,20 @@ public boolean isAvailable() {
return getHelper() != null;
}
+ /**
+ * Sets the context, and stores the helper in its "org.restlet.engine.helper" attribute.
+ *
+ * @param context The context.
+ */
+ @Override
+ public void setContext(Context context) {
+ super.setContext(context);
+
+ if (context != null && getHelper() != null) {
+ context.getAttributes().put("org.restlet.engine.helper", getHelper());
+ }
+ }
+
@Override
public synchronized void start() throws Exception {
if (isStopped()) {
diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java
index 632e3bd099..2b45c6e3dc 100644
--- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java
+++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java
@@ -913,6 +913,20 @@ public int getEventLoopWorkerThreads() {
getHelpedParameters().getFirstValue("eventLoop.workerThreads", "0"));
}
+ /**
+ * Returns the event loop group doing network I/O, so that other Netty channels can share it.
+ * The group belongs to this connector: it is created on start and shut down on stop, closing
+ * the channels registered with it. Callers must not shut it down.
+ *
+ * This is an engine extension point, meant for Restlet extensions.
+ *
+ * @return The event loop group, or null if the connector is not started.
+ */
+ @SuppressWarnings("exports")
+ public EventLoopGroup getEventLoopGroup() {
+ return group;
+ }
+
/**
* Returns the HTTP client transport mode among the following options: "HTTP1_1", "HTTP2",
* "HTTP3" (experimental) and "DYNAMIC". Default to "HTTP1_1".