diff --git a/README.md b/README.md index 47c3c8c..8e404a3 100644 --- a/README.md +++ b/README.md @@ -182,12 +182,26 @@ app.onException((ex, req, res) -> { | `h2MaxConnectionLifetimeMs` | `0` | Optional connection lifetime; `0` disables it. | | `h2StreamIdleTimeoutMs` | `60000` | Inactive open-stream deadline. | +## WebSockets over HTTP/2 + +The same `ws(path, handler)` route serves WebSockets over HTTP/1.1 and HTTP/2. When HTTP/2 is +enabled, Flash advertises RFC 8441 extended CONNECT support and carries WebSocket frames inside +flow-controlled DATA frames. No alternate handler, route, or session API is required: + +```java +app.ws("/live", handler); +``` + +HTTP/1.1 clients use the ordinary `101 Switching Protocols` upgrade. HTTP/2 clients use an +extended CONNECT and receive status `200`; Flash applies the same RFC 6455 framing, masking, +fragmentation, close, and callback behavior on both transports. Client support for negotiating +WebSockets over HTTP/2 varies, so clients without RFC 8441 support continue to use HTTP/1.1. + ## TLS -HTTPS and WSS are a transport-layer concern only: once a listener is bound, the accepted -`Socket` is either plain or an `SSLSocket` indistinguishably from `HttpServer`'s point of view -onward — the request parser, router, and WebSocket upgrade never branch on it. WSS is therefore -not a separate feature; it's a WebSocket upgrade running over whatever transport it was handed. +HTTPS and WSS are a transport-layer concern only. Once a listener is bound, the accepted socket +is plain or TLS; the selected HTTP connection implementation then performs either the HTTP/1.1 +upgrade or the HTTP/2 extended CONNECT. WSS does not require a separate route or handler API. ### Quick start diff --git a/flash/docs/http2/DECISIONS.md b/flash/docs/http2/DECISIONS.md index 639356f..b19e27d 100644 --- a/flash/docs/http2/DECISIONS.md +++ b/flash/docs/http2/DECISIONS.md @@ -1042,3 +1042,25 @@ general-purpose stack. bounded pool or client-side multiplexing without changing the proxy-facing API. --- + +## DEC-32 — Reuse the WebSocket router and session for extended CONNECT + +**Context.** RFC 8441 changes the HTTP handshake and transport framing, but not the application +route, RFC 6455 message semantics, or handler lifecycle. Introducing an HTTP/2-specific router, +handler, or session would duplicate public and internal behavior. + +**Decision.** Validate CONNECT and `:protocol` at the HTTP/2 wire boundary, then expose a +`websocket` extended CONNECT as GET only while resolving the existing `AbstractWsRouter` route. +Feed request DATA to the existing `WebSocketSession` and adapt the protocol-neutral +`ResponseStream` to its `OutputStream` contract. Publish response HEADERS in their own first batch +so the full-duplex producer cannot block the handshake while waiting for request DATA. + +**Consequence.** One `ws(path, handler)` registration behaves the same on HTTP/1.1 and HTTP/2; +masking, fragmentation, callbacks, and close handling have one implementation. HTTP/2 contributes +only pseudo-header validation and DATA flow control, while the shared response bridge remains +usable by other streaming adapters. + +**Revisit when.** Only if a future WebSocket transport cannot be represented by the existing +stream pair without losing protocol semantics. + +--- diff --git a/flash/docs/http2/IMPLEMENTATION-PLAN.md b/flash/docs/http2/IMPLEMENTATION-PLAN.md index fc0d042..663881b 100644 --- a/flash/docs/http2/IMPLEMENTATION-PLAN.md +++ b/flash/docs/http2/IMPLEMENTATION-PLAN.md @@ -76,7 +76,7 @@ Status values: `not started` / `in progress` / `blocked` / `done`. | 12 — Trailers, half-close, gRPC | done | `feature/core/http2` | Protocol-neutral request/response trailers, bounded push streaming, four half-close orderings and authority-form CONNECT tunnels complete. Real grpcurl 1.9.3 unary/server-streaming/error interop passes. EX-48/49 fixed. HTTP/2 remains opt-in until the Phase 13 hostile-peer gate (DEC-29). 649/649 tests green from a clean `-Pjmh` build. | | 13 — Security hardening & abuse resistance | done | `feature/core/http2` | Two-bucket Rapid Reset/stream/settings/ping/aggregate counters, control/write queue bounds, optional stream/byte/lifetime budgets, absolute header and idle-stream deadlines, and hostile-peer suite complete. Security review found+fixed EX-50/51. JMH counter: 38.083 ns/op, ~10^-4 B/op, no GC. 100k-CONTINUATION attack terminates in under 2 s with bounded retained heap. 663/663 tests green from a clean `-Pjmh` build. | | 14 — h2c prior knowledge + proxy support | done | `feature/core/http2` | Independent TLS/h2c gates, pooled proxy-oriented h2 client with TLS ALPN and h2c, bidirectional h1/h2 trailer relay, shared four-direction hop-by-hop policy and certificate-backed 421 handling complete. Real grpcurl h2c interop passes. 670/670 tests green from a clean `-Pjmh` build. | -| 15 — RFC 8441 extended CONNECT (WS over h2) | not started | — | — | +| 15 — RFC 8441 extended CONNECT (WS over h2) | done | `feature/core/http2` | SETTINGS_ENABLE_CONNECT_PROTOCOL, shared WS router/session, DATA flow control, >1 MiB message, h1/h2 parity and lifecycle hardening complete. EX-52/53 fixed; DEC-32 recorded. 675/675 tests green from a clean `-Pjmh` build; real grpcurl interop remains green. | | 16 — Compliance test suite | not started | — | — | | 17 — Benchmarks, allocation gates, tuning | not started | — | — | | 18 — Documentation | not started | — | — | @@ -813,6 +813,25 @@ stream retirement atomic in `Http2StreamTable` and require both the expected str identity to match the live table entry. A regression test proves that a stale retirement cannot remove the next generation of the same pooled object. **Phase**: 13. +### EX-52 — WebSocket `onOpen` failures bypassed lifecycle cleanup + +Found while routing extended CONNECT through the existing WebSocket loop. `onOpen` ran before the +loop's `try/finally`, and runtime failures from application callbacks were not handled alongside +I/O failures. An exception could therefore escape without `onError`, `onClose`, or guaranteed +transport release. **Fix**: include `onOpen` and all callback dispatch in the guarded lifecycle, +report runtime failures, and force-close in a nested `finally` even if `onClose` fails. +`WebSocketLoopTest` is the regression test. **Phase**: 15. + +### EX-53 — Push-streaming HTTP/2 responses could deadlock before response headers + +Found in the first live extended-CONNECT test. `Http2ResponseWriter.startFlowControlled` tried to +read the first push-streaming body byte while constructing the same batch as the response HEADERS. +A full-duplex producer waiting for request DATA therefore blocked before the client could receive +the successful response and send that DATA. **Fix**: publish push-streaming HEADERS as the first +batch and start body reads only from the post-write resume batch. `WebSocketOverH2Test` proves the +handshake completes before sending a message and then carries a message beyond the flow window. +**Phase**: 15. + --- # PART III — The phases @@ -2956,8 +2975,9 @@ defines the h2 mechanism. - `flash/docs/http2/WEBSOCKET.md`. ### DoD -- [ ] A browser negotiating h2 can open a WebSocket to a Flash `ws()` route. -- [ ] `AbstractWsRouter` and `FastPathWsRouterImpl` unchanged. +- [x] An RFC 8441 client negotiating h2 can open a WebSocket to a Flash `ws()` route + (`WebSocketOverH2Test`; the release-browser matrix remains Phase 16 scope). +- [x] `AbstractWsRouter` and `FastPathWsRouterImpl` unchanged. --- diff --git a/flash/docs/http2/WEBSOCKET.md b/flash/docs/http2/WEBSOCKET.md new file mode 100644 index 0000000..2e228e2 --- /dev/null +++ b/flash/docs/http2/WEBSOCKET.md @@ -0,0 +1,52 @@ +# WebSockets over HTTP/2 + +Flash implements RFC 8441 extended CONNECT alongside the existing HTTP/1.1 WebSocket upgrade. +Both transports resolve the same `ws(path, handler)` registration through `AbstractWsRouter` and +run the same `WebSocketSession`, frame parser, handler callbacks, and close lifecycle. + +## Protocol negotiation + +Every HTTP/2 server connection advertises `SETTINGS_ENABLE_CONNECT_PROTOCOL` (`0x8`) with value +`1`. A WebSocket request uses this pseudo-header shape: + +```text +:method CONNECT +:protocol websocket +:scheme https # or http +:authority example.com +:path /live +``` + +The normal HTTP/1.1 upgrade fields (`Connection`, `Upgrade`, `Sec-WebSocket-Key`, and +`Sec-WebSocket-Accept`) are neither required nor permitted on this path. A matched route receives +status `200`; a missing route receives `404`. + +## Shared application behavior + +At the router boundary, an extended CONNECT for `websocket` is represented as a GET so the +existing WebSocket router can be reused without a second registration table or protocol-specific +handler API. The wire validator retains the original CONNECT semantics and rejects malformed +pseudo-header combinations before dispatch. + +Request DATA is exposed through the existing streaming `RequestBody`. WebSocket output passes +through the common push-style `ResponseStream`, so HTTP/2 stream and connection flow-control +windows apply without changing the WebSocket codec. Messages may cross any number of DATA-frame +boundaries; those boundaries are invisible to RFC 6455 framing. Client-to-server masking remains +mandatory and is validated by the same frame parser used for HTTP/1.1. + +## Lifecycle and backpressure + +Response HEADERS are sent before the push producer is allowed to wait for request DATA. This is +required for a full-duplex protocol: waiting for the first WebSocket frame before publishing the +successful CONNECT response would deadlock compliant clients. Subsequent response batches block +behind the bounded response bridge and resume when HTTP/2 flow-control credit becomes available. + +Handler failures from `onOpen` or `onMessage` are reported through `onError`; `onClose` is invoked +once and the transport is released even if the close callback itself fails. + +## Verification + +`WebSocketOverH2Test` exercises the extended CONNECT exchange, fragmented text, masking, graceful +close, and a binary message larger than the initial one-mebibyte stream window. +`WebSocketParityTest` sends the same message through one route and handler over HTTP/1.1 and +HTTP/2 and compares the result byte for byte. diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2Preface.java b/flash/src/main/java/dev/relism/flash/http2/Http2Preface.java index c198c0b..7c0169d 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2Preface.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2Preface.java @@ -55,6 +55,7 @@ public final class Http2Preface { setting(bytes, Http2Settings.MAX_CONCURRENT_STREAMS, Http2Limits.MAX_CONCURRENT_STREAMS); setting(bytes, Http2Settings.INITIAL_WINDOW_SIZE, Http2Limits.INITIAL_WINDOW_SIZE_LOCAL); setting(bytes, Http2Settings.MAX_HEADER_LIST_SIZE, Http2Limits.MAX_HEADER_LIST_SIZE); + setting(bytes, Http2Settings.ENABLE_CONNECT_PROTOCOL, 1); frame.endFrame(); return copy(bytes); } diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2Settings.java b/flash/src/main/java/dev/relism/flash/http2/Http2Settings.java index 8f1a278..75b1636 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2Settings.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2Settings.java @@ -11,6 +11,7 @@ public final class Http2Settings { public static final int INITIAL_WINDOW_SIZE = 0x4; public static final int MAX_FRAME_SIZE = 0x5; public static final int MAX_HEADER_LIST_SIZE = 0x6; + public static final int ENABLE_CONNECT_PROTOCOL = 0x8; public static final int DEFAULT_HEADER_TABLE_SIZE = 4_096; public static final int DEFAULT_INITIAL_WINDOW_SIZE = 65_535; @@ -78,7 +79,7 @@ public final class Http2Settings { private static void validate(int id, long value) { switch (id) { - case ENABLE_PUSH -> { + case ENABLE_PUSH, ENABLE_CONNECT_PROTOCOL -> { if (value > 1) throw Http2Exception.PROTOCOL_ERROR; } case INITIAL_WINDOW_SIZE -> { diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java b/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java index 0c8e79e..a99e3a0 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java @@ -1,5 +1,6 @@ package dev.relism.flash.http2; +import dev.relism.flash.http.ContentType; import dev.relism.flash.http.HttpMethod; import dev.relism.flash.http.HttpStatus; import dev.relism.flash.http2.frame.Http2FrameWriter; @@ -11,7 +12,11 @@ import dev.relism.flash.http2.stream.Http2StreamTable; import dev.relism.flash.models.Request; import dev.relism.flash.models.RequestHandler; import dev.relism.flash.models.Response; +import dev.relism.flash.models.ResponseStreamOutputStream; import dev.relism.flash.transport.ConnectionContext; +import dev.relism.flash.websocket.WebSocketHandler; +import dev.relism.flash.websocket.WebSocketLoop; +import dev.relism.flash.websocket.WebSocketSession; import java.io.IOException; import java.util.concurrent.RejectedExecutionException; import lombok.extern.slf4j.Slf4j; @@ -103,10 +108,29 @@ final class Http2StreamDispatcher implements Http2Stream.ResponseSink { Request request = stream.assembleRequest(context.remoteAddress(), context.sslSocket()); Response pooled = stream.resetResponse(); Response response = pooled; - Object routeScratch = stream.routeScratch(context.router()); if (!Http2Authority.isServed(request.header("host"), request.sslSession())) { response.status(HttpStatus.MISDIRECTED_REQUEST); + if (stream.websocketConnect()) response.type(ContentType.NONE).streaming(output -> {}); + } else if (stream.websocketConnect()) { + WebSocketHandler handler = + context.wsRouter().route(request, stream.wsRouteScratch(context.wsRouter())); + response.type(ContentType.NONE); + if (handler == null) { + response.status(HttpStatus.NOT_FOUND).streaming(output -> {}); + } else { + response.streaming( + output -> + WebSocketLoop.run( + new WebSocketSession( + request.body().stream(), + new ResponseStreamOutputStream(output), + context.configuration().getWsFrameBufferSize(), + request, + false), + handler)); + } } else { + Object routeScratch = stream.routeScratch(context.router()); RequestHandler handler = context.router().route(request, routeScratch); if (handler == null) handler = context.router().getNotFoundHandler(); try { diff --git a/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java b/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java index 3733c45..137646a 100644 --- a/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java +++ b/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java @@ -151,7 +151,8 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize throws IOException { if (streamId <= 0) throw new IllegalArgumentException("streamId must be positive"); if (maxFrameSize <= 0 || availableFlowWindow < 0) { - throw new IllegalArgumentException("frame size must be positive and flow window non-negative"); + throw new IllegalArgumentException( + "frame size must be positive and flow window non-negative"); } headerBlock.reset(); output.reset(); @@ -212,7 +213,7 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize finished = true; endStreamInBatch = true; } - if (hasBody && availableFlowWindow > 0) { + if (hasBody && availableFlowWindow > 0 && !pushBody) { appendData(maxFrameSize, availableFlowWindow); } return dataBytesInBatch; @@ -282,7 +283,8 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize end = unknownLength ? eof : bodyRemaining == 0; boolean trailersFollow = end && response.hasTrailers(); if (count != 0 || !trailersFollow) { - frames.beginFrame(FrameType.DATA, end && !trailersFollow ? FrameFlags.END_STREAM : 0, streamId); + frames.beginFrame( + FrameType.DATA, end && !trailersFollow ? FrameFlags.END_STREAM : 0, streamId); output.writeBytes(relay, 0, count); frames.endFrame(); } diff --git a/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java b/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java index bb46284..1dd63b4 100644 --- a/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java +++ b/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java @@ -12,6 +12,7 @@ public final class PseudoHeaders { private static final int SCHEME = 2; private static final int PATH = 4; private static final int AUTHORITY = 8; + private static final int PROTOCOL = 16; private final PooledSlice name = new PooledSlice(); private final PooledSlice value = new PooledSlice(); @@ -19,6 +20,7 @@ public final class PseudoHeaders { private final PooledSlice scheme = new PooledSlice(); private final PooledSlice path = new PooledSlice(); private final PooledSlice authority = new PooledSlice(); + private final PooledSlice protocol = new PooledSlice(); private final PooledSlice host = new PooledSlice(); private int present; @@ -28,6 +30,7 @@ public final class PseudoHeaders { scheme.reset(null, 0, 0); path.reset(null, 0, 0); authority.reset(null, 0, 0); + protocol.reset(null, 0, 0); host.reset(null, 0, 0); boolean regularSeen = false; @@ -51,7 +54,15 @@ public final class PseudoHeaders { if ((present & METHOD) == 0) fail(streamId, "missing :method"); boolean connect = equals(method, "CONNECT"); - if (connect) { + boolean extendedConnect = (present & PROTOCOL) != 0; + if (extendedConnect) { + if (!connect) fail(streamId, ":protocol requires CONNECT"); + int required = METHOD | SCHEME | PATH | AUTHORITY | PROTOCOL; + if ((present & required) != required) { + fail(streamId, "extended CONNECT missing pseudo-header"); + } + if (path.length() == 0) fail(streamId, "empty :path"); + } else if (connect) { if ((present & AUTHORITY) == 0) fail(streamId, "CONNECT requires :authority"); if ((present & (SCHEME | PATH)) != 0) fail(streamId, "CONNECT forbids :scheme and :path"); } else { @@ -94,11 +105,16 @@ public final class PseudoHeaders { return authority; } + public boolean websocket() { + return protocol.array() != null && equals(protocol, "websocket"); + } + private void copySlice(int bit, PooledSlice source) { if (bit == METHOD) copy(source, method); else if (bit == SCHEME) copy(source, scheme); else if (bit == PATH) copy(source, path); - else copy(source, authority); + else if (bit == AUTHORITY) copy(source, authority); + else copy(source, protocol); } private static void copy(PooledSlice source, PooledSlice target) { @@ -110,6 +126,7 @@ public final class PseudoHeaders { if (equals(name, ":scheme")) return SCHEME; if (equals(name, ":path")) return PATH; if (equals(name, ":authority")) return AUTHORITY; + if (equals(name, ":protocol")) return PROTOCOL; return 0; } diff --git a/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java b/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java index f6f10d0..1c7808d 100644 --- a/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java +++ b/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java @@ -17,6 +17,7 @@ import dev.relism.flash.models.RequestBody; import dev.relism.flash.models.RequestLine; import dev.relism.flash.models.Response; import dev.relism.flash.routing.AbstractRouter; +import dev.relism.flash.routing.AbstractWsRouter; import dev.relism.fpr.core.ByteView; import java.io.IOException; import java.net.InetSocketAddress; @@ -60,6 +61,7 @@ public final class Http2Stream private int emptyDataFrames; private Http2StreamTable owner; private Object routeScratch; + private Object wsRouteScratch; private volatile boolean dispatched; private volatile boolean cancelled; private boolean headersValidated; @@ -135,9 +137,11 @@ public final class Http2Stream throw new Http2StreamException( id, Http2ErrorCode.PROTOCOL_ERROR, "unsupported request method"); } + if (pseudoHeaders.websocket()) method = HttpMethod.GET; requestLine.reset(method, path, question < 0 ? null : query, protocol, headers); requestBody.reset(http2Body, http2Body.declaredLength(), null, 0, 0); - Request assembled = Request.forParsed(request, requestLine, requestBody, remoteAddress, sslSocket); + Request assembled = + Request.forParsed(request, requestLine, requestBody, remoteAddress, sslSocket); assembled.setTrailers(trailers); return assembled; } @@ -279,6 +283,15 @@ public final class Http2Stream return routeScratch; } + public Object wsRouteScratch(AbstractWsRouter router) { + if (wsRouteScratch == null) wsRouteScratch = router.newScratch(); + return wsRouteScratch; + } + + public boolean websocketConnect() { + return pseudoHeaders.websocket(); + } + public void markDispatched() { dispatched = true; } diff --git a/flash/src/main/java/dev/relism/flash/models/ResponseStreamOutputStream.java b/flash/src/main/java/dev/relism/flash/models/ResponseStreamOutputStream.java new file mode 100644 index 0000000..34eb05e --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/models/ResponseStreamOutputStream.java @@ -0,0 +1,35 @@ +package dev.relism.flash.models; + +import java.io.IOException; +import java.io.OutputStream; + +/** Adapts a flow-controlled response stream to APIs that write to an {@link OutputStream}. */ +public final class ResponseStreamOutputStream extends OutputStream { + private final ResponseStream stream; + private final byte[] single = new byte[1]; + + public ResponseStreamOutputStream(ResponseStream stream) { + this.stream = stream; + } + + @Override + public void write(int value) throws IOException { + single[0] = (byte) value; + stream.write(single, 0, 1); + } + + @Override + public void write(byte[] bytes, int offset, int length) throws IOException { + stream.write(bytes, offset, length); + } + + @Override + public void flush() throws IOException { + stream.flush(); + } + + @Override + public void close() throws IOException { + stream.close(); + } +} diff --git a/flash/src/main/java/dev/relism/flash/websocket/WebSocketLoop.java b/flash/src/main/java/dev/relism/flash/websocket/WebSocketLoop.java index 2c04c47..df388b6 100644 --- a/flash/src/main/java/dev/relism/flash/websocket/WebSocketLoop.java +++ b/flash/src/main/java/dev/relism/flash/websocket/WebSocketLoop.java @@ -3,39 +3,44 @@ package dev.relism.flash.websocket; import java.io.IOException; /** - * Drives one {@link WebSocketSession}'s read loop until the session closes, dispatching frames - * responsibility is this loop; the handshake and upgrade detection live in - * {@link WebSocketUpgrade}. + * Drives one {@link WebSocketSession}'s read loop until the session closes. The handshake and + * upgrade detection live in {@link WebSocketUpgrade}. */ public final class WebSocketLoop { - private WebSocketLoop() { - } + private WebSocketLoop() {} - public static void run(WebSocketSession session, WebSocketHandler handler) { - handler.onOpen(session); - WebSocketFrame frame = new WebSocketFrame(); - try { - while (session.isOpen()) { - if (!session.readFrame(frame)) break; - switch (frame.opcode()) { - case WebSocketFrame.OP_TEXT, WebSocketFrame.OP_BINARY - -> handler.onMessage(session, frame); - case WebSocketFrame.OP_CLOSE - -> session.closeFromPeer(frame); - case WebSocketFrame.OP_PING - -> session.sendPong(frame); - case WebSocketFrame.OP_PONG -> { /* heartbeat ack, no-op */ } - } - } - } catch (WebSocketProtocolException e) { - try { session.close(e.closeCode()); } catch (IOException ignored) { } - handler.onError(session, e); - } catch (IOException e) { - handler.onError(session, e); - } finally { - handler.onClose(session, session.closeCode()); - session.forceClose(); + public static void run(WebSocketSession session, WebSocketHandler handler) { + WebSocketFrame frame = new WebSocketFrame(); + try { + handler.onOpen(session); + while (session.isOpen()) { + if (!session.readFrame(frame)) break; + switch (frame.opcode()) { + case WebSocketFrame.OP_TEXT, WebSocketFrame.OP_BINARY -> + handler.onMessage(session, frame); + case WebSocketFrame.OP_CLOSE -> session.closeFromPeer(frame); + case WebSocketFrame.OP_PING -> session.sendPong(frame); + case WebSocketFrame.OP_PONG -> { + // Heartbeat acknowledgement; no action is required. + } } + } + } catch (WebSocketProtocolException failure) { + try { + session.close(failure.closeCode()); + } catch (IOException ignored) { + // The peer may already have closed the transport. + } + handler.onError(session, failure); + } catch (IOException | RuntimeException failure) { + handler.onError(session, failure); + } finally { + try { + handler.onClose(session, session.closeCode()); + } finally { + session.forceClose(); + } } + } } diff --git a/flash/src/test/java/dev/relism/flash/http2/H2WebSocketTestClient.java b/flash/src/test/java/dev/relism/flash/http2/H2WebSocketTestClient.java new file mode 100644 index 0000000..179c743 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/H2WebSocketTestClient.java @@ -0,0 +1,292 @@ +package dev.relism.flash.http2; + +import dev.relism.flash.bytes.ByteWriter; +import dev.relism.flash.http2.frame.FrameFlags; +import dev.relism.flash.http2.frame.FrameType; +import dev.relism.flash.http2.frame.FrameWriteBuffer; +import dev.relism.flash.http2.hpack.HpackDecoder; +import dev.relism.flash.http2.hpack.HpackEncoder; +import dev.relism.flash.websocket.WebSocketFrame; +import java.io.ByteArrayOutputStream; +import java.io.Closeable; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.Socket; +import java.nio.charset.StandardCharsets; + +/** Minimal RFC 8441 peer used only by the live WebSocket-over-h2 tests. */ +final class H2WebSocketTestClient implements Closeable { + private static final int WINDOW = 2 * 1024 * 1024; + + private final Socket socket; + private final InputStream input; + private final OutputStream output; + private final ByteArrayOutputStream responseData = new ByteArrayOutputStream(); + private int connectionWindow = 65_535; + private int streamWindow = 65_535; + private int peerMaxFrame = 16_384; + private boolean connectProtocolAdvertised; + private boolean responseEnded; + + H2WebSocketTestClient(String host, int port, String path) throws Exception { + socket = new Socket(host, port); + socket.setSoTimeout(5_000); + input = socket.getInputStream(); + output = socket.getOutputStream(); + writePreface(); + awaitSettings(); + writeConnect(host + ":" + port, path); + int status = awaitStatus(); + if (status != 200) throw new IOException("extended CONNECT returned " + status); + } + + boolean connectProtocolAdvertised() { + return connectProtocolAdvertised; + } + + void sendText(String value) throws Exception { + sendWebSocketFrame(true, WebSocketFrame.OP_TEXT, value.getBytes(StandardCharsets.UTF_8), false); + } + + void sendFragmentedText(String first, String second) throws Exception { + sendWebSocketFrame( + false, WebSocketFrame.OP_TEXT, first.getBytes(StandardCharsets.UTF_8), false); + sendWebSocketFrame( + true, WebSocketFrame.OP_CONTINUATION, second.getBytes(StandardCharsets.UTF_8), false); + } + + void sendBinary(byte[] value) throws Exception { + sendWebSocketFrame(true, WebSocketFrame.OP_BINARY, value, false); + } + + byte[] readMessage(byte expectedOpcode) throws Exception { + responseData.reset(); + while (true) { + readAndHandleFrame(); + byte[] bytes = responseData.toByteArray(); + if (bytes.length < 2) continue; + int opcode = bytes[0] & 0x0f; + int marker = bytes[1] & 0x7f; + int headerLength; + long payloadLength; + if (marker < 126) { + headerLength = 2; + payloadLength = marker; + } else if (marker == 126) { + if (bytes.length < 4) continue; + headerLength = 4; + payloadLength = ((bytes[2] & 0xff) << 8) | (bytes[3] & 0xff); + } else { + if (bytes.length < 10) continue; + headerLength = 10; + payloadLength = 0; + for (int i = 2; i < 10; i++) payloadLength = (payloadLength << 8) | (bytes[i] & 0xffL); + } + if (payloadLength > Integer.MAX_VALUE || bytes.length < headerLength + payloadLength) { + continue; + } + if (opcode != expectedOpcode) throw new IOException("unexpected WebSocket opcode " + opcode); + byte[] payload = new byte[(int) payloadLength]; + System.arraycopy(bytes, headerLength, payload, 0, payload.length); + return payload; + } + } + + void closeGracefully() throws Exception { + sendWebSocketFrame(true, WebSocketFrame.OP_CLOSE, new byte[] {3, (byte) 232}, true); + while (!responseEnded) readAndHandleFrame(); + } + + @Override + public void close() throws IOException { + socket.close(); + } + + private void writePreface() throws IOException { + output.write(Http2Preface.clientPreface()); + ByteWriter bytes = new ByteWriter(64); + FrameWriteBuffer frames = new FrameWriteBuffer(bytes); + frames.beginFrame(FrameType.SETTINGS, 0, 0); + bytes.writeUInt16(Http2Settings.ENABLE_PUSH); + bytes.writeUInt32(0); + bytes.writeUInt16(Http2Settings.INITIAL_WINDOW_SIZE); + bytes.writeUInt32(WINDOW); + frames.endFrame(); + frames.beginFrame(FrameType.WINDOW_UPDATE, 0, 0); + bytes.writeUInt31(WINDOW - 65_535); + frames.endFrame(); + output.write(bytes.array(), 0, bytes.length()); + } + + private void awaitSettings() throws Exception { + while (!connectProtocolAdvertised) { + WireFrame frame = readFrame(); + if (frame.type == FrameType.SETTINGS.code() && (frame.flags & FrameFlags.ACK) == 0) { + for (int offset = 0; offset < frame.payload.length; offset += 6) { + int id = ((frame.payload[offset] & 0xff) << 8) | (frame.payload[offset + 1] & 0xff); + int value = readInt(frame.payload, offset + 2); + if (id == Http2Settings.ENABLE_CONNECT_PROTOCOL && value == 1) { + connectProtocolAdvertised = true; + } else if (id == Http2Settings.INITIAL_WINDOW_SIZE) { + streamWindow = value; + } else if (id == Http2Settings.MAX_FRAME_SIZE) { + peerMaxFrame = value; + } + } + writeEmpty(FrameType.SETTINGS, FrameFlags.ACK, 0); + } else { + handle(frame); + } + } + } + + private void writeConnect(String authority, String path) throws IOException { + ByteWriter bytes = new ByteWriter(256); + FrameWriteBuffer frame = new FrameWriteBuffer(bytes); + frame.beginFrame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1); + HpackEncoder.writeLiteralWithNameIndex( + bytes, 2, "CONNECT".getBytes(StandardCharsets.US_ASCII), false); + HpackEncoder.writeIndexed(bytes, 6); + HpackEncoder.writeLiteralWithNameIndex( + bytes, 1, authority.getBytes(StandardCharsets.US_ASCII), false); + HpackEncoder.writeLiteralWithNameIndex( + bytes, 4, path.getBytes(StandardCharsets.US_ASCII), false); + HpackEncoder.writeLiteral( + bytes, + ":protocol".getBytes(StandardCharsets.US_ASCII), + "websocket".getBytes(StandardCharsets.US_ASCII)); + frame.endFrame(); + output.write(bytes.array(), 0, bytes.length()); + } + + private int awaitStatus() throws Exception { + HpackDecoder decoder = new HpackDecoder(); + while (true) { + WireFrame frame = readFrame(); + if (frame.type != FrameType.HEADERS.code() || frame.streamId != 1) { + handle(frame); + continue; + } + int[] status = {0}; + decoder.decode( + frame.payload, + 0, + frame.payload.length, + (name, value, never) -> { + if (name.length() == 7 && name.byteAt(0) == ':') { + status[0] = + (value.byteAt(0) - '0') * 100 + + (value.byteAt(1) - '0') * 10 + + value.byteAt(2) + - '0'; + } + }); + return status[0]; + } + } + + private void sendWebSocketFrame(boolean fin, byte opcode, byte[] payload, boolean endStream) + throws Exception { + byte[] encoded = maskedFrame(fin, opcode, payload); + int offset = 0; + while (offset < encoded.length) { + while (connectionWindow <= 0 || streamWindow <= 0) readAndHandleFrame(); + int count = + Math.min( + encoded.length - offset, + Math.min(peerMaxFrame, Math.min(connectionWindow, streamWindow))); + writeData(encoded, offset, count, endStream && offset + count == encoded.length); + offset += count; + connectionWindow -= count; + streamWindow -= count; + } + } + + private void readAndHandleFrame() throws Exception { + handle(readFrame()); + } + + private void handle(WireFrame frame) throws IOException { + if (frame.type == FrameType.WINDOW_UPDATE.code()) { + int increment = readInt(frame.payload, 0) & 0x7fff_ffff; + if (frame.streamId == 0) connectionWindow += increment; + else if (frame.streamId == 1) streamWindow += increment; + } else if (frame.type == FrameType.DATA.code() && frame.streamId == 1) { + responseData.write(frame.payload); + responseEnded = (frame.flags & FrameFlags.END_STREAM) != 0; + } else if (frame.type == FrameType.SETTINGS.code() && (frame.flags & FrameFlags.ACK) == 0) { + writeEmpty(FrameType.SETTINGS, FrameFlags.ACK, 0); + } else if (frame.type == FrameType.RST_STREAM.code() && frame.streamId == 1) { + throw new IOException("WebSocket stream reset with " + readInt(frame.payload, 0)); + } else if (frame.type == FrameType.GOAWAY.code()) { + throw new IOException("HTTP/2 connection closed with " + readInt(frame.payload, 4)); + } + } + + private void writeData(byte[] payload, int offset, int length, boolean endStream) + throws IOException { + ByteWriter bytes = new ByteWriter(length + 9); + FrameWriteBuffer frame = new FrameWriteBuffer(bytes); + frame.beginFrame(FrameType.DATA, endStream ? FrameFlags.END_STREAM : 0, 1); + bytes.writeBytes(payload, offset, length); + frame.endFrame(); + output.write(bytes.array(), 0, bytes.length()); + } + + private void writeEmpty(FrameType type, int flags, int streamId) throws IOException { + byte[] frame = {0, 0, 0, (byte) type.code(), (byte) flags, 0, 0, 0, (byte) streamId}; + output.write(frame); + } + + private WireFrame readFrame() throws IOException { + byte[] header = input.readNBytes(9); + if (header.length != 9) throw new EOFException("HTTP/2 connection closed between frames"); + int length = ((header[0] & 0xff) << 16) | ((header[1] & 0xff) << 8) | (header[2] & 0xff); + int streamId = + ((header[5] & 0x7f) << 24) + | ((header[6] & 0xff) << 16) + | ((header[7] & 0xff) << 8) + | (header[8] & 0xff); + byte[] payload = input.readNBytes(length); + if (payload.length != length) throw new EOFException("HTTP/2 frame truncated"); + return new WireFrame(header[3] & 0xff, header[4] & 0xff, streamId, payload); + } + + private static byte[] maskedFrame(boolean fin, byte opcode, byte[] payload) { + int lengthBytes = payload.length <= 125 ? 0 : payload.length <= 0xffff ? 2 : 8; + byte[] frame = new byte[2 + lengthBytes + 4 + payload.length]; + int position = 0; + frame[position++] = (byte) ((fin ? 0x80 : 0) | opcode); + if (lengthBytes == 0) { + frame[position++] = (byte) (0x80 | payload.length); + } else if (lengthBytes == 2) { + frame[position++] = (byte) (0x80 | 126); + frame[position++] = (byte) (payload.length >>> 8); + frame[position++] = (byte) payload.length; + } else { + frame[position++] = (byte) (0x80 | 127); + long payloadLength = payload.length; + for (int shift = 56; shift >= 0; shift -= 8) { + frame[position++] = (byte) (payloadLength >>> shift); + } + } + byte[] mask = {1, 2, 3, 4}; + System.arraycopy(mask, 0, frame, position, mask.length); + position += mask.length; + for (int i = 0; i < payload.length; i++) { + frame[position + i] = (byte) (payload[i] ^ mask[i & 3]); + } + return frame; + } + + private static int readInt(byte[] bytes, int offset) { + return ((bytes[offset] & 0xff) << 24) + | ((bytes[offset + 1] & 0xff) << 16) + | ((bytes[offset + 2] & 0xff) << 8) + | (bytes[offset + 3] & 0xff); + } + + private record WireFrame(int type, int flags, int streamId, byte[] payload) {} +} diff --git a/flash/src/test/java/dev/relism/flash/http2/Http2SettingsTest.java b/flash/src/test/java/dev/relism/flash/http2/Http2SettingsTest.java index 8fabab6..38eb658 100644 --- a/flash/src/test/java/dev/relism/flash/http2/Http2SettingsTest.java +++ b/flash/src/test/java/dev/relism/flash/http2/Http2SettingsTest.java @@ -39,8 +39,10 @@ class Http2SettingsTest { } @Test - void validatesEnablePushInitialWindowAndFrameSize() { + void validatesBooleanSettingsInitialWindowAndFrameSize() { assertCode(Http2ErrorCode.PROTOCOL_ERROR, payload(Http2Settings.ENABLE_PUSH, 2)); + assertCode( + Http2ErrorCode.PROTOCOL_ERROR, payload(Http2Settings.ENABLE_CONNECT_PROTOCOL, 2)); assertCode( Http2ErrorCode.FLOW_CONTROL_ERROR, payload(Http2Settings.INITIAL_WINDOW_SIZE, 0x8000_0000)); assertCode(Http2ErrorCode.PROTOCOL_ERROR, payload(Http2Settings.MAX_FRAME_SIZE, 16_383)); diff --git a/flash/src/test/java/dev/relism/flash/http2/WebSocketOverH2Test.java b/flash/src/test/java/dev/relism/flash/http2/WebSocketOverH2Test.java new file mode 100644 index 0000000..d269f51 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/WebSocketOverH2Test.java @@ -0,0 +1,75 @@ +package dev.relism.flash.http2; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.relism.flash.extension.FlashApp; +import dev.relism.flash.extension.FlashConfiguration; +import dev.relism.flash.websocket.WebSocketFrame; +import dev.relism.flash.websocket.WebSocketHandler; +import dev.relism.flash.websocket.WebSocketSession; +import java.net.ServerSocket; +import java.nio.charset.StandardCharsets; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +class WebSocketOverH2Test { + private FlashApp app; + + @AfterEach + void stop() { + if (app != null) app.stop().join(); + } + + @Test + void opensEchoesFragmentsAndCarriesAMessageLargerThanTheFlowWindow() throws Exception { + int port = freePort(); + app = + FlashApp.create( + FlashConfiguration.builder() + .host("127.0.0.1") + .port(port) + .http2CleartextEnabled(true) + .wsFrameBufferSize(2 * 1024 * 1024) + .build()); + app.ws( + "/chat", + new WebSocketHandler() { + @Override + public void onOpen(WebSocketSession session) {} + + @Override + public void onMessage(WebSocketSession session, WebSocketFrame frame) { + try { + session.echo(frame); + } catch (Exception failure) { + throw new RuntimeException(failure); + } + } + }); + app.start(); + + try (H2WebSocketTestClient client = + new H2WebSocketTestClient("127.0.0.1", port, "/chat")) { + assertTrue(client.connectProtocolAdvertised()); + + client.sendFragmentedText("hel", "lo"); + assertArrayEquals( + "hello".getBytes(StandardCharsets.UTF_8), + client.readMessage(WebSocketFrame.OP_TEXT)); + + byte[] large = new byte[Http2Limits.INITIAL_WINDOW_SIZE_LOCAL + 128 * 1024 + 17]; + for (int i = 0; i < large.length; i++) large[i] = (byte) (i * 31); + client.sendBinary(large); + assertArrayEquals(large, client.readMessage(WebSocketFrame.OP_BINARY)); + + client.closeGracefully(); + } + } + + private static int freePort() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/WebSocketParityTest.java b/flash/src/test/java/dev/relism/flash/http2/WebSocketParityTest.java new file mode 100644 index 0000000..fb120ae --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/WebSocketParityTest.java @@ -0,0 +1,143 @@ +package dev.relism.flash.http2; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.relism.flash.extension.FlashApp; +import dev.relism.flash.extension.FlashConfiguration; +import dev.relism.flash.websocket.WebSocketFrame; +import dev.relism.flash.websocket.WebSocketHandler; +import dev.relism.flash.websocket.WebSocketSession; +import java.io.ByteArrayOutputStream; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +class WebSocketParityTest { + private FlashApp app; + + @AfterEach + void stop() { + if (app != null) app.stop().join(); + } + + @Test + void oneRouteAndHandlerEchoTheSameMessageOverHttp1AndHttp2() throws Exception { + int port = freePort(); + app = + FlashApp.create( + FlashConfiguration.builder() + .host("127.0.0.1") + .port(port) + .http2CleartextEnabled(true) + .build()); + app.ws( + "/parity", + new WebSocketHandler() { + @Override + public void onOpen(WebSocketSession session) {} + + @Override + public void onMessage(WebSocketSession session, WebSocketFrame frame) { + try { + session.echo(frame); + } catch (Exception failure) { + throw new RuntimeException(failure); + } + } + }); + app.start(); + + byte[] expected = "same-handler".getBytes(StandardCharsets.UTF_8); + byte[] overHttp1 = exchangeOverHttp1(port, expected); + byte[] overHttp2; + try (H2WebSocketTestClient client = + new H2WebSocketTestClient("127.0.0.1", port, "/parity")) { + client.sendText(new String(expected, StandardCharsets.UTF_8)); + overHttp2 = client.readMessage(WebSocketFrame.OP_TEXT); + client.closeGracefully(); + } + + assertArrayEquals(expected, overHttp1); + assertArrayEquals(overHttp1, overHttp2); + } + + private static byte[] exchangeOverHttp1(int port, byte[] payload) throws Exception { + try (Socket socket = new Socket("127.0.0.1", port)) { + socket.setSoTimeout(5_000); + InputStream input = socket.getInputStream(); + OutputStream output = socket.getOutputStream(); + String key = + Base64.getEncoder() + .encodeToString("flash-parity-key".getBytes(StandardCharsets.US_ASCII)); + String request = + "GET /parity HTTP/1.1\r\n" + + "Host: 127.0.0.1:" + + port + + "\r\nUpgrade: websocket\r\n" + + "Connection: Upgrade\r\nSec-WebSocket-Key: " + + key + + "\r\nSec-WebSocket-Version: 13\r\n\r\n"; + output.write(request.getBytes(StandardCharsets.US_ASCII)); + output.flush(); + assertTrue(readHeaders(input).startsWith("HTTP/1.1 101 Switching Protocols")); + + output.write(maskedFrame(WebSocketFrame.OP_TEXT, payload)); + output.flush(); + byte[] echoed = readServerFrame(input, WebSocketFrame.OP_TEXT); + output.write(maskedFrame(WebSocketFrame.OP_CLOSE, new byte[] {3, (byte) 232})); + output.flush(); + return echoed; + } + } + + private static String readHeaders(InputStream input) throws Exception { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + int previous3 = -1; + int previous2 = -1; + int previous1 = -1; + int current; + while ((current = input.read()) >= 0) { + bytes.write(current); + if (previous3 == '\r' && previous2 == '\n' && previous1 == '\r' && current == '\n') { + break; + } + previous3 = previous2; + previous2 = previous1; + previous1 = current; + } + return bytes.toString(StandardCharsets.US_ASCII); + } + + private static byte[] maskedFrame(byte opcode, byte[] payload) { + byte[] encoded = new byte[6 + payload.length]; + encoded[0] = (byte) (0x80 | opcode); + encoded[1] = (byte) (0x80 | payload.length); + byte[] mask = {1, 2, 3, 4}; + System.arraycopy(mask, 0, encoded, 2, mask.length); + for (int i = 0; i < payload.length; i++) { + encoded[6 + i] = (byte) (payload[i] ^ mask[i & 3]); + } + return encoded; + } + + private static byte[] readServerFrame(InputStream input, byte expectedOpcode) throws Exception { + byte[] header = input.readNBytes(2); + if (header.length != 2 || (header[0] & 0x0f) != expectedOpcode) { + throw new AssertionError("unexpected WebSocket response frame"); + } + int length = header[1] & 0x7f; + return input.readNBytes(length); + } + + private static int freePort() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/message/PseudoHeaderValidationTest.java b/flash/src/test/java/dev/relism/flash/http2/message/PseudoHeaderValidationTest.java index 4ec4e3f..a724323 100644 --- a/flash/src/test/java/dev/relism/flash/http2/message/PseudoHeaderValidationTest.java +++ b/flash/src/test/java/dev/relism/flash/http2/message/PseudoHeaderValidationTest.java @@ -42,6 +42,28 @@ class PseudoHeaderValidationTest { rejects(":method", "CONNECT", ":scheme", "https", ":authority", "example.com:443"); } + @Test + void validatesExtendedConnectShape() { + assertDoesNotThrow( + () -> + validate( + ":method", "CONNECT", + ":protocol", "websocket", + ":scheme", "https", + ":path", "/chat", + ":authority", "example.com")); + rejects( + ":method", "GET", + ":protocol", "websocket", + ":scheme", "https", + ":path", "/chat", + ":authority", "example.com"); + rejects( + ":method", "CONNECT", + ":protocol", "websocket", + ":authority", "example.com"); + } + @Test void rejectsUppercaseForbiddenAndInvalidTeFields() { rejects(validWith("X-Test", "1")); diff --git a/flash/src/test/java/dev/relism/flash/websocket/WebSocketLoopTest.java b/flash/src/test/java/dev/relism/flash/websocket/WebSocketLoopTest.java new file mode 100644 index 0000000..bb87a59 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/websocket/WebSocketLoopTest.java @@ -0,0 +1,92 @@ +package dev.relism.flash.websocket; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.InputStream; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; + +class WebSocketLoopTest { + @Test + void onOpenFailureStillReportsErrorClosesAndReleasesSession() { + RuntimeException failure = new RuntimeException("open failed"); + AtomicReference reported = new AtomicReference<>(); + AtomicInteger closes = new AtomicInteger(); + WebSocketSession session = + new WebSocketSession( + new ByteArrayInputStream(new byte[0]), new ByteArrayOutputStream(), 128); + + WebSocketLoop.run( + session, + new WebSocketHandler() { + @Override + public void onOpen(WebSocketSession opened) { + throw failure; + } + + @Override + public void onMessage(WebSocketSession opened, WebSocketFrame frame) {} + + @Override + public void onError(WebSocketSession opened, Throwable error) { + reported.set(error); + } + + @Override + public void onClose(WebSocketSession opened, int code) { + closes.incrementAndGet(); + } + }); + + assertSame(failure, reported.get()); + assertEquals(1, closes.get()); + assertFalse(session.isOpen()); + } + + @Test + void onCloseFailureCannotPreventTransportRelease() { + AtomicInteger inputCloses = new AtomicInteger(); + WebSocketSession session = + new WebSocketSession( + new InputStream() { + @Override + public int read() { + return -1; + } + + @Override + public void close() { + inputCloses.incrementAndGet(); + } + }, + new ByteArrayOutputStream(), + 128); + + assertThrows( + RuntimeException.class, + () -> + WebSocketLoop.run( + session, + new WebSocketHandler() { + @Override + public void onOpen(WebSocketSession opened) {} + + @Override + public void onMessage(WebSocketSession opened, WebSocketFrame frame) {} + + @Override + public void onClose(WebSocketSession opened, int code) { + throw new RuntimeException("close failed"); + } + })); + + assertEquals(1, inputCloses.get()); + assertFalse(session.isOpen()); + } +}