connect(
if (cfg.isOfferClientNoContextTakeover()) {
ext.append("; client_no_context_takeover");
}
- if (cfg.getOfferClientMaxWindowBits() != null && cfg.getOfferClientMaxWindowBits() == 15) {
- ext.append("; client_max_window_bits=15");
- }
+ // client_max_window_bits is never offered: the JDK Deflater cannot honor a
+ // server-selected window smaller than 15 (RFC 7692 7.2.1).
if (cfg.getOfferServerMaxWindowBits() != null) {
ext.append("; server_max_window_bits=").append(cfg.getOfferServerMaxWindowBits());
}
@@ -211,8 +210,9 @@ public void produceRequest(final RequestChannel channel, final HttpContext conte
public void consumeResponse(final HttpResponse response, final EntityDetails entityDetails,
final HttpContext context) throws HttpException, IOException {
- if (response.getCode() != HttpStatus.SC_OK) {
- failFuture(new IllegalStateException("Unexpected status: " + response.getCode()));
+ final int code = response.getCode();
+ if (code < HttpStatus.SC_OK || code >= HttpStatus.SC_REDIRECTION) {
+ failFuture(new IllegalStateException("Unexpected status: " + code));
return;
}
@@ -321,8 +321,7 @@ public void cancel() {
}
private static String headerValue(final HttpResponse r, final String name) {
- final Header h = r.getFirstHeader(name);
- return h != null ? h.getValue() : null;
+ return Http1UpgradeProtocol.headerValue(r, name);
}
private void failFuture(final Exception ex) {
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransport.java b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransport.java
index d002d32e2f..2a8c980139 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransport.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransport.java
@@ -28,6 +28,7 @@
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.channels.CancelledKeyException;
import org.apache.hc.core5.annotation.Internal;
import org.apache.hc.core5.http.nio.DataStreamChannel;
@@ -51,14 +52,26 @@ public int write(final ByteBuffer src) throws IOException {
if (ch == null) {
return 0;
}
- return ch.write(src);
+ try {
+ return ch.write(src);
+ } catch (final CancelledKeyException ex) {
+ // A cancelled selection key is a terminal state, not transient back-pressure. Surface it
+ // as an I/O error so the engine takes its close path instead of retaining the frame and
+ // waiting for an output callback that will never arrive.
+ throw new IOException("WebSocket transport channel is closed", ex);
+ }
}
@Override
public void requestOutput() {
final DataStreamChannel ch = channel;
if (ch != null) {
- ch.requestOutput();
+ try {
+ ch.requestOutput();
+ } catch (final CancelledKeyException ignore) {
+ // requestOutput is a best-effort nudge; if the channel's selection key was
+ // cancelled by a concurrent shutdown there is nothing left to flush.
+ }
}
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/WebSocketSessionEngine.java b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/WebSocketSessionEngine.java
index f2e9b3a27c..4ab599a024 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/WebSocketSessionEngine.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/client5/http/websocket/transport/WebSocketSessionEngine.java
@@ -109,6 +109,7 @@ public final class WebSocketSessionEngine {
final AtomicBoolean open = new AtomicBoolean(true);
final AtomicBoolean closeSent = new AtomicBoolean(false);
private final AtomicBoolean closeReceived = new AtomicBoolean(false);
+ private final AtomicBoolean released = new AtomicBoolean(false);
volatile boolean closeAfterFlush;
private volatile ScheduledFuture closeTimeoutFuture;
@@ -332,6 +333,12 @@ private void handleFrame() {
return;
}
if (FrameOpcode.isControl(op)) {
+ if (r1) {
+ // Control frames are never compressed, so an extension never defines RSV1 for them.
+ initiateClose(1002, "RSV1 set on control frame");
+ inbuf.clear();
+ return;
+ }
if (!fin) {
initiateClose(1002, "fragmented control frame");
inbuf.clear();
@@ -379,8 +386,7 @@ private void handleFrame() {
inbuf.clear();
return;
}
- appendToMessage(payload);
- if (fin) {
+ if (appendToMessage(payload) && fin) {
deliverAssembledMessage();
}
break;
@@ -484,15 +490,16 @@ private void startMessage(final int opcode, final ByteBuffer payload, final bool
appendToMessage(payload);
}
- private void appendToMessage(final ByteBuffer payload) {
+ private boolean appendToMessage(final ByteBuffer payload) {
final int n = payload.remaining();
assemblingSize += n;
if (cfg.getMaxMessageSize() > 0 && assemblingSize > cfg.getMaxMessageSize()) {
initiateClose(1009, "Message too big");
- return;
+ return false;
}
assemblingBuf = WebSocketBufferOps.ensureCapacity(assemblingBuf, n);
assemblingBuf.put(payload.asReadOnlyBuffer());
+ return true;
}
private void deliverAssembledMessage() {
@@ -673,6 +680,9 @@ boolean enqueueData(final OutFrame frame) {
}
private void drainAndRelease() {
+ if (!released.compareAndSet(false, true)) {
+ return;
+ }
if (activeWrite != null) {
if (activeWrite.dataFrame) {
dataQueuedBytes.addAndGet(-activeWrite.size);
@@ -690,7 +700,22 @@ private void drainAndRelease() {
dataQueuedBytes.addAndGet(-f.size);
}
}
- cancelCloseTimeout();
+ // Close the encoder and decoder chains coordinated with try/finally so a failure closing
+ // one still closes the other, then cancel the close timeout unconditionally. Each chain's
+ // close() is itself exception-safe and releases every codec it holds.
+ try {
+ if (encChain != null) {
+ encChain.close();
+ }
+ } finally {
+ try {
+ if (decChain != null) {
+ decChain.close();
+ }
+ } finally {
+ cancelCloseTimeout();
+ }
+ }
}
// ==================================================================
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtension.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtension.java
index 6c1068f6bf..957a7f600c 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtension.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtension.java
@@ -144,18 +144,25 @@ public ByteBuffer encode(final WebSocketFrameType type, final boolean fin, final
deflater.setInput(input);
final ByteArrayOutputStream out = new ByteArrayOutputStream(Math.max(128, input.length / 2));
final byte[] buffer = new byte[Math.min(16384, Math.max(1024, input.length))];
- while (!deflater.needsInput()) {
- final int count = deflater.deflate(buffer, 0, buffer.length, Deflater.SYNC_FLUSH);
+ // Drain until a SYNC_FLUSH leaves the output buffer partly filled; testing needsInput()
+ // would stop once the input is consumed and could drop the trailing 00 00 FF FF flush bytes.
+ int count;
+ do {
+ count = deflater.deflate(buffer, 0, buffer.length, Deflater.SYNC_FLUSH);
if (count > 0) {
out.write(buffer, 0, count);
- } else {
- break;
}
- }
+ } while (count == buffer.length);
final byte[] data = out.toByteArray();
final ByteBuffer encoded;
- if (data.length >= 4) {
- encoded = ByteBuffer.wrap(data, 0, data.length - 4);
+ // Strip the 00 00 FF FF flush trailer only on the final fragment; non-final fragments
+ // must keep it so each intermediate empty stored block stays valid and the reassembled
+ // DEFLATE stream decodes (RFC 7692 section 7.2.1). When stripping leaves nothing the final
+ // fragment is empty, which RFC 7692 section 7.2.3.6 represents as the single octet 0x00.
+ if (fin) {
+ encoded = data.length > 4
+ ? ByteBuffer.wrap(data, 0, data.length - 4)
+ : ByteBuffer.wrap(new byte[]{0x00});
} else {
encoded = ByteBuffer.wrap(data);
}
@@ -183,6 +190,12 @@ public WebSocketExtensionData getResponseData() {
return new WebSocketExtensionData(getName(), params);
}
+ @Override
+ public void close() {
+ deflater.end();
+ inflater.end();
+ }
+
private static boolean isDataFrame(final WebSocketFrameType type) {
return type == WebSocketFrameType.TEXT || type == WebSocketFrameType.BINARY;
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactory.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactory.java
index 6166efe842..d486156d04 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactory.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactory.java
@@ -26,14 +26,18 @@
*/
package org.apache.hc.core5.websocket;
+import java.util.Map;
+
/**
* Factory for {@code permessage-deflate} extensions (RFC 7692).
*
- * Note: the JDK {@link java.util.zip.Deflater} / {@link java.util.zip.Inflater}
- * only support a 15-bit window size. This factory therefore accepts
- * {@code client_max_window_bits} / {@code server_max_window_bits} only when
- * they are either absent or explicitly set to {@code 15}. Other values are
- * rejected during negotiation.
+ * An offer is accepted only when every parameter is recognised and well-formed. The
+ * {@code *_no_context_takeover} parameters must appear without a value, and the window-bits
+ * parameters must be exactly {@code 15} because the JDK {@link java.util.zip.Deflater} /
+ * {@link java.util.zip.Inflater} only support a 15-bit window. {@code client_max_window_bits} may
+ * also be offered without a value, in which case the client advertises support and the server
+ * selects the window. Any unknown parameter, any value on a valueless parameter, and any window
+ * size other than {@code 15} cause the offer to be declined (RFC 7692 section 7.1).
*/
public final class PerMessageDeflateExtensionFactory implements WebSocketExtensionFactory {
@@ -44,33 +48,46 @@ public String getName() {
@Override
public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) {
- if (request == null) {
- return null;
- }
- final String name = request.getName();
- if (!"permessage-deflate".equals(name)) {
+ if (request == null || !"permessage-deflate".equals(request.getName())) {
return null;
}
- final boolean serverNoContextTakeover = request.getParameters().containsKey("server_no_context_takeover");
- final boolean clientNoContextTakeover = request.getParameters().containsKey("client_no_context_takeover");
- final boolean clientMaxBitsPresent = request.getParameters().containsKey("client_max_window_bits");
- final boolean serverMaxBitsPresent = request.getParameters().containsKey("server_max_window_bits");
- Integer clientMaxWindowBits = parseWindowBits(request.getParameters().get("client_max_window_bits"));
- Integer serverMaxWindowBits = parseWindowBits(request.getParameters().get("server_max_window_bits"));
- if (clientMaxBitsPresent && clientMaxWindowBits == null) {
- clientMaxWindowBits = 15;
- }
- if (serverMaxBitsPresent && serverMaxWindowBits == null) {
- serverMaxWindowBits = 15;
- }
-
- if (!PerMessageDeflateExtension.isValidWindowBits(clientMaxWindowBits)
- || !PerMessageDeflateExtension.isValidWindowBits(serverMaxWindowBits)) {
- return null;
- }
- // JDK Deflater/Inflater only supports 15-bit window size.
- if (!isSupportedWindowBits(clientMaxWindowBits) || !isSupportedWindowBits(serverMaxWindowBits)) {
- return null;
+ boolean serverNoContextTakeover = false;
+ boolean clientNoContextTakeover = false;
+ Integer clientMaxWindowBits = null;
+ Integer serverMaxWindowBits = null;
+ for (final Map.Entry param : request.getParameters().entrySet()) {
+ final String value = param.getValue();
+ switch (param.getKey()) {
+ case "server_no_context_takeover":
+ if (value != null) {
+ return null;
+ }
+ serverNoContextTakeover = true;
+ break;
+ case "client_no_context_takeover":
+ if (value != null) {
+ return null;
+ }
+ clientNoContextTakeover = true;
+ break;
+ case "client_max_window_bits":
+ // A valueless parameter advertises support; the server selects the window (15).
+ if (value != null && !isWindowBits15(value)) {
+ return null;
+ }
+ clientMaxWindowBits = 15;
+ break;
+ case "server_max_window_bits":
+ // Must carry a value, and only 15 is supported.
+ if (!isWindowBits15(value)) {
+ return null;
+ }
+ serverMaxWindowBits = 15;
+ break;
+ default:
+ // Unknown parameters make the offer invalid.
+ return null;
+ }
}
return new PerMessageDeflateExtension(
serverNoContextTakeover,
@@ -79,18 +96,13 @@ public WebSocketExtension create(final WebSocketExtensionData request, final boo
serverMaxWindowBits);
}
- private static Integer parseWindowBits(final String value) {
- if (value == null || value.isEmpty()) {
- return null;
- }
- try {
- return Integer.parseInt(value);
- } catch (final NumberFormatException ignore) {
- return null;
- }
- }
-
- private static boolean isSupportedWindowBits(final Integer bits) {
- return bits == null || bits == 15;
+ /**
+ * The only window size the JDK Deflater/Inflater supports. A valid value is exactly the two
+ * decimal digits {@code "15"}; this rejects a leading zero ({@code "015"}), a sign
+ * ({@code "+15"}) and any missing or non-numeric value, which {@link Integer#parseInt} would
+ * otherwise accept.
+ */
+ private static boolean isWindowBits15(final String value) {
+ return "15".equals(value);
}
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtension.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtension.java
index 4364a5008f..98d213c857 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtension.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtension.java
@@ -77,4 +77,13 @@ default ByteBuffer encode(
default WebSocketExtensionData getResponseData() {
return new WebSocketExtensionData(getName(), null);
}
+
+ /**
+ * Releases any native resources held by this extension (e.g. a {@code Deflater} or
+ * {@code Inflater}). Called once when the owning session terminates.
+ *
+ * @since 5.7
+ */
+ default void close() {
+ }
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiation.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiation.java
index 4acf9779ce..768c6aa4c5 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiation.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiation.java
@@ -29,12 +29,13 @@
import java.util.Collections;
import java.util.List;
import java.util.StringJoiner;
+import java.util.concurrent.atomic.AtomicBoolean;
-
-public final class WebSocketExtensionNegotiation {
+public final class WebSocketExtensionNegotiation implements AutoCloseable {
private final List extensions;
private final List responseData;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
WebSocketExtensionNegotiation(
final List extensions,
@@ -58,4 +59,32 @@ public String formatResponseHeader() {
}
return joiner.length() > 0 ? joiner.toString() : null;
}
+
+ /**
+ * Releases the native resources held by the negotiated extensions. Idempotent; only the first
+ * invocation closes the extensions and each extension is closed exactly once. A failure to
+ * close one extension does not prevent the remaining extensions from being closed.
+ *
+ * @since 5.7
+ */
+ @Override
+ public void close() {
+ if (closed.compareAndSet(false, true)) {
+ RuntimeException failure = null;
+ for (final WebSocketExtension extension : extensions) {
+ try {
+ extension.close();
+ } catch (final RuntimeException ex) {
+ if (failure == null) {
+ failure = ex;
+ } else {
+ failure.addSuppressed(ex);
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+ }
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistry.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistry.java
index 12fa259f3f..95203a4b84 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistry.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistry.java
@@ -27,9 +27,11 @@
package org.apache.hc.core5.websocket;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.hc.core5.util.Args;
@@ -52,17 +54,39 @@ public WebSocketExtensionNegotiation negotiate(
final boolean server) throws WebSocketException {
final List extensions = new ArrayList<>();
final List responseData = new ArrayList<>();
- if (requested != null) {
- for (final WebSocketExtensionData request : requested) {
- final WebSocketExtensionFactory factory = factories.get(request.getName());
- if (factory != null) {
- final WebSocketExtension extension = factory.create(request, server);
- if (extension != null) {
- extensions.add(extension);
- responseData.add(extension.getResponseData());
+ final Set accepted = new HashSet<>();
+ try {
+ if (requested != null) {
+ for (final WebSocketExtensionData request : requested) {
+ // At most one offer per extension name is accepted; a second accepted offer of the
+ // same extension would claim the same RSV bit and break both the response header and
+ // the RSV bookkeeping in the frame reader.
+ if (accepted.contains(request.getName())) {
+ continue;
}
+ final WebSocketExtensionFactory factory = factories.get(request.getName());
+ if (factory != null) {
+ final WebSocketExtension extension = factory.create(request, server);
+ if (extension != null) {
+ extensions.add(extension);
+ responseData.add(extension.getResponseData());
+ accepted.add(request.getName());
+ }
+ }
+ }
+ }
+ } catch (final WebSocketException | RuntimeException ex) {
+ // A later factory failed after earlier extensions were already created; close them so
+ // their native resources are not leaked, preserving the original failure and recording
+ // any close failure as a suppressed exception.
+ for (final WebSocketExtension extension : extensions) {
+ try {
+ extension.close();
+ } catch (final RuntimeException closeEx) {
+ ex.addSuppressed(closeEx);
}
}
+ throw ex;
}
return new WebSocketExtensionNegotiation(extensions, responseData);
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensions.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensions.java
index 6c853b50fb..38ee16dddc 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensions.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketExtensions.java
@@ -27,6 +27,7 @@
package org.apache.hc.core5.websocket;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -44,8 +45,30 @@ private WebSocketExtensions() {
public static List parse(final Header header) {
final List extensions = new ArrayList<>();
+ addOffers(header, extensions);
+ return extensions;
+ }
+
+ /**
+ * Parses and combines every {@code Sec-WebSocket-Extensions} header field. RFC 6455 permits the
+ * header to be split across multiple fields and requires them to be treated as a single combined
+ * value, so a client's fallback offers carried in a later field are examined as well.
+ *
+ * @since 5.7
+ */
+ public static List parse(final Iterator headers) {
+ final List extensions = new ArrayList<>();
+ if (headers != null) {
+ while (headers.hasNext()) {
+ addOffers(headers.next(), extensions);
+ }
+ }
+ return extensions;
+ }
+
+ private static void addOffers(final Header header, final List extensions) {
if (header == null) {
- return extensions;
+ return;
}
for (final HeaderElement element : MessageSupport.parseElements(header)) {
final String name = element.getName();
@@ -53,11 +76,22 @@ public static List parse(final Header header) {
continue;
}
final Map params = new LinkedHashMap<>();
+ boolean duplicateParameter = false;
for (final NameValuePair param : element.getParameters()) {
+ // RFC 7692 section 7.1: an extension parameter MUST NOT appear more than once in a
+ // negotiation offer. A repeated parameter makes the whole offer invalid, so it is
+ // dropped here and the extension is left un-negotiated rather than silently collapsed
+ // into a single map entry.
+ if (params.containsKey(param.getName())) {
+ duplicateParameter = true;
+ break;
+ }
params.put(param.getName(), param.getValue());
}
+ if (duplicateParameter) {
+ continue;
+ }
extensions.add(new WebSocketExtensionData(name, params));
}
- return extensions;
}
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java
index fb14e941b7..2bbdf8cf40 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameReader.java
@@ -98,7 +98,8 @@ WebSocketFrame readFrame() throws IOException {
len = (len << 8) | (readByte() & 0xFF);
}
if (len < 0) {
- throw new WebSocketException("Negative frame payload length");
+ // RFC 6455 section 5.2: the most significant bit of a 64-bit length MUST be 0.
+ throw new WebSocketException("64-bit frame length must have the most significant bit set to 0");
}
}
if (len > Integer.MAX_VALUE) {
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameWriter.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameWriter.java
index 69ef31e5b8..abc4a2c9c6 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameWriter.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/WebSocketFrameWriter.java
@@ -34,6 +34,7 @@
import java.util.List;
import org.apache.hc.core5.util.Args;
+import org.apache.hc.core5.websocket.message.CloseCodec;
class WebSocketFrameWriter {
@@ -63,13 +64,9 @@ void writePong(final ByteBuffer payload) throws IOException {
}
void writeClose(final int statusCode, final String reason) throws IOException {
- final byte[] reasonBytes = reason != null ? reason.getBytes(StandardCharsets.UTF_8) : new byte[0];
- final int len = 2 + reasonBytes.length;
- final ByteBuffer buffer = ByteBuffer.allocate(len);
- buffer.put((byte) ((statusCode >> 8) & 0xFF));
- buffer.put((byte) (statusCode & 0xFF));
- buffer.put(reasonBytes);
- buffer.flip();
+ // CloseCodec truncates the reason to 123 UTF-8 bytes so the payload never
+ // exceeds the 125-byte control frame limit (RFC 6455 section 5.5).
+ final ByteBuffer buffer = ByteBuffer.wrap(CloseCodec.encode(statusCode, reason));
writeFrame(WebSocketFrameType.CLOSE, buffer, false, false, false);
}
@@ -102,6 +99,9 @@ private void writeFrame(
Args.notNull(type, "Frame type");
final ByteBuffer buffer = payload != null ? payload.asReadOnlyBuffer() : ByteBuffer.allocate(0);
final int payloadLen = buffer.remaining();
+ if (type.isControl() && payloadLen > 125) {
+ throw new IllegalArgumentException("Control frame payload > 125 bytes: " + payloadLen);
+ }
int firstByte = 0x80 | (type.getOpcode() & 0x0F);
if (rsv1) {
firstByte |= 0x40;
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/ExtensionChain.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/ExtensionChain.java
index 3b73baf01b..5fcc1fef5b 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/ExtensionChain.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/ExtensionChain.java
@@ -112,6 +112,30 @@ public WebSocketExtensionChain.Encoded encode(final byte[] data, final boolean f
}
return new WebSocketExtensionChain.Encoded(out, setRsv1);
}
+
+ /**
+ * Releases native resources held by the encoders in this chain. Every encoder is closed even
+ * if a previous one fails; the first failure is rethrown with the rest attached as suppressed.
+ *
+ * @since 5.7
+ */
+ public void close() {
+ RuntimeException failure = null;
+ for (final WebSocketExtensionChain.Encoder e : encs) {
+ try {
+ e.close();
+ } catch (final RuntimeException ex) {
+ if (failure == null) {
+ failure = ex;
+ } else {
+ failure.addSuppressed(ex);
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
}
public static final class DecodeChain {
@@ -143,5 +167,29 @@ public byte[] decode(final byte[] data, final long maxDecodedSize) throws Except
}
return out;
}
+
+ /**
+ * Releases native resources held by the decoders in this chain. Every decoder is closed even
+ * if a previous one fails; the first failure is rethrown with the rest attached as suppressed.
+ *
+ * @since 5.7
+ */
+ public void close() {
+ RuntimeException failure = null;
+ for (final WebSocketExtensionChain.Decoder d : decs) {
+ try {
+ d.close();
+ } catch (final RuntimeException ex) {
+ if (failure == null) {
+ failure = ex;
+ } else {
+ failure.addSuppressed(ex);
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
}
}
\ No newline at end of file
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/PerMessageDeflate.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/PerMessageDeflate.java
index 002b9d486e..61b64c480a 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/PerMessageDeflate.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/PerMessageDeflate.java
@@ -87,44 +87,57 @@ public Encoded encode(final byte[] data, final boolean first, final boolean fin)
return new Encoded(out, first);
}
+ @Override
+ public void close() {
+ def.end();
+ }
+
private byte[] compressMessage(final byte[] data) {
return doDeflate(data, true, true, clientNoContextTakeover);
}
private byte[] compressFragment(final byte[] data, final boolean fin) {
- return doDeflate(data, fin, true, fin && clientNoContextTakeover);
+ // Strip the 00 00 FF FF flush trailer only on the final fragment; non-final
+ // fragments must keep it so each intermediate empty stored block stays valid
+ // and the reassembled DEFLATE stream decodes (RFC 7692 section 7.2.1).
+ return doDeflate(data, fin, fin, fin && clientNoContextTakeover);
}
private byte[] doDeflate(final byte[] data,
final boolean fin,
final boolean stripTail,
final boolean maybeReset) {
- if (data == null || data.length == 0) {
- if (fin && maybeReset) {
- def.reset();
- }
- return new byte[0];
- }
- def.setInput(data);
- final ByteArrayOutputStream out = new ByteArrayOutputStream(Math.max(128, data.length / 2));
- final byte[] buf = new byte[Math.min(16384, Math.max(1024, data.length))];
- while (!def.needsInput()) {
- final int n = def.deflate(buf, 0, buf.length, Deflater.SYNC_FLUSH);
+ // Empty input is deflated through SYNC_FLUSH as well: an empty message or empty
+ // final fragment must compress to the single octet 0x00 once the 00 00 FF FF flush
+ // trailer is stripped (RFC 7692 section 7.2.3.6), not to an empty payload.
+ final byte[] input = data != null ? data : new byte[0];
+ def.setInput(input);
+ final ByteArrayOutputStream out = new ByteArrayOutputStream(Math.max(128, input.length / 2));
+ final byte[] buf = new byte[Math.min(16384, Math.max(1024, input.length))];
+ // Drain until a SYNC_FLUSH leaves the output buffer partly filled. Testing
+ // needsInput() would stop as soon as the input is consumed, which can drop the
+ // trailing 00 00 FF FF flush bytes when the buffer fills exactly and then have
+ // stripTail cut into real data.
+ int n;
+ do {
+ n = def.deflate(buf, 0, buf.length, Deflater.SYNC_FLUSH);
if (n > 0) {
out.write(buf, 0, n);
- } else {
- break;
}
- }
+ } while (n == buf.length);
byte[] all = out.toByteArray();
- if (stripTail && all.length >= 4) {
- final int newLen = all.length - 4; // strip 00 00 FF FF
- if (newLen <= 0) {
- all = new byte[0];
- } else {
- final byte[] trimmed = new byte[newLen];
- System.arraycopy(all, 0, trimmed, 0, newLen);
+ if (stripTail) {
+ // Strip the trailing 00 00 FF FF SYNC_FLUSH marker. When nothing remains the
+ // message (or final fragment) is empty; RFC 7692 section 7.2.3.6 represents an
+ // empty payload as the single octet 0x00, not as an empty payload. A fresh
+ // Deflater already yields 0x00 here, but one holding context (context takeover)
+ // emits only the 4-octet marker, so the 0x00 must be supplied explicitly.
+ if (all.length > 4) {
+ final byte[] trimmed = new byte[all.length - 4];
+ System.arraycopy(all, 0, trimmed, 0, trimmed.length);
all = trimmed;
+ } else {
+ all = new byte[]{0x00};
}
}
if (fin && maybeReset) {
@@ -182,6 +195,11 @@ public byte[] decode(final byte[] compressedMessage, final long maxDecodedSize)
}
return out.toByteArray();
}
+
+ @Override
+ public void close() {
+ inf.end();
+ }
};
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/WebSocketExtensionChain.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/WebSocketExtensionChain.java
index 92ad719015..ad082c44a0 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/WebSocketExtensionChain.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/extension/WebSocketExtensionChain.java
@@ -69,6 +69,14 @@ interface Encoder {
* Encode one fragment; return transformed payload and whether to set RSV on FIRST frame.
*/
Encoded encode(byte[] data, boolean first, boolean fin);
+
+ /**
+ * Releases any native resources held by this encoder (e.g. a {@code Deflater}).
+ *
+ * @since 5.7
+ */
+ default void close() {
+ }
}
interface Decoder {
@@ -88,5 +96,13 @@ interface Decoder {
default byte[] decode(final byte[] payload, final long maxDecodedSize) throws Exception {
return decode(payload);
}
+
+ /**
+ * Releases any native resources held by this decoder (e.g. an {@code Inflater}).
+ *
+ * @since 5.7
+ */
+ default void close() {
+ }
}
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/frame/WebSocketFrameWriter.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/frame/WebSocketFrameWriter.java
index 5839f0caca..0126a513c4 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/frame/WebSocketFrameWriter.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/frame/WebSocketFrameWriter.java
@@ -35,7 +35,7 @@
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
-import java.util.concurrent.ThreadLocalRandom;
+import java.security.SecureRandom;
import org.apache.hc.core5.annotation.Internal;
import org.apache.hc.core5.websocket.message.CloseCodec;
@@ -48,6 +48,9 @@
@Internal
public final class WebSocketFrameWriter {
+ // RFC 6455 section 5.3 requires the masking key to be derived from a strong source of entropy.
+ private static final SecureRandom MASK_RNG = new SecureRandom();
+
// -- Text/Binary -----------------------------------------------------------
public ByteBuffer text(final CharSequence data, final boolean fin) {
@@ -164,7 +167,7 @@ public ByteBuffer frameIntoWithRSV(final int opcode, final ByteBuffer payload, f
int maskInt = 0;
if (mask) {
- maskInt = ThreadLocalRandom.current().nextInt();
+ maskInt = MASK_RNG.nextInt();
out.put((byte) (maskInt >>> 24)).put((byte) (maskInt >>> 16))
.put((byte) (maskInt >>> 8)).put((byte) maskInt);
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandler.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandler.java
index 64cb417143..7d424faf81 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandler.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandler.java
@@ -35,6 +35,8 @@
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hc.core5.concurrent.DefaultThreadFactory;
@@ -55,6 +57,7 @@
import org.apache.hc.core5.websocket.WebSocketCloseStatus;
import org.apache.hc.core5.websocket.WebSocketConfig;
import org.apache.hc.core5.websocket.WebSocketConstants;
+import org.apache.hc.core5.websocket.WebSocketException;
import org.apache.hc.core5.websocket.WebSocketExtensionNegotiation;
import org.apache.hc.core5.websocket.WebSocketExtensionRegistry;
import org.apache.hc.core5.websocket.WebSocketExtensions;
@@ -68,6 +71,15 @@ final class WebSocketH2ServerExchangeHandler implements AsyncServerExchangeHandl
private static final byte[] END_INBOUND = new byte[0];
private static final ByteBuffer END_OUTBOUND = ByteBuffer.allocate(0);
+ /** Bounded in-flight inbound byte budget advertised via HTTP/2 flow control. */
+ private static final int INBOUND_WINDOW = 256 * 1024;
+
+ /** Default budget for buffered outbound bytes before the application writer is back-pressured. */
+ private static final int DEFAULT_OUTBOUND_BUFFER_SIZE = 1024 * 1024;
+
+ /** Default maximum size of a single buffered outbound chunk (an implementation detail). */
+ private static final int DEFAULT_OUTBOUND_CHUNK_SIZE = 64 * 1024;
+
/**
* Default execution strategy (no explicit thread creation in the handler).
* Note: tasks are typically long-lived (one per WS session). The bootstrap should ideally inject an executor.
@@ -81,15 +93,26 @@ final class WebSocketH2ServerExchangeHandler implements AsyncServerExchangeHandl
private final Executor executor;
private final BlockingQueue inbound = new LinkedBlockingQueue<>();
+ // Entry count is unbounded; outbound memory is bounded by a byte budget (see reserveOutboundBudget).
private final BlockingQueue outbound = new LinkedBlockingQueue<>();
private final ReentrantLock outLock = new ReentrantLock();
private ByteBuffer currentOutbound;
+ // Byte-budget back-pressure for the outbound queue: the application writer blocks once the
+ // in-flight outbound bytes would exceed the budget and resumes as produce() drains chunks.
+ private final ReentrantLock outboundBudgetLock = new ReentrantLock();
+ private final Condition outboundBudgetAvailable = outboundBudgetLock.newCondition();
+ private final int outboundBudgetBytes;
+ private final int outboundChunkSize;
+ private long pendingOutboundBytes;
+
private volatile boolean responseSent;
private volatile boolean outboundEnd;
private volatile boolean shutdown;
private volatile DataStreamChannel dataChannel;
+ private volatile CapacityChannel capacityChannel;
+ private final AtomicBoolean initialCreditGranted = new AtomicBoolean(false);
WebSocketH2ServerExchangeHandler(
final WebSocketHandler handler,
@@ -103,10 +126,25 @@ final class WebSocketH2ServerExchangeHandler implements AsyncServerExchangeHandl
final WebSocketConfig config,
final WebSocketExtensionRegistry extensionRegistry,
final Executor executor) {
+ this(handler, config, extensionRegistry, executor,
+ DEFAULT_OUTBOUND_BUFFER_SIZE, DEFAULT_OUTBOUND_CHUNK_SIZE);
+ }
+
+ WebSocketH2ServerExchangeHandler(
+ final WebSocketHandler handler,
+ final WebSocketConfig config,
+ final WebSocketExtensionRegistry extensionRegistry,
+ final Executor executor,
+ final int outboundBufferSize,
+ final int outboundChunkSize) {
this.handler = Args.notNull(handler, "WebSocket handler");
this.config = config != null ? config : WebSocketConfig.DEFAULT;
this.extensionRegistry = extensionRegistry != null ? extensionRegistry : WebSocketExtensionRegistry.createDefault();
this.executor = executor != null ? executor : DEFAULT_EXECUTOR;
+ this.outboundBudgetBytes = Args.positive(outboundBufferSize, "Outbound buffer size");
+ // A single chunk must never exceed the budget, or a large write could block forever waiting
+ // for capacity that can never be freed; cap the chunk size at the budget.
+ this.outboundChunkSize = Math.min(Args.positive(outboundChunkSize, "Outbound chunk size"), this.outboundBudgetBytes);
this.responseSent = false;
this.outboundEnd = false;
this.shutdown = false;
@@ -133,64 +171,109 @@ public void handleRequest(
}
final WebSocketExtensionNegotiation negotiation = extensionRegistry.negotiate(
- WebSocketExtensions.parse(request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER)),
+ WebSocketExtensions.parse(request.headerIterator(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER)),
true);
- final BasicHttpResponse response = new BasicHttpResponse(HttpStatus.SC_OK);
- final String extensionsHeader = negotiation.formatResponseHeader();
- if (extensionsHeader != null) {
- response.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, extensionsHeader);
- }
+ // The worker thread takes ownership of the negotiated extensions once the task is accepted
+ // and releases them in its finally after the stream has been torn down. If the hand-off
+ // never happens (a failed sendResponse or a rejected executor) this method releases them,
+ // attaching any close failure as a suppressed exception rather than masking the original.
+ try {
+ final BasicHttpResponse response = new BasicHttpResponse(HttpStatus.SC_OK);
+ final String extensionsHeader = negotiation.formatResponseHeader();
+ if (extensionsHeader != null) {
+ response.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, extensionsHeader);
+ }
- final List offeredProtocols = WebSocketHandshake.parseSubprotocols(
- request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL_LOWER));
- final String protocolResponse = handler.selectSubprotocol(offeredProtocols);
- if (protocolResponse != null) {
- response.addHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL_LOWER, protocolResponse);
- }
+ final List offeredProtocols = WebSocketHandshake.parseSubprotocols(
+ request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL_LOWER));
+ final String protocolResponse = handler.selectSubprotocol(offeredProtocols);
+ if (protocolResponse != null) {
+ response.addHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL_LOWER, protocolResponse);
+ }
- responseChannel.sendResponse(response, new BasicEntityDetails(-1, null), context);
- responseSent = true;
+ responseChannel.sendResponse(response, new BasicEntityDetails(-1, null), context);
+ responseSent = true;
- final InputStream inputStream = new QueueInputStream(inbound);
- final OutputStream outputStream = new QueueOutputStream(outbound);
- final WebSocketSession session = new WebSocketSession(
- config, inputStream, outputStream, null, null, negotiation.getExtensions());
+ final InputStream inputStream = new QueueInputStream(inbound);
+ final OutputStream outputStream = new QueueOutputStream();
+ final WebSocketSession session = new WebSocketSession(
+ config, inputStream, outputStream, null, null, negotiation.getExtensions());
- executor.execute(() -> {
- try {
- handler.onOpen(session);
- new WebSocketServerProcessor(session, handler, config.getMaxMessageSize()).process();
- } catch (final WebSocketProtocolException ex) {
- handler.onError(session, ex);
+ executor.execute(() -> {
try {
- session.close(ex.closeCode, ex.getMessage());
- } catch (final IOException ignore) {
- // ignore
- }
- } catch (final Exception ex) {
- handler.onError(session, ex);
- try {
- session.close(WebSocketCloseStatus.INTERNAL_ERROR.getCode(), "WebSocket error");
- } catch (final IOException ignore) {
- // ignore
- }
- } finally {
- shutdown = true;
- outbound.offer(END_OUTBOUND);
- inbound.offer(END_INBOUND);
-
- final DataStreamChannel channel = dataChannel;
- if (channel != null) {
- channel.requestOutput();
+ handler.onOpen(session);
+ new WebSocketServerProcessor(session, handler, config.getMaxMessageSize()).process();
+ } catch (final WebSocketProtocolException ex) {
+ handler.onError(session, ex);
+ try {
+ session.close(ex.closeCode, ex.getMessage());
+ } catch (final IOException ignore) {
+ // ignore
+ }
+ } catch (final WebSocketException ex) {
+ handler.onError(session, ex);
+ try {
+ session.close(WebSocketCloseStatus.PROTOCOL_ERROR.getCode(), ex.getMessage());
+ } catch (final IOException ignore) {
+ // ignore
+ }
+ } catch (final Exception ex) {
+ handler.onError(session, ex);
+ try {
+ session.close(WebSocketCloseStatus.INTERNAL_ERROR.getCode(), "WebSocket error");
+ } catch (final IOException ignore) {
+ // ignore
+ }
+ } finally {
+ // Tear the stream down first so a failing extension close cannot leave the
+ // HTTP/2 stream without an END_STREAM; release the extensions last, always.
+ try {
+ shutdown = true;
+ abortOutboundBudget();
+ enqueueSentinel(END_OUTBOUND);
+ inbound.offer(END_INBOUND);
+
+ final DataStreamChannel channel = dataChannel;
+ if (channel != null) {
+ channel.requestOutput();
+ }
+ } finally {
+ negotiation.close();
+ }
}
+ });
+ } catch (final Exception primary) {
+ // The task was never handed off to the worker (sendResponse failed or the executor
+ // rejected it); release the negotiated extensions here, preserving the original failure.
+ try {
+ negotiation.close();
+ } catch (final RuntimeException closeEx) {
+ primary.addSuppressed(closeEx);
}
- });
+ throw primary;
+ }
}
@Override
public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
- capacityChannel.update(Integer.MAX_VALUE);
+ this.capacityChannel = capacityChannel;
+ // Advertise a bounded window once; further credit is replenished as the worker thread
+ // drains buffered bytes, so inbound memory stays bounded even if the worker stalls.
+ if (initialCreditGranted.compareAndSet(false, true)) {
+ capacityChannel.update(INBOUND_WINDOW);
+ }
+ }
+
+ private void replenishInbound(final int n) {
+ final CapacityChannel channel = capacityChannel;
+ if (channel != null && n > 0) {
+ try {
+ channel.update(n);
+ } catch (final IOException ignore) {
+ // channel already gone; nothing to replenish
+ }
+ }
}
@Override
@@ -271,6 +354,7 @@ public void produce(final DataStreamChannel channel) throws IOException {
} finally {
outLock.unlock();
}
+ releaseOutboundBudget(buf.capacity());
continue;
}
@@ -290,12 +374,19 @@ public void produce(final DataStreamChannel channel) throws IOException {
} finally {
outLock.unlock();
}
+ // The chunk has been fully written; release its bytes so a back-pressured writer resumes.
+ releaseOutboundBudget(buf.capacity());
}
}
@Override
public void failed(final Exception cause) {
shutdown = true;
+ // Release any writer blocked on the outbound byte budget so it observes the shutdown, then
+ // clear the queues and post the sentinels so the worker's blocking reads/writes unwind.
+ abortOutboundBudget();
+ outbound.clear();
+ inbound.clear();
outbound.offer(END_OUTBOUND);
inbound.offer(END_INBOUND);
@@ -308,8 +399,10 @@ public void failed(final Exception cause) {
@Override
public void releaseResources() {
shutdown = true;
+ abortOutboundBudget();
outbound.clear();
inbound.clear();
+ inbound.offer(END_INBOUND);
outLock.lock();
try {
currentOutbound = null;
@@ -318,7 +411,97 @@ public void releaseResources() {
}
}
- private static final class QueueInputStream extends InputStream {
+ private void enqueueOutbound(final byte[] data, final int off, final int len) throws IOException {
+ // Once the exchange has failed or been released nothing drains the queue any more, so a
+ // blocking reservation would wedge the worker thread; fail the write instead.
+ if (shutdown) {
+ throw new IOException("WebSocket stream already terminated");
+ }
+ // Split the write into budget-sized chunks. Each chunk reserves its bytes against the
+ // outbound budget and blocks the worker until the reactor has drained enough previously
+ // queued bytes, so a single large write cannot buffer an unbounded amount of memory.
+ int position = off;
+ int remaining = len;
+ while (remaining > 0) {
+ final int n = Math.min(remaining, outboundChunkSize);
+ reserveOutboundBudget(n);
+ final byte[] chunk = new byte[n];
+ System.arraycopy(data, position, chunk, 0, n);
+ outbound.add(ByteBuffer.wrap(chunk));
+ // Signal after every chunk, before the next reservation can block: a write larger than
+ // the budget must let the reactor drain and free capacity, otherwise the worker would
+ // park with the reactor never told there is output to drain (deadlock).
+ requestOutput();
+ position += n;
+ remaining -= n;
+ }
+ }
+
+ /**
+ * Enqueues a zero-length control sentinel (END_OUTBOUND). Sentinels carry no payload and never
+ * block on the byte budget, so the HTTP/2 stream can always be terminated.
+ */
+ private void enqueueSentinel(final ByteBuffer sentinel) {
+ outbound.add(sentinel);
+ requestOutput();
+ }
+
+ private void requestOutput() {
+ final DataStreamChannel channel = dataChannel;
+ if (responseSent && channel != null) {
+ channel.requestOutput();
+ }
+ }
+
+ private void reserveOutboundBudget(final int n) throws IOException {
+ outboundBudgetLock.lock();
+ try {
+ // Wait until this chunk fits within the budget. A single chunk is always admitted when
+ // the queue is empty (pendingOutboundBytes == 0) so an oversized chunk cannot deadlock.
+ while (!shutdown && pendingOutboundBytes > 0 && pendingOutboundBytes + n > outboundBudgetBytes) {
+ try {
+ outboundBudgetAvailable.await();
+ } catch (final InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while waiting for outbound capacity", ex);
+ }
+ }
+ if (shutdown) {
+ throw new IOException("WebSocket stream already terminated");
+ }
+ pendingOutboundBytes += n;
+ } finally {
+ outboundBudgetLock.unlock();
+ }
+ }
+
+ private void releaseOutboundBudget(final int n) {
+ if (n <= 0) {
+ return;
+ }
+ outboundBudgetLock.lock();
+ try {
+ pendingOutboundBytes -= n;
+ if (pendingOutboundBytes < 0) {
+ pendingOutboundBytes = 0;
+ }
+ outboundBudgetAvailable.signalAll();
+ } finally {
+ outboundBudgetLock.unlock();
+ }
+ }
+
+ private void abortOutboundBudget() {
+ outboundBudgetLock.lock();
+ try {
+ pendingOutboundBytes = 0;
+ outboundBudgetAvailable.signalAll();
+ } finally {
+ outboundBudgetLock.unlock();
+ }
+ }
+
+ private final class QueueInputStream extends InputStream {
private final BlockingQueue queue;
private byte[] current;
@@ -331,6 +514,11 @@ private static final class QueueInputStream extends InputStream {
@Override
public int read() throws IOException {
if (current == null || pos >= current.length) {
+ // The previous chunk is fully consumed: replenish inbound flow-control credit for
+ // its bytes so the peer may send more, keeping buffered memory bounded.
+ if (current != null && current != END_INBOUND && current.length > 0) {
+ replenishInbound(current.length);
+ }
try {
current = queue.take();
} catch (final InterruptedException ex) {
@@ -348,40 +536,31 @@ public int read() throws IOException {
private final class QueueOutputStream extends OutputStream {
- private final BlockingQueue queue;
-
- QueueOutputStream(final BlockingQueue queue) {
- this.queue = queue;
- }
-
@Override
public void write(final int b) throws IOException {
- queue.offer(ByteBuffer.wrap(new byte[]{(byte) b}));
- requestOutput();
+ // enqueueOutbound signals the reactor after the chunk is queued.
+ enqueueOutbound(new byte[]{(byte) b}, 0, 1);
}
@Override
public void write(final byte[] b, final int off, final int len) throws IOException {
+ // Validate the range before reserving budget: an invalid range must fail cleanly rather
+ // than reserve bytes and then throw in arraycopy, which would leak the reservation and
+ // wedge later writes.
+ if ((off | len) < 0 || len > b.length - off) {
+ throw new IndexOutOfBoundsException();
+ }
if (len == 0) {
return;
}
- final byte[] copy = new byte[len];
- System.arraycopy(b, off, copy, 0, len);
- queue.offer(ByteBuffer.wrap(copy));
- requestOutput();
+ enqueueOutbound(b, off, len);
}
@Override
- public void close() {
- queue.offer(END_OUTBOUND);
- requestOutput();
- }
-
- private void requestOutput() {
- final DataStreamChannel channel = dataChannel;
- if (responseSent && channel != null) {
- channel.requestOutput();
- }
+ public void close() throws IOException {
+ // The END_OUTBOUND sentinel must always terminate the HTTP/2 stream, so it bypasses the
+ // byte budget even after a shutdown has cleared the queue.
+ enqueueSentinel(END_OUTBOUND);
}
}
}
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessor.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessor.java
index 97e1be1bf9..066de09319 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessor.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessor.java
@@ -53,6 +53,10 @@ class WebSocketServerProcessor {
}
void process() throws IOException {
+ readLoop();
+ }
+
+ private void readLoop() throws IOException {
ByteBuffer continuationBuf = null;
WebSocketFrameType continuationType = null;
while (true) {
@@ -85,6 +89,9 @@ void process() throws IOException {
return;
case TEXT:
case BINARY:
+ if (continuationBuf != null) {
+ throw new WebSocketProtocolException(1002, "Data frame during fragmented message");
+ }
if (frame.isFin()) {
dispatchMessage(type, frame.getPayload());
} else {
diff --git a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandler.java b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandler.java
index 4a7adb556f..62edae6040 100644
--- a/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandler.java
+++ b/httpclient5-websocket/src/main/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandler.java
@@ -81,7 +81,13 @@ public void handle(
return;
}
if (!WebSocketHandshake.isWebSocketUpgrade(request)) {
- trigger.submitResponse(new BasicClassicHttpResponse(HttpStatus.SC_UPGRADE_REQUIRED));
+ final ClassicHttpResponse rejection = new BasicClassicHttpResponse(HttpStatus.SC_UPGRADE_REQUIRED);
+ // RFC 9110 section 15.5.22 requires a 426 response to carry an Upgrade header field, and
+ // section 7.8 the matching Connection: Upgrade option; RFC 6455 section 4.4 adds the version.
+ rejection.addHeader(HttpHeaders.CONNECTION, HeaderElements.UPGRADE);
+ rejection.addHeader(HttpHeaders.UPGRADE, "websocket");
+ rejection.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13");
+ trigger.submitResponse(rejection);
return;
}
final WebSocketHandler handler = supplier.get();
@@ -96,42 +102,47 @@ public void handle(
response.addHeader(HttpHeaders.CONNECTION, HeaderElements.UPGRADE);
response.addHeader(HttpHeaders.UPGRADE, "websocket");
response.addHeader(WebSocketConstants.SEC_WEBSOCKET_ACCEPT, accept);
- final WebSocketExtensionNegotiation negotiation = extensionRegistry.negotiate(
- WebSocketExtensions.parse(request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS)),
- true);
- final String extensionsHeader = negotiation.formatResponseHeader();
- if (extensionsHeader != null) {
- response.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, extensionsHeader);
- }
- final List offeredProtocols = WebSocketHandshake.parseSubprotocols(
- request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL));
- final String protocol = handler.selectSubprotocol(offeredProtocols);
- if (protocol != null) {
- response.addHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL, protocol);
- }
- trigger.submitResponse(response);
- final InputStream inputStream = connection.getSocketInputStream();
- final OutputStream outputStream = connection.getSocketOutputStream();
- if (inputStream == null || outputStream == null) {
- connection.close();
- return;
- }
- final WebSocketSession session = new WebSocketSession(config, inputStream, outputStream,
- connection.getRemoteAddress(), connection.getLocalAddress(), negotiation.getExtensions());
- try {
- handler.onOpen(session);
- new WebSocketServerProcessor(session, handler, config.getMaxMessageSize()).process();
- } catch (final WebSocketProtocolException ex) {
- handler.onError(session, ex);
- session.close(ex.closeCode, ex.getMessage());
- } catch (final WebSocketException ex) {
- handler.onError(session, ex);
- session.close(WebSocketCloseStatus.PROTOCOL_ERROR.getCode(), ex.getMessage());
- } catch (final Exception ex) {
- handler.onError(session, ex);
- session.close(WebSocketCloseStatus.INTERNAL_ERROR.getCode(), "WebSocket error");
- } finally {
- connection.close();
+ // Try-with-resources transfers ownership of the negotiated extensions to this scope: they
+ // are released on every exit path (a failed submitResponse, unavailable streams, a throwing
+ // onOpen and normal or exceptional processor termination), and a close failure is attached
+ // as a suppressed exception rather than masking the primary failure.
+ try (final WebSocketExtensionNegotiation negotiation = extensionRegistry.negotiate(
+ WebSocketExtensions.parse(request.headerIterator(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS)),
+ true)) {
+ final String extensionsHeader = negotiation.formatResponseHeader();
+ if (extensionsHeader != null) {
+ response.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, extensionsHeader);
+ }
+ final List offeredProtocols = WebSocketHandshake.parseSubprotocols(
+ request.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL));
+ final String protocol = handler.selectSubprotocol(offeredProtocols);
+ if (protocol != null) {
+ response.addHeader(WebSocketConstants.SEC_WEBSOCKET_PROTOCOL, protocol);
+ }
+ trigger.submitResponse(response);
+ final InputStream inputStream = connection.getSocketInputStream();
+ final OutputStream outputStream = connection.getSocketOutputStream();
+ if (inputStream == null || outputStream == null) {
+ connection.close();
+ return;
+ }
+ final WebSocketSession session = new WebSocketSession(config, inputStream, outputStream,
+ connection.getRemoteAddress(), connection.getLocalAddress(), negotiation.getExtensions());
+ try {
+ handler.onOpen(session);
+ new WebSocketServerProcessor(session, handler, config.getMaxMessageSize()).process();
+ } catch (final WebSocketProtocolException ex) {
+ handler.onError(session, ex);
+ session.close(ex.closeCode, ex.getMessage());
+ } catch (final WebSocketException ex) {
+ handler.onError(session, ex);
+ session.close(WebSocketCloseStatus.PROTOCOL_ERROR.getCode(), ex.getMessage());
+ } catch (final Exception ex) {
+ handler.onError(session, ex);
+ session.close(WebSocketCloseStatus.INTERNAL_ERROR.getCode(), "WebSocket error");
+ } finally {
+ connection.close();
+ }
}
}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocolExtensionTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocolExtensionTest.java
index 153a632cdd..aa5359eca1 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocolExtensionTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/client/impl/protocol/Http1UpgradeProtocolExtensionTest.java
@@ -29,8 +29,10 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.hc.client5.http.websocket.api.WebSocketClientConfig;
+import org.apache.hc.core5.http.message.BasicHttpResponse;
import org.apache.hc.core5.websocket.extension.ExtensionChain;
import org.apache.hc.core5.websocket.frame.FrameHeaderBits;
import org.junit.jupiter.api.Test;
@@ -47,36 +49,102 @@ void pmce_rejectedWhenDisabled() {
}
@Test
- void pmce_rejectedWhenParametersNotOffered() {
+ void pmce_rejectedWhenClientWindowBitsNotOffered() {
final WebSocketClientConfig cfg = WebSocketClientConfig.custom()
.offerServerNoContextTakeover(false)
.offerClientNoContextTakeover(false)
- .offerClientMaxWindowBits(null)
.offerServerMaxWindowBits(null)
.build();
- assertThrows(IllegalStateException.class, () ->
- Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate; server_no_context_takeover"));
+ // The client never offers client_max_window_bits, so a response that echoes it must be rejected.
assertThrows(IllegalStateException.class, () ->
Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate; client_max_window_bits=15"));
}
@Test
- void pmce_rejectedOnUnknownOrDuplicate() {
+ void pmce_acceptsUninvitedNoContextTakeover() {
+ // RFC 7692 sections 7.1.1.1 and 7.1.1.2: the server may include either
+ // no-context-takeover parameter in the response even when the offer did not.
+ final WebSocketClientConfig cfg = WebSocketClientConfig.custom()
+ .offerServerNoContextTakeover(false)
+ .offerClientNoContextTakeover(false)
+ .build();
+ final ExtensionChain chain = Http1UpgradeProtocol.buildExtensionChain(cfg,
+ "permessage-deflate; server_no_context_takeover; client_no_context_takeover");
+ assertFalse(chain.isEmpty());
+ }
+
+ @Test
+ void pmce_acceptsUninvitedServerMaxWindowBits() {
+ // RFC 7692 section 7.1.2.1: the server may constrain its own window uninvited.
+ final WebSocketClientConfig cfg = WebSocketClientConfig.custom()
+ .offerServerMaxWindowBits(null)
+ .build();
+ final ExtensionChain chain = Http1UpgradeProtocol.buildExtensionChain(cfg,
+ "permessage-deflate; server_max_window_bits=12");
+ assertFalse(chain.isEmpty());
+ }
+
+ @Test
+ void headerValue_combinesRepeatedHeaders() {
+ // RFC 6455 section 9.1: Sec-WebSocket-Extensions may be split across multiple
+ // header fields; all instances must be taken into account.
+ final BasicHttpResponse response = new BasicHttpResponse(101);
+ response.addHeader("Sec-WebSocket-Extensions", "permessage-deflate");
+ response.addHeader("Sec-WebSocket-Extensions", "x-unknown");
+ assertEquals("permessage-deflate, x-unknown",
+ Http1UpgradeProtocol.headerValue(response, "Sec-WebSocket-Extensions"));
+ // The combined value must flow into validation: the unknown extension is rejected.
final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build();
assertThrows(IllegalStateException.class, () ->
- Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate; unknown=1"));
+ Http1UpgradeProtocol.buildExtensionChain(cfg,
+ Http1UpgradeProtocol.headerValue(response, "Sec-WebSocket-Extensions")));
+ }
+
+ @Test
+ void pmce_rejectsDuplicateServerNoContextTakeover() {
+ // RFC 7692 section 7.1: a parameter name must not appear more than once in a response.
+ final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build();
+ final IllegalStateException ex = assertThrows(IllegalStateException.class, () ->
+ Http1UpgradeProtocol.buildExtensionChain(cfg,
+ "permessage-deflate; server_no_context_takeover; server_no_context_takeover"));
+ assertTrue(ex.getMessage().startsWith("Duplicate permessage-deflate parameter"), ex.getMessage());
+ }
+
+ @Test
+ void pmce_rejectsDuplicateClientNoContextTakeover() {
+ final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build();
+ final IllegalStateException ex = assertThrows(IllegalStateException.class, () ->
+ Http1UpgradeProtocol.buildExtensionChain(cfg,
+ "permessage-deflate; client_no_context_takeover; client_no_context_takeover"));
+ assertTrue(ex.getMessage().startsWith("Duplicate permessage-deflate parameter"), ex.getMessage());
+ }
+
+ @Test
+ void pmce_rejectsDuplicateServerMaxWindowBits() {
+ final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build();
+ final IllegalStateException ex = assertThrows(IllegalStateException.class, () ->
+ Http1UpgradeProtocol.buildExtensionChain(cfg,
+ "permessage-deflate; server_max_window_bits=12; server_max_window_bits=10"));
+ assertTrue(ex.getMessage().startsWith("Duplicate permessage-deflate parameter"), ex.getMessage());
+ }
+
+ @Test
+ void pmce_rejectsDuplicateClientMaxWindowBits() {
+ // client_max_window_bits can no longer be offered, so the first occurrence is already
+ // rejected as "not offered"; a syntactically invalid duplicated response must still fail.
+ final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build();
assertThrows(IllegalStateException.class, () ->
- Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate, permessage-deflate"));
+ Http1UpgradeProtocol.buildExtensionChain(cfg,
+ "permessage-deflate; client_max_window_bits=15; client_max_window_bits=12"));
}
@Test
- void pmce_rejectedOnUnsupportedClientWindowBits() {
- final WebSocketClientConfig cfg = WebSocketClientConfig.custom()
- .offerClientMaxWindowBits(15)
- .offerServerMaxWindowBits(15)
- .build();
+ void pmce_rejectedOnUnknownOrDuplicate() {
+ final WebSocketClientConfig cfg = WebSocketClientConfig.custom().build();
assertThrows(IllegalStateException.class, () ->
- Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate; client_max_window_bits=12"));
+ Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate; unknown=1"));
+ assertThrows(IllegalStateException.class, () ->
+ Http1UpgradeProtocol.buildExtensionChain(cfg, "permessage-deflate, permessage-deflate"));
}
@Test
@@ -93,13 +161,12 @@ void pmce_acceptsServerWindowBitsBelow15() {
@Test
void pmce_validNegotiation_buildsChain() {
final WebSocketClientConfig cfg = WebSocketClientConfig.custom()
- .offerClientMaxWindowBits(15)
.offerServerMaxWindowBits(15)
.offerClientNoContextTakeover(true)
.offerServerNoContextTakeover(true)
.build();
final ExtensionChain chain = Http1UpgradeProtocol.buildExtensionChain(cfg,
- "permessage-deflate; client_no_context_takeover; server_no_context_takeover; client_max_window_bits=15");
+ "permessage-deflate; client_no_context_takeover; server_no_context_takeover; server_max_window_bits=15");
assertFalse(chain.isEmpty());
assertEquals(FrameHeaderBits.RSV1, chain.rsvMask());
}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/example/WebSocketEchoClient.java b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/example/WebSocketEchoClient.java
index e21526690f..1f131eb846 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/example/WebSocketEchoClient.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/example/WebSocketEchoClient.java
@@ -54,7 +54,6 @@
*
* - Per-message deflate enabled
* - Server and client context takeover disabled (stateless compression)
- * - Client maximum window bits set to 15
*
*
* The lifecycle follows these steps:
@@ -89,7 +88,6 @@ public static void main(final String[] args) throws Exception {
.enablePerMessageDeflate(true)
.offerServerNoContextTakeover(true)
.offerClientNoContextTakeover(true)
- .offerClientMaxWindowBits(15)
.setCloseWaitTimeout(Timeout.ofSeconds(2))
.build();
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransportTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransportTest.java
new file mode 100644
index 0000000000..e65826f459
--- /dev/null
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/DataStreamChannelTransportTest.java
@@ -0,0 +1,93 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * .
+ *
+ */
+package org.apache.hc.client5.http.websocket.transport;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.CancelledKeyException;
+import java.util.List;
+
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.nio.DataStreamChannel;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies the transport's behaviour when the channel's selection key has been cancelled by a
+ * concurrent shutdown: {@code requestOutput()} tolerates it as a best-effort nudge, while
+ * {@code write(ByteBuffer)} surfaces it as an {@link IOException} so the engine takes its close
+ * path instead of misreading it as transient back-pressure.
+ */
+class DataStreamChannelTransportTest {
+
+ private static final class CancelledChannel implements DataStreamChannel {
+
+ @Override
+ public int write(final ByteBuffer src) {
+ throw new CancelledKeyException();
+ }
+
+ @Override
+ public void requestOutput() {
+ throw new CancelledKeyException();
+ }
+
+ @Override
+ public void endStream() {
+ }
+
+ @Override
+ public void endStream(final List trailers) {
+ }
+ }
+
+ @Test
+ void requestOutputSwallowsCancelledKeyDuringShutdown() {
+ final DataStreamChannelTransport transport = new DataStreamChannelTransport();
+ transport.setChannel(new CancelledChannel());
+ assertDoesNotThrow(transport::requestOutput);
+ }
+
+ @Test
+ void writeThrowsOnCancelledChannel() {
+ final DataStreamChannelTransport transport = new DataStreamChannelTransport();
+ transport.setChannel(new CancelledChannel());
+ // A cancelled selection key is terminal; write() must surface an IOException rather than
+ // report written == 0, which the engine would misread as transient back-pressure.
+ final IOException ex = assertThrows(IOException.class,
+ () -> transport.write(ByteBuffer.allocate(4)));
+ assertTrue(ex.getCause() instanceof CancelledKeyException);
+ }
+
+ @Test
+ void requestOutputIsNoOpWithoutChannel() {
+ assertDoesNotThrow(new DataStreamChannelTransport()::requestOutput);
+ }
+}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/WebSocketInboundRfcGapTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/WebSocketInboundRfcGapTest.java
new file mode 100644
index 0000000000..2d9f5427d6
--- /dev/null
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/client5/http/websocket/transport/WebSocketInboundRfcGapTest.java
@@ -0,0 +1,127 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * .
+ *
+ */
+package org.apache.hc.client5.http.websocket.transport;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.hc.client5.http.websocket.api.WebSocketClientConfig;
+import org.apache.hc.client5.http.websocket.api.WebSocketListener;
+import org.apache.hc.core5.websocket.extension.ExtensionChain;
+import org.apache.hc.core5.websocket.extension.PerMessageDeflate;
+import org.apache.hc.core5.websocket.frame.FrameOpcode;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Reproduces two inbound RFC 6455 conformance gaps in the live NIO client engine that
+ * {@link WebSocketInboundRfcTest} does not cover: a control frame carrying RSV1 while an extension
+ * is negotiated (RFC 6455 ยง5.2), and delivery of a truncated over-limit message to the listener
+ * after the connection has already been failed with 1009 (RFC 6455 ยง5.4 / ยง7.1.7).
+ */
+class WebSocketInboundRfcGapTest {
+
+ @Test
+ void rsv1OnControlFrameWithNegotiatedExtension_closesWith1002() {
+ final CaptureListener listener = new CaptureListener();
+ // permessage-deflate negotiated: the engine owns RSV1 for DATA frames, but a control frame
+ // is never compressed, so RSV1 on a PING has no defined meaning and MUST fail the connection.
+ final ExtensionChain chain = new ExtensionChain();
+ chain.add(new PerMessageDeflate(true, false, false, null, null));
+ final WebSocketSessionEngine engine = newEngine(listener, 0L, chain);
+
+ engine.onData(controlFrameWithRsv1(FrameOpcode.PING));
+
+ assertEquals(1002, listener.closeCode.get(),
+ "RSV1 on a control frame must fail the connection with 1002");
+ }
+
+ @Test
+ void overLimitFinalFragment_doesNotDeliverTruncatedMessageAfter1009() {
+ final CaptureListener listener = new CaptureListener();
+ final WebSocketSessionEngine engine = newEngine(listener, 4L, null);
+
+ engine.onData(unmaskedFrame(FrameOpcode.BINARY, false, false, new byte[]{1, 2, 3}));
+ engine.onData(unmaskedFrame(FrameOpcode.CONT, true, false, new byte[]{4, 5}));
+
+ assertEquals(1009, listener.closeCode.get(), "over-limit message must fail with 1009");
+ assertFalse(listener.messageDelivered.get(),
+ "a message must not be delivered to the listener after the connection is failed");
+ }
+
+ private static WebSocketSessionEngine newEngine(final CaptureListener listener,
+ final long maxMessageSize,
+ final ExtensionChain chain) {
+ final WebSocketClientConfig.Builder cfg = WebSocketClientConfig.custom()
+ .enablePerMessageDeflate(false);
+ if (maxMessageSize > 0) {
+ cfg.setMaxMessageSize(maxMessageSize);
+ }
+ return new WebSocketSessionEngine(new StubTransport(), listener, cfg.build(), chain, null);
+ }
+
+ private static ByteBuffer unmaskedFrame(final int opcode, final boolean fin, final boolean rsv1,
+ final byte[] payload) {
+ final int len = payload != null ? payload.length : 0;
+ final ByteBuffer buf = ByteBuffer.allocate(2 + len);
+ buf.put((byte) ((fin ? 0x80 : 0x00) | (rsv1 ? 0x40 : 0x00) | (opcode & 0x0F)));
+ buf.put((byte) len);
+ if (len > 0) {
+ buf.put(payload);
+ }
+ buf.flip();
+ return buf;
+ }
+
+ private static ByteBuffer controlFrameWithRsv1(final int opcode) {
+ return unmaskedFrame(opcode, true, true, new byte[0]);
+ }
+
+ private static final class CaptureListener implements WebSocketListener {
+ private final AtomicInteger closeCode = new AtomicInteger(-1);
+ private final AtomicBoolean messageDelivered = new AtomicBoolean(false);
+
+ @Override
+ public void onText(final CharBuffer data, final boolean last) {
+ messageDelivered.set(true);
+ }
+
+ @Override
+ public void onBinary(final ByteBuffer data, final boolean last) {
+ messageDelivered.set(true);
+ }
+
+ @Override
+ public void onClose(final int statusCode, final String reason) {
+ closeCode.compareAndSet(-1, statusCode);
+ }
+ }
+}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactoryTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactoryTest.java
index 964ed201ce..d034ba2473 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactoryTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionFactoryTest.java
@@ -32,8 +32,10 @@
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
+import org.apache.hc.core5.http.message.BasicHeader;
import org.junit.jupiter.api.Test;
class PerMessageDeflateExtensionFactoryTest {
@@ -69,4 +71,83 @@ void echoesNegotiatedWindowBitsWhenRequested() {
assertEquals("15", response.getParameters().get("client_max_window_bits"));
assertEquals("15", response.getParameters().get("server_max_window_bits"));
}
+
+ @Test
+ void rejectsServerMaxWindowBitsWithoutValue() {
+ assertNull(negotiateServer("permessage-deflate; server_max_window_bits"),
+ "server_max_window_bits must carry a value");
+ }
+
+ @Test
+ void rejectsServerMaxWindowBitsNonNumeric() {
+ final Map params = new LinkedHashMap<>();
+ params.put("server_max_window_bits", "abc");
+ assertNull(create(params), "a non-numeric server_max_window_bits must be rejected");
+ }
+
+ @Test
+ void rejectsServerMaxWindowBitsBelowRange() {
+ final Map params = new LinkedHashMap<>();
+ params.put("server_max_window_bits", "7");
+ assertNull(create(params), "server_max_window_bits below 8 must be rejected");
+ }
+
+ @Test
+ void rejectsClientMaxWindowBitsNonNumeric() {
+ final Map params = new LinkedHashMap<>();
+ params.put("client_max_window_bits", "abc");
+ assertNull(create(params), "a present but non-numeric client_max_window_bits must be rejected");
+ }
+
+ @Test
+ void acceptsValuelessClientMaxWindowBits() {
+ assertNotNull(negotiateServer("permessage-deflate; client_max_window_bits"),
+ "a valueless client_max_window_bits merely advertises support and is accepted");
+ }
+
+ @Test
+ void rejectsUnknownParameter() {
+ final Map params = new LinkedHashMap<>();
+ params.put("unknown", "1");
+ assertNull(create(params), "an unknown parameter must be rejected");
+ }
+
+ @Test
+ void rejectsValuedNoContextTakeover() {
+ final Map params = new LinkedHashMap<>();
+ params.put("server_no_context_takeover", "1");
+ assertNull(create(params), "server_no_context_takeover must not carry a value");
+ }
+
+ @Test
+ void acceptsValuelessNoContextTakeover() {
+ assertNotNull(negotiateServer("permessage-deflate; server_no_context_takeover; client_no_context_takeover"),
+ "valueless no-context-takeover parameters are accepted");
+ }
+
+ @Test
+ void rejectsLeadingZeroWindowBits() {
+ final Map params = new LinkedHashMap<>();
+ params.put("server_max_window_bits", "015");
+ assertNull(create(params), "a leading-zero window-bits value must be rejected");
+ }
+
+ @Test
+ void rejectsPlusSignedWindowBits() {
+ final Map params = new LinkedHashMap<>();
+ params.put("server_max_window_bits", "+15");
+ assertNull(create(params), "a signed window-bits value must be rejected");
+ }
+
+ private static WebSocketExtension create(final Map params) {
+ return new PerMessageDeflateExtensionFactory().create(
+ new WebSocketExtensionData("permessage-deflate", params), true);
+ }
+
+ private static WebSocketExtension negotiateServer(final String headerValue) {
+ final List offers = WebSocketExtensions.parse(
+ new BasicHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, headerValue));
+ assertEquals(1, offers.size(), "the header must parse to a single offer");
+ return new PerMessageDeflateExtensionFactory().create(offers.get(0), true);
+ }
}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionTest.java
index 20fd8b019b..b561fbd700 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/PerMessageDeflateExtensionTest.java
@@ -96,6 +96,73 @@ void decodeZeroLimitMeansUnlimited() throws Exception {
assertArrayEquals(plain, toBytes(out));
}
+ @Test
+ void fragmentedEncodeRoundTripsThroughDecode() throws Exception {
+ // Encode a message as three frames (fin=false, false, true), then decode them back.
+ // Non-final fragments must retain the 00 00 FF FF flush trailer so the reassembled
+ // DEFLATE stream stays valid (RFC 7692 section 7.2.1).
+ final PerMessageDeflateExtension enc = new PerMessageDeflateExtension();
+ final byte[] p1 = "The quick brown fox ".getBytes(StandardCharsets.UTF_8);
+ final byte[] p2 = "jumps over the lazy ".getBytes(StandardCharsets.UTF_8);
+ final byte[] p3 = "dog, and again the fox.".getBytes(StandardCharsets.UTF_8);
+
+ final ByteBuffer f1 = enc.encode(WebSocketFrameType.TEXT, false, ByteBuffer.wrap(p1));
+ final ByteBuffer f2 = enc.encode(WebSocketFrameType.CONTINUATION, false, ByteBuffer.wrap(p2));
+ final ByteBuffer f3 = enc.encode(WebSocketFrameType.CONTINUATION, true, ByteBuffer.wrap(p3));
+
+ final PerMessageDeflateExtension dec = new PerMessageDeflateExtension();
+ final ByteArrayOutputStream joined = new ByteArrayOutputStream();
+ joined.write(toBytes(dec.decode(WebSocketFrameType.TEXT, false, f1)));
+ joined.write(toBytes(dec.decode(WebSocketFrameType.CONTINUATION, false, f2)));
+ joined.write(toBytes(dec.decode(WebSocketFrameType.CONTINUATION, true, f3)));
+
+ final byte[] expected = new byte[p1.length + p2.length + p3.length];
+ System.arraycopy(p1, 0, expected, 0, p1.length);
+ System.arraycopy(p2, 0, expected, p1.length, p2.length);
+ System.arraycopy(p3, 0, expected, p1.length + p2.length, p3.length);
+ assertArrayEquals(expected, joined.toByteArray());
+ }
+
+ @Test
+ void encodesEmptyMessageAsSingleZeroOctet() throws Exception {
+ final PerMessageDeflateExtension enc = new PerMessageDeflateExtension();
+ final ByteBuffer encoded = enc.encode(WebSocketFrameType.BINARY, true, ByteBuffer.allocate(0));
+
+ // RFC 7692 section 7.2.3.6: an empty compressed message is the single octet 0x00.
+ assertArrayEquals(new byte[]{0x00}, toBytes(encoded));
+
+ final PerMessageDeflateExtension dec = new PerMessageDeflateExtension();
+ assertArrayEquals(new byte[0],
+ toBytes(dec.decode(WebSocketFrameType.BINARY, true, ByteBuffer.wrap(new byte[]{0x00}))));
+ }
+
+ @Test
+ void encodesEmptyMessageAsZeroOctetWithContextTakeover() throws Exception {
+ // The default extension keeps deflate/inflate context across messages.
+ final PerMessageDeflateExtension enc = new PerMessageDeflateExtension();
+ final PerMessageDeflateExtension dec = new PerMessageDeflateExtension();
+
+ final byte[] hello = "hello".getBytes(StandardCharsets.UTF_8);
+ assertArrayEquals(hello, toBytes(dec.decode(WebSocketFrameType.TEXT, true,
+ enc.encode(WebSocketFrameType.TEXT, true, ByteBuffer.wrap(hello)))));
+
+ // The deflater now holds context, so an empty message flushes to only 00 00 FF FF; it must
+ // still encode to the single octet 0x00 (RFC 7692 section 7.2.3.6), not to an empty payload.
+ final ByteBuffer emptyEncoded = enc.encode(WebSocketFrameType.TEXT, true, ByteBuffer.allocate(0));
+ assertArrayEquals(new byte[]{0x00}, toBytes(emptyEncoded));
+ assertArrayEquals(new byte[0], toBytes(dec.decode(WebSocketFrameType.TEXT, true, emptyEncoded)));
+ }
+
+ @Test
+ void closeReleasesNativeCodecs() throws Exception {
+ final PerMessageDeflateExtension ext = new PerMessageDeflateExtension();
+ ext.encode(WebSocketFrameType.TEXT, true, ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)));
+ ext.close();
+ // After close() the Deflater has been ended; reusing the extension must fail.
+ assertThrows(Exception.class,
+ () -> ext.encode(WebSocketFrameType.TEXT, true, ByteBuffer.wrap("again".getBytes(StandardCharsets.UTF_8))));
+ }
+
private static byte[] deflateWithSyncFlush(final byte[] input) {
final Deflater deflater = new Deflater(Deflater.DEFAULT_COMPRESSION, true);
deflater.setInput(input);
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiationTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiationTest.java
index c1c5a415d2..6697567310 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiationTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionNegotiationTest.java
@@ -30,6 +30,7 @@
import static org.junit.jupiter.api.Assertions.assertNull;
import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
@@ -51,4 +52,27 @@ void formatsEmptyHeaderAsNull() {
Collections.emptyList(), Collections.emptyList());
assertNull(negotiation.formatResponseHeader());
}
+
+ @Test
+ void closeIsIdempotentAndClosesEachExtensionOnce() {
+ final AtomicInteger closed = new AtomicInteger();
+ final WebSocketExtension extension = new WebSocketExtension() {
+ @Override
+ public String getName() {
+ return "x-test";
+ }
+
+ @Override
+ public void close() {
+ closed.incrementAndGet();
+ }
+ };
+ final WebSocketExtensionNegotiation negotiation = new WebSocketExtensionNegotiation(
+ Collections.singletonList(extension), Collections.emptyList());
+
+ negotiation.close();
+ negotiation.close();
+
+ assertEquals(1, closed.get(), "each extension must be closed exactly once");
+ }
}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistryTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistryTest.java
index f3b6b7418b..16b21512ea 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistryTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionRegistryTest.java
@@ -28,9 +28,15 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import java.nio.ByteBuffer;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
@@ -102,4 +108,112 @@ void defaultRegistryContainsPerMessageDeflate() throws Exception {
true);
assertEquals(1, negotiation.getExtensions().size());
}
+
+ @Test
+ void duplicateOffersAreDeduplicated() throws Exception {
+ final WebSocketExtensionRegistry registry = WebSocketExtensionRegistry.createDefault();
+ // A client may offer the same extension more than once; accepting both would claim RSV1
+ // twice and later break the frame reader with an IllegalStateException.
+ final WebSocketExtensionNegotiation negotiation = registry.negotiate(
+ Arrays.asList(
+ new WebSocketExtensionData("permessage-deflate", Collections.emptyMap()),
+ new WebSocketExtensionData("permessage-deflate", Collections.emptyMap())),
+ true);
+ assertEquals(1, negotiation.getExtensions().size(), "duplicate offers must collapse to one extension");
+ assertEquals(1, negotiation.getResponseData().size(), "response must list the extension once");
+ }
+
+ @Test
+ void fallbackOfferIsAcceptedWhenTheFirstIsDeclined() throws Exception {
+ final WebSocketExtensionRegistry registry = WebSocketExtensionRegistry.createDefault();
+ final Map unsupported = new LinkedHashMap<>();
+ unsupported.put("server_max_window_bits", "12");
+ // The first offer asks for a 12-bit window the JDK Deflater cannot honour and is declined;
+ // the registry must move on and accept the client's plain fallback offer.
+ final WebSocketExtensionNegotiation negotiation = registry.negotiate(
+ Arrays.asList(
+ new WebSocketExtensionData("permessage-deflate", unsupported),
+ new WebSocketExtensionData("permessage-deflate", Collections.emptyMap())),
+ true);
+ assertEquals(1, negotiation.getExtensions().size(), "the fallback offer must be negotiated");
+ }
+
+ @Test
+ void negotiateClosesCreatedExtensionsWhenLaterFactoryThrows() {
+ final AtomicInteger closed = new AtomicInteger();
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(closingFactory("ext-a", closed, null))
+ .register(throwingFactory("ext-b", new WebSocketException("boom")));
+
+ final WebSocketException ex = assertThrows(WebSocketException.class, () ->
+ registry.negotiate(Arrays.asList(
+ new WebSocketExtensionData("ext-a", Collections.emptyMap()),
+ new WebSocketExtensionData("ext-b", Collections.emptyMap())),
+ true));
+
+ assertEquals("boom", ex.getMessage());
+ assertEquals(1, closed.get(), "the already-created extension must be closed exactly once");
+ }
+
+ @Test
+ void negotiateSuppressesCloseFailureOnOriginalException() {
+ final WebSocketException primary = new WebSocketException("primary");
+ final IllegalStateException closeFailure = new IllegalStateException("close failed");
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(closingFactory("ext-a", new AtomicInteger(), closeFailure))
+ .register(throwingFactory("ext-b", primary));
+
+ final WebSocketException ex = assertThrows(WebSocketException.class, () ->
+ registry.negotiate(Arrays.asList(
+ new WebSocketExtensionData("ext-a", Collections.emptyMap()),
+ new WebSocketExtensionData("ext-b", Collections.emptyMap())),
+ true));
+
+ assertSame(primary, ex, "the original negotiation failure must be preserved");
+ assertEquals(1, ex.getSuppressed().length, "the close failure must be attached as suppressed");
+ assertSame(closeFailure, ex.getSuppressed()[0]);
+ }
+
+ private static WebSocketExtensionFactory closingFactory(
+ final String name, final AtomicInteger closed, final RuntimeException closeFailure) {
+ return new WebSocketExtensionFactory() {
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) {
+ return new WebSocketExtension() {
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public void close() {
+ closed.incrementAndGet();
+ if (closeFailure != null) {
+ throw closeFailure;
+ }
+ }
+ };
+ }
+ };
+ }
+
+ private static WebSocketExtensionFactory throwingFactory(final String name, final WebSocketException failure) {
+ return new WebSocketExtensionFactory() {
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public WebSocketExtension create(final WebSocketExtensionData request, final boolean server)
+ throws WebSocketException {
+ throw failure;
+ }
+ };
+ }
}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionsTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionsTest.java
index 5b48cae8f8..03da54f692 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionsTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketExtensionsTest.java
@@ -27,10 +27,13 @@
package org.apache.hc.core5.websocket;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import org.apache.hc.core5.http.Header;
import org.apache.hc.core5.http.message.BasicHeader;
import org.junit.jupiter.api.Test;
@@ -49,4 +52,37 @@ void parsesExtensionsHeader() {
final Map params = data.get(1).getParameters();
assertEquals("1", params.get("bar"));
}
+
+ @Test
+ void dropsOfferWithDuplicateParameter() {
+ final BasicHeader header = new BasicHeader(
+ WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS,
+ "permessage-deflate; client_no_context_takeover; client_no_context_takeover");
+ assertTrue(WebSocketExtensions.parse(header).isEmpty(),
+ "a repeated parameter makes the offer invalid and it must be dropped, not collapsed");
+ }
+
+ @Test
+ void dropsOnlyTheOfferWithDuplicateParameter() {
+ final BasicHeader header = new BasicHeader(
+ WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS,
+ "permessage-deflate; server_max_window_bits=10; server_max_window_bits=12, foo; bar=1");
+ final List data = WebSocketExtensions.parse(header);
+ assertEquals(1, data.size(), "the malformed permessage-deflate offer is dropped, the valid one kept");
+ assertEquals("foo", data.get(0).getName());
+ }
+
+ @Test
+ void combinesMultipleExtensionHeaders() {
+ // RFC 6455: multiple Sec-WebSocket-Extensions fields are a single combined value, so a
+ // client's fallback offer in a later field must still be examined.
+ final List data = WebSocketExtensions.parse(Arrays.asList(
+ new BasicHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS,
+ "permessage-deflate; server_max_window_bits=12"),
+ new BasicHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS,
+ "permessage-deflate")).iterator());
+ assertEquals(2, data.size(), "offers from every Sec-WebSocket-Extensions field must be combined");
+ assertEquals("12", data.get(0).getParameters().get("server_max_window_bits"));
+ assertTrue(data.get(1).getParameters().isEmpty(), "the fallback offer must be preserved");
+ }
}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameWriterTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameWriterTest.java
index 0e80917d70..1f1fbf79cd 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameWriterTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/WebSocketFrameWriterTest.java
@@ -27,6 +27,7 @@
package org.apache.hc.core5.websocket;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.ByteArrayOutputStream;
@@ -89,6 +90,32 @@ void setsRsv1WhenExtensionUsesIt() throws Exception {
assertTrue((out[0] & 0x40) != 0, "RSV1 must be set");
}
+ @Test
+ void rejectsOversizedControlFrame() {
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ final WebSocketFrameWriter writer = new WebSocketFrameWriter(baos, Collections.emptyList());
+ assertThrows(IllegalArgumentException.class,
+ () -> writer.writePing(ByteBuffer.wrap(new byte[126])));
+ }
+
+ @Test
+ void closeReasonTruncatedToControlFrameLimit() throws Exception {
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ final WebSocketFrameWriter writer = new WebSocketFrameWriter(baos, Collections.emptyList());
+
+ final StringBuilder reason = new StringBuilder();
+ for (int i = 0; i < 200; i++) {
+ reason.append('x');
+ }
+ writer.writeClose(1000, reason.toString());
+
+ final byte[] out = baos.toByteArray();
+ assertEquals((byte) 0x88, out[0]); // FIN + CLOSE
+ final int payloadLen = out[1] & 0xFF; // no MASK bit on server frames
+ assertTrue(payloadLen <= 125, "CLOSE payload must fit the 125-byte control frame limit, was " + payloadLen);
+ assertEquals(out.length - 2, payloadLen, "single-byte length header, no extended length");
+ }
+
private static byte[] writeBinary(final byte[] payload, final List extensions) throws Exception {
final ByteArrayOutputStream baos = new ByteArrayOutputStream();
final WebSocketFrameWriter writer = new WebSocketFrameWriter(baos, extensions);
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/ExtensionChainTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/ExtensionChainTest.java
index a79abeb83d..8ef3ff3929 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/ExtensionChainTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/ExtensionChainTest.java
@@ -27,8 +27,13 @@
package org.apache.hc.core5.websocket.extension;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
@@ -47,4 +52,72 @@ void addAndUsePmce_decodeRoundTrip() throws Exception {
assertArrayEquals(data, back);
}
+
+ @Test
+ void encodeChainClosesAllEncodersEvenWhenOneThrows() {
+ final AtomicInteger closed = new AtomicInteger();
+ final RuntimeException boom = new IllegalStateException("boom");
+ final WebSocketExtensionChain.Encoder throwing = new WebSocketExtensionChain.Encoder() {
+ @Override
+ public WebSocketExtensionChain.Encoded encode(final byte[] data, final boolean first, final boolean fin) {
+ return new WebSocketExtensionChain.Encoded(data, first);
+ }
+
+ @Override
+ public void close() {
+ throw boom;
+ }
+ };
+ final WebSocketExtensionChain.Encoder counting = new WebSocketExtensionChain.Encoder() {
+ @Override
+ public WebSocketExtensionChain.Encoded encode(final byte[] data, final boolean first, final boolean fin) {
+ return new WebSocketExtensionChain.Encoded(data, first);
+ }
+
+ @Override
+ public void close() {
+ closed.incrementAndGet();
+ }
+ };
+ final ExtensionChain.EncodeChain chain = new ExtensionChain.EncodeChain(Arrays.asList(throwing, counting));
+
+ final RuntimeException ex = assertThrows(RuntimeException.class, chain::close);
+
+ assertSame(boom, ex);
+ assertEquals(1, closed.get(), "the second encoder must still be closed");
+ }
+
+ @Test
+ void decodeChainClosesAllDecodersEvenWhenOneThrows() {
+ final AtomicInteger closed = new AtomicInteger();
+ final RuntimeException boom = new IllegalStateException("boom");
+ final WebSocketExtensionChain.Decoder throwing = new WebSocketExtensionChain.Decoder() {
+ @Override
+ public byte[] decode(final byte[] payload) {
+ return payload;
+ }
+
+ @Override
+ public void close() {
+ throw boom;
+ }
+ };
+ final WebSocketExtensionChain.Decoder counting = new WebSocketExtensionChain.Decoder() {
+ @Override
+ public byte[] decode(final byte[] payload) {
+ return payload;
+ }
+
+ @Override
+ public void close() {
+ closed.incrementAndGet();
+ }
+ };
+ final ExtensionChain.DecodeChain chain = new ExtensionChain.DecodeChain(Arrays.asList(throwing, counting));
+
+ final RuntimeException ex = assertThrows(RuntimeException.class, chain::close);
+
+ assertSame(boom, ex);
+ assertEquals(1, closed.get(), "the second decoder must still be closed");
+ }
}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/MessageDeflateTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/MessageDeflateTest.java
index 284566dac4..b1235cf05d 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/MessageDeflateTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/MessageDeflateTest.java
@@ -84,6 +84,47 @@ void roundTrip_message() throws Exception {
assertArrayEquals(plain, roundTrip);
}
+ @Test
+ void encode_emptyMessage_isSingleZeroOctet() throws Exception {
+ final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null);
+ final WebSocketExtensionChain.Encoder enc = pmce.newEncoder();
+
+ final WebSocketExtensionChain.Encoded encoded = enc.encode(new byte[0], true, true);
+
+ // RFC 7692 section 7.2.3.6: an empty compressed message is the single octet 0x00.
+ assertArrayEquals(new byte[]{0x00}, encoded.payload);
+ assertTrue(encoded.setRsvOnFirst, "RSV1 on the first (and only) frame");
+ assertArrayEquals(new byte[0], pmce.newDecoder().decode(encoded.payload),
+ "an empty compressed message must round-trip to empty");
+ }
+
+ @Test
+ void encode_emptyMessage_withContextTakeover_roundTrips() throws Exception {
+ // clientNoContextTakeover = false: the deflater keeps its context across messages.
+ final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null);
+ final WebSocketExtensionChain.Encoder enc = pmce.newEncoder();
+ final WebSocketExtensionChain.Decoder dec = pmce.newDecoder();
+
+ final byte[] hello = "hello".getBytes(StandardCharsets.UTF_8);
+ assertArrayEquals(hello, dec.decode(enc.encode(hello, true, true).payload));
+
+ final byte[] empty = enc.encode(new byte[0], true, true).payload;
+ assertArrayEquals(new byte[]{0x00}, empty, "empty message compresses to 0x00 with context takeover");
+ assertArrayEquals(new byte[0], dec.decode(empty));
+ }
+
+ @Test
+ void encode_emptyMessage_withNoContextTakeover_roundTrips() throws Exception {
+ // clientNoContextTakeover = true: the deflater is reset after each message.
+ final PerMessageDeflate pmce = new PerMessageDeflate(true, false, true, null, null);
+ final WebSocketExtensionChain.Encoder enc = pmce.newEncoder();
+ final WebSocketExtensionChain.Decoder dec = pmce.newDecoder();
+
+ final byte[] empty = enc.encode(new byte[0], true, true).payload;
+ assertArrayEquals(new byte[]{0x00}, empty, "empty message compresses to 0x00 with no-context-takeover");
+ assertArrayEquals(new byte[0], dec.decode(empty));
+ }
+
@Test
void decode_withinLimit_succeeds() throws Exception {
final PerMessageDeflate pmce = new PerMessageDeflate(true, true, true, null, null);
@@ -134,6 +175,28 @@ void decode_zeroLimitMeansUnlimited() throws Exception {
assertArrayEquals(plain, roundTrip);
}
+ @Test
+ void encoderCloseReleasesDeflater() {
+ final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null);
+ final WebSocketExtensionChain.Encoder enc = pmce.newEncoder();
+ enc.encode("hello".getBytes(StandardCharsets.UTF_8), true, true);
+ enc.close();
+ // After close() the Deflater has been ended; reusing the encoder must fail.
+ assertThrows(Exception.class,
+ () -> enc.encode("again".getBytes(StandardCharsets.UTF_8), true, true));
+ }
+
+ @Test
+ void decoderCloseReleasesInflater() throws Exception {
+ final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null);
+ final WebSocketExtensionChain.Encoder enc = pmce.newEncoder();
+ final WebSocketExtensionChain.Decoder dec = pmce.newDecoder();
+ final byte[] wire = enc.encode("hello".getBytes(StandardCharsets.UTF_8), true, true).payload;
+ dec.decode(wire);
+ dec.close();
+ assertThrows(Exception.class, () -> dec.decode(wire));
+ }
+
private static boolean endsWithTail(final byte[] b) {
if (b.length < 4) {
return false;
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/PerMessageDeflateFragmentedRoundTripTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/PerMessageDeflateFragmentedRoundTripTest.java
new file mode 100644
index 0000000000..cb7d69e33b
--- /dev/null
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/extension/PerMessageDeflateFragmentedRoundTripTest.java
@@ -0,0 +1,122 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * .
+ *
+ */
+package org.apache.hc.core5.websocket.extension;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.hc.core5.websocket.extension.WebSocketExtensionChain.Decoder;
+import org.apache.hc.core5.websocket.extension.WebSocketExtensionChain.Encoder;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Proves that a permessage-deflate message split across several WebSocket data frames survives a
+ * compress/decompress round trip (RFC 7692 ยง7.2). The existing {@link MessageDeflateTest} only
+ * exercises the single-frame path ({@code encode(data, true, true)}); the fragmented path
+ * ({@code compressFragment}) is never decoded back.
+ */
+final class PerMessageDeflateFragmentedRoundTripTest {
+
+ private static final String TEXT =
+ "The quick brown fox jumps over the lazy dog. "
+ + "Pack my box with five dozen liquor jugs. "
+ + "How vexingly quick daft zebras jump! "
+ + "The five boxing wizards jump quickly.";
+
+ private static byte[] encodeFragmented(final Encoder enc, final byte[] plain, final int fragments) {
+ final ByteArrayOutputStream wire = new ByteArrayOutputStream();
+ final int chunk = (plain.length + fragments - 1) / fragments;
+ for (int i = 0; i < fragments; i++) {
+ final int off = i * chunk;
+ final int len = Math.min(chunk, plain.length - off);
+ final byte[] part = new byte[len];
+ System.arraycopy(plain, off, part, 0, len);
+ final boolean first = i == 0;
+ final boolean fin = i == fragments - 1;
+ final byte[] out = enc.encode(part, first, fin).payload;
+ wire.write(out, 0, out.length);
+ }
+ return wire.toByteArray();
+ }
+
+ @Test
+ void fragmentedMessageRoundTripsAcrossThreeFrames() throws Exception {
+ final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null);
+ final Encoder enc = pmce.newEncoder();
+ final Decoder dec = pmce.newDecoder();
+
+ final byte[] plain = TEXT.getBytes(StandardCharsets.UTF_8);
+
+ final byte[] wire = encodeFragmented(enc, plain, 3);
+ final byte[] roundTrip = dec.decode(wire);
+
+ assertArrayEquals(plain, roundTrip,
+ "A fragmented compressed message must reassemble to the original bytes");
+ }
+
+ @Test
+ void fragmentedMessageWithEmptyFinalFragmentRoundTrips() throws Exception {
+ final PerMessageDeflate pmce = new PerMessageDeflate(true, false, false, null, null);
+ final Encoder enc = pmce.newEncoder();
+ final Decoder dec = pmce.newDecoder();
+
+ final byte[] head = "hello".getBytes(StandardCharsets.UTF_8);
+ final ByteArrayOutputStream wire = new ByteArrayOutputStream();
+ final byte[] firstFragment = enc.encode(head, true, false).payload;
+ wire.write(firstFragment, 0, firstFragment.length);
+ final byte[] finalFragment = enc.encode(new byte[0], false, true).payload;
+ wire.write(finalFragment, 0, finalFragment.length);
+
+ // RFC 7692 section 7.2.3.6: an empty final fragment is the single octet 0x00.
+ assertArrayEquals(new byte[]{0x00}, finalFragment,
+ "an empty final fragment must be encoded as the single octet 0x00");
+ assertArrayEquals(head, dec.decode(wire.toByteArray()),
+ "a fragmented message whose final fragment is empty must round-trip");
+ }
+
+ @Test
+ void fragmentedMessageMatchesSingleFrameEncoding() throws Exception {
+ final byte[] plain = TEXT.getBytes(StandardCharsets.UTF_8);
+
+ final byte[] fragmentedWire = encodeFragmented(
+ new PerMessageDeflate(true, false, false, null, null).newEncoder(), plain, 4);
+ final byte[] singleWire =
+ new PerMessageDeflate(true, false, false, null, null).newEncoder()
+ .encode(plain, true, true).payload;
+
+ // The wire bytes may differ (block boundaries), but both must decode to the same plaintext.
+ final byte[] fromFragmented =
+ new PerMessageDeflate(true, false, false, null, null).newDecoder().decode(fragmentedWire);
+ final byte[] fromSingle =
+ new PerMessageDeflate(true, false, false, null, null).newDecoder().decode(singleWire);
+
+ assertArrayEquals(plain, fromSingle, "single-frame control must round-trip");
+ assertArrayEquals(plain, fromFragmented, "fragmented message must round-trip to the same plaintext");
+ }
+}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandlerTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandlerTest.java
index ae88106701..e54b6aa334 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandlerTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketH2ServerExchangeHandlerTest.java
@@ -26,22 +26,49 @@
*/
package org.apache.hc.core5.websocket.server;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.hc.core5.http.EntityDetails;
+import org.apache.hc.core5.http.Header;
import org.apache.hc.core5.http.HttpRequest;
import org.apache.hc.core5.http.HttpResponse;
import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.http.Method;
import org.apache.hc.core5.http.message.BasicHttpRequest;
import org.apache.hc.core5.http.nio.AsyncPushProducer;
+import org.apache.hc.core5.http.nio.CapacityChannel;
+import org.apache.hc.core5.http.nio.DataStreamChannel;
import org.apache.hc.core5.http.nio.ResponseChannel;
import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.http.protocol.HttpCoreContext;
import org.apache.hc.core5.websocket.WebSocketConstants;
+import org.apache.hc.core5.websocket.WebSocketExtension;
+import org.apache.hc.core5.websocket.WebSocketExtensionData;
+import org.apache.hc.core5.websocket.WebSocketExtensionFactory;
import org.apache.hc.core5.websocket.WebSocketExtensionRegistry;
import org.apache.hc.core5.websocket.WebSocketHandler;
+import org.apache.hc.core5.websocket.WebSocketSession;
import org.junit.jupiter.api.Test;
class WebSocketH2ServerExchangeHandlerTest {
@@ -69,6 +96,35 @@ HttpResponse getResponse() {
}
}
+ private static final class CountingExtensionFactory implements WebSocketExtensionFactory {
+
+ private final AtomicInteger closed;
+
+ CountingExtensionFactory(final AtomicInteger closed) {
+ this.closed = closed;
+ }
+
+ @Override
+ public String getName() {
+ return "x-test";
+ }
+
+ @Override
+ public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) {
+ return new WebSocketExtension() {
+ @Override
+ public String getName() {
+ return "x-test";
+ }
+
+ @Override
+ public void close() {
+ closed.incrementAndGet();
+ }
+ };
+ }
+ }
+
@Test
void rejectsNonConnectMethod() throws Exception {
final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler(
@@ -115,4 +171,360 @@ void rejectsUnknownProtocol() throws Exception {
assertNotNull(channel.getResponse());
assertEquals(HttpStatus.SC_BAD_REQUEST, channel.getResponse().getCode());
}
+
+ @Test
+ void releasesExtensionsWhenOnOpenThrows() throws Exception {
+ final AtomicInteger closed = new AtomicInteger();
+ final AtomicReference worker = new AtomicReference<>();
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(new CountingExtensionFactory(closed));
+ final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler(
+ new WebSocketHandler() {
+ @Override
+ public void onOpen(final WebSocketSession session) {
+ throw new IllegalStateException("onOpen failed");
+ }
+ }, null, registry, worker::set);
+
+ final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo");
+ request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, "x-test");
+ handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create());
+
+ worker.get().run();
+
+ assertEquals(1, closed.get(), "onOpen failure must release negotiated extensions exactly once");
+ }
+
+ @Test
+ void releasesExtensionsWhenExecutorRejects() throws Exception {
+ final AtomicInteger closed = new AtomicInteger();
+ final Executor rejecting = command -> {
+ throw new RejectedExecutionException("rejected");
+ };
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(new CountingExtensionFactory(closed));
+ final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler(
+ new WebSocketHandler() {
+ }, null, registry, rejecting);
+
+ final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo");
+ request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, "x-test");
+
+ assertThrows(RejectedExecutionException.class, () ->
+ handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create()));
+
+ assertEquals(1, closed.get(), "a rejected executor must release negotiated extensions exactly once");
+ }
+
+ @Test
+ void releasesExtensionsOnNormalCompletion() throws Exception {
+ final AtomicInteger closed = new AtomicInteger();
+ final AtomicReference worker = new AtomicReference<>();
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(new CountingExtensionFactory(closed));
+ final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler(
+ new WebSocketHandler() {
+ }, null, registry, worker::set);
+
+ final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo");
+ request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, "x-test");
+ handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create());
+
+ // A clean end-of-stream lets the processor read loop terminate normally.
+ handler.streamEnd(null);
+ worker.get().run();
+
+ assertEquals(1, closed.get(), "normal processor completion must release negotiated extensions exactly once");
+ }
+
+ @Test
+ void executorRejectionIsNotMaskedByExtensionCloseFailure() throws Exception {
+ final RuntimeException closeFailure = new IllegalStateException("close failed");
+ final Executor rejecting = command -> {
+ throw new RejectedExecutionException("rejected");
+ };
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(throwingCloseFactory(closeFailure));
+ final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler(
+ new WebSocketHandler() {
+ }, null, registry, rejecting);
+
+ final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo");
+ request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, "x-test");
+
+ final RejectedExecutionException ex = assertThrows(RejectedExecutionException.class, () ->
+ handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create()));
+ assertEquals("rejected", ex.getMessage(), "the original rejection must be preserved");
+ assertEquals(1, ex.getSuppressed().length, "the extension close failure must be suppressed, not masking");
+ assertSame(closeFailure, ex.getSuppressed()[0]);
+ }
+
+ @Test
+ void streamIsTornDownEvenWhenExtensionCloseThrows() throws Exception {
+ final RuntimeException closeFailure = new IllegalStateException("close failed");
+ final AtomicReference worker = new AtomicReference<>();
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(throwingCloseFactory(closeFailure));
+ final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler(
+ new WebSocketHandler() {
+ }, null, registry, worker::set);
+
+ final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo");
+ request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS_LOWER, "x-test");
+ handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create());
+
+ handler.streamEnd(null);
+ // The worker's extension close() throws, but the stream teardown must still have run first.
+ assertThrows(RuntimeException.class, () -> worker.get().run());
+
+ final CollectingDataStreamChannel channel = new CollectingDataStreamChannel();
+ while (handler.available() > 0) {
+ handler.produce(channel);
+ }
+ assertTrue(channel.endStreamCalled(),
+ "the HTTP/2 stream must be terminated with END_STREAM despite the extension close failure");
+ }
+
+ @Test
+ void protocolViolationClosesWith1002() throws Exception {
+ final AtomicReference worker = new AtomicReference<>();
+ final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler(
+ new WebSocketHandler() {
+ }, null, WebSocketExtensionRegistry.createDefault(), worker::set);
+
+ final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo");
+ request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket");
+ handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create());
+
+ // A fragmented (FIN=0) masked PING violates RFC 6455 section 5.5 and raises a
+ // checked WebSocketException, which must map to close code 1002, not 1011.
+ handler.consume(ByteBuffer.wrap(new byte[]{0x09, (byte) 0x80, 1, 2, 3, 4}));
+ worker.get().run();
+
+ final CollectingDataStreamChannel channel = new CollectingDataStreamChannel();
+ while (handler.available() > 0) {
+ handler.produce(channel);
+ }
+ final byte[] out = channel.bytes();
+ assertEquals((byte) 0x88, out[0], "expected a CLOSE frame");
+ final int code = ((out[2] & 0xFF) << 8) | (out[3] & 0xFF);
+ assertEquals(1002, code, "protocol violation must close with 1002");
+ }
+
+ private static WebSocketExtensionFactory throwingCloseFactory(final RuntimeException closeFailure) {
+ return new WebSocketExtensionFactory() {
+ @Override
+ public String getName() {
+ return "x-test";
+ }
+
+ @Override
+ public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) {
+ return new WebSocketExtension() {
+ @Override
+ public String getName() {
+ return "x-test";
+ }
+
+ @Override
+ public void close() {
+ throw closeFailure;
+ }
+ };
+ }
+ };
+ }
+
+ private static final class CollectingDataStreamChannel implements DataStreamChannel {
+ private final ByteArrayOutputStream collected = new ByteArrayOutputStream();
+ private boolean endStreamCalled;
+
+ @Override
+ public void requestOutput() {
+ }
+
+ @Override
+ public int write(final ByteBuffer src) {
+ final int n = src.remaining();
+ final byte[] chunk = new byte[n];
+ src.get(chunk);
+ collected.write(chunk, 0, n);
+ return n;
+ }
+
+ @Override
+ public void endStream() {
+ endStreamCalled = true;
+ }
+
+ @Override
+ public void endStream(final List trailers) {
+ endStreamCalled = true;
+ }
+
+ byte[] bytes() {
+ return collected.toByteArray();
+ }
+
+ boolean endStreamCalled() {
+ return endStreamCalled;
+ }
+ }
+
+ @Test
+ void writeAfterStreamFailureFailsInsteadOfBlocking() throws Exception {
+ final AtomicReference worker = new AtomicReference<>();
+ final AtomicReference error = new AtomicReference<>();
+ final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler(
+ new WebSocketHandler() {
+ @Override
+ public void onOpen(final WebSocketSession session) {
+ try {
+ session.sendText("hello");
+ } catch (final Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ @Override
+ public void onError(final WebSocketSession session, final Exception cause) {
+ error.set(cause);
+ }
+ }, null, WebSocketExtensionRegistry.createDefault(), worker::set);
+
+ final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo");
+ request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket");
+ handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create());
+
+ // The stream fails before the worker gets to run; a write must then surface an
+ // error through the session instead of blocking on the dead outbound queue.
+ handler.failed(new IOException("stream reset"));
+ worker.get().run();
+
+ assertNotNull(error.get(), "write after stream failure must fail, not block");
+ }
+
+ @Test
+ void largeOutboundWriteCompletesViaRequestOutputSignals() throws Exception {
+ // A tiny budget (8 bytes, 4-byte chunks) cannot hold the whole 66-byte frame, so the writer
+ // blocks mid-write. Draining is driven ONLY by requestOutput() signals here, never by
+ // spontaneous polling, so the write can only complete if requestOutput() fires per chunk.
+ final StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < 64; i++) {
+ sb.append((char) ('a' + (i % 26)));
+ }
+ final String text = sb.toString();
+
+ final CountDownLatch writeDone = new CountDownLatch(1);
+ final AtomicReference writeError = new AtomicReference<>();
+ final AtomicReference worker = new AtomicReference<>();
+ final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler(
+ new WebSocketHandler() {
+ @Override
+ public void onOpen(final WebSocketSession session) {
+ try {
+ session.sendText(text);
+ } catch (final Exception ex) {
+ writeError.set(ex);
+ } finally {
+ writeDone.countDown();
+ }
+ }
+ }, null, WebSocketExtensionRegistry.createDefault(), worker::set, 8, 4);
+
+ final HttpRequest request = new BasicHttpRequest(Method.CONNECT, "/echo");
+ request.addHeader(WebSocketConstants.PSEUDO_PROTOCOL, "websocket");
+ handler.handleRequest(request, null, new CapturingResponseChannel(), HttpCoreContext.create());
+
+ final Semaphore outputReady = new Semaphore(0);
+ final AtomicInteger requestOutputCalls = new AtomicInteger();
+ final AtomicBoolean ended = new AtomicBoolean();
+ final ByteArrayOutputStream collected = new ByteArrayOutputStream();
+ final DataStreamChannel channel = new DataStreamChannel() {
+ @Override
+ public void requestOutput() {
+ requestOutputCalls.incrementAndGet();
+ outputReady.release();
+ }
+
+ @Override
+ public int write(final ByteBuffer src) {
+ final int n = src.remaining();
+ final byte[] c = new byte[n];
+ src.get(c);
+ collected.write(c, 0, n);
+ return n;
+ }
+
+ @Override
+ public void endStream() {
+ ended.set(true);
+ outputReady.release();
+ }
+
+ @Override
+ public void endStream(final List trailers) {
+ endStream();
+ }
+ };
+
+ final Thread reactor = new Thread(() -> {
+ try {
+ handler.produce(channel); // HttpCore drives the first produce() after the response
+ while (!ended.get() && outputReady.tryAcquire(5, TimeUnit.SECONDS)) {
+ handler.produce(channel);
+ }
+ } catch (final IOException | InterruptedException ex) {
+ throw new RuntimeException(ex);
+ }
+ }, "ws-reactor");
+ reactor.setDaemon(true);
+ final Thread writerThread = new Thread(worker.get(), "ws-writer");
+ writerThread.setDaemon(true);
+
+ reactor.start();
+ writerThread.start();
+
+ // Without a per-chunk requestOutput() the writer parks and the reactor is never re-signalled,
+ // so this await would time out; with the signal it completes.
+ assertTrue(writeDone.await(10, TimeUnit.SECONDS),
+ "a write larger than the budget must complete via requestOutput() signals, not deadlock");
+ assertNull(writeError.get(), "the write must succeed");
+
+ handler.streamEnd(null); // end the read loop so the worker enqueues END_OUTBOUND
+ writerThread.join(TimeUnit.SECONDS.toMillis(5));
+ reactor.join(TimeUnit.SECONDS.toMillis(5));
+
+ assertTrue(requestOutputCalls.get() > 0, "draining must be driven by requestOutput() signals");
+ final byte[] out = collected.toByteArray();
+ assertEquals(66, out.length, "the 64-byte text frame must be reassembled losslessly from 4-byte chunks");
+ assertEquals((byte) 0x81, out[0], "FIN + text opcode");
+ assertArrayEquals(text.getBytes(StandardCharsets.UTF_8), Arrays.copyOfRange(out, 2, out.length));
+ }
+
+ @Test
+ void advertisesBoundedInboundCapacity() throws Exception {
+ final WebSocketH2ServerExchangeHandler handler = new WebSocketH2ServerExchangeHandler(
+ new WebSocketHandler() {
+ }, null, WebSocketExtensionRegistry.createDefault());
+
+ final AtomicInteger granted = new AtomicInteger();
+ final CapacityChannel channel = new CapacityChannel() {
+ @Override
+ public void update(final int increment) {
+ granted.addAndGet(increment);
+ }
+ };
+
+ handler.updateCapacity(channel);
+ handler.updateCapacity(channel); // a repeated query must not re-grant the initial window
+
+ assertEquals(256 * 1024, granted.get(),
+ "inbound credit must be a bounded window, not Integer.MAX_VALUE");
+ }
}
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessorTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessorTest.java
index 31ee63b14a..e921a21b29 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessorTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerProcessorTest.java
@@ -27,6 +27,7 @@
package org.apache.hc.core5.websocket.server;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.ByteArrayInputStream;
@@ -39,6 +40,7 @@
import org.apache.hc.core5.websocket.WebSocketConfig;
import org.apache.hc.core5.websocket.WebSocketHandler;
import org.apache.hc.core5.websocket.WebSocketSession;
+import org.apache.hc.core5.websocket.exceptions.WebSocketProtocolException;
import org.junit.jupiter.api.Test;
class WebSocketServerProcessorTest {
@@ -106,6 +108,37 @@ void processesTextAndCloseFrames() throws Exception {
assertTrue(out.size() > 0, "server should send close response");
}
+ @Test
+ void dataFrameDuringFragmentedMessage_isRejected() throws Exception {
+ final ByteArrayOutputStream frames = new ByteArrayOutputStream();
+ // TEXT, not final -> starts a fragmented message
+ frames.write(maskedFrame(0x1, false, "a".getBytes(StandardCharsets.UTF_8)));
+ // a new TEXT frame while the fragmented message is still in progress (RFC 6455 section 5.4)
+ frames.write(maskedFrame(0x1, true, "b".getBytes(StandardCharsets.UTF_8)));
+
+ final WebSocketSession session = new WebSocketSession(
+ WebSocketConfig.DEFAULT,
+ new ByteArrayInputStream(frames.toByteArray()),
+ new ByteArrayOutputStream(),
+ null,
+ null,
+ Collections.emptyList());
+ final WebSocketServerProcessor processor =
+ new WebSocketServerProcessor(session, new TrackingHandler(), 1024);
+
+ final WebSocketProtocolException ex = assertThrows(WebSocketProtocolException.class, processor::process);
+ assertEquals(1002, ex.closeCode);
+ assertTrue(ex.getMessage().contains("Data frame during fragmented message"), ex.getMessage());
+ }
+
+ private static byte[] maskedFrame(final int opcode, final boolean fin, final byte[] payload) {
+ final byte[] framed = maskedFrame(opcode, payload);
+ if (!fin) {
+ framed[0] &= 0x7F; // clear the FIN bit
+ }
+ return framed;
+ }
+
private static final class TrackingHandler implements WebSocketHandler {
private String text;
private int closeCode;
diff --git a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandlerTest.java b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandlerTest.java
index 817f093ad5..d1ceb6811f 100644
--- a/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandlerTest.java
+++ b/httpclient5-websocket/src/test/java/org/apache/hc/core5/websocket/server/WebSocketServerRequestHandlerTest.java
@@ -28,6 +28,8 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.IOException;
@@ -36,9 +38,11 @@
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.HeaderElements;
import org.apache.hc.core5.http.HttpHeaders;
import org.apache.hc.core5.http.HttpRequestMapper;
import org.apache.hc.core5.http.HttpStatus;
@@ -48,6 +52,9 @@
import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.websocket.WebSocketConfig;
import org.apache.hc.core5.websocket.WebSocketConstants;
+import org.apache.hc.core5.websocket.WebSocketExtension;
+import org.apache.hc.core5.websocket.WebSocketExtensionData;
+import org.apache.hc.core5.websocket.WebSocketExtensionFactory;
import org.apache.hc.core5.websocket.WebSocketExtensionRegistry;
import org.apache.hc.core5.websocket.WebSocketHandler;
import org.apache.hc.core5.websocket.WebSocketSession;
@@ -120,6 +127,152 @@ public String selectSubprotocol(final List protocols) {
connection.close();
}
+ @Test
+ void releasesExtensionsWhenOnOpenThrows() throws Exception {
+ final AtomicInteger closed = new AtomicInteger();
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(new CountingExtensionFactory(closed));
+ final Supplier supplier = () -> new WebSocketHandler() {
+ @Override
+ public void onOpen(final WebSocketSession session) {
+ throw new IllegalStateException("onOpen failed");
+ }
+ };
+ final HttpRequestMapper> mapper = (request, context) -> supplier;
+ final WebSocketServerRequestHandler handler = new WebSocketServerRequestHandler(
+ mapper, WebSocketConfig.DEFAULT, registry);
+
+ final BasicClassicHttpRequest request = new BasicClassicHttpRequest("GET", "/ws");
+ request.addHeader(HttpHeaders.CONNECTION, "Upgrade");
+ request.addHeader(HttpHeaders.UPGRADE, "websocket");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_KEY, "dGhlIHNhbXBsZSBub25jZQ==");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, "x-test");
+
+ final RecordingTrigger trigger = new RecordingTrigger();
+ final HttpContext context = HttpCoreContext.create();
+ final WebSocketServerConnection connection = createConnection();
+ context.setAttribute(WebSocketContextKeys.CONNECTION, connection);
+ try {
+ handler.handle(request, trigger, context);
+ } catch (final IOException ignore) {
+ // A failed handshake may surface an I/O error while writing the close frame; this
+ // is independent of whether the negotiated extensions were released.
+ }
+
+ assertEquals(1, closed.get(), "onOpen failure must release negotiated extensions exactly once");
+ connection.close();
+ }
+
+ @Test
+ void releasesExtensionsWhenSubmitResponseFails() throws Exception {
+ final AtomicInteger closed = new AtomicInteger();
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(new CountingExtensionFactory(closed));
+ final HttpRequestMapper> mapper =
+ (request, context) -> () -> new WebSocketHandler() {
+ };
+ final WebSocketServerRequestHandler handler = new WebSocketServerRequestHandler(
+ mapper, WebSocketConfig.DEFAULT, registry);
+
+ final BasicClassicHttpRequest request = new BasicClassicHttpRequest("GET", "/ws");
+ request.addHeader(HttpHeaders.CONNECTION, "Upgrade");
+ request.addHeader(HttpHeaders.UPGRADE, "websocket");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_KEY, "dGhlIHNhbXBsZSBub25jZQ==");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, "x-test");
+
+ final ResponseTrigger throwingTrigger = new ResponseTrigger() {
+ @Override
+ public void sendInformation(final ClassicHttpResponse response) {
+ }
+
+ @Override
+ public void submitResponse(final ClassicHttpResponse response) throws IOException {
+ throw new IOException("submit failed");
+ }
+ };
+ final HttpContext context = HttpCoreContext.create();
+ final WebSocketServerConnection connection = createConnection();
+ context.setAttribute(WebSocketContextKeys.CONNECTION, connection);
+
+ assertThrows(IOException.class, () -> handler.handle(request, throwingTrigger, context));
+
+ assertEquals(1, closed.get(), "a failed submitResponse must release negotiated extensions exactly once");
+ connection.close();
+ }
+
+ @Test
+ void releasesExtensionsOnNormalCompletion() throws Exception {
+ final AtomicInteger closed = new AtomicInteger();
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(new CountingExtensionFactory(closed));
+ final HttpRequestMapper> mapper =
+ (request, context) -> () -> new WebSocketHandler() {
+ };
+ final WebSocketServerRequestHandler handler = new WebSocketServerRequestHandler(
+ mapper, WebSocketConfig.DEFAULT, registry);
+
+ final BasicClassicHttpRequest request = new BasicClassicHttpRequest("GET", "/ws");
+ request.addHeader(HttpHeaders.CONNECTION, "Upgrade");
+ request.addHeader(HttpHeaders.UPGRADE, "websocket");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_KEY, "dGhlIHNhbXBsZSBub25jZQ==");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, "x-test");
+
+ final RecordingTrigger trigger = new RecordingTrigger();
+ final HttpContext context = HttpCoreContext.create();
+ // The peer end of the socket is already closed, so the processor reads EOF and the
+ // read loop terminates normally.
+ final WebSocketServerConnection connection = createConnection();
+ context.setAttribute(WebSocketContextKeys.CONNECTION, connection);
+
+ handler.handle(request, trigger, context);
+
+ assertEquals(1, closed.get(), "normal processor completion must release negotiated extensions exactly once");
+ connection.close();
+ }
+
+ @Test
+ void submitResponseFailureIsNotMaskedByExtensionCloseFailure() throws Exception {
+ final RuntimeException closeFailure = new IllegalStateException("close failed");
+ final WebSocketExtensionRegistry registry = new WebSocketExtensionRegistry()
+ .register(throwingCloseFactory(closeFailure));
+ final HttpRequestMapper> mapper =
+ (request, context) -> () -> new WebSocketHandler() {
+ };
+ final WebSocketServerRequestHandler handler = new WebSocketServerRequestHandler(
+ mapper, WebSocketConfig.DEFAULT, registry);
+
+ final BasicClassicHttpRequest request = new BasicClassicHttpRequest("GET", "/ws");
+ request.addHeader(HttpHeaders.CONNECTION, "Upgrade");
+ request.addHeader(HttpHeaders.UPGRADE, "websocket");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION, "13");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_KEY, "dGhlIHNhbXBsZSBub25jZQ==");
+ request.addHeader(WebSocketConstants.SEC_WEBSOCKET_EXTENSIONS, "x-test");
+
+ final ResponseTrigger throwingTrigger = new ResponseTrigger() {
+ @Override
+ public void sendInformation(final ClassicHttpResponse response) {
+ }
+
+ @Override
+ public void submitResponse(final ClassicHttpResponse response) throws IOException {
+ throw new IOException("submit failed");
+ }
+ };
+ final HttpContext context = HttpCoreContext.create();
+ final WebSocketServerConnection connection = createConnection();
+ context.setAttribute(WebSocketContextKeys.CONNECTION, connection);
+
+ final IOException ex = assertThrows(IOException.class,
+ () -> handler.handle(request, throwingTrigger, context));
+ assertEquals("submit failed", ex.getMessage(), "the original failure must be preserved");
+ assertEquals(1, ex.getSuppressed().length, "the extension close failure must be suppressed, not masking");
+ assertSame(closeFailure, ex.getSuppressed()[0]);
+ connection.close();
+ }
+
@Test
void returnsUpgradeRequiredForNonUpgradeRequest() throws Exception {
final WebSocketServerRequestHandler handler = new WebSocketServerRequestHandler(
@@ -132,6 +285,16 @@ void returnsUpgradeRequiredForNonUpgradeRequest() throws Exception {
handler.handle(request, trigger, HttpCoreContext.create());
assertEquals(HttpStatus.SC_UPGRADE_REQUIRED, trigger.response.getCode());
+ // RFC 9110 section 15.5.22: a 426 response MUST carry an Upgrade header field, with the
+ // matching Connection: Upgrade option (section 7.8).
+ assertNotNull(trigger.response.getFirstHeader(HttpHeaders.UPGRADE));
+ assertEquals("websocket", trigger.response.getFirstHeader(HttpHeaders.UPGRADE).getValue());
+ assertNotNull(trigger.response.getFirstHeader(HttpHeaders.CONNECTION));
+ assertTrue(trigger.response.getFirstHeader(HttpHeaders.CONNECTION).getValue()
+ .equalsIgnoreCase(HeaderElements.UPGRADE));
+ // RFC 6455 section 4.4: the rejection must advertise the supported version.
+ assertNotNull(trigger.response.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION));
+ assertEquals("13", trigger.response.getFirstHeader(WebSocketConstants.SEC_WEBSOCKET_VERSION).getValue());
}
@Test
@@ -162,6 +325,59 @@ private static WebSocketServerConnection createConnection() throws IOException {
return factory.createConnection(socket);
}
+ private static final class CountingExtensionFactory implements WebSocketExtensionFactory {
+
+ private final AtomicInteger closed;
+
+ CountingExtensionFactory(final AtomicInteger closed) {
+ this.closed = closed;
+ }
+
+ @Override
+ public String getName() {
+ return "x-test";
+ }
+
+ @Override
+ public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) {
+ return new WebSocketExtension() {
+ @Override
+ public String getName() {
+ return "x-test";
+ }
+
+ @Override
+ public void close() {
+ closed.incrementAndGet();
+ }
+ };
+ }
+ }
+
+ private static WebSocketExtensionFactory throwingCloseFactory(final RuntimeException closeFailure) {
+ return new WebSocketExtensionFactory() {
+ @Override
+ public String getName() {
+ return "x-test";
+ }
+
+ @Override
+ public WebSocketExtension create(final WebSocketExtensionData request, final boolean server) {
+ return new WebSocketExtension() {
+ @Override
+ public String getName() {
+ return "x-test";
+ }
+
+ @Override
+ public void close() {
+ throw closeFailure;
+ }
+ };
+ }
+ };
+ }
+
private static final class RecordingTrigger implements ResponseTrigger {
ClassicHttpResponse response;