diff --git a/changes.md b/changes.md index fea3b6525b..40da6670f9 100644 --- a/changes.md +++ b/changes.md @@ -74,10 +74,14 @@ 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, with or without a context + (issue #1599). `Client.getHelper()` is now public, as an engine extension point. - 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. @@ -97,6 +101,9 @@ Changes log - Fixed the OpenAPI extension build path for Commons Lang types inherited from Swagger. - Slf4j bridge reports wrong line number and caller class in log4j. Issue #233. - When GET is an allowed method of a server resource, HEAD is now also listed (e.g. in the Allow header of a 405 response). Issue #266. + - `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. diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketClient.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketClient.java index ab3ccc1e63..7d15465b47 100644 --- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketClient.java +++ b/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,50 +18,255 @@ 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. Close the client to release its I/O threads. + * its own connection. + * + *

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 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 { - private final NettyWebSocketConnector connector; + private SSLContext sslContext; - private WebSocketClient(Builder builder) throws IOException { - this.connector = - new NettyWebSocketConnector( - builder.sslContext, - builder.connectTimeout.toMillis(), - builder.idleTimeout.toMillis(), - builder.maxMessageSize, - (builder.maxFrameSize > 0) ? builder.maxFrameSize : builder.maxMessageSize, - builder.compression, - builder.hostnameVerification, - Logger.getLogger(WebSocketClient.class.getName())); + private Duration connectTimeout = Duration.ofSeconds(15); + + private Duration idleTimeout = Duration.ofSeconds(30); + + private int maxMessageSize = 65536; + + private int maxFrameSize; + + private boolean compression = true; + + private boolean hostnameVerification = true; + + private Client client; + + private NettyWebSocketConnector connector; + + /** Constructor using the default settings. */ + public WebSocketClient() {} + + /** + * Returns the SSL context used for "wss" URIs. + * + * @return The SSL context, or null for the JVM default context. + */ + public synchronized SSLContext getSslContext() { + return sslContext; + } + + /** + * Sets the SSL context used for "wss" URIs. Defaults to the JVM default context. + * + * @param sslContext The SSL context. + */ + public synchronized void setSslContext(SSLContext sslContext) { + requireStopped(); + this.sslContext = sslContext; + } + + /** + * Returns the connect and handshake timeout. + * + * @return The timeout; zero for none. + */ + public synchronized Duration getConnectTimeout() { + return connectTimeout; + } + + /** + * Sets the connect and handshake timeout. Defaults to 15 seconds; zero for none. + * + * @param connectTimeout The timeout. + */ + public synchronized void setConnectTimeout(Duration connectTimeout) { + requireStopped(); + this.connectTimeout = requireNonNegative(connectTimeout, "connectTimeout"); + } + + /** + * Returns the idle time after which the server is pinged. + * + * @return The idle timeout; zero when disabled. + */ + public synchronized Duration getIdleTimeout() { + return idleTimeout; + } + + /** + * Sets the idle time after which the server is pinged, the connection being closed if it stays + * idle as long again. Defaults to 30 seconds; zero to disable. + * + * @param idleTimeout The idle timeout. + */ + public synchronized void setIdleTimeout(Duration idleTimeout) { + requireStopped(); + this.idleTimeout = requireNonNegative(idleTimeout, "idleTimeout"); + } + + /** + * Returns the maximum size in bytes of a received message. + * + * @return The maximum message size. + */ + public synchronized int getMaxMessageSize() { + return maxMessageSize; + } + + /** + * Sets the maximum size in bytes of a received message. Defaults to 65536. + * + * @param maxMessageSize The maximum message size. + */ + public synchronized void setMaxMessageSize(int maxMessageSize) { + requireStopped(); + if (maxMessageSize <= 0) { + throw new IllegalArgumentException("maxMessageSize must be positive"); + } + this.maxMessageSize = maxMessageSize; + } + + /** + * Returns the maximum size in bytes of a received frame. + * + * @return The maximum frame size, the maximum message size when not set. + */ + public synchronized int getMaxFrameSize() { + return (maxFrameSize > 0) ? maxFrameSize : maxMessageSize; + } + + /** + * Sets the maximum size in bytes of a received frame. A message may be fragmented in several + * frames. Defaults to the maximum message size, and is capped by it. + * + * @param maxFrameSize The maximum frame size. + */ + public synchronized void setMaxFrameSize(int maxFrameSize) { + requireStopped(); + if (maxFrameSize <= 0) { + throw new IllegalArgumentException("maxFrameSize must be positive"); + } + this.maxFrameSize = maxFrameSize; + } + + /** + * Indicates whether "permessage-deflate" compression (RFC 7692) is offered to the server. + * + * @return True if compression is offered. + */ + public synchronized boolean isCompression() { + return compression; + } + + /** + * Sets whether "permessage-deflate" compression (RFC 7692) is offered to the server. Defaults + * to true; it is used only if the server accepts it. + * + * @param compression True to offer compression. + */ + public synchronized void setCompression(boolean compression) { + requireStopped(); + this.compression = compression; + } + + /** + * Indicates whether the server host name is verified for "wss" URIs. + * + * @return True if the host name is verified. + */ + public synchronized boolean isHostnameVerification() { + return hostnameVerification; + } + + /** + * Sets whether the server host name is verified for "wss" URIs. Defaults to true. + * + * @param hostnameVerification True to verify the host name. + */ + public synchronized void setHostnameVerification(boolean hostnameVerification) { + requireStopped(); + this.hostnameVerification = hostnameVerification; } /** - * Creates a client with the default settings. + * Returns the borrowed HTTP client whose I/O threads are shared. * - * @return The client. - * @throws IOException If the client cannot be created. + * @return The borrowed client, or null if the WebSocket client creates its own I/O threads. */ - public static WebSocketClient create() throws IOException { - return builder().build(); + 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 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(), + maxMessageSize, + getMaxFrameSize(), + compression, + hostnameVerification, + Logger.getLogger(WebSocketClient.class.getName())); + } } /** - * Returns a builder. + * Indicates whether the client is started. * - * @return The builder. + * @return True if the client is started. */ - public static Builder builder() { - return new Builder(); + public synchronized boolean isStarted() { + return connector != null; } /** - * Opens a session, blocking until the handshake completes. + * 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(); + connector = null; + } + } + + /** + * Opens a session, blocking until the handshake completes. Starts the client if needed. * * @param uri The "ws" or "wss" URI. * @return The open session. @@ -71,7 +277,7 @@ public WebSocketSession connect(String uri) throws IOException { } /** - * Opens a session, blocking until the handshake completes. + * Opens a session, blocking until the handshake completes. Starts the client if needed. * * @param uri The "ws" or "wss" URI. * @param subprotocols The subprotocols to offer, in preference order. @@ -81,138 +287,56 @@ public WebSocketSession connect(String uri) throws IOException { */ public WebSocketSession connect(URI uri, List subprotocols, Map headers) throws IOException { - return connector.connect( + NettyWebSocketConnector current; + synchronized (this) { + start(); + current = connector; + } + return current.connect( Objects.requireNonNull(uri, "uri"), List.copyOf(subprotocols), new LinkedHashMap<>(headers)); } + /** Stops the client. */ @Override public void close() { - connector.close(); + stop(); } - /** Builds a {@link WebSocketClient}. */ - public static final class Builder { - - private SSLContext sslContext; - - private Duration connectTimeout = Duration.ofSeconds(15); - - private Duration idleTimeout = Duration.ofSeconds(30); - - private int maxMessageSize = 65536; - - private int maxFrameSize; - - private boolean compression = true; - - private boolean hostnameVerification = true; - - private Builder() {} - - /** - * Sets the SSL context used for "wss" URIs. Defaults to the JVM default context. - * - * @param sslContext The SSL context. - * @return This builder. - */ - public Builder sslContext(SSLContext sslContext) { - this.sslContext = sslContext; - return this; - } - - /** - * Sets the connect and handshake timeout. Defaults to 15 seconds; zero for none. - * - * @param connectTimeout The timeout. - * @return This builder. - */ - public Builder connectTimeout(Duration connectTimeout) { - this.connectTimeout = requireNonNegative(connectTimeout, "connectTimeout"); - return this; - } - - /** - * Sets the idle time after which the server is pinged, the connection being closed if it - * stays idle as long again. Defaults to 30 seconds; zero to disable. - * - * @param idleTimeout The idle timeout. - * @return This builder. - */ - public Builder idleTimeout(Duration idleTimeout) { - this.idleTimeout = requireNonNegative(idleTimeout, "idleTimeout"); - return this; - } - - /** - * Sets the maximum size in bytes of a received message. Defaults to 65536. - * - * @param maxMessageSize The maximum message size. - * @return This builder. - */ - public Builder maxMessageSize(int maxMessageSize) { - if (maxMessageSize <= 0) { - throw new IllegalArgumentException("maxMessageSize must be positive"); - } - this.maxMessageSize = maxMessageSize; - return this; - } - - /** - * Sets the maximum size in bytes of a received frame. A message may be fragmented in - * several frames. Defaults to the maximum message size, and is capped by it. - * - * @param maxFrameSize The maximum frame size. - * @return This builder. - */ - public Builder maxFrameSize(int maxFrameSize) { - if (maxFrameSize <= 0) { - throw new IllegalArgumentException("maxFrameSize must be positive"); - } - this.maxFrameSize = maxFrameSize; - return this; + /** + * 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; } - - /** - * Sets whether "permessage-deflate" compression (RFC 7692) is offered to the server. - * Defaults to true; it is used only if the server accepts it. - * - * @param compression True to offer compression. - * @return This builder. - */ - public Builder compression(boolean compression) { - this.compression = compression; - return this; + if (!(client.getHelper() instanceof NettyClientHelper nettyHelper)) { + throw new IllegalStateException( + "The borrowed client must use the Netty HTTP connector"); } - - /** - * Sets whether the server host name is verified for "wss" URIs. Defaults to true. - * - * @param hostnameVerification True to verify the host name. - * @return This builder. - */ - public Builder hostnameVerification(boolean hostnameVerification) { - this.hostnameVerification = hostnameVerification; - return this; + EventLoopGroup group = nettyHelper.getEventLoopGroup(); + if (group == null || group.isShuttingDown()) { + throw new IllegalStateException( + "Start the borrowed client before the WebSocket client"); } + return group; + } - /** - * Builds the client. - * - * @return The client. - * @throws IOException If the client cannot be created. - */ - public WebSocketClient build() throws IOException { - return new WebSocketClient(this); + private void requireStopped() { + if (connector != null) { + throw new IllegalStateException( + "Stop the WebSocket client before changing its settings"); } + } - private static Duration requireNonNegative(Duration value, String name) { - Objects.requireNonNull(value, name); - if (value.isNegative()) { - throw new IllegalArgumentException(name + " cannot be negative"); - } - return value; + private static Duration requireNonNegative(Duration value, String name) { + Objects.requireNonNull(value, name); + if (value.isNegative()) { + throw new IllegalArgumentException(name + " cannot be negative"); } + return value; } } 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..04d591ce66 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,13 @@ 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. */ + private final ChannelGroup channels = + new DefaultChannelGroup("restlet-websocket-client", GlobalEventExecutor.INSTANCE); + private final SslContext sslContext; private final long connectTimeoutMs; @@ -85,6 +95,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 +109,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 +141,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 +175,11 @@ public WebSocketSession connect(URI uri, List subprotocols, Map 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 +201,7 @@ public WebSocketSession connect(URI uri, List subprotocols, Map() { @Override protected void initChannel(SocketChannel ch) { + channels.add(ch); initPipeline(ch, secure, host, port, handshaker, ready); } }); @@ -273,7 +296,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. */ diff --git a/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketCompressionTestCase.java b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketCompressionTestCase.java index 0cb1b5304f..3e0e47882d 100644 --- a/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketCompressionTestCase.java +++ b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketCompressionTestCase.java @@ -145,7 +145,9 @@ void compressedMessagesRoundTrip() throws Exception { int port = start(new Context()); String large = "compressible ".repeat(10_000); - try (WebSocketClient client = WebSocketClient.builder().maxMessageSize(1 << 20).build(); + WebSocketClient client = new WebSocketClient(); + client.setMaxMessageSize(1 << 20); + try (client; WebSocketSession session = client.connect("ws://localhost:" + port + "/")) { // 130 KB of text: larger than the server default limit (64 KB) once inflated, so the // server would close the session if the limit were applied to compressed frames @@ -166,7 +168,7 @@ void inflatedMessagesAreCappedByTheMessageSize() throws Exception { context.getParameters().add("websocket.maxMessageSize", "1024"); int port = start(context); - try (WebSocketClient client = WebSocketClient.create()) { + try (WebSocketClient client = new WebSocketClient()) { WebSocketSession session = client.connect("ws://localhost:" + port + "/"); // Compresses to far less than 1 KB on the wire, but inflates to 100 KB. session.sendText("a".repeat(100_000)); @@ -182,7 +184,9 @@ void framesLargerThanTheFrameSizeCloseTheSession() throws Exception { context.getParameters().add("websocket.maxFrameSize", "16"); int port = start(context); - try (WebSocketClient client = WebSocketClient.builder().compression(false).build()) { + WebSocketClient client = new WebSocketClient(); + client.setCompression(false); + try (client) { WebSocketSession session = client.connect("ws://localhost:" + port + "/"); session.sendText("small"); assertEquals("small", session.receive().getText()); @@ -197,8 +201,10 @@ void framesLargerThanTheFrameSizeCloseTheSession() throws Exception { void clientFrameSizeLimitAppliesToServerMessages() throws Exception { int port = start(new Context()); - try (WebSocketClient client = - WebSocketClient.builder().compression(false).maxFrameSize(16).build()) { + WebSocketClient client = new WebSocketClient(); + client.setCompression(false); + client.setMaxFrameSize(16); + try (client) { WebSocketSession session = client.connect("ws://localhost:" + port + "/"); session.sendText("y".repeat(64)); assertNull(session.receive()); diff --git a/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketSessionTestCase.java b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketSessionTestCase.java index 482afa241e..247854bd1e 100644 --- a/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketSessionTestCase.java +++ b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketSessionTestCase.java @@ -9,6 +9,7 @@ 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.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -113,12 +114,11 @@ void echoesOverTls() throws Exception { context.getParameters().add("keyStoreType", "PKCS12"); int port = start(Protocol.HTTPS, context, ECHO); - try (WebSocketClient client = - WebSocketClient.builder() - .sslContext(trustingContext()) - .hostnameVerification(true) - .maxMessageSize(1024) - .build(); + WebSocketClient client = new WebSocketClient(); + client.setSslContext(trustingContext()); + client.setHostnameVerification(true); + client.setMaxMessageSize(1024); + try (client; WebSocketSession session = client.connect("wss://localhost:" + port + "/")) { session.sendText("secure"); assertEquals("secure", session.receive().getText()); @@ -131,8 +131,9 @@ void idleSessionsArePingedAndKeptAliveByPongs() throws Exception { context.getParameters().add("connector.idleTimeout", "100"); int port = start(Protocol.HTTP, context, ECHO); - try (WebSocketClient client = - WebSocketClient.builder().idleTimeout(Duration.ofMillis(100)).build(); + WebSocketClient client = new WebSocketClient(); + client.setIdleTimeout(Duration.ofMillis(100)); + try (client; WebSocketSession session = client.connect("ws://localhost:" + port + "/")) { // Several idle periods: both sides ping, and the pongs keep the session open. Thread.sleep(600); @@ -152,7 +153,7 @@ void handlerFailureClosesWithInternalError() throws Exception { throw new IllegalStateException("boom"); }); - try (WebSocketClient client = WebSocketClient.create()) { + try (WebSocketClient client = new WebSocketClient()) { WebSocketSession session = client.connect("ws://localhost:" + port + "/"); assertNull(session.receive()); assertEquals(WebSocketSession.INTERNAL_ERROR, session.getCloseCode()); @@ -166,8 +167,9 @@ void connectionFailuresAreReported() throws Exception { try (ServerSocket socket = new ServerSocket(0)) { freePort = socket.getLocalPort(); } - try (WebSocketClient client = - WebSocketClient.builder().connectTimeout(Duration.ofSeconds(2)).build()) { + WebSocketClient client = new WebSocketClient(); + client.setConnectTimeout(Duration.ofSeconds(2)); + try (client) { String refusedUri = "ws://localhost:" + freePort + "/"; String httpUri = "http://localhost:" + freePort + "/"; assertThrows(ConnectException.class, () -> client.connect(refusedUri)); @@ -177,14 +179,37 @@ void connectionFailuresAreReported() throws Exception { } @Test - void builderValidatesSettings() { - WebSocketClient.Builder builder = WebSocketClient.builder(); - assertThrows(IllegalArgumentException.class, () -> builder.maxMessageSize(0)); + void clientValidatesSettings() { + WebSocketClient client = new WebSocketClient(); + assertThrows(IllegalArgumentException.class, () -> client.setMaxMessageSize(0)); + assertThrows(IllegalArgumentException.class, () -> client.setMaxFrameSize(0)); assertThrows( - IllegalArgumentException.class, () -> builder.idleTimeout(Duration.ofSeconds(-1))); + IllegalArgumentException.class, + () -> client.setIdleTimeout(Duration.ofSeconds(-1))); assertThrows( IllegalArgumentException.class, - () -> builder.connectTimeout(Duration.ofSeconds(-1))); + () -> client.setConnectTimeout(Duration.ofSeconds(-1))); + } + + @Test + void settingsAreFixedWhileStarted() throws Exception { + WebSocketClient client = new WebSocketClient(); + assertEquals(65536, client.getMaxFrameSize()); + client.setMaxMessageSize(1024); + assertEquals(1024, client.getMaxFrameSize()); + + client.start(); + try { + assertTrue(client.isStarted()); + assertThrows(IllegalStateException.class, () -> client.setMaxMessageSize(2048)); + assertThrows(IllegalStateException.class, () -> client.setCompression(false)); + } finally { + client.stop(); + } + + assertFalse(client.isStarted()); + client.setMaxMessageSize(2048); + assertEquals(2048, client.getMaxMessageSize()); } @Test @@ -202,7 +227,7 @@ void invalidServerParametersAreRejected() throws Exception { context.getParameters().add("websocket.maxMessageSize", "0"); int port = start(Protocol.HTTP, context, ECHO); - try (WebSocketClient client = WebSocketClient.create()) { + try (WebSocketClient client = new WebSocketClient()) { assertThrows(IOException.class, () -> client.connect("ws://localhost:" + port + "/")); } } diff --git a/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketSharedEventLoopTestCase.java b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketSharedEventLoopTestCase.java new file mode 100644 index 0000000000..082d79ea66 --- /dev/null +++ b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketSharedEventLoopTestCase.java @@ -0,0 +1,168 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * 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 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; + +/** 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 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 acceptsAClientWithoutContext() throws Exception { + Client plain = new Client(Protocol.HTTP); + plain.start(); + try (WebSocketClient client = new WebSocketClient()) { + client.setClient(plain); + WebSocketSession session = client.connect(uri); + session.sendText("plain"); + assertEquals("plain", session.receive().getText()); + } finally { + plain.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.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketTestCase.java b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketTestCase.java index 7a33d55ce2..8e5160502f 100644 --- a/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketTestCase.java +++ b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketTestCase.java @@ -60,7 +60,8 @@ void setUp() throws IOException { Engine.register(false); Engine.getInstance().getRegisteredServers().add(new HttpServerHelper(null)); Engine.getInstance().getRegisteredClients().add(new HttpClientHelper(null)); - client = WebSocketClient.builder().connectTimeout(Duration.ofSeconds(5)).build(); + client = new WebSocketClient(); + client.setConnectTimeout(Duration.ofSeconds(5)); } @AfterEach diff --git a/org.restlet/src/main/java/org/restlet/Client.java b/org.restlet/src/main/java/org/restlet/Client.java index 79d008356b..f24c9c7b21 100644 --- a/org.restlet/src/main/java/org/restlet/Client.java +++ b/org.restlet/src/main/java/org/restlet/Client.java @@ -107,11 +107,14 @@ public Client(String protocolName) { } /** - * Returns the helper provided by the implementation. + * Returns the helper provided by the implementation. This is an engine extension point, meant + * for Restlet extensions that share the connector's resources, such as its I/O threads. It + * doesn't depend on the client having a context. * - * @return The helper provided by the implementation. + * @return The helper provided by the implementation, or null if no connector supports the + * protocols. */ - private RestletHelper getHelper() { + public RestletHelper getHelper() { return this.helper; } @@ -142,6 +145,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 862361fb40..758eb6a10a 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 @@ -911,6 +911,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".