diff --git a/changes.md b/changes.md index fea3b6525b..695e7c88a3 100644 --- a/changes.md +++ b/changes.md @@ -70,7 +70,9 @@ Changes log `WebSocketUpgrader` (or `WebSocketUpgrade.accept`) accepts upgrades that went through the normal routing, filters and authenticators, and runs a `WebSocketHandler` on its own virtual thread with a blocking `WebSocketSession` (`receive`, `sendText`, `sendBinary`, - `close`). `WebSocketClient` opens client sessions on "ws" and "wss" URIs. The server + `close`). `WebSocketClient` opens client sessions on "ws" and "wss" URIs; it is configured + with setters, which are rejected while it is started, and can be stopped, reconfigured + and started again (`connect` starts a stopped client). The server 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), 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..a8f21476d8 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 @@ -21,198 +21,263 @@ /** * 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 release + * its I/O threads; it can be started again afterwards. */ 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 NettyWebSocketConnector connector; + + /** Constructor using the default settings. */ + public WebSocketClient() {} /** - * Creates a client with the default settings. + * Returns the SSL context used for "wss" URIs. * - * @return The client. - * @throws IOException If the client cannot be created. + * @return The SSL context, or null for the JVM default context. */ - public static WebSocketClient create() throws IOException { - return builder().build(); + public synchronized SSLContext getSslContext() { + return sslContext; } /** - * Returns a builder. + * Sets the SSL context used for "wss" URIs. Defaults to the JVM default context. * - * @return The builder. + * @param sslContext The SSL context. */ - public static Builder builder() { - return new Builder(); + public synchronized void setSslContext(SSLContext sslContext) { + requireStopped(); + this.sslContext = sslContext; } /** - * Opens a session, blocking until the handshake completes. + * Returns the connect and handshake timeout. * - * @param uri The "ws" or "wss" URI. - * @return The open session. - * @throws IOException If the connection or the handshake fails. + * @return The timeout; zero for none. */ - public WebSocketSession connect(String uri) throws IOException { - return connect(URI.create(uri), List.of(), Map.of()); + public synchronized Duration getConnectTimeout() { + return connectTimeout; } /** - * Opens a session, blocking until the handshake completes. + * Sets the connect and handshake timeout. Defaults to 15 seconds; zero for none. * - * @param uri The "ws" or "wss" URI. - * @param subprotocols The subprotocols to offer, in preference order. - * @param headers Additional handshake request headers, for example "Authorization". - * @return The open session. - * @throws IOException If the connection or the handshake fails. + * @param connectTimeout The timeout. */ - public WebSocketSession connect(URI uri, List subprotocols, Map headers) - throws IOException { - return connector.connect( - Objects.requireNonNull(uri, "uri"), - List.copyOf(subprotocols), - new LinkedHashMap<>(headers)); + public synchronized void setConnectTimeout(Duration connectTimeout) { + requireStopped(); + this.connectTimeout = requireNonNegative(connectTimeout, "connectTimeout"); } - @Override - public void close() { - connector.close(); + /** + * Returns the idle time after which the server is pinged. + * + * @return The idle timeout; zero when disabled. + */ + public synchronized Duration getIdleTimeout() { + return idleTimeout; } - /** Builds a {@link WebSocketClient}. */ - public static final class Builder { - - private SSLContext sslContext; + /** + * 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"); + } - private Duration connectTimeout = Duration.ofSeconds(15); + /** + * Returns the maximum size in bytes of a received message. + * + * @return The maximum message size. + */ + public synchronized int getMaxMessageSize() { + return maxMessageSize; + } - private Duration idleTimeout = Duration.ofSeconds(30); + /** + * 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; + } - private int maxMessageSize = 65536; + /** + * 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; + } - private int maxFrameSize; + /** + * 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; + } - private boolean compression = true; + /** + * Indicates whether "permessage-deflate" compression (RFC 7692) is offered to the server. + * + * @return True if compression is offered. + */ + public synchronized boolean isCompression() { + return compression; + } - private boolean hostnameVerification = true; + /** + * 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; + } - private Builder() {} + /** + * 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 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 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; + } - /** - * 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; + /** + * Starts the client, creating its I/O threads. Does nothing if already started. + * + * @throws IOException If the client cannot be started. + */ + public synchronized void start() throws IOException { + if (connector == null) { + connector = + new NettyWebSocketConnector( + sslContext, + connectTimeout.toMillis(), + idleTimeout.toMillis(), + maxMessageSize, + getMaxFrameSize(), + compression, + hostnameVerification, + Logger.getLogger(WebSocketClient.class.getName())); } + } - /** - * 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; - } + /** + * Indicates whether the client is started. + * + * @return True if the client is started. + */ + public synchronized boolean isStarted() { + return connector != null; + } - /** - * 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; + /** Stops the client, releasing its I/O threads. Does nothing if already stopped. */ + public synchronized void stop() { + if (connector != null) { + connector.close(); + connector = null; } + } - /** - * 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; - } + /** + * Opens a session, blocking until the handshake completes. Starts the client if needed. + * + * @param uri The "ws" or "wss" URI. + * @return The open session. + * @throws IOException If the connection or the handshake fails. + */ + public WebSocketSession connect(String uri) throws IOException { + return connect(URI.create(uri), List.of(), Map.of()); + } - /** - * 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; + /** + * 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. + * @param headers Additional handshake request headers, for example "Authorization". + * @return The open session. + * @throws IOException If the connection or the handshake fails. + */ + public WebSocketSession connect(URI uri, List subprotocols, Map headers) + throws IOException { + NettyWebSocketConnector current; + synchronized (this) { + start(); + current = connector; } + return current.connect( + Objects.requireNonNull(uri, "uri"), + List.copyOf(subprotocols), + new LinkedHashMap<>(headers)); + } - /** - * 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; - } + /** Stops the client. */ + @Override + public void close() { + stop(); + } - /** - * 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/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..0c6dac5a8d 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,67 @@ 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 + void stoppedClientReconnectsWithUpdatedSettings() throws Exception { + int port = start(Protocol.HTTP, new Context(), ECHO); + String uri = "ws://localhost:" + port + "/"; + String large = "x".repeat(64); + + try (WebSocketClient client = new WebSocketClient()) { + client.setMaxMessageSize(1024); + WebSocketSession first = client.connect(uri); + first.sendText(large); + assertEquals(large, first.receive().getText()); + + client.stop(); + assertNull(first.receive()); + assertFalse(first.isOpen()); + + client.setMaxMessageSize(16); + try (WebSocketSession second = client.connect(uri)) { + assertTrue(client.isStarted()); + second.sendText("small"); + assertEquals("small", second.receive().getText()); + + // The echo exceeds the new client limit. + second.sendText(large); + assertNull(second.receive()); + assertEquals(WebSocketSession.MESSAGE_TOO_BIG, second.getCloseCode()); + } + } } @Test @@ -202,7 +257,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/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