From 3b10bbcb7bf2203bc2360123184abb49e560239a Mon Sep 17 00:00:00 2001
From: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
Date: Sun, 18 Oct 2026 14:33:47 -0400
Subject: [PATCH 1/7] feat: add WebSocket API
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
---
changes.md | 7 +
org.restlet/src/main/java/module-info.java | 1 +
.../restlet/engine/netty/NettyServerCall.java | 25 ++
.../engine/netty/NettyServerHelper.java | 62 ++++
.../engine/netty/NettyWebSocketConnector.java | 276 +++++++++++++++
.../engine/netty/NettyWebSocketSession.java | 324 ++++++++++++++++++
.../engine/netty/NettyWebSocketUpgrader.java | 149 ++++++++
.../restlet/websocket/WebSocketClient.java | 185 ++++++++++
.../restlet/websocket/WebSocketHandler.java | 26 ++
.../restlet/websocket/WebSocketMessage.java | 99 ++++++
.../restlet/websocket/WebSocketResource.java | 80 +++++
.../restlet/websocket/WebSocketSession.java | 128 +++++++
.../restlet/websocket/WebSocketUpgrade.java | 145 ++++++++
.../org/restlet/websocket/package-info.java | 14 +
.../engine/connector/WebSocketTestCase.java | 316 +++++++++++++++++
.../websocket/WebSocketUpgradeTestCase.java | 116 +++++++
16 files changed, 1953 insertions(+)
create mode 100644 org.restlet/src/main/java/org/restlet/engine/netty/NettyWebSocketConnector.java
create mode 100644 org.restlet/src/main/java/org/restlet/engine/netty/NettyWebSocketSession.java
create mode 100644 org.restlet/src/main/java/org/restlet/engine/netty/NettyWebSocketUpgrader.java
create mode 100644 org.restlet/src/main/java/org/restlet/websocket/WebSocketClient.java
create mode 100644 org.restlet/src/main/java/org/restlet/websocket/WebSocketHandler.java
create mode 100644 org.restlet/src/main/java/org/restlet/websocket/WebSocketMessage.java
create mode 100644 org.restlet/src/main/java/org/restlet/websocket/WebSocketResource.java
create mode 100644 org.restlet/src/main/java/org/restlet/websocket/WebSocketSession.java
create mode 100644 org.restlet/src/main/java/org/restlet/websocket/WebSocketUpgrade.java
create mode 100644 org.restlet/src/main/java/org/restlet/websocket/package-info.java
create mode 100644 org.restlet/src/test/java/org/restlet/engine/connector/WebSocketTestCase.java
create mode 100644 org.restlet/src/test/java/org/restlet/websocket/WebSocketUpgradeTestCase.java
diff --git a/changes.md b/changes.md
index b17e9584ee..502cf304ee 100644
--- a/changes.md
+++ b/changes.md
@@ -66,6 +66,13 @@ Changes log
streaming and reading Server-Sent Events over the HTTP connectors, including optional
comment heartbeats and reconnects honoring `retry:` and `Last-Event-ID`. The
`server.maxEventStreams` parameter limits concurrent event streams.
+ - Added experimental WebSocket support in the new `org.restlet.websocket` package, on
+ HTTP/1.1 upgrades. `WebSocketResource` (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 answers pings, pings idle sessions, and
+ supports the "server.maxWebSockets" and "websocket.maxMessageSize" parameters.
- 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.
diff --git a/org.restlet/src/main/java/module-info.java b/org.restlet/src/main/java/module-info.java
index 520bad7227..500f3a7060 100644
--- a/org.restlet/src/main/java/module-info.java
+++ b/org.restlet/src/main/java/module-info.java
@@ -52,4 +52,5 @@
exports org.restlet.security;
exports org.restlet.service;
exports org.restlet.util;
+ exports org.restlet.websocket;
}
diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java
index f4b0623a5e..d9023add39 100644
--- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java
+++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java
@@ -64,6 +64,7 @@
import org.restlet.representation.Representation;
import org.restlet.service.ConnectorService;
import org.restlet.util.Series;
+import org.restlet.websocket.WebSocketUpgrade;
/**
* Server exchange bridging a Netty HTTP request (HTTP/1.1, or one HTTP/2 or HTTP/3 stream converted
@@ -356,6 +357,30 @@ private InputStream peek(InputStream stream) {
* @param response The Restlet response.
*/
void commit(Response response) {
+ WebSocketUpgrade upgrade = WebSocketUpgrade.get(response);
+ if (upgrade != null && Status.INFO_SWITCHING_PROTOCOL.equals(response.getStatus())) {
+ if (kind != StreamKind.HTTP1) {
+ response.setStatus(
+ Status.SERVER_ERROR_NOT_IMPLEMENTED,
+ "WebSocket is only supported over HTTP/1.1");
+ } else if (!helper.tryAcquireWebSocket()) {
+ response.setStatus(Status.SERVER_ERROR_SERVICE_UNAVAILABLE);
+ } else {
+ completed = true;
+ responseHeadWritten = true;
+ requestBody.close();
+ try {
+ NettyWebSocketUpgrader.upgrade(helper, channel, request, confidential, upgrade);
+ } finally {
+ if (response.getOnSent() != null) {
+ response.getOnSent().handle(response.getRequest(), response);
+ }
+ }
+ return;
+ }
+ response.setEntity(null);
+ }
+
try {
responseTrailers = response.getTrailers();
if (isEventStream(response.getEntity())) {
diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java
index f9f32d5a66..f9a1edbab7 100644
--- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java
+++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java
@@ -117,6 +117,20 @@
* When 0, there is no limit.
*
*
+ *
server.maxWebSockets
+ *
int
+ *
0
+ *
Maximum number of open WebSocket sessions; extra upgrades receive a 503 response. When 0,
+ * there is no limit.
+ *
+ *
+ *
websocket.maxMessageSize
+ *
int
+ *
65536
+ *
Maximum size in bytes of a received WebSocket message; larger messages close the session
+ * with status 1009.
+ *
+ *
*
eventLoop.bossThreads
*
int
*
1
@@ -211,6 +225,9 @@ public abstract class NettyServerHelper extends ServerHelper {
/** Limits concurrent SSE streams, or null when unlimited. */
private volatile Semaphore eventStreamPermits;
+ /** Limits open WebSocket sessions, or null when unlimited. */
+ private volatile Semaphore webSocketPermits;
+
/** The number of calls in progress. */
private final AtomicInteger activeCalls = new AtomicInteger();
@@ -522,6 +539,20 @@ void releaseEventStream() {
}
}
+ /** Tries to acquire a permit for a WebSocket session. */
+ boolean tryAcquireWebSocket() {
+ Semaphore permits = webSocketPermits;
+ return !stopping && (permits == null || permits.tryAcquire());
+ }
+
+ /** Releases an acquired WebSocket session permit. */
+ void releaseWebSocket() {
+ Semaphore permits = webSocketPermits;
+ if (permits != null) {
+ permits.release();
+ }
+ }
+
/**
* Returns the executor running Restlet calls.
*
@@ -546,6 +577,8 @@ public synchronized void start() throws Exception {
callPermits = (maxCalls > 0) ? new Semaphore(maxCalls) : null;
int maxEventStreams = getServerMaxEventStreams();
eventStreamPermits = (maxEventStreams > 0) ? new Semaphore(maxEventStreams) : null;
+ int maxWebSockets = getServerMaxWebSockets();
+ webSocketPermits = (maxWebSockets > 0) ? new Semaphore(maxWebSockets) : null;
callExecutor =
CallExecutors.create(
"restlet-server",
@@ -860,6 +893,35 @@ public int getServerMaxEventStreams() {
return result;
}
+ /**
+ * Maximum number of open WebSocket sessions. Defaults to 0 (unlimited).
+ *
+ * @return Maximum number of open WebSocket sessions.
+ */
+ public int getServerMaxWebSockets() {
+ int result =
+ Integer.parseInt(getHelpedParameters().getFirstValue("server.maxWebSockets", "0"));
+ if (result < 0) {
+ throw new IllegalArgumentException("'server.maxWebSockets' cannot be negative");
+ }
+ return result;
+ }
+
+ /**
+ * Maximum size in bytes of a received WebSocket message. Defaults to 65536.
+ *
+ * @return Maximum WebSocket message size.
+ */
+ public int getWebSocketMaxMessageSize() {
+ int result =
+ Integer.parseInt(
+ getHelpedParameters().getFirstValue("websocket.maxMessageSize", "65536"));
+ if (result <= 0) {
+ throw new IllegalArgumentException("'websocket.maxMessageSize' must be positive");
+ }
+ return result;
+ }
+
/**
* Number of event loop threads accepting connections. Defaults to 1.
*
diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyWebSocketConnector.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyWebSocketConnector.java
new file mode 100644
index 0000000000..14c7b4e824
--- /dev/null
+++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyWebSocketConnector.java
@@ -0,0 +1,276 @@
+/**
+ * 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
+ *
This class is internal to the Netty connectors; use {@link org.restlet.websocket
+ * .WebSocketClient}.
+ */
+public final class NettyWebSocketConnector implements AutoCloseable {
+
+ private final EventLoopGroup group;
+
+ private final SslContext sslContext;
+
+ private final long connectTimeoutMs;
+
+ private final long idleTimeoutMs;
+
+ private final int maxMessageSize;
+
+ private final boolean hostnameVerification;
+
+ private final Logger logger;
+
+ /**
+ * Constructor.
+ *
+ * @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
+ * none.
+ * @param maxMessageSize The maximum size in bytes of a received message.
+ * @param hostnameVerification Whether the server host name is verified for "wss" URIs.
+ * @param logger The logger.
+ * @throws IOException If the SSL context cannot be created.
+ */
+ public NettyWebSocketConnector(
+ SSLContext sslContext,
+ long connectTimeoutMs,
+ long idleTimeoutMs,
+ int maxMessageSize,
+ boolean hostnameVerification,
+ Logger logger)
+ throws IOException {
+ try {
+ SSLContext jdkContext = (sslContext != null) ? sslContext : SSLContext.getDefault();
+ this.sslContext =
+ NettySsl.newContext(
+ jdkContext, true, ApplicationProtocolConfig.DISABLED, ClientAuth.NONE);
+ } catch (Exception e) {
+ throw new IOException("Unable to create the WebSocket SSL context", e);
+ }
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.idleTimeoutMs = idleTimeoutMs;
+ this.maxMessageSize = maxMessageSize;
+ this.hostnameVerification = hostnameVerification;
+ this.logger = logger;
+ this.group = new MultiThreadIoEventLoopGroup(0, NioIoHandler.newFactory());
+ }
+
+ /**
+ * Opens a session, blocking until the handshake completes.
+ *
+ * @param uri The "ws" or "wss" URI.
+ * @param subprotocols The subprotocols to offer, in preference order, possibly empty.
+ * @param headers Additional handshake request headers.
+ * @return The open session.
+ * @throws IOException If the connection or the handshake fails.
+ */
+ public WebSocketSession connect(URI uri, List subprotocols, Map headers)
+ throws IOException {
+ String scheme = uri.getScheme() == null ? "" : uri.getScheme().toLowerCase();
+ boolean secure;
+ if ("wss".equals(scheme)) {
+ secure = true;
+ } else if ("ws".equals(scheme)) {
+ secure = false;
+ } else {
+ throw new IllegalArgumentException("Unsupported WebSocket URI scheme: " + scheme);
+ }
+ String host = uri.getHost();
+ if (host == null) {
+ throw new IllegalArgumentException("The WebSocket URI has no host: " + uri);
+ }
+ int port = (uri.getPort() > 0) ? uri.getPort() : (secure ? 443 : 80);
+
+ HttpHeaders extraHeaders = new DefaultHttpHeaders();
+ headers.forEach(extraHeaders::add);
+ WebSocketClientHandshaker handshaker =
+ WebSocketClientHandshakerFactory.newHandshaker(
+ uri,
+ WebSocketVersion.V13,
+ subprotocols.isEmpty() ? null : String.join(",", subprotocols),
+ false,
+ extraHeaders,
+ maxMessageSize);
+
+ Promise ready = group.next().newPromise();
+ Bootstrap bootstrap =
+ new Bootstrap()
+ .group(group)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .handler(
+ new ChannelInitializer() {
+ @Override
+ protected void initChannel(SocketChannel ch) {
+ initPipeline(ch, secure, host, port, handshaker, ready);
+ }
+ });
+ if (connectTimeoutMs > 0) {
+ bootstrap.option(
+ ChannelOption.CONNECT_TIMEOUT_MILLIS,
+ (int) Math.min(Integer.MAX_VALUE, connectTimeoutMs));
+ }
+ ChannelFuture connect = bootstrap.connect(host, port);
+ Channel channel =
+ NettyFutures.await(connect, connectTimeoutMs, "Unable to connect to " + uri);
+ try {
+ NettyWebSocketSession session =
+ NettyFutures.await(ready, connectTimeoutMs, "WebSocket handshake with " + uri);
+ session.start();
+ return session;
+ } catch (IOException e) {
+ channel.close();
+ throw e;
+ }
+ }
+
+ private void initPipeline(
+ SocketChannel ch,
+ boolean secure,
+ String host,
+ int port,
+ WebSocketClientHandshaker handshaker,
+ Promise ready) {
+ ChannelPipeline p = ch.pipeline();
+ if (idleTimeoutMs > 0) {
+ p.addLast("idle", new IdleStateHandler(0, 0, idleTimeoutMs, TimeUnit.MILLISECONDS));
+ }
+ p.addLast("writability", Writability.HANDLER);
+ if (secure) {
+ SSLEngine engine = sslContext.newEngine(ch.alloc(), host, port);
+ if (hostnameVerification) {
+ SSLParameters parameters = engine.getSSLParameters();
+ parameters.setEndpointIdentificationAlgorithm("HTTPS");
+ engine.setSSLParameters(parameters);
+ }
+ SslHandler sslHandler = new SslHandler(engine);
+ sslHandler.setHandshakeTimeoutMillis(connectTimeoutMs);
+ p.addLast("ssl", sslHandler);
+ }
+ p.addLast("http-codec", new HttpClientCodec());
+ p.addLast("http-aggregator", new HttpObjectAggregator(8192));
+ p.addLast("ws-handshake", new HandshakeHandler(handshaker, ready));
+ }
+
+ @Override
+ public void close() {
+ group.shutdownGracefully(0, 2, TimeUnit.SECONDS).awaitUninterruptibly(5, TimeUnit.SECONDS);
+ }
+
+ /** Sends the handshake request and installs the session once the response is accepted. */
+ private final class HandshakeHandler extends SimpleChannelInboundHandler {
+
+ private final WebSocketClientHandshaker handshaker;
+
+ private final Promise ready;
+
+ HandshakeHandler(
+ WebSocketClientHandshaker handshaker, Promise ready) {
+ this.handshaker = handshaker;
+ this.ready = ready;
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ handshaker
+ .handshake(ctx.channel())
+ .addListener(
+ f -> {
+ if (!f.isSuccess()) {
+ ready.tryFailure(f.cause());
+ ctx.close();
+ }
+ });
+ super.channelActive(ctx);
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse response) {
+ Channel channel = ctx.channel();
+ try {
+ handshaker.finishHandshake(channel, response);
+ } catch (RuntimeException e) {
+ ready.tryFailure(
+ new IOException("WebSocket handshake rejected: " + response.status(), e));
+ ctx.close();
+ return;
+ }
+ NettyWebSocketSession session =
+ new NettyWebSocketSession(
+ channel,
+ handshaker.actualSubprotocol(),
+ maxMessageSize,
+ connectTimeoutMs,
+ logger);
+ ChannelPipeline p = ctx.pipeline();
+ p.addAfter(ctx.name(), "ws-aggregator", new WebSocketFrameAggregator(maxMessageSize));
+ p.addAfter("ws-aggregator", "ws-session", session.newHandler());
+ p.remove(this);
+ if (!ready.trySuccess(session)) {
+ ctx.close();
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ ready.tryFailure(new IOException("Connection closed during the WebSocket handshake"));
+ super.channelInactive(ctx);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ ready.tryFailure(cause);
+ ctx.close();
+ }
+ }
+}
diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyWebSocketSession.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyWebSocketSession.java
new file mode 100644
index 0000000000..b0848eade9
--- /dev/null
+++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyWebSocketSession.java
@@ -0,0 +1,324 @@
+/**
+ * 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.engine.netty;
+
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.TooLongFrameException;
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.CorruptedWebSocketFrameException;
+import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketCloseStatus;
+import io.netty.handler.codec.http.websocketx.WebSocketFrame;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.util.ReferenceCountUtil;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.restlet.websocket.WebSocketMessage;
+import org.restlet.websocket.WebSocketSession;
+
+/**
+ * WebSocket session over a Netty channel whose pipeline decodes WebSocket frames and aggregates
+ * fragmented messages. Received messages are queued for a blocking consumer; the channel is read on
+ * demand, so a slow consumer slows down the peer. Pings are answered with pongs, and an idle
+ * connection is pinged once before being closed.
+ *
+ *
This class is internal to the Netty connectors.
+ */
+final class NettyWebSocketSession implements WebSocketSession {
+
+ /** Marks the end of the inbound messages. */
+ private static final Object END = new Object();
+
+ /** Number of queued messages above which the channel is no longer read. */
+ private static final int MAX_QUEUED_MESSAGES = 16;
+
+ /** Maximum time to wait for the peer to complete the close handshake. */
+ private static final long CLOSE_TIMEOUT_MS = 5000;
+
+ private final Channel channel;
+
+ private volatile String subprotocol;
+
+ private final int maxMessageSize;
+
+ private final long writeTimeoutMs;
+
+ private final Logger logger;
+
+ private final LinkedBlockingQueue
*
- *
server.maxWebSockets
- *
int
- *
0
- *
Maximum number of open WebSocket sessions; extra upgrades receive a 503 response. When 0,
- * there is no limit.
- *
- *
- *
websocket.maxMessageSize
- *
int
- *
65536
- *
Maximum size in bytes of a received WebSocket message; larger messages close the session
- * with status 1009.
- *
- *
*
eventLoop.bossThreads
*
int
*
1
@@ -225,9 +211,6 @@ public abstract class NettyServerHelper extends ServerHelper {
/** Limits concurrent SSE streams, or null when unlimited. */
private volatile Semaphore eventStreamPermits;
- /** Limits open WebSocket sessions, or null when unlimited. */
- private volatile Semaphore webSocketPermits;
-
/** The number of calls in progress. */
private final AtomicInteger activeCalls = new AtomicInteger();
@@ -539,20 +522,6 @@ void releaseEventStream() {
}
}
- /** Tries to acquire a permit for a WebSocket session. */
- boolean tryAcquireWebSocket() {
- Semaphore permits = webSocketPermits;
- return !stopping && (permits == null || permits.tryAcquire());
- }
-
- /** Releases an acquired WebSocket session permit. */
- void releaseWebSocket() {
- Semaphore permits = webSocketPermits;
- if (permits != null) {
- permits.release();
- }
- }
-
/**
* Returns the executor running Restlet calls.
*
@@ -577,8 +546,6 @@ public synchronized void start() throws Exception {
callPermits = (maxCalls > 0) ? new Semaphore(maxCalls) : null;
int maxEventStreams = getServerMaxEventStreams();
eventStreamPermits = (maxEventStreams > 0) ? new Semaphore(maxEventStreams) : null;
- int maxWebSockets = getServerMaxWebSockets();
- webSocketPermits = (maxWebSockets > 0) ? new Semaphore(maxWebSockets) : null;
callExecutor =
CallExecutors.create(
"restlet-server",
@@ -893,35 +860,6 @@ public int getServerMaxEventStreams() {
return result;
}
- /**
- * Maximum number of open WebSocket sessions. Defaults to 0 (unlimited).
- *
- * @return Maximum number of open WebSocket sessions.
- */
- public int getServerMaxWebSockets() {
- int result =
- Integer.parseInt(getHelpedParameters().getFirstValue("server.maxWebSockets", "0"));
- if (result < 0) {
- throw new IllegalArgumentException("'server.maxWebSockets' cannot be negative");
- }
- return result;
- }
-
- /**
- * Maximum size in bytes of a received WebSocket message. Defaults to 65536.
- *
- * @return Maximum WebSocket message size.
- */
- public int getWebSocketMaxMessageSize() {
- int result =
- Integer.parseInt(
- getHelpedParameters().getFirstValue("websocket.maxMessageSize", "65536"));
- if (result <= 0) {
- throw new IllegalArgumentException("'websocket.maxMessageSize' must be positive");
- }
- return result;
- }
-
/**
* Number of event loop threads accepting connections. Defaults to 1.
*
diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyUpgradeContext.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyUpgradeContext.java
new file mode 100644
index 0000000000..d5d92baeb4
--- /dev/null
+++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyUpgradeContext.java
@@ -0,0 +1,106 @@
+/**
+ * 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.engine.netty;
+
+import io.netty.channel.Channel;
+import io.netty.handler.codec.http.HttpRequest;
+import java.util.Map;
+import java.util.logging.Logger;
+import org.restlet.data.Parameter;
+import org.restlet.util.Series;
+
+/**
+ * Connection handed to a {@link NettyConnectionUpgrade}. It exposes Netty types: extensions using
+ * it must require the Netty modules themselves.
+ */
+@SuppressWarnings("exports")
+public final class NettyUpgradeContext {
+
+ private final Channel channel;
+
+ private final HttpRequest request;
+
+ private final boolean confidential;
+
+ private final NettyServerHelper helper;
+
+ NettyUpgradeContext(
+ Channel channel, HttpRequest request, boolean confidential, NettyServerHelper helper) {
+ this.channel = channel;
+ this.request = request;
+ this.confidential = confidential;
+ this.helper = helper;
+ }
+
+ /**
+ * Returns the HTTP/1.1 connection. Its pipeline still holds the HTTP server codec and the idle
+ * state handler, if any; the Restlet HTTP handlers were removed. Reading is on demand
+ * (auto-read is off).
+ *
+ * @return The connection.
+ */
+ public Channel getChannel() {
+ return channel;
+ }
+
+ /**
+ * Returns the head of the upgrade request.
+ *
+ * @return The upgrade request head.
+ */
+ public HttpRequest getRequest() {
+ return request;
+ }
+
+ /**
+ * Indicates whether the connection is secured with TLS.
+ *
+ * @return True if the connection is secured with TLS.
+ */
+ public boolean isConfidential() {
+ return confidential;
+ }
+
+ /**
+ * Returns the parameters of the server connector.
+ *
+ * @return The parameters of the server connector.
+ */
+ public Series getParameters() {
+ return helper.getHelpedParameters();
+ }
+
+ /**
+ * Returns the attributes of the server connector, shared by all its connections, for example to
+ * keep per-server state.
+ *
+ * @return The attributes of the server connector.
+ */
+ public Map getAttributes() {
+ return helper.getAttributes();
+ }
+
+ /**
+ * Returns the connection idle timeout of the server connector.
+ *
+ * @return The idle timeout in milliseconds, 0 for none.
+ */
+ public long getIdleTimeout() {
+ return helper.getConnectorIdleTimeout();
+ }
+
+ /**
+ * Returns the logger of the server connector.
+ *
+ * @return The logger.
+ */
+ public Logger getLogger() {
+ return helper.getLogger();
+ }
+}
diff --git a/org.restlet/src/main/java/org/restlet/websocket/package-info.java b/org.restlet/src/main/java/org/restlet/websocket/package-info.java
deleted file mode 100644
index 1fe1eecdca..0000000000
--- a/org.restlet/src/main/java/org/restlet/websocket/package-info.java
+++ /dev/null
@@ -1,14 +0,0 @@
-/**
- * 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.
- */
-/**
- * WebSocket support (experimental): {@link org.restlet.websocket.WebSocketResource} accepts
- * upgrades on the HTTP server connector, {@link org.restlet.websocket.WebSocketClient} opens client
- * sessions, and both use the blocking {@link org.restlet.websocket.WebSocketSession}.
- */
-package org.restlet.websocket;
diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/NettyConnectionUpgradeTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/NettyConnectionUpgradeTestCase.java
new file mode 100644
index 0000000000..4e21a021ab
--- /dev/null
+++ b/org.restlet/src/test/java/org/restlet/engine/connector/NettyConnectionUpgradeTestCase.java
@@ -0,0 +1,147 @@
+/**
+ * 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.engine.connector;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.restlet.Context;
+import org.restlet.Request;
+import org.restlet.Response;
+import org.restlet.Restlet;
+import org.restlet.Server;
+import org.restlet.data.Protocol;
+import org.restlet.data.Status;
+import org.restlet.engine.Engine;
+import org.restlet.engine.netty.NettyConnectionUpgrade;
+import org.restlet.engine.netty.NettyUpgradeContext;
+
+/** Tests the connection upgrade hook of the Netty server connector. */
+class NettyConnectionUpgradeTestCase {
+
+ private Server server;
+
+ @BeforeEach
+ void setUp() {
+ Engine.clearThreadLocalVariables();
+ Engine.register(false);
+ Engine.getInstance().getRegisteredServers().add(new HttpServerHelper(null));
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (server != null) {
+ server.stop();
+ }
+ Engine.clearThreadLocalVariables();
+ Engine.register();
+ }
+
+ private int start(NettyConnectionUpgrade upgrade, String protocol) throws Exception {
+ Context context = new Context();
+ if (protocol != null) {
+ context.getParameters().add("http.transport.protocol", protocol);
+ }
+ server =
+ new Server(
+ context,
+ Protocol.HTTP,
+ 0,
+ new Restlet() {
+ @Override
+ public void handle(Request request, Response response) {
+ response.getAttributes()
+ .put(NettyConnectionUpgrade.ATTRIBUTE, upgrade);
+ response.setStatus(Status.INFO_SWITCHING_PROTOCOL);
+ }
+ });
+ server.start();
+ return server.getActualPort();
+ }
+
+ private static String exchange(int port) throws Exception {
+ try (Socket socket = new Socket("localhost", port)) {
+ OutputStream out = socket.getOutputStream();
+ out.write(
+ ("GET /up HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\n"
+ + "Upgrade: test\r\n\r\n")
+ .getBytes(StandardCharsets.US_ASCII));
+ out.flush();
+ BufferedReader in =
+ new BufferedReader(
+ new InputStreamReader(
+ socket.getInputStream(), StandardCharsets.US_ASCII));
+ StringBuilder result = new StringBuilder();
+ String line;
+ while ((line = in.readLine()) != null) {
+ result.append(line).append('\n');
+ }
+ return result.toString();
+ }
+ }
+
+ @Test
+ void handsTheConnectionOverToTheUpgrade() throws Exception {
+ AtomicReference seen = new AtomicReference<>();
+ int port =
+ start(
+ context -> {
+ seen.set(context);
+ FullHttpResponse response =
+ new DefaultFullHttpResponse(
+ HttpVersion.HTTP_1_1,
+ HttpResponseStatus.SWITCHING_PROTOCOLS,
+ Unpooled.EMPTY_BUFFER);
+ response.headers().set("Upgrade", "test");
+ context.getChannel()
+ .writeAndFlush(response)
+ .addListener(ChannelFutureListener.CLOSE);
+ },
+ null);
+
+ String reply = exchange(port);
+
+ assertTrue(reply.startsWith("HTTP/1.1 101 Switching Protocols"), reply);
+ NettyUpgradeContext context = seen.get();
+ assertEquals("/up", context.getRequest().uri());
+ assertFalse(context.isConfidential());
+ assertEquals(30000, context.getIdleTimeout());
+ assertTrue(context.getParameters().isEmpty());
+ context.getAttributes().put("seen", true);
+ assertTrue(context.getLogger() != null);
+ }
+
+ @Test
+ void failingUpgradeClosesTheConnection() throws Exception {
+ int port =
+ start(
+ context -> {
+ throw new IllegalStateException("refused");
+ },
+ null);
+
+ assertEquals("", exchange(port));
+ }
+}
diff --git a/pom.xml b/pom.xml
index 69f042a2db..8170543aba 100644
--- a/pom.xml
+++ b/pom.xml
@@ -85,6 +85,7 @@
org.restlet.ext.springorg.restlet.ext.thymeleaforg.restlet.ext.velocity
+ org.restlet.ext.websocketorg.restlet.ext.xml
From f4a9b3bb51b677f28b79d4c0b1607b1a3cf61081 Mon Sep 17 00:00:00 2001
From: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
Date: Sun, 18 Oct 2026 18:38:45 -0400
Subject: [PATCH 3/7] refactor: rename WebSocketResource to WebSocketUpgrader
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
---
changes.md | 2 +-
...a => NettyWebSocketConnectionUpgrade.java} | 4 ++--
.../ext/websocket/WebSocketUpgrade.java | 4 +++-
...etResource.java => WebSocketUpgrader.java} | 6 ++---
.../restlet/ext/websocket/package-info.java | 2 +-
.../websocket/WebSocketSessionTestCase.java | 6 ++---
.../ext/websocket/WebSocketTestCase.java | 22 +++++++++----------
.../websocket/WebSocketUpgradeTestCase.java | 4 ++--
8 files changed, 26 insertions(+), 24 deletions(-)
rename org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/{NettyWebSocketUpgrader.java => NettyWebSocketConnectionUpgrade.java} (98%)
rename org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/{WebSocketResource.java => WebSocketUpgrader.java} (94%)
diff --git a/changes.md b/changes.md
index 9b97fb34d0..85fc3356b3 100644
--- a/changes.md
+++ b/changes.md
@@ -67,7 +67,7 @@ Changes log
comment heartbeats and reconnects honoring `retry:` and `Last-Event-ID`. The
`server.maxEventStreams` parameter limits concurrent event streams.
- Added the experimental `org.restlet.ext.websocket` extension, on HTTP/1.1 upgrades.
- `WebSocketResource` (or `WebSocketUpgrade.accept`) accepts upgrades that went through
+ `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
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketUpgrader.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketConnectionUpgrade.java
similarity index 98%
rename from org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketUpgrader.java
rename to org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketConnectionUpgrade.java
index ac17a8d1e6..9a8b083739 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketUpgrader.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketConnectionUpgrade.java
@@ -38,7 +38,7 @@
* Switches an HTTP/1.1 server connection to WebSocket after a Restlet accepted the upgrade, then
* runs the session handler on its own virtual thread.
*/
-final class NettyWebSocketUpgrader implements NettyConnectionUpgrade {
+final class NettyWebSocketConnectionUpgrade implements NettyConnectionUpgrade {
/** Name of the server connector attribute holding the session permits. */
private static final String PERMITS_ATTRIBUTE = "org.restlet.ext.websocket.permits";
@@ -53,7 +53,7 @@ final class NettyWebSocketUpgrader implements NettyConnectionUpgrade {
*
* @param upgrade The accepted upgrade.
*/
- NettyWebSocketUpgrader(WebSocketUpgrade upgrade) {
+ NettyWebSocketConnectionUpgrade(WebSocketUpgrade upgrade) {
this.upgrade = upgrade;
}
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrade.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrade.java
index 2ce6afeaf6..27790a9668 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrade.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrade.java
@@ -100,7 +100,9 @@ public static void accept(Response response, WebSocketHandler handler, String su
WebSocketUpgrade upgrade = new WebSocketUpgrade(handler, subprotocol);
response.getAttributes().put(ATTRIBUTE, upgrade);
response.getAttributes()
- .put(NettyConnectionUpgrade.ATTRIBUTE, new NettyWebSocketUpgrader(upgrade));
+ .put(
+ NettyConnectionUpgrade.ATTRIBUTE,
+ new NettyWebSocketConnectionUpgrade(upgrade));
response.setStatus(Status.INFO_SWITCHING_PROTOCOL);
response.setEntity(null);
}
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketResource.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrader.java
similarity index 94%
rename from org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketResource.java
rename to org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrader.java
index ba30ba9058..28b9b644e5 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketResource.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrader.java
@@ -21,10 +21,10 @@
* authenticators in front of it apply to the upgrade request. Requests that do not ask for an
* upgrade receive a 400 (Bad Request) response.
*/
-public abstract class WebSocketResource extends Restlet {
+public abstract class WebSocketUpgrader extends Restlet {
/** Constructor. */
- protected WebSocketResource() {
+ protected WebSocketUpgrader() {
super();
}
@@ -33,7 +33,7 @@ protected WebSocketResource() {
*
* @param context The context.
*/
- protected WebSocketResource(Context context) {
+ protected WebSocketUpgrader(Context context) {
super(context);
}
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/package-info.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/package-info.java
index befb67dc05..f51cece052 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/package-info.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/package-info.java
@@ -7,7 +7,7 @@
*
Restlet is a registered trademark of QlikTech International AB.
*/
/**
- * WebSocket support (experimental): {@link org.restlet.ext.websocket.WebSocketResource} accepts
+ * WebSocket support (experimental): {@link org.restlet.ext.websocket.WebSocketUpgrader} accepts
* upgrades on the Netty HTTP server connector of the core, {@link
* org.restlet.ext.websocket.WebSocketClient} opens client sessions, and both use the blocking
* {@link org.restlet.ext.websocket.WebSocketSession}.
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 248656b509..aefed28f32 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
@@ -67,8 +67,8 @@ void tearDown() throws Exception {
Engine.register();
}
- private static WebSocketResource resource(WebSocketHandler handler) {
- return new WebSocketResource() {
+ private static WebSocketUpgrader upgrader(WebSocketHandler handler) {
+ return new WebSocketUpgrader() {
@Override
protected WebSocketHandler accept(Request request, Response response) {
return handler;
@@ -86,7 +86,7 @@ protected WebSocketHandler accept(Request request, Response response) {
private int start(Protocol protocol, Context context, WebSocketHandler handler)
throws Exception {
- server = new Server(context, protocol, 0, resource(handler));
+ server = new Server(context, protocol, 0, upgrader(handler));
server.start();
return server.getActualPort();
}
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 1f32d60c56..f60324cd1b 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
@@ -74,8 +74,8 @@ void tearDown() throws Exception {
Engine.register();
}
- private static WebSocketResource resource(WebSocketHandler handler) {
- return new WebSocketResource() {
+ private static WebSocketUpgrader upgrader(WebSocketHandler handler) {
+ return new WebSocketUpgrader() {
@Override
protected WebSocketHandler accept(Request request, Response response) {
return handler;
@@ -104,7 +104,7 @@ private String start(Restlet root, Context context) throws Exception {
@Test
void echoesTextAndBinaryMessagesWithSubprotocol() throws Exception {
- String base = start(resource(ECHO), new Context());
+ String base = start(upgrader(ECHO), new Context());
try (WebSocketSession session =
client.connect(URI.create(base + "/echo"), List.of("other", "chat"), Map.of())) {
@@ -126,7 +126,7 @@ void echoesTextAndBinaryMessagesWithSubprotocol() throws Exception {
void serverHandlerReturningClosesTheSessionNormally() throws Exception {
String base =
start(
- resource(
+ upgrader(
session -> {
session.sendText("bye");
}),
@@ -146,7 +146,7 @@ void clientCloseIsReportedToTheServer() throws Exception {
CountDownLatch closed = new CountDownLatch(1);
String base =
start(
- resource(
+ upgrader(
session -> {
assertNull(session.receive());
serverCode.set(session.getCloseCode());
@@ -169,7 +169,7 @@ void clientCloseIsReportedToTheServer() throws Exception {
void largerMessagesThanTheLimitCloseTheSession() throws Exception {
Context context = new Context();
context.getParameters().add("websocket.maxMessageSize", "16");
- String base = start(resource(ECHO), context);
+ String base = start(upgrader(ECHO), context);
WebSocketSession session = client.connect(base + "/");
session.sendText("x".repeat(64));
@@ -184,7 +184,7 @@ void routingAndAuthenticationApplyToTheUpgrade() throws Exception {
@Override
public Restlet createInboundRoot() {
Router router = new Router(getContext());
- router.attach("/echo", resource(ECHO));
+ router.attach("/echo", upgrader(ECHO));
Authenticator guard =
new Authenticator(getContext()) {
@Override
@@ -239,8 +239,8 @@ protected boolean authenticate(
}
@Test
- void plainRequestsToAWebSocketResourceAreRejected() throws Exception {
- String base = start(resource(ECHO), new Context());
+ void plainRequestsToAWebSocketUpgraderAreRejected() throws Exception {
+ String base = start(upgrader(ECHO), new Context());
Client http = new Client(Protocol.HTTP);
http.start();
try {
@@ -256,7 +256,7 @@ void plainRequestsToAWebSocketResourceAreRejected() throws Exception {
void enforcesTheOpenSessionLimit() throws Exception {
Context context = new Context();
context.getParameters().add("websocket.maxSessions", "1");
- String base = start(resource(ECHO), context);
+ String base = start(upgrader(ECHO), context);
WebSocketSession first = client.connect(base + "/");
IOException rejected = assertThrows(IOException.class, () -> client.connect(base + "/"));
@@ -278,7 +278,7 @@ void enforcesTheOpenSessionLimit() throws Exception {
@Test
void interoperatesWithTheJdkWebSocketClient() throws Exception {
- String base = start(resource(ECHO), new Context());
+ String base = start(upgrader(ECHO), new Context());
CompletableFuture reply = new CompletableFuture<>();
HttpClient jdk = HttpClient.newHttpClient();
WebSocket ws =
diff --git a/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketUpgradeTestCase.java b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketUpgradeTestCase.java
index 4f08b5b4c1..2fb2acb8d4 100644
--- a/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketUpgradeTestCase.java
+++ b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketUpgradeTestCase.java
@@ -86,8 +86,8 @@ void acceptRejectsInvalidUpgrades() {
@Test
void resourceRejectsWhenNoHandler() {
- WebSocketResource refusing =
- new WebSocketResource() {
+ WebSocketUpgrader refusing =
+ new WebSocketUpgrader() {
@Override
protected WebSocketHandler accept(Request request, Response response) {
return null;
From b1d2d862f77eb8b5f292b45511176b5911edb638 Mon Sep 17 00:00:00 2001
From: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
Date: Sun, 18 Oct 2026 18:46:59 -0400
Subject: [PATCH 4/7] refactor: move Netty WebSocket classes to an internal
package
org.restlet.ext.websocket.internal is not exported by the module.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
---
.../org/restlet/ext/websocket/WebSocketClient.java | 1 +
.../org/restlet/ext/websocket/WebSocketUpgrade.java | 1 +
.../NettyWebSocketConnectionUpgrade.java | 9 ++++++---
.../{ => internal}/NettyWebSocketConnector.java | 12 +++++++-----
.../{ => internal}/NettyWebSocketSession.java | 4 +++-
.../ext/websocket/internal/package-info.java | 13 +++++++++++++
6 files changed, 31 insertions(+), 9 deletions(-)
rename org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/{ => internal}/NettyWebSocketConnectionUpgrade.java (95%)
rename org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/{ => internal}/NettyWebSocketConnector.java (96%)
rename org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/{ => internal}/NettyWebSocketSession.java (98%)
create mode 100644 org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/package-info.java
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 ec6c4b70f7..62f42e75d8 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
@@ -17,6 +17,7 @@
import java.util.Objects;
import java.util.logging.Logger;
import javax.net.ssl.SSLContext;
+import org.restlet.ext.websocket.internal.NettyWebSocketConnector;
/**
* Opens WebSocket client sessions ("ws" and "wss" URIs) over HTTP/1.1 upgrades. Each session uses
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrade.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrade.java
index 27790a9668..2a7d60c20b 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrade.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketUpgrade.java
@@ -18,6 +18,7 @@
import org.restlet.data.Method;
import org.restlet.data.Status;
import org.restlet.engine.netty.NettyConnectionUpgrade;
+import org.restlet.ext.websocket.internal.NettyWebSocketConnectionUpgrade;
/**
* Accepts a WebSocket upgrade on a Restlet response. The upgrade request goes through the normal
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketConnectionUpgrade.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnectionUpgrade.java
similarity index 95%
rename from org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketConnectionUpgrade.java
rename to org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnectionUpgrade.java
index 9a8b083739..fec6c3968a 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketConnectionUpgrade.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnectionUpgrade.java
@@ -6,7 +6,7 @@
*
* Restlet is a registered trademark of QlikTech International AB.
*/
-package org.restlet.ext.websocket;
+package org.restlet.ext.websocket.internal;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
@@ -32,13 +32,16 @@
import org.restlet.engine.Engine;
import org.restlet.engine.netty.NettyConnectionUpgrade;
import org.restlet.engine.netty.NettyUpgradeContext;
+import org.restlet.ext.websocket.WebSocketHandler;
+import org.restlet.ext.websocket.WebSocketSession;
+import org.restlet.ext.websocket.WebSocketUpgrade;
import org.restlet.util.Series;
/**
* Switches an HTTP/1.1 server connection to WebSocket after a Restlet accepted the upgrade, then
* runs the session handler on its own virtual thread.
*/
-final class NettyWebSocketConnectionUpgrade implements NettyConnectionUpgrade {
+public final class NettyWebSocketConnectionUpgrade implements NettyConnectionUpgrade {
/** Name of the server connector attribute holding the session permits. */
private static final String PERMITS_ATTRIBUTE = "org.restlet.ext.websocket.permits";
@@ -53,7 +56,7 @@ final class NettyWebSocketConnectionUpgrade implements NettyConnectionUpgrade {
*
* @param upgrade The accepted upgrade.
*/
- NettyWebSocketConnectionUpgrade(WebSocketUpgrade upgrade) {
+ public NettyWebSocketConnectionUpgrade(WebSocketUpgrade upgrade) {
this.upgrade = upgrade;
}
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketConnector.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnector.java
similarity index 96%
rename from org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketConnector.java
rename to org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnector.java
index a6ded39fcc..6b3343d26d 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketConnector.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnector.java
@@ -6,7 +6,7 @@
*
* Restlet is a registered trademark of QlikTech International AB.
*/
-package org.restlet.ext.websocket;
+package org.restlet.ext.websocket.internal;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
@@ -50,13 +50,15 @@
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLParameters;
+import org.restlet.ext.websocket.WebSocketSession;
/**
* Netty implementation of the WebSocket client: one connection per session, HTTP/1.1 upgrade.
*
- *
This class is internal to the extension; use {@link WebSocketClient}.
+ *
This class is internal to the extension; use {@link
+ * org.restlet.ext.websocket.WebSocketClient}.
*/
-final class NettyWebSocketConnector implements AutoCloseable {
+public final class NettyWebSocketConnector implements AutoCloseable {
private final EventLoopGroup group;
@@ -84,7 +86,7 @@ final class NettyWebSocketConnector implements AutoCloseable {
* @param logger The logger.
* @throws IOException If the SSL context cannot be created.
*/
- NettyWebSocketConnector(
+ public NettyWebSocketConnector(
SSLContext sslContext,
long connectTimeoutMs,
long idleTimeoutMs,
@@ -124,7 +126,7 @@ final class NettyWebSocketConnector implements AutoCloseable {
* @return The open session.
* @throws IOException If the connection or the handshake fails.
*/
- WebSocketSession connect(URI uri, List subprotocols, Map headers)
+ public WebSocketSession connect(URI uri, List subprotocols, Map headers)
throws IOException {
String scheme = uri.getScheme() == null ? "" : uri.getScheme().toLowerCase();
boolean secure;
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketSession.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java
similarity index 98%
rename from org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketSession.java
rename to org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java
index cd853e9c37..6cff20d400 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/NettyWebSocketSession.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java
@@ -6,7 +6,7 @@
*
* Restlet is a registered trademark of QlikTech International AB.
*/
-package org.restlet.ext.websocket;
+package org.restlet.ext.websocket.internal;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
@@ -36,6 +36,8 @@
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;
+import org.restlet.ext.websocket.WebSocketMessage;
+import org.restlet.ext.websocket.WebSocketSession;
/**
* WebSocket session over a Netty channel whose pipeline decodes WebSocket frames and aggregates
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/package-info.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/package-info.java
new file mode 100644
index 0000000000..541fea0652
--- /dev/null
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/package-info.java
@@ -0,0 +1,13 @@
+/**
+ * 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.
+ */
+/**
+ * Netty implementation of the WebSocket extension. This package is not exported by the module and
+ * is not part of the API.
+ */
+package org.restlet.ext.websocket.internal;
From 032295d0684cd0c095e5c98532bb812ea6f2da0b Mon Sep 17 00:00:00 2001
From: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
Date: Sun, 18 Oct 2026 19:45:50 -0400
Subject: [PATCH 5/7] feat: add WebSocket compression and frame size limit
Negotiate permessage-deflate (RFC 7692) on the server and the client, cap inflated messages by the message size, and add a separate frame size limit.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
---
changes.md | 6 +-
.../src/main/java/module-info.java | 1 +
.../ext/websocket/WebSocketClient.java | 33 +++
.../ext/websocket/WebSocketSession.java | 6 +-
.../NettyWebSocketConnectionUpgrade.java | 84 ++++++-
.../internal/NettyWebSocketConnector.java | 32 ++-
.../internal/NettyWebSocketSession.java | 9 +
.../restlet/ext/websocket/package-info.java | 16 +-
.../WebSocketCompressionTestCase.java | 208 ++++++++++++++++++
9 files changed, 382 insertions(+), 13 deletions(-)
create mode 100644 org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketCompressionTestCase.java
diff --git a/changes.md b/changes.md
index 85fc3356b3..6b3e16d514 100644
--- a/changes.md
+++ b/changes.md
@@ -71,8 +71,10 @@ Changes log
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
- answers pings, pings idle sessions, and supports the "websocket.maxSessions" and
- "websocket.maxMessageSize" parameters.
+ 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.
- 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.
diff --git a/org.restlet.ext.websocket/src/main/java/module-info.java b/org.restlet.ext.websocket/src/main/java/module-info.java
index 0037928816..00aa4c4517 100644
--- a/org.restlet.ext.websocket/src/main/java/module-info.java
+++ b/org.restlet.ext.websocket/src/main/java/module-info.java
@@ -13,6 +13,7 @@
// Netty is an implementation detail of the extension: required, not re-exported.
requires io.netty.buffer;
requires io.netty.codec;
+ requires io.netty.codec.compression;
requires io.netty.codec.http;
requires io.netty.common;
requires io.netty.handler;
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 62f42e75d8..ab3ccc1e63 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
@@ -34,6 +34,8 @@ private WebSocketClient(Builder builder) throws IOException {
builder.connectTimeout.toMillis(),
builder.idleTimeout.toMillis(),
builder.maxMessageSize,
+ (builder.maxFrameSize > 0) ? builder.maxFrameSize : builder.maxMessageSize,
+ builder.compression,
builder.hostnameVerification,
Logger.getLogger(WebSocketClient.class.getName()));
}
@@ -101,6 +103,10 @@ public static final class Builder {
private int maxMessageSize = 65536;
+ private int maxFrameSize;
+
+ private boolean compression = true;
+
private boolean hostnameVerification = true;
private Builder() {}
@@ -153,6 +159,33 @@ public Builder maxMessageSize(int 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;
+ }
+
+ /**
+ * 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;
+ }
+
/**
* Sets whether the server host name is verified for "wss" URIs. Defaults to true.
*
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketSession.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketSession.java
index 1be7e4196c..31d4f70a7f 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketSession.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketSession.java
@@ -113,9 +113,11 @@ default void close() throws IOException {
/**
* Returns the status code of the close frame received from the peer, {@link #NO_STATUS} if it
- * had none, or {@link #ABNORMAL_CLOSURE} if the connection closed without one.
+ * had none, or {@link #ABNORMAL_CLOSURE} if the connection closed without one. When this side
+ * fails the session because of a peer error (for example {@link #MESSAGE_TOO_BIG} or {@link
+ * #PROTOCOL_ERROR}), the status code it sent is returned.
*
- * @return The close status code, or -1 while no close frame was received.
+ * @return The close status code, or -1 while the session was not closed.
*/
int getCloseCode();
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnectionUpgrade.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnectionUpgrade.java
index fec6c3968a..79dc5ca634 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnectionUpgrade.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketConnectionUpgrade.java
@@ -12,12 +12,15 @@
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.compression.ZlibCodecFactory;
import io.netty.handler.codec.http.DefaultFullHttpRequest;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
@@ -25,6 +28,11 @@
import io.netty.handler.codec.http.websocketx.WebSocketFrameAggregator;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionData;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionUtil;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketServerExtension;
+import io.netty.handler.codec.http.websocketx.extensions.compression.PerMessageDeflateServerExtensionHandshaker;
+import java.util.Map;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
@@ -74,6 +82,9 @@ public void upgrade(NettyUpgradeContext context) throws Exception {
}
int maxMessageSize = getMaxMessageSize(parameters);
+ int maxFrameSize = getMaxFrameSize(parameters, maxMessageSize);
+ boolean compression =
+ Boolean.parseBoolean(parameters.getFirstValue("websocket.compression", "true"));
HttpRequest request = context.getRequest();
FullHttpRequest full =
new DefaultFullHttpRequest(
@@ -87,8 +98,8 @@ public void upgrade(NettyUpgradeContext context) throws Exception {
location,
upgrade.getSubprotocol(),
WebSocketDecoderConfig.newBuilder()
- .maxFramePayloadLength(maxMessageSize)
- .allowExtensions(false)
+ .maxFramePayloadLength(maxFrameSize)
+ .allowExtensions(compression)
.build());
WebSocketServerHandshaker handshaker = factory.newHandshaker(full);
if (handshaker == null) {
@@ -107,10 +118,23 @@ public void upgrade(NettyUpgradeContext context) throws Exception {
context.getLogger());
// Installed before the handshake so that no frame sent right after it is missed.
ChannelPipeline pipeline = channel.pipeline();
+ HttpHeaders responseHeaders = new DefaultHttpHeaders();
+ if (compression) {
+ WebSocketServerExtension extension = negotiateCompression(request, maxMessageSize);
+ if (extension != null) {
+ responseHeaders.set(
+ HttpHeaderNames.SEC_WEBSOCKET_EXTENSIONS,
+ toHeaderValue(extension.newReponseData()));
+ // The frame codecs installed by the handshake pass through these positions.
+ pipeline.addLast("ws-inflater", extension.newExtensionDecoder());
+ pipeline.addLast("ws-deflater", extension.newExtensionEncoder());
+ }
+ }
pipeline.addLast("ws-aggregator", new WebSocketFrameAggregator(maxMessageSize));
pipeline.addLast("ws-session", session.newHandler());
try {
- ChannelFuture future = handshaker.handshake(channel, full);
+ ChannelFuture future =
+ handshaker.handshake(channel, full, responseHeaders, channel.newPromise());
if (!future.await(HANDSHAKE_TIMEOUT_MS, TimeUnit.MILLISECONDS) || !future.isSuccess()) {
context.getLogger().log(Level.FINE, "WebSocket handshake failed", future.cause());
channel.close();
@@ -159,6 +183,48 @@ private static Semaphore getPermits(NettyUpgradeContext context, Series new Semaphore(max));
}
+ /**
+ * Accepts the first "permessage-deflate" offer of the request that can be honored.
+ *
+ * @param request The upgrade request.
+ * @param maxMessageSize The maximum size of an inflated message.
+ * @return The negotiated extension, or null.
+ */
+ private static WebSocketServerExtension negotiateCompression(
+ HttpRequest request, int maxMessageSize) {
+ String offers = request.headers().getAsString(HttpHeaderNames.SEC_WEBSOCKET_EXTENSIONS);
+ if (offers == null) {
+ return null;
+ }
+ PerMessageDeflateServerExtensionHandshaker handshaker =
+ new PerMessageDeflateServerExtensionHandshaker(
+ 6,
+ ZlibCodecFactory.isSupportingWindowSizeAndMemLevel(),
+ PerMessageDeflateServerExtensionHandshaker.MAX_WINDOW_SIZE,
+ false,
+ false,
+ maxMessageSize);
+ for (WebSocketExtensionData offer : WebSocketExtensionUtil.extractExtensions(offers)) {
+ WebSocketServerExtension extension = handshaker.handshakeExtension(offer);
+ if (extension != null) {
+ return extension;
+ }
+ }
+ return null;
+ }
+
+ /** Formats extension data as a "Sec-WebSocket-Extensions" header value. */
+ static String toHeaderValue(WebSocketExtensionData data) {
+ StringBuilder result = new StringBuilder(data.name());
+ for (Map.Entry parameter : data.parameters().entrySet()) {
+ result.append("; ").append(parameter.getKey());
+ if (parameter.getValue() != null) {
+ result.append('=').append(parameter.getValue());
+ }
+ }
+ return result.toString();
+ }
+
private static int getMaxMessageSize(Series parameters) {
int result =
Integer.parseInt(parameters.getFirstValue("websocket.maxMessageSize", "65536"));
@@ -168,6 +234,18 @@ private static int getMaxMessageSize(Series parameters) {
return result;
}
+ private static int getMaxFrameSize(Series parameters, int maxMessageSize) {
+ String value = parameters.getFirstValue("websocket.maxFrameSize");
+ if (value == null) {
+ return maxMessageSize;
+ }
+ int result = Integer.parseInt(value);
+ if (result <= 0) {
+ throw new IllegalArgumentException("'websocket.maxFrameSize' must be positive");
+ }
+ return Math.min(result, maxMessageSize);
+ }
+
private static void sendError(Channel channel, HttpResponseStatus status) {
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status);
response.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0);
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 6b3343d26d..d0ff99b43c 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
@@ -21,6 +21,7 @@
import io.netty.channel.nio.NioIoHandler;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.compression.ZlibCodecFactory;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpClientCodec;
@@ -30,6 +31,9 @@
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import io.netty.handler.codec.http.websocketx.WebSocketFrameAggregator;
import io.netty.handler.codec.http.websocketx.WebSocketVersion;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandler;
+import io.netty.handler.codec.http.websocketx.extensions.compression.PerMessageDeflateClientExtensionHandshaker;
+import io.netty.handler.codec.http.websocketx.extensions.compression.PerMessageDeflateServerExtensionHandshaker;
import io.netty.handler.ssl.ApplicationProtocolConfig;
import io.netty.handler.ssl.ClientAuth;
import io.netty.handler.ssl.IdentityCipherSuiteFilter;
@@ -70,6 +74,10 @@ public final class NettyWebSocketConnector implements AutoCloseable {
private final int maxMessageSize;
+ private final int maxFrameSize;
+
+ private final boolean compression;
+
private final boolean hostnameVerification;
private final Logger logger;
@@ -82,6 +90,8 @@ public final class NettyWebSocketConnector implements AutoCloseable {
* @param idleTimeoutMs The idle time after which the connection is pinged, then closed, 0 for
* none.
* @param maxMessageSize The maximum size in bytes of a received message.
+ * @param maxFrameSize The maximum size in bytes of a received frame.
+ * @param compression Whether "permessage-deflate" compression is offered to the server.
* @param hostnameVerification Whether the server host name is verified for "wss" URIs.
* @param logger The logger.
* @throws IOException If the SSL context cannot be created.
@@ -91,6 +101,8 @@ public NettyWebSocketConnector(
long connectTimeoutMs,
long idleTimeoutMs,
int maxMessageSize,
+ int maxFrameSize,
+ boolean compression,
boolean hostnameVerification,
Logger logger)
throws IOException {
@@ -112,6 +124,8 @@ public NettyWebSocketConnector(
this.connectTimeoutMs = connectTimeoutMs;
this.idleTimeoutMs = idleTimeoutMs;
this.maxMessageSize = maxMessageSize;
+ this.maxFrameSize = Math.min(maxFrameSize, maxMessageSize);
+ this.compression = compression;
this.hostnameVerification = hostnameVerification;
this.logger = logger;
this.group = new MultiThreadIoEventLoopGroup(0, NioIoHandler.newFactory());
@@ -150,9 +164,9 @@ public WebSocketSession connect(URI uri, List subprotocols, Map ready = group.next().newPromise();
Bootstrap bootstrap =
@@ -239,6 +253,20 @@ private void initPipeline(
}
p.addLast("http-codec", new HttpClientCodec());
p.addLast("http-aggregator", new HttpObjectAggregator(8192));
+ if (compression) {
+ // Offers permessage-deflate, then installs the compression codecs if accepted.
+ // Inflated messages are capped by the message size limit.
+ p.addLast(
+ "ws-compression",
+ new WebSocketClientExtensionHandler(
+ new PerMessageDeflateClientExtensionHandshaker(
+ 6,
+ ZlibCodecFactory.isSupportingWindowSizeAndMemLevel(),
+ PerMessageDeflateServerExtensionHandshaker.MAX_WINDOW_SIZE,
+ false,
+ false,
+ maxMessageSize)));
+ }
p.addLast("ws-handshake", new HandshakeHandler(handshaker, ready));
}
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java
index 6cff20d400..34724126ac 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java
@@ -17,6 +17,7 @@
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.TooLongFrameException;
+import io.netty.handler.codec.compression.DecompressionException;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.CorruptedWebSocketFrameException;
@@ -283,6 +284,10 @@ private void onClose(ChannelHandlerContext ctx, int status, String reason) {
}
private void failSession(ChannelHandlerContext ctx, int code) {
+ if (!closeReceived) {
+ closeCode = code;
+ closeReceived = true;
+ }
if (closeSent.compareAndSet(false, true) && ctx.channel().isActive()) {
ctx.writeAndFlush(new CloseWebSocketFrame(code, null))
.addListener(ChannelFutureListener.CLOSE);
@@ -372,6 +377,10 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
code = corrupted.closeStatus().code();
} else if (cause instanceof TooLongFrameException) {
code = MESSAGE_TOO_BIG;
+ } else if (cause instanceof DecompressionException) {
+ // Raised by the permessage-deflate decoder when an inflated message exceeds the
+ // maximum message size.
+ code = MESSAGE_TOO_BIG;
} else {
logger.log(Level.FINE, "WebSocket session failure", cause);
}
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/package-info.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/package-info.java
index f51cece052..7fe3aa700a 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/package-info.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/package-info.java
@@ -12,9 +12,17 @@
* org.restlet.ext.websocket.WebSocketClient} opens client sessions, and both use the blocking
* {@link org.restlet.ext.websocket.WebSocketSession}.
*
- *
Server connector parameters read by this extension: "websocket.maxSessions" (maximum number of
- * open sessions, extra upgrades receive a 503 response; 0, the default, for no limit) and
- * "websocket.maxMessageSize" (maximum size in bytes of a received message, 65536 by default; larger
- * messages close the session with status 1009).
+ *
Server connector parameters read by this extension:
+ *
+ *
+ *
"websocket.maxSessions": maximum number of open sessions, extra upgrades receive a 503
+ * response; 0, the default, for no limit.
+ *
"websocket.maxMessageSize": maximum size in bytes of a received message, after
+ * decompression, 65536 by default; larger messages close the session with status 1009.
+ *
"websocket.maxFrameSize": maximum size in bytes of a received frame, the maximum message
+ * size by default (and at most); larger frames close the session with status 1009.
+ *
"websocket.compression": whether "permessage-deflate" compression (RFC 7692) is accepted
+ * when the client offers it, true by default.
+ *
+ * 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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+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.HttpServerHelper;
+
+/** Tests permessage-deflate compression and the frame size limit. */
+class WebSocketCompressionTestCase {
+
+ private Server server;
+
+ @BeforeEach
+ void setUp() {
+ Engine.clearThreadLocalVariables();
+ Engine.register(false);
+ Engine.getInstance().getRegisteredServers().add(new HttpServerHelper(null));
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (server != null) {
+ server.stop();
+ }
+ Engine.clearThreadLocalVariables();
+ Engine.register();
+ }
+
+ private static final WebSocketHandler ECHO =
+ session -> {
+ WebSocketMessage message;
+ while ((message = session.receive()) != null) {
+ session.send(message);
+ }
+ };
+
+ private int start(Context context) throws Exception {
+ server =
+ new Server(
+ context,
+ Protocol.HTTP,
+ 0,
+ new WebSocketUpgrader() {
+ @Override
+ protected WebSocketHandler accept(Request request, Response response) {
+ return ECHO;
+ }
+ });
+ server.start();
+ return server.getActualPort();
+ }
+
+ /** Sends a raw handshake and returns the response head lines. */
+ private static List handshake(int port, String extensions) throws Exception {
+ try (Socket socket = new Socket("localhost", port)) {
+ String key = Base64.getEncoder().encodeToString(new byte[16]);
+ OutputStream out = socket.getOutputStream();
+ out.write(
+ ("GET / HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\n"
+ + "Connection: Upgrade\r\nSec-WebSocket-Version: 13\r\n"
+ + "Sec-WebSocket-Key: "
+ + key
+ + "\r\n"
+ + (extensions == null
+ ? ""
+ : "Sec-WebSocket-Extensions: " + extensions + "\r\n")
+ + "\r\n")
+ .getBytes(StandardCharsets.US_ASCII));
+ out.flush();
+ BufferedReader in =
+ new BufferedReader(
+ new InputStreamReader(
+ socket.getInputStream(), StandardCharsets.US_ASCII));
+ List lines = new ArrayList<>();
+ String line;
+ while ((line = in.readLine()) != null && !line.isEmpty()) {
+ lines.add(line.toLowerCase());
+ }
+ return lines;
+ }
+ }
+
+ @Test
+ void serverAcceptsPermessageDeflate() throws Exception {
+ int port = start(new Context());
+
+ List head = handshake(port, "permessage-deflate; client_max_window_bits");
+
+ assertTrue(head.get(0).startsWith("http/1.1 101"), head.toString());
+ assertTrue(
+ head.stream()
+ .anyMatch(
+ h ->
+ h.startsWith("sec-websocket-extensions:")
+ && h.contains("permessage-deflate")),
+ head.toString());
+ }
+
+ @Test
+ void serverIgnoresCompressionWhenDisabledOrNotOffered() throws Exception {
+ Context context = new Context();
+ context.getParameters().add("websocket.compression", "false");
+ int port = start(context);
+
+ List disabled = handshake(port, "permessage-deflate");
+ assertTrue(disabled.get(0).startsWith("http/1.1 101"), disabled.toString());
+ assertTrue(
+ disabled.stream().noneMatch(h -> h.startsWith("sec-websocket-extensions:")),
+ disabled.toString());
+
+ List notOffered = handshake(port, null);
+ assertTrue(
+ notOffered.stream().noneMatch(h -> h.startsWith("sec-websocket-extensions:")),
+ notOffered.toString());
+ }
+
+ @Test
+ void compressedMessagesRoundTrip() throws Exception {
+ int port = start(new Context());
+ String large = "compressible ".repeat(10_000);
+
+ try (WebSocketClient client = WebSocketClient.builder().maxMessageSize(1 << 20).build();
+ 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
+ // only. Use a smaller payload against the default limit instead.
+ String text = large.substring(0, 60_000);
+ session.sendText(text);
+ assertEquals(text, session.receive().getText());
+
+ byte[] binary = text.getBytes(StandardCharsets.UTF_8);
+ session.sendBinary(binary);
+ assertEquals(text, session.receive().getText());
+ }
+ }
+
+ @Test
+ void inflatedMessagesAreCappedByTheMessageSize() throws Exception {
+ Context context = new Context();
+ context.getParameters().add("websocket.maxMessageSize", "1024");
+ int port = start(context);
+
+ try (WebSocketClient client = WebSocketClient.create()) {
+ 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));
+ assertNull(session.receive());
+ assertEquals(WebSocketSession.MESSAGE_TOO_BIG, session.getCloseCode());
+ }
+ }
+
+ @Test
+ void framesLargerThanTheFrameSizeCloseTheSession() throws Exception {
+ Context context = new Context();
+ context.getParameters().add("websocket.compression", "false");
+ context.getParameters().add("websocket.maxFrameSize", "16");
+ int port = start(context);
+
+ try (WebSocketClient client = WebSocketClient.builder().compression(false).build()) {
+ WebSocketSession session = client.connect("ws://localhost:" + port + "/");
+ session.sendText("small");
+ assertEquals("small", session.receive().getText());
+
+ session.sendText("x".repeat(64));
+ assertNull(session.receive());
+ assertEquals(WebSocketSession.MESSAGE_TOO_BIG, session.getCloseCode());
+ }
+ }
+
+ @Test
+ void clientFrameSizeLimitAppliesToServerMessages() throws Exception {
+ int port = start(new Context());
+
+ try (WebSocketClient client =
+ WebSocketClient.builder().compression(false).maxFrameSize(16).build()) {
+ WebSocketSession session = client.connect("ws://localhost:" + port + "/");
+ session.sendText("y".repeat(64));
+ assertNull(session.receive());
+ assertEquals(WebSocketSession.MESSAGE_TOO_BIG, session.getCloseCode());
+ }
+ }
+}
From 6b6e4d13a48eafc7e0a42239837addecd3a887ad Mon Sep 17 00:00:00 2001
From: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
Date: Sun, 18 Oct 2026 23:33:20 -0400
Subject: [PATCH 6/7] fix: address SonarCloud findings on WebSocket extension
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
.../ext/websocket/WebSocketMessage.java | 42 ++++++----
.../internal/NettyWebSocketConnector.java | 3 +-
.../internal/NettyWebSocketSession.java | 80 ++++++++++---------
.../websocket/WebSocketSessionTestCase.java | 12 ++-
.../ext/websocket/WebSocketTestCase.java | 2 +-
.../websocket/WebSocketUpgradeTestCase.java | 14 ++--
.../NettyConnectionUpgradeTestCase.java | 16 ++--
7 files changed, 90 insertions(+), 79 deletions(-)
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketMessage.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketMessage.java
index 7b1b3d0530..92d687b11e 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketMessage.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/WebSocketMessage.java
@@ -15,13 +15,11 @@
/** An immutable, complete WebSocket data message: either text or binary. */
public final class WebSocketMessage {
- private final String text;
+ /** Either a {@link String} or a {@code byte[]}, never null. */
+ private final Object payload;
- private final byte[] data;
-
- private WebSocketMessage(String text, byte[] data) {
- this.text = text;
- this.data = data;
+ private WebSocketMessage(Object payload) {
+ this.payload = payload;
}
/**
@@ -31,7 +29,7 @@ private WebSocketMessage(String text, byte[] data) {
* @return The message.
*/
public static WebSocketMessage text(String text) {
- return new WebSocketMessage(Objects.requireNonNull(text, "text"), null);
+ return new WebSocketMessage(Objects.requireNonNull(text, "text"));
}
/**
@@ -41,7 +39,7 @@ public static WebSocketMessage text(String text) {
* @return The message.
*/
public static WebSocketMessage binary(byte[] data) {
- return new WebSocketMessage(null, Objects.requireNonNull(data, "data").clone());
+ return new WebSocketMessage(Objects.requireNonNull(data, "data").clone());
}
/**
@@ -50,7 +48,7 @@ public static WebSocketMessage binary(byte[] data) {
* @return True for a text message.
*/
public boolean isText() {
- return text != null;
+ return payload instanceof String;
}
/**
@@ -59,7 +57,7 @@ public boolean isText() {
* @return True for a binary message.
*/
public boolean isBinary() {
- return data != null;
+ return payload instanceof byte[];
}
/**
@@ -68,7 +66,9 @@ public boolean isBinary() {
* @return The text.
*/
public String getText() {
- return isText() ? text : new String(data, StandardCharsets.UTF_8);
+ return (payload instanceof String text)
+ ? text
+ : new String((byte[]) payload, StandardCharsets.UTF_8);
}
/**
@@ -77,23 +77,31 @@ public String getText() {
* @return A copy of the bytes.
*/
public byte[] getData() {
- return isBinary() ? data.clone() : text.getBytes(StandardCharsets.UTF_8);
+ return (payload instanceof byte[] data)
+ ? data.clone()
+ : ((String) payload).getBytes(StandardCharsets.UTF_8);
}
@Override
public boolean equals(Object obj) {
- return obj instanceof WebSocketMessage that
- && Objects.equals(text, that.text)
- && Arrays.equals(data, that.data);
+ if (!(obj instanceof WebSocketMessage that)) {
+ return false;
+ }
+ if (payload instanceof byte[] data) {
+ return that.payload instanceof byte[] other && Arrays.equals(data, other);
+ }
+ return payload.equals(that.payload);
}
@Override
public int hashCode() {
- return 31 * Objects.hashCode(text) + Arrays.hashCode(data);
+ return (payload instanceof byte[] data) ? Arrays.hashCode(data) : payload.hashCode();
}
@Override
public String toString() {
- return isText() ? "text[" + text.length() + " chars]" : "binary[" + data.length + " bytes]";
+ return (payload instanceof String text)
+ ? "text[" + text.length() + " chars]"
+ : "binary[" + ((byte[]) payload).length + " bytes]";
}
}
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 d0ff99b43c..90a12de5f4 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
@@ -155,7 +155,8 @@ public WebSocketSession connect(URI uri, List subprotocols, Map 0) ? uri.getPort() : (secure ? 443 : 80);
+ int defaultPort = secure ? 443 : 80;
+ int port = (uri.getPort() > 0) ? uri.getPort() : defaultPort;
HttpHeaders extraHeaders = new DefaultHttpHeaders();
headers.forEach(extraHeaders::add);
diff --git a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java
index 34724126ac..144a1634e0 100644
--- a/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java
+++ b/org.restlet.ext.websocket/src/main/java/org/restlet/ext/websocket/internal/NettyWebSocketSession.java
@@ -151,7 +151,7 @@ public WebSocketMessage receive() throws IOException {
throw new InterruptedIOException("Interrupted while receiving a WebSocket message");
}
if (next == END) {
- inbound.offer(END);
+ inbound.add(END);
return null;
}
if (inbound.size() < MAX_QUEUED_MESSAGES && readPaused.compareAndSet(true, false)) {
@@ -224,15 +224,6 @@ private void awaitWritable() throws IOException {
}
}
- private void signalWritability() {
- writabilityLock.lock();
- try {
- writable.signalAll();
- } finally {
- writabilityLock.unlock();
- }
- }
-
@Override
public void close(int code, String reason) throws IOException {
if (closeSent.compareAndSet(false, true) && channel.isActive()) {
@@ -264,40 +255,51 @@ public String getCloseReason() {
return closeReason;
}
- private void enqueue(WebSocketMessage message) {
- inbound.offer(message);
- }
+ /** Converts frames into queued messages and handles the control frames. */
+ private final class Handler extends ChannelInboundHandlerAdapter {
- private void onClose(ChannelHandlerContext ctx, int status, String reason) {
- closeCode = (status < 0) ? NO_STATUS : status;
- closeReason = (reason == null || reason.isEmpty()) ? null : reason;
- closeReceived = true;
- if (closeSent.compareAndSet(false, true)) {
- int echo =
- WebSocketCloseStatus.isValidStatusCode(closeCode) ? closeCode : NORMAL_CLOSURE;
- ctx.writeAndFlush(new CloseWebSocketFrame(echo, null))
- .addListener(ChannelFutureListener.CLOSE);
- } else {
- ctx.close();
+ private void signalWritability() {
+ writabilityLock.lock();
+ try {
+ writable.signalAll();
+ } finally {
+ writabilityLock.unlock();
+ }
}
- inbound.offer(END);
- }
- private void failSession(ChannelHandlerContext ctx, int code) {
- if (!closeReceived) {
- closeCode = code;
- closeReceived = true;
+ private void enqueue(WebSocketMessage message) {
+ inbound.add(message);
}
- if (closeSent.compareAndSet(false, true) && ctx.channel().isActive()) {
- ctx.writeAndFlush(new CloseWebSocketFrame(code, null))
- .addListener(ChannelFutureListener.CLOSE);
- } else {
- ctx.close();
+
+ private void onClose(ChannelHandlerContext ctx, int status, String reason) {
+ closeCode = (status < 0) ? NO_STATUS : status;
+ closeReason = (reason == null || reason.isEmpty()) ? null : reason;
+ closeReceived = true;
+ if (closeSent.compareAndSet(false, true)) {
+ int echo =
+ WebSocketCloseStatus.isValidStatusCode(closeCode)
+ ? closeCode
+ : NORMAL_CLOSURE;
+ ctx.writeAndFlush(new CloseWebSocketFrame(echo, null))
+ .addListener(ChannelFutureListener.CLOSE);
+ } else {
+ ctx.close();
+ }
+ inbound.add(END);
}
- }
- /** Converts frames into queued messages and handles the control frames. */
- private final class Handler extends ChannelInboundHandlerAdapter {
+ private void failSession(ChannelHandlerContext ctx, int code) {
+ if (!closeReceived) {
+ closeCode = code;
+ closeReceived = true;
+ }
+ if (closeSent.compareAndSet(false, true) && ctx.channel().isActive()) {
+ ctx.writeAndFlush(new CloseWebSocketFrame(code, null))
+ .addListener(ChannelFutureListener.CLOSE);
+ } else {
+ ctx.close();
+ }
+ }
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
@@ -352,7 +354,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
closeCode = ABNORMAL_CLOSURE;
closeReceived = true;
}
- inbound.offer(END);
+ inbound.add(END);
super.channelInactive(ctx);
}
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 aefed28f32..482afa241e 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
@@ -168,12 +168,10 @@ void connectionFailuresAreReported() throws Exception {
}
try (WebSocketClient client =
WebSocketClient.builder().connectTimeout(Duration.ofSeconds(2)).build()) {
- assertThrows(
- ConnectException.class,
- () -> client.connect("ws://localhost:" + freePort + "/"));
- assertThrows(
- IllegalArgumentException.class,
- () -> client.connect("http://localhost:" + freePort + "/"));
+ String refusedUri = "ws://localhost:" + freePort + "/";
+ String httpUri = "http://localhost:" + freePort + "/";
+ assertThrows(ConnectException.class, () -> client.connect(refusedUri));
+ assertThrows(IllegalArgumentException.class, () -> client.connect(httpUri));
assertThrows(IllegalArgumentException.class, () -> client.connect("ws:///nohost"));
}
}
@@ -194,7 +192,7 @@ void messagesHaveValueSemantics() {
assertEquals(WebSocketMessage.text("a"), WebSocketMessage.text("a"));
assertEquals(WebSocketMessage.text("a").hashCode(), WebSocketMessage.text("a").hashCode());
assertNotEquals(WebSocketMessage.text("a"), WebSocketMessage.binary(new byte[] {'a'}));
- assertNotEquals(WebSocketMessage.text("a"), "a");
+ assertNotEquals(new Object(), WebSocketMessage.text("a"));
assertEquals("text[1 chars]", WebSocketMessage.text("a").toString());
}
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 f60324cd1b..7a33d55ce2 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
@@ -268,7 +268,7 @@ void enforcesTheOpenSessionLimit() throws Exception {
for (int i = 0; i < 50 && second == null; i++) {
try {
second = client.connect(base + "/");
- } catch (IOException e) {
+ } catch (IOException _) {
Thread.sleep(50);
}
}
diff --git a/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketUpgradeTestCase.java b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketUpgradeTestCase.java
index 2fb2acb8d4..c1966382a9 100644
--- a/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketUpgradeTestCase.java
+++ b/org.restlet.ext.websocket/src/test/java/org/restlet/ext/websocket/WebSocketUpgradeTestCase.java
@@ -69,18 +69,14 @@ void acceptRecordsTheUpgradeOnTheResponse() {
@Test
void acceptRejectsInvalidUpgrades() {
+ Response plain = new Response(new Request(Method.GET, "http://localhost/"));
+ WebSocketHandler handler = session -> {};
assertThrows(
- IllegalStateException.class,
- () ->
- WebSocketUpgrade.accept(
- new Response(new Request(Method.GET, "http://localhost/")),
- session -> {},
- null));
+ IllegalStateException.class, () -> WebSocketUpgrade.accept(plain, handler, null));
+ Response chat = new Response(upgradeRequest("chat"));
assertThrows(
IllegalArgumentException.class,
- () ->
- WebSocketUpgrade.accept(
- new Response(upgradeRequest("chat")), session -> {}, "other"));
+ () -> WebSocketUpgrade.accept(chat, handler, "other"));
assertNull(WebSocketUpgrade.get(new Response(upgradeRequest())));
}
diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/NettyConnectionUpgradeTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/NettyConnectionUpgradeTestCase.java
index 4e21a021ab..fd6cf3edd4 100644
--- a/org.restlet/src/test/java/org/restlet/engine/connector/NettyConnectionUpgradeTestCase.java
+++ b/org.restlet/src/test/java/org/restlet/engine/connector/NettyConnectionUpgradeTestCase.java
@@ -10,6 +10,7 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import io.netty.buffer.Unpooled;
@@ -84,10 +85,15 @@ public void handle(Request request, Response response) {
private static String exchange(int port) throws Exception {
try (Socket socket = new Socket("localhost", port)) {
OutputStream out = socket.getOutputStream();
- out.write(
- ("GET /up HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\n"
- + "Upgrade: test\r\n\r\n")
- .getBytes(StandardCharsets.US_ASCII));
+ String request =
+ """
+ GET /up HTTP/1.1\r
+ Host: localhost\r
+ Connection: Upgrade\r
+ Upgrade: test\r
+ \r
+ """;
+ out.write(request.getBytes(StandardCharsets.US_ASCII));
out.flush();
BufferedReader in =
new BufferedReader(
@@ -130,7 +136,7 @@ void handsTheConnectionOverToTheUpgrade() throws Exception {
assertEquals(30000, context.getIdleTimeout());
assertTrue(context.getParameters().isEmpty());
context.getAttributes().put("seen", true);
- assertTrue(context.getLogger() != null);
+ assertNotNull(context.getLogger());
}
@Test
From 455daff0ee3f68c7ca8792173e26801c463bd0ac Mon Sep 17 00:00:00 2001
From: Jerome Louvel <374450+jlouvel@users.noreply.github.com>
Date: Tue, 20 Oct 2026 16:55:46 -0400
Subject: [PATCH 7/7] fix: don't read the next HTTP/1.1 request before the
response is complete
NettyServerHandler requested a channel read right after dispatching each
request. When the decoder had already produced the whole request, that
read could take the next pipelined request on the same connection while
the current response was still being written, interleaving responses and
corrupting the stream (flaky AsyncTestCase: 'Line Feed must be preceded by
Carriage Return'). On HTTP/1.1 the read is now deferred to
channelReadComplete and skipped once the request content has arrived.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
.../engine/netty/NettyServerHandler.java | 25 +++++++++++++++++--
1 file changed, 23 insertions(+), 2 deletions(-)
diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java
index 3cec34a87a..6ca8775f02 100644
--- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java
+++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java
@@ -55,6 +55,9 @@ class NettyServerHandler extends ChannelInboundHandlerAdapter {
/** The call receiving the current request body. */
private NettyServerCall currentCall;
+ /** Whether to read the rest of the current HTTP/1.1 request once the channel read completes. */
+ private boolean readRequestBody;
+
NettyServerHandler(NettyServerHelper helper, boolean confidential, StreamKind kind) {
this.helper = helper;
this.confidential = confidential;
@@ -136,11 +139,29 @@ private void onRequest(ChannelHandlerContext ctx, HttpRequest request) {
sendError(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE);
}
- // Keep reading the body of this request (if any) without waiting for the call.
- ctx.read();
+ // Keep reading the body of this request (if any) without waiting for the call. On HTTP/1.1,
+ // the read is deferred to channelReadComplete: the decoder may already have produced the
+ // whole request, and reading now could take the next request before this response is
+ // complete.
+ if (kind.isStream()) {
+ ctx.read();
+ } else {
+ readRequestBody = true;
+ }
+ }
+
+ @Override
+ public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+ if (readRequestBody) {
+ readRequestBody = false;
+ ctx.read();
+ }
+ super.channelReadComplete(ctx);
}
private void onContent(ChannelHandlerContext ctx, HttpContent content) {
+ // Request content drives the next reads from now on.
+ readRequestBody = false;
ChannelInputStream body = currentBody;
if (body == null) {
content.release();