stop() {
- return CompletableFuture.runAsync(() -> {
- stopped = true;
- for (BoundListener bl : boundListeners) {
- try { bl.socket().close(); } catch (IOException e) { log.error("Error closing server socket", e); }
- }
- activeSockets.forEach(s -> { try { s.close(); } catch (IOException ignored) {} });
- executorService.shutdown();
- try {
- if (!executorService.awaitTermination(30, TimeUnit.SECONDS))
- executorService.shutdownNow();
- } catch (InterruptedException e) {
- executorService.shutdownNow();
- Thread.currentThread().interrupt();
- }
- });
- }
-
- // ── Hot-path ──────────────────────────────────────────────────────────────
-
- private void process(Socket socket) {
- try {
- executorService.submit(() -> {
- activeSockets.add(socket);
- try (socket;
- OutputStream out = new BufferedOutputStream(socket.getOutputStream())) {
-
- // TCP_NODELAY: disable Nagle's algorithm.
- // Small WS frames (< MSS) are sent immediately rather than
- // waiting up to 200 ms for more data to coalesce. Latency
- // drops significantly at the cost of slightly more TCP segments
- // under sustained bulk transfer — acceptable for interactive WS.
- socket.setTcpNoDelay(true);
- socket.setSendBufferSize(SOCKET_BUF_SIZE);
-
- // EX-30: force the TLS handshake explicitly, under a bounded timeout,
- // before any protocol decision is made. SSLSocket#getApplicationProtocol()
- // (which ProtocolNegotiator relies on) returns null until the handshake has
- // actually completed; nothing previously forced that before the first read,
- // which happened to work by accident (the JDK triggers it lazily on read)
- // but left ALPN unreadable at exactly the point negotiation needs it.
- if (socket instanceof SSLSocket sslSocketForHandshake) {
- socket.setSoTimeout(configuration.getHeaderReadTimeoutMs());
- sslSocketForHandshake.startHandshake();
- socket.setSoTimeout(0); // BufferedByteSource's deadline takes over below
- }
-
- // rawOut is the unbuffered socket stream — passed to WebSocketSession
- // directly. WS writes are already bulk (header + payload in two calls);
- // with TCP_NODELAY the kernel ships them without Nagle delay, so no
- // userspace buffer is needed and no flush() is required per frame.
- // HTTP responses continue to use the BufferedOutputStream (out) because
- // writeResponse() does many small individual writes that benefit from
- // userspace coalescing before a single syscall.
- OutputStream rawOut = socket.getOutputStream();
-
- // EX-10: the single buffered, deadline-aware view over this connection's
- // inbound bytes — see BufferedByteSource's Javadoc. Not part of the
- // try-with-resources list above because closing `socket` already closes
- // the stream it wraps (same reasoning that already applied to rawOut).
- BufferedByteSource in = new BufferedByteSource(socket.getInputStream(), socket);
-
- NegotiatedProtocol negotiated = negotiateProtocol(socket, in);
- if (negotiated == NegotiatedProtocol.H2) {
- // No Http2Connection exists yet (lands in Phase 8) — close cleanly
- // rather than attempt to speak a protocol this version cannot serve.
- return;
- }
-
- RequestParser parser = new RequestParser(
- configuration.getMaxHeaderBufferSize(),
- (InetSocketAddress) socket.getRemoteSocketAddress(),
- socket instanceof SSLSocket sslSocket ? sslSocket : null);
-
- byte[] idleProbe = new byte[1];
-
- while (!stopped) {
- // EX-07: wait for the next request to begin, bounded by the generous
- // idle-keep-alive timeout — sitting idle between keep-alive requests is
- // normal, not an attack. peek() lets us detect "bytes have started
- // arriving" without handing them to the parser under the wrong deadline.
- //
- // Skipped entirely when the parser already has bytes buffered from a
- // previous read (HTTP pipelining: a client that sent two requests back
- // to back before reading either response). In that case the next
- // request has, by definition, already started — peeking the *source*
- // for a fresh byte would wait for something that is never coming there,
- // since it already arrived and is sitting in the parser's own buffer.
- if (!parser.hasBufferedBytes()) {
- in.setDeadline(System.nanoTime() + configuration.getIdleKeepAliveTimeoutMs() * 1_000_000L);
- int firstByteSeen;
- try {
- firstByteSeen = in.peek(idleProbe, 0, 1);
- } catch (SocketTimeoutException e) {
- break; // idle timeout — nothing pending; close quietly, like EOF
- }
- if (firstByteSeen <= 0) break; // clean EOF
- }
-
- // Bytes have started arriving: tighten to the slowloris-specific bound
- // for the rest of the header block. A per-read SO_TIMEOUT alone would
- // never trip here — see BufferedByteSource's Javadoc.
- in.setDeadline(System.nanoTime() + configuration.getHeaderReadTimeoutMs() * 1_000_000L);
- Request request;
- try {
- request = parser.parse(in);
- } catch (MalformedRequestException e) {
- // EX-02/03/08/18: a fixed, minimal, non-customizable rejection —
- // never routed through the handler or the user's exception handler
- // (see MalformedRequestException's Javadoc) — and the connection is
- // always closed afterwards, never kept alive.
- Response rejection = new Response(e.status(), e.getMessage(), ContentType.TEXT_PLAIN);
- writeResponse(out, rejection, false);
- break;
- } catch (SocketTimeoutException e) {
- break; // header-read deadline exceeded — close
- }
- if (request == null) break;
-
- if (request.method() == HttpMethod.GET && isWebSocketUpgrade(request)) {
- in.clearDeadline(); // the WS session loop is long-lived; it paces itself
- WebSocketHandler wsHandler = wsRouter.route(request);
- if (wsHandler == null) {
- out.write(WS_REJECT_400);
- out.flush();
- break;
- }
- // Flush buffered HTTP bytes (the 101 response) before WebSocketSession
- // takes over rawOut — otherwise the handshake reply stays stuck in
- // the BufferedOutputStream buffer and the client never sees it.
- performHandshake(out, request);
- out.flush();
- request.drain();
- WebSocketSession session = new WebSocketSession(
- in, rawOut, configuration.getWsFrameBufferSize(), request, false);
- runWsLoop(session, wsHandler);
- return;
- }
-
- // Headers are fully read; the body (if any) may still be pending —
- // whether the handler consumes it or the automatic drain() below does,
- // bound it by the same deadline (EX-07).
- in.setDeadline(System.nanoTime() + configuration.getBodyReadTimeoutMs() * 1_000_000L);
-
- boolean keepAlive = isKeepAlive(request);
- Response response = new Response(200, ContentType.TEXT_PLAIN);
-
- RequestHandler handler = router.route(request);
- if (handler == null) handler = router.getNotFoundHandler();
-
- try {
- Object result = handler.handle(request, response);
- if (result instanceof Response r) response = r;
- else if (result != null) response.setBody(result);
- } catch (Exception ex) {
- Object result = router.getExceptionHandler().handle(ex, request, response);
- if (result instanceof Response r) response = r;
- else if (result != null) response.setBody(result);
- }
-
- writeResponse(out, response, keepAlive);
- request.drain();
- in.clearDeadline();
- if (!keepAlive) break;
- }
-
- } catch (IOException e) {
- if (!stopped) {
- if (e instanceof java.net.SocketException)
- log.debug("Connection closed: {}", e.getMessage());
- else
- log.error("I/O error handling request", e);
- }
- } catch (Exception e) {
- // Anything not an IOException here means a collaborator misbehaved on the TLS
- // handshake path — most likely a custom TlsConfig#ofContext KeyManager/
- // TrustManager throwing (e.g. a failed DB lookup or on-demand cert issuance).
- // That failure is isolated to this one virtual thread/connection: the
- // try-with-resources above still closes the socket, the finally below still
- // runs, and the accept loop (a different thread entirely) never sees this.
- if (!stopped) log.error("Unexpected error handling connection", e);
- } finally {
- activeSockets.remove(socket);
- }
- });
- } catch (RejectedExecutionException ignored) {
- try { socket.close(); } catch (IOException e) { log.debug("Error closing socket on shutdown", e); }
- }
- }
-
- // ── Protocol negotiation ───────────────────────────────────────────────────
-
- /**
- * Decides h1 vs h2 for one connection, applying {@link FlashConfiguration#isHttp2Enabled()}
- * to the plaintext (h2c) path — see {@link ProtocolNegotiator}'s Javadoc for why the flag is
- * applied here rather than inside the negotiator itself. TLS/ALPN detection costs nothing
- * (the handshake already resolved it) and is therefore always performed, regardless of the
- * flag: what the flag gates is whether Flash even attempts the h2c preface peek on a
- * plaintext socket, so that a plaintext connection with the feature left at its default
- * behaves byte-for-byte like pre-HTTP/2 Flash.
- */
- private NegotiatedProtocol negotiateProtocol(Socket socket, BufferedByteSource in) throws IOException {
- if (socket instanceof SSLSocket) {
- return ProtocolNegotiator.negotiate(socket, in);
- }
- if (!configuration.isHttp2Enabled()) {
- return NegotiatedProtocol.HTTP_1_1;
- }
- in.setDeadline(System.nanoTime() + configuration.getIdleKeepAliveTimeoutMs() * 1_000_000L);
- try {
- return ProtocolNegotiator.negotiate(socket, in);
- } finally {
- in.clearDeadline();
- }
- }
-
- // ── WebSocket upgrade detection (zero-alloc) ──────────────────────────────
-
- private static boolean isWebSocketUpgrade(Request request) {
- ByteView upgrade = request.getRequestLine().getHeaders().view("Upgrade");
- if (upgrade == null) return false;
- if (!tokenEqualsIgnoreCase(upgrade, 0, upgrade.length(), "websocket")) return false;
- return connectionContainsUpgrade(request);
- }
-
- private static boolean connectionContainsUpgrade(Request request) {
- ByteView conn = request.getRequestLine().getHeaders().view("Connection");
- if (conn == null) return false;
- int len = conn.length(), i = 0;
- while (i < len) {
- while (i < len && conn.byteAt(i) == ' ') i++;
- int start = i;
- while (i < len && conn.byteAt(i) != ',') i++;
- if (tokenEqualsIgnoreCase(conn, start, i, "upgrade")) return true;
- i++;
- }
- return false;
- }
-
- private static boolean tokenEqualsIgnoreCase(ByteView view, int start, int end, String token) {
- int tlen = token.length();
- int wlen = end - start;
- while (wlen > 0 && view.byteAt(start + wlen - 1) == ' ') wlen--;
- if (wlen != tlen) return false;
- for (int i = 0; i < tlen; i++) {
- byte b = view.byteAt(start + i);
- if (b >= 'A' && b <= 'Z') b += 32;
- if (b != (byte) token.charAt(i)) return false;
- }
- return true;
- }
-
- // ── WebSocket handshake ───────────────────────────────────────────────────
-
- private void performHandshake(OutputStream out, Request request) throws IOException {
- ByteView keyView = request.getRequestLine().getHeaders().view("Sec-WebSocket-Key");
- if (keyView == null) throw new IOException("Missing Sec-WebSocket-Key header");
-
- MessageDigest sha1 = SHA1.get();
- sha1.reset();
- for (int i = 0, len = keyView.length(); i < len; i++) sha1.update(keyView.byteAt(i));
- sha1.update(WS_GUID_BYTES);
-
- byte[] accept = Base64.getEncoder().encode(sha1.digest());
-
- out.write(WS_HANDSHAKE_PREFIX);
- out.write(accept);
- out.write(WS_HANDSHAKE_SUFFIX);
- out.flush();
- }
-
- // ── WebSocket session loop ────────────────────────────────────────────────
-
- private void runWsLoop(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 (IOException e) {
- handler.onError(session, e);
- } finally {
- handler.onClose(session, session.closeCode());
- session.forceClose();
- }
- }
-
- // ── HTTP keep-alive detection ─────────────────────────────────────────────
-
- private static boolean isKeepAlive(Request request) {
- if (request.headerEquals("Connection", "close")) return false;
- ByteView protocol = request.getRequestLine().getProtocol();
- int plen = protocol.length();
- if (plen == 8) {
- byte minor = protocol.byteAt(7);
- if (minor == '1') return true;
- if (minor == '0') return request.headerEquals("Connection", "keep-alive");
- }
- log.debug("Unrecognised protocol '{}', treating as close", protocol);
- return false;
- }
-
- // ── Response serialisation ────────────────────────────────────────────────
-
- private static void writeResponse(OutputStream out, Response response, boolean keepAlive) throws IOException {
- out.write(HTTP_1_1);
- byte[] statusBytes = response.getStatusBytes();
- if (statusBytes != null) out.write(statusBytes);
- else writeStatusPhrase(out, response.getStatusCode());
- out.write(CRLF);
- out.write(CONTENT_TYPE);
- out.write(response.getContentType());
- out.write(CRLF);
- response.writeHeaders(out);
-
- if (response.isStreaming()) {
- writeStreamingBody(out, response, keepAlive);
- } else {
- byte[] body = response.getBody();
- out.write(CONTENT_LENGTH);
- writeLong(out, body != null ? body.length : 0);
- out.write(CRLF);
- out.write(keepAlive ? CONNECTION_KEEPALIVE : CONNECTION_CLOSE);
- out.write(CRLF);
- if (body != null) out.write(body);
- }
- out.flush();
- }
-
- private static void writeStreamingBody(OutputStream out, Response response, boolean keepAlive) throws IOException {
- if (!response.isChunked()) {
- out.write(CONTENT_LENGTH);
- writeLong(out, response.getStreamLength());
- out.write(CRLF);
- out.write(keepAlive ? CONNECTION_KEEPALIVE : CONNECTION_CLOSE);
- out.write(CRLF);
- relay(response.getStream(), out);
- } else {
- out.write(TRANSFER_CHUNKED);
- out.write(keepAlive ? CONNECTION_KEEPALIVE : CONNECTION_CLOSE);
- out.write(CRLF);
- writeChunked(out, response.getStream());
- }
- }
-
- /**
- * Copies {@code in} to {@code out} until EOF, same contract as {@link InputStream#transferTo}
- * — but via {@link #STREAM_RELAY_BUFFER} instead of a fresh {@code byte[]} per call, which is
- * what {@code transferTo}'s own (JDK-internal) implementation would otherwise allocate on
- * every streamed response.
- */
- private static void relay(InputStream in, OutputStream out) throws IOException {
- byte[] buf = STREAM_RELAY_BUFFER.get();
- int n;
- while ((n = in.read(buf)) > 0) out.write(buf, 0, n);
- }
-
- private static void writeStatusPhrase(OutputStream out, int statusCode) throws IOException {
- byte[] phrase = HttpStatus.bytesForCode(statusCode);
- if (phrase != null) out.write(phrase);
- else { writeLong(out, statusCode); out.write(UNKNOWN_STATUS_SUFFIX); }
- }
-
- private static void writeLong(OutputStream out, long value) throws IOException {
- if (value == 0) { out.write('0'); return; }
- byte[] buf = LONG_BUF.get();
- int pos = buf.length;
- boolean neg = value < 0;
- if (neg) value = -value;
- do { buf[--pos] = (byte) ('0' + value % 10); value /= 10; } while (value > 0);
- if (neg) buf[--pos] = '-';
- out.write(buf, pos, buf.length - pos);
- }
-
- private static void writeChunked(OutputStream out, InputStream stream) throws IOException {
- byte[] buf = STREAM_RELAY_BUFFER.get();
- int n;
- while ((n = stream.read(buf)) > 0) {
- writeHex(out, n);
- out.write(CRLF);
- out.write(buf, 0, n);
- out.write(CRLF);
- }
- out.write(FINAL_CHUNK);
- }
-
- private static void writeHex(OutputStream out, int value) throws IOException {
- int shift = 28;
- boolean leading = true;
- while (shift >= 0) {
- int digit = (value >>> shift) & 0xF;
- if (digit != 0 || !leading) {
- leading = false;
- out.write(digit < 10 ? '0' + digit : 'a' + digit - 10);
- }
- shift -= 4;
- }
- if (leading) out.write('0');
- }
-}
\ No newline at end of file
diff --git a/flash/src/main/java/dev/relism/flash/RequestParser.java b/flash/src/main/java/dev/relism/flash/RequestParser.java
index c1a2d99..dcfaa76 100644
--- a/flash/src/main/java/dev/relism/flash/RequestParser.java
+++ b/flash/src/main/java/dev/relism/flash/RequestParser.java
@@ -106,7 +106,7 @@ public class RequestParser {
* complete), so an idle-timeout wait on the underlying source would wait for bytes that
* were never going to arrive there — they are already here.
*/
- boolean hasBufferedBytes() {
+ public boolean hasBufferedBytes() {
return bufLen > 0;
}
diff --git a/flash/src/main/java/dev/relism/flash/ServerHandle.java b/flash/src/main/java/dev/relism/flash/ServerHandle.java
index d1d3cc5..a4926c5 100644
--- a/flash/src/main/java/dev/relism/flash/ServerHandle.java
+++ b/flash/src/main/java/dev/relism/flash/ServerHandle.java
@@ -4,6 +4,7 @@ import dev.relism.flash.extension.FlashApp;
import dev.relism.flash.extension.FlashConfiguration;
import dev.relism.flash.routing.AbstractRouter;
import dev.relism.flash.routing.AbstractWsRouter;
+import dev.relism.flash.transport.TransportFactory;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
@@ -11,7 +12,8 @@ import java.util.concurrent.CompletableFuture;
/**
* Public handle to the underlying HTTP transport. Returned by {@link #create}
* so that {@link FlashApp} can start and stop the server
- * without holding a direct reference to the package-private {@link HttpServer}.
+ * without holding a direct reference to the transport's internal composition
+ * ({@link TransportFactory}, {@code EX-34}).
*/
public interface ServerHandle {
@@ -30,6 +32,6 @@ public interface ServerHandle {
static ServerHandle create(FlashConfiguration config,
AbstractRouter httpRouter,
AbstractWsRouter wsRouter) throws IOException {
- return new HttpServer(config, httpRouter, wsRouter);
+ return TransportFactory.create(config, httpRouter, wsRouter);
}
}
diff --git a/flash/src/main/java/dev/relism/flash/extension/FlashConfiguration.java b/flash/src/main/java/dev/relism/flash/extension/FlashConfiguration.java
index 14cdb7e..31d14cd 100644
--- a/flash/src/main/java/dev/relism/flash/extension/FlashConfiguration.java
+++ b/flash/src/main/java/dev/relism/flash/extension/FlashConfiguration.java
@@ -109,6 +109,14 @@ public class FlashConfiguration {
@Builder.Default
boolean http2Enabled = false;
+ /**
+ * Whether every response includes a {@code Date} header (RFC 9110 §6.6.1). Default
+ * {@code true}; set {@code false} if Flash sits behind a reverse proxy that already adds
+ * one, to skip the (already cheap — see {@code dev.relism.flash.http.DateHeader}) write.
+ */
+ @Builder.Default
+ boolean sendDate = true;
+
/** One bind target: a TCP port, an optional bind host (default: all interfaces), and optional TLS. */
public record Listener(int port, String host, TlsConfig tls) {
public Listener(int port) { this(port, null, null); }
diff --git a/flash/src/main/java/dev/relism/flash/http/DateHeader.java b/flash/src/main/java/dev/relism/flash/http/DateHeader.java
new file mode 100644
index 0000000..679c703
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/http/DateHeader.java
@@ -0,0 +1,56 @@
+package dev.relism.flash.http;
+
+import java.nio.charset.StandardCharsets;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+
+/**
+ * {@code EX-16}: RFC 9110 §6.6.1 — an origin server with a clock SHOULD send {@code Date}.
+ * Flash never emitted it. Rather than formatting a timestamp on every response, a single
+ * daemon thread refreshes a pre-encoded {@code "Date: ...\r\n"} field line once per second into
+ * a {@code volatile byte[]}; {@code Http1ResponseWriter} writes it with one
+ * {@code OutputStream#write(byte[])} — the cost per response is one volatile read and one
+ * write, never a format call (R4).
+ *
+ * The parallel HPACK-encoded rendering for HTTP/2 responses is added in Phase 9.
+ */
+public final class DateHeader {
+
+ private DateHeader() {
+ }
+
+ private static final DateTimeFormatter FORMATTER =
+ DateTimeFormatter.RFC_1123_DATE_TIME.withZone(ZoneOffset.UTC);
+
+ private static volatile byte[] current = encode();
+
+ static {
+ Thread refresher = new Thread(() -> {
+ while (true) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ current = encode();
+ }
+ }, "flash-date-header");
+ refresher.setDaemon(true);
+ refresher.start();
+ }
+
+ private static byte[] encode() {
+ String line = "Date: " + FORMATTER.format(ZonedDateTime.now(ZoneOffset.UTC)) + "\r\n";
+ return line.getBytes(StandardCharsets.US_ASCII);
+ }
+
+ /**
+ * The current pre-encoded {@code "Date: ...\r\n"} field line, accurate to within one
+ * second. Never allocates — the same array is returned until the next refresh.
+ */
+ public static byte[] bytes() {
+ return current;
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/http1/Http1Connection.java b/flash/src/main/java/dev/relism/flash/http1/Http1Connection.java
new file mode 100644
index 0000000..43cc581
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/http1/Http1Connection.java
@@ -0,0 +1,131 @@
+package dev.relism.flash.http1;
+
+import dev.relism.flash.RequestParser;
+import dev.relism.flash.exceptions.MalformedRequestException;
+import dev.relism.flash.http.ContentType;
+import dev.relism.flash.http.HttpMethod;
+import dev.relism.flash.models.Request;
+import dev.relism.flash.models.RequestHandler;
+import dev.relism.flash.models.Response;
+import dev.relism.flash.transport.BufferedByteSource;
+import dev.relism.flash.transport.ConnectionContext;
+import dev.relism.flash.transport.ConnectionProtocol;
+import dev.relism.flash.websocket.WebSocketHandler;
+import dev.relism.flash.websocket.WebSocketLoop;
+import dev.relism.flash.websocket.WebSocketSession;
+import dev.relism.flash.websocket.WebSocketUpgrade;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.SocketTimeoutException;
+
+/**
+ * The HTTP/1.1 keep-alive request loop: parse → route → handle → respond, repeated until the
+ * connection closes. Sole responsibility: drive that loop for one connection; parsing lives in
+ * {@link RequestParser}, serialization in {@link Http1ResponseWriter}, and the WebSocket upgrade
+ * path hands off to {@link WebSocketUpgrade}/{@link WebSocketLoop} entirely — once a connection
+ * upgrades, this class has nothing further to do with it.
+ */
+public final class Http1Connection implements ConnectionProtocol {
+
+ @Override
+ public void run(ConnectionContext ctx) throws IOException {
+ RequestParser parser = new RequestParser(
+ ctx.configuration().getMaxHeaderBufferSize(),
+ ctx.remoteAddress(),
+ ctx.sslSocket());
+
+ BufferedByteSource in = ctx.in();
+ OutputStream out = ctx.out();
+ byte[] idleProbe = new byte[1];
+
+ while (!ctx.stopped().getAsBoolean()) {
+ // EX-07: wait for the next request to begin, bounded by the generous
+ // idle-keep-alive timeout — sitting idle between keep-alive requests is normal, not
+ // an attack. Skipped when the parser already has bytes buffered from a previous
+ // read (HTTP pipelining): the next request has, by definition, already started, so
+ // waiting on the *source* for a fresh byte would wait for something that already
+ // arrived and is sitting in the parser's own buffer.
+ if (!parser.hasBufferedBytes()) {
+ in.setDeadline(System.nanoTime() + ctx.configuration().getIdleKeepAliveTimeoutMs() * 1_000_000L);
+ int firstByteSeen;
+ try {
+ firstByteSeen = in.peek(idleProbe, 0, 1);
+ } catch (SocketTimeoutException e) {
+ break; // idle timeout — nothing pending; close quietly, like EOF
+ }
+ if (firstByteSeen <= 0) break; // clean EOF
+ }
+
+ // Bytes have started arriving: tighten to the slowloris-specific bound for the rest
+ // of the header block.
+ in.setDeadline(System.nanoTime() + ctx.configuration().getHeaderReadTimeoutMs() * 1_000_000L);
+ Request request;
+ try {
+ request = parser.parse(in);
+ } catch (MalformedRequestException e) {
+ // EX-02/03/08/18: a fixed, minimal, non-customizable rejection — never routed
+ // through a handler or the user's exception handler — and the connection is
+ // always closed afterwards, never kept alive.
+ Response rejection = new Response(e.status(), e.getMessage(), ContentType.TEXT_PLAIN);
+ Http1ResponseWriter.writeResponse(out, rejection, null, false, ctx.configuration().isSendDate(), ctx.scratch());
+ break;
+ } catch (SocketTimeoutException e) {
+ break; // header-read deadline exceeded — close
+ }
+ if (request == null) break;
+
+ if (request.method() == HttpMethod.GET && WebSocketUpgrade.isWebSocketUpgrade(request)) {
+ in.clearDeadline(); // the WS session loop is long-lived; it paces itself
+ WebSocketHandler wsHandler = ctx.wsRouter().route(request);
+ if (wsHandler == null) {
+ out.write(WebSocketUpgrade.REJECT_400);
+ out.flush();
+ break;
+ }
+ // Flush buffered HTTP bytes (the 101 response) before WebSocketSession takes
+ // over rawOut — otherwise the handshake reply stays stuck in the buffered
+ // stream and the client never sees it.
+ WebSocketUpgrade.performHandshake(out, request, ctx.scratch());
+ out.flush();
+ request.drain();
+ WebSocketSession session = new WebSocketSession(
+ in, ctx.rawOut(), ctx.configuration().getWsFrameBufferSize(), request, false);
+ WebSocketLoop.run(session, wsHandler);
+ return;
+ }
+
+ // Headers are fully read; the body (if any) may still be pending — whether the
+ // handler consumes it or the automatic drain() below does, bound it by the same
+ // deadline.
+ in.setDeadline(System.nanoTime() + ctx.configuration().getBodyReadTimeoutMs() * 1_000_000L);
+
+ boolean keepAlive = Http1KeepAlive.isKeepAlive(request);
+ Response response = new Response(200, ContentType.TEXT_PLAIN);
+
+ RequestHandler handler = ctx.router().route(request);
+ if (handler == null) handler = ctx.router().getNotFoundHandler();
+
+ try {
+ Object result = handler.handle(request, response);
+ if (result instanceof Response r) response = r;
+ else if (result != null) response.setBody(result);
+ } catch (Exception ex) {
+ Object result = ctx.router().getExceptionHandler().handle(ex, request, response);
+ if (result instanceof Response r) response = r;
+ else if (result != null) response.setBody(result);
+ }
+
+ // EX-32: re-checked here, not just before dispatch — a shutdown that begins while
+ // this handler was running (the common case: draining connections mid-request) must
+ // still force this response to Connection: close, not whatever was decided before
+ // the handler ran.
+ boolean actuallyKeepAlive = keepAlive && !ctx.stopped().getAsBoolean();
+ Http1ResponseWriter.writeResponse(out, response, request.method(), actuallyKeepAlive,
+ ctx.configuration().isSendDate(), ctx.scratch());
+ request.drain();
+ in.clearDeadline();
+ if (!actuallyKeepAlive) break;
+ }
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/http1/Http1KeepAlive.java b/flash/src/main/java/dev/relism/flash/http1/Http1KeepAlive.java
new file mode 100644
index 0000000..30857e9
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/http1/Http1KeepAlive.java
@@ -0,0 +1,71 @@
+package dev.relism.flash.http1;
+
+import dev.relism.flash.models.Request;
+import dev.relism.fpr.core.ByteView;
+
+/**
+ * HTTP/1.1 keep-alive decision (RFC 9110 §7.6.1) and the shared {@code Connection} header
+ * token-list scanner both it and WebSocket upgrade detection need.
+ *
+ *
{@code EX-13}: {@code Connection} is a comma-separated token list
+ * (e.g. {@code "Connection: keep-alive, Upgrade"}), not a single value — a whole-value compare
+ * against {@code "close"} misses exactly that case. {@link #tokenListContains} is the one
+ * scanner both this class's {@link #isKeepAlive} and {@code WebSocketUpgrade}'s
+ * {@code Connection: Upgrade} check use, so the two can never drift apart again.
+ */
+public final class Http1KeepAlive {
+
+ private Http1KeepAlive() {
+ }
+
+ /**
+ * Whether the connection should remain open after this response. HTTP/1.1 defaults to
+ * keep-alive unless {@code Connection} lists {@code close}; HTTP/1.0 defaults to close
+ * unless it lists {@code keep-alive}.
+ */
+ public static boolean isKeepAlive(Request request) {
+ if (connectionContainsToken(request, "close")) return false;
+ ByteView protocol = request.getRequestLine().getProtocol();
+ int plen = protocol.length();
+ if (plen == 8) {
+ byte minor = protocol.byteAt(7);
+ if (minor == '1') return true;
+ if (minor == '0') return connectionContainsToken(request, "keep-alive");
+ }
+ return false;
+ }
+
+ /** Whether the request's {@code Connection} header lists {@code token} (case-insensitive). */
+ public static boolean connectionContainsToken(Request request, String token) {
+ ByteView conn = request.getRequestLine().getHeaders().view("Connection");
+ if (conn == null) return false;
+ return tokenListContains(conn, token);
+ }
+
+ /** Scans a comma-separated token list for {@code token} (case-insensitive, OWS-tolerant). */
+ public static boolean tokenListContains(ByteView view, String token) {
+ int len = view.length(), i = 0;
+ while (i < len) {
+ while (i < len && view.byteAt(i) == ' ') i++;
+ int start = i;
+ while (i < len && view.byteAt(i) != ',') i++;
+ if (tokenEqualsIgnoreCase(view, start, i, token)) return true;
+ i++;
+ }
+ return false;
+ }
+
+ /** Case-insensitive compare of {@code view[start, end)}, trimming trailing spaces, against {@code token}. */
+ public static boolean tokenEqualsIgnoreCase(ByteView view, int start, int end, String token) {
+ int tlen = token.length();
+ int wlen = end - start;
+ while (wlen > 0 && view.byteAt(start + wlen - 1) == ' ') wlen--;
+ if (wlen != tlen) return false;
+ for (int i = 0; i < tlen; i++) {
+ byte b = view.byteAt(start + i);
+ if (b >= 'A' && b <= 'Z') b += 32;
+ if (b != (byte) token.charAt(i)) return false;
+ }
+ return true;
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java b/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java
new file mode 100644
index 0000000..9dd8ef5
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java
@@ -0,0 +1,170 @@
+package dev.relism.flash.http1;
+
+import dev.relism.flash.http.DateHeader;
+import dev.relism.flash.http.HttpMethod;
+import dev.relism.flash.http.HttpStatus;
+import dev.relism.flash.models.Response;
+import dev.relism.flash.transport.ConnectionScratch;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * Serializes a {@link Response} as an HTTP/1.1 message. Sole responsibility: response
+ * serialization — routing, handler dispatch, and the request loop live in
+ * {@link Http1Connection}.
+ *
+ *
Zero-allocation: the decimal encoding of the status code / {@code Content-Length} and the
+ * relay buffer used for streaming bodies both come from the connection's {@link ConnectionScratch}
+ * ({@code EX-06}) instead of a per-call allocation or a {@code ThreadLocal}.
+ */
+public final class Http1ResponseWriter {
+
+ private Http1ResponseWriter() {
+ }
+
+ private static final byte[] HTTP_1_1 = "HTTP/1.1 ".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] CRLF = "\r\n".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] CONTENT_TYPE = "Content-Type: ".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] CONTENT_LENGTH = "Content-Length: ".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] CONNECTION_CLOSE = "Connection: close\r\n".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] CONNECTION_KEEPALIVE = "Connection: keep-alive\r\n".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] TRANSFER_CHUNKED = "Transfer-Encoding: chunked\r\n".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] FINAL_CHUNK = "0\r\n\r\n".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] UNKNOWN_STATUS_SUFFIX = " Unknown".getBytes(StandardCharsets.UTF_8);
+
+ /**
+ * Writes {@code response} to {@code out} as a complete HTTP/1.1 message.
+ *
+ * @param method the request method — {@code null} is treated as "not HEAD" (used for
+ * parser-rejection responses, which never reach a handler and so have no
+ * associated method)
+ * @param sendDate whether to include the {@code Date} header ({@code FlashConfiguration#isSendDate()})
+ */
+ public static void writeResponse(OutputStream out, Response response, HttpMethod method,
+ boolean keepAlive, boolean sendDate, ConnectionScratch scratch) throws IOException {
+ int statusCode = response.getStatusCode();
+ // RFC 9110 §8.6/§15: 204, 304 and all 1xx responses MUST NOT carry Content-Length or a
+ // body at all — not "an empty one", none (EX-15). A HEAD response (RFC 9110 §9.3.2)
+ // still reports the Content-Length GET would have, but never writes body bytes.
+ boolean noContentAllowed = statusCode == 204 || statusCode == 304 || (statusCode >= 100 && statusCode < 200);
+ boolean suppressBody = noContentAllowed || method == HttpMethod.HEAD;
+
+ out.write(HTTP_1_1);
+ byte[] statusBytes = response.getStatusBytes();
+ if (statusBytes != null) out.write(statusBytes);
+ else writeStatusPhrase(out, statusCode, scratch);
+ out.write(CRLF);
+
+ // EX-15: a Content-Type of ContentType.NONE (empty byte[]) used to still emit the line
+ // "Content-Type: \r\n" — a header with no value. Skip the line entirely instead.
+ byte[] contentType = response.getContentType();
+ if (contentType != null && contentType.length > 0) {
+ out.write(CONTENT_TYPE);
+ out.write(contentType);
+ out.write(CRLF);
+ }
+
+ // EX-16: precomputed once per second by a shared daemon thread — one volatile read,
+ // one write(byte[]), never a per-response format call.
+ if (sendDate) out.write(DateHeader.bytes());
+
+ response.writeHeaders(out);
+
+ if (response.isStreaming()) {
+ writeStreamingBody(out, response, keepAlive, noContentAllowed, suppressBody, scratch);
+ } else {
+ byte[] body = response.getBody();
+ int len = body != null ? body.length : 0;
+ if (!noContentAllowed) {
+ out.write(CONTENT_LENGTH);
+ writeLong(out, len, scratch);
+ out.write(CRLF);
+ }
+ out.write(keepAlive ? CONNECTION_KEEPALIVE : CONNECTION_CLOSE);
+ out.write(CRLF);
+ // EX-14: HEAD reports the Content-Length GET would have (above) but never writes
+ // the body itself.
+ if (body != null && !suppressBody) out.write(body);
+ }
+ out.flush();
+ }
+
+ private static void writeStreamingBody(OutputStream out, Response response, boolean keepAlive,
+ boolean noContentAllowed, boolean suppressBody,
+ ConnectionScratch scratch) throws IOException {
+ if (!response.isChunked()) {
+ if (!noContentAllowed) {
+ out.write(CONTENT_LENGTH);
+ writeLong(out, response.getStreamLength(), scratch);
+ out.write(CRLF);
+ }
+ out.write(keepAlive ? CONNECTION_KEEPALIVE : CONNECTION_CLOSE);
+ out.write(CRLF);
+ if (!suppressBody) relay(response.getStream(), out, scratch);
+ } else {
+ out.write(TRANSFER_CHUNKED);
+ out.write(keepAlive ? CONNECTION_KEEPALIVE : CONNECTION_CLOSE);
+ out.write(CRLF);
+ // A HEAD response still declares the Transfer-Encoding GET would have used (RFC
+ // 9110 §9.3.2) but writes zero body bytes — not even the final-chunk marker, since
+ // there is no chunk framing at all for a message with no body.
+ if (!suppressBody) writeChunked(out, response.getStream(), scratch);
+ }
+ }
+
+ /**
+ * Copies {@code in} to {@code out} until EOF, via {@link ConnectionScratch#relayBuffer}
+ * instead of a fresh {@code byte[]} per call.
+ */
+ private static void relay(InputStream in, OutputStream out, ConnectionScratch scratch) throws IOException {
+ byte[] buf = scratch.relayBuffer;
+ int n;
+ while ((n = in.read(buf)) > 0) out.write(buf, 0, n);
+ }
+
+ private static void writeStatusPhrase(OutputStream out, int statusCode, ConnectionScratch scratch) throws IOException {
+ byte[] phrase = HttpStatus.bytesForCode(statusCode);
+ if (phrase != null) out.write(phrase);
+ else { writeLong(out, statusCode, scratch); out.write(UNKNOWN_STATUS_SUFFIX); }
+ }
+
+ private static void writeLong(OutputStream out, long value, ConnectionScratch scratch) throws IOException {
+ if (value == 0) { out.write('0'); return; }
+ byte[] buf = scratch.decimalBuffer;
+ int pos = buf.length;
+ boolean neg = value < 0;
+ if (neg) value = -value;
+ do { buf[--pos] = (byte) ('0' + value % 10); value /= 10; } while (value > 0);
+ if (neg) buf[--pos] = '-';
+ out.write(buf, pos, buf.length - pos);
+ }
+
+ private static void writeChunked(OutputStream out, InputStream stream, ConnectionScratch scratch) throws IOException {
+ byte[] buf = scratch.relayBuffer;
+ int n;
+ while ((n = stream.read(buf)) > 0) {
+ writeHex(out, n);
+ out.write(CRLF);
+ out.write(buf, 0, n);
+ out.write(CRLF);
+ }
+ out.write(FINAL_CHUNK);
+ }
+
+ private static void writeHex(OutputStream out, int value) throws IOException {
+ int shift = 28;
+ boolean leading = true;
+ while (shift >= 0) {
+ int digit = (value >>> shift) & 0xF;
+ if (digit != 0 || !leading) {
+ leading = false;
+ out.write(digit < 10 ? '0' + digit : 'a' + digit - 10);
+ }
+ shift -= 4;
+ }
+ if (leading) out.write('0');
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/AcceptLoop.java b/flash/src/main/java/dev/relism/flash/transport/AcceptLoop.java
new file mode 100644
index 0000000..42bea09
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/AcceptLoop.java
@@ -0,0 +1,29 @@
+package dev.relism.flash.transport;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.function.BooleanSupplier;
+
+/**
+ * Single accept-loop body — runs on each of a listener's accept threads. All threads for the
+ * same listener block on the same {@link java.net.ServerSocket}; the JVM ensures only one wakes
+ * per incoming connection (no thundering herd). Other listeners' accept threads are entirely
+ * independent.
+ */
+@Slf4j
+public final class AcceptLoop {
+
+ private AcceptLoop() {
+ }
+
+ public static void run(BoundListener listener, ConnectionRunner runner, BooleanSupplier stopped) {
+ while (!stopped.getAsBoolean()) {
+ try {
+ runner.accept(listener.socket().accept(), stopped);
+ } catch (IOException e) {
+ if (!stopped.getAsBoolean()) log.error("Accept error", e);
+ }
+ }
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/BoundListener.java b/flash/src/main/java/dev/relism/flash/transport/BoundListener.java
new file mode 100644
index 0000000..6fd2fe6
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/BoundListener.java
@@ -0,0 +1,7 @@
+package dev.relism.flash.transport;
+
+import java.net.ServerSocket;
+
+/** One bound listener socket (plain or TLS) plus whether it is TLS, for logging only. */
+public record BoundListener(ServerSocket socket, boolean secure) {
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/ConnectionContext.java b/flash/src/main/java/dev/relism/flash/transport/ConnectionContext.java
new file mode 100644
index 0000000..f7024d3
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/ConnectionContext.java
@@ -0,0 +1,49 @@
+package dev.relism.flash.transport;
+
+import dev.relism.flash.extension.FlashConfiguration;
+import dev.relism.flash.routing.AbstractRouter;
+import dev.relism.flash.routing.AbstractWsRouter;
+
+import javax.net.ssl.SSLSocket;
+
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.function.BooleanSupplier;
+
+/**
+ * Everything a {@link ConnectionProtocol} implementation needs to serve one connection, bundled
+ * into a single object instead of a long parameter list.
+ *
+ * @param socket the accepted socket — owns its lifecycle (closing it is the caller's,
+ * i.e. {@link ConnectionRunner}'s, responsibility, not the protocol's)
+ * @param sslSocket {@code socket} narrowed to {@link SSLSocket}, or {@code null} for a
+ * plaintext connection
+ * @param in the single buffered, deadline-aware source for this connection's
+ * inbound bytes
+ * @param out the buffered output stream — for header/body writes that benefit from
+ * userspace coalescing before a single syscall
+ * @param rawOut the unbuffered output stream — for WebSocket, whose writes are already
+ * bulk (see {@code WebSocketSession})
+ * @param remoteAddress the client's address, or {@code null} if unavailable
+ * @param scratch this connection's reusable buffers ({@code EX-06})
+ * @param router the HTTP router
+ * @param wsRouter the WebSocket router
+ * @param configuration the server configuration (timeouts, limits, feature flags)
+ * @param stopped {@code true} once the server has begun shutting down — a protocol
+ * implementation's request loop must check this and exit promptly
+ */
+public record ConnectionContext(
+ Socket socket,
+ SSLSocket sslSocket,
+ BufferedByteSource in,
+ OutputStream out,
+ OutputStream rawOut,
+ InetSocketAddress remoteAddress,
+ ConnectionScratch scratch,
+ AbstractRouter router,
+ AbstractWsRouter wsRouter,
+ FlashConfiguration configuration,
+ BooleanSupplier stopped
+) {
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/ConnectionProtocol.java b/flash/src/main/java/dev/relism/flash/transport/ConnectionProtocol.java
new file mode 100644
index 0000000..c4fe26c
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/ConnectionProtocol.java
@@ -0,0 +1,14 @@
+package dev.relism.flash.transport;
+
+import java.io.IOException;
+
+/**
+ * The h1/h2 seam R1 requires: the protocol decision is made once, immediately after
+ * ALPN/preface detection ({@link ConnectionRunner}), and dispatches to one implementation of
+ * this interface. After that point neither implementation knows the other exists.
+ */
+public interface ConnectionProtocol {
+
+ /** Runs this connection to completion. Returns when the connection should be closed. */
+ void run(ConnectionContext ctx) throws IOException;
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java b/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java
new file mode 100644
index 0000000..056760e
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java
@@ -0,0 +1,142 @@
+package dev.relism.flash.transport;
+
+import dev.relism.flash.extension.FlashConfiguration;
+import dev.relism.flash.routing.AbstractRouter;
+import dev.relism.flash.routing.AbstractWsRouter;
+
+import lombok.extern.slf4j.Slf4j;
+
+import javax.net.ssl.SSLSocket;
+
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketException;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.function.BooleanSupplier;
+
+/**
+ * Owns one connection's socket lifecycle from accept to close: configures socket options,
+ * forces the TLS handshake if applicable ({@code EX-30}), negotiates the protocol, and
+ * dispatches to the matching {@link ConnectionProtocol} — guaranteeing cleanup (scratch
+ * release, active-socket tracking) regardless of how the protocol implementation exits.
+ *
+ *
Sole responsibility: connection setup/teardown. It contains no HTTP semantics at all —
+ * those live entirely inside whichever {@link ConnectionProtocol} it dispatches to (today,
+ * always {@code Http1Connection}; an {@code H2} negotiation result is closed cleanly, since
+ * {@code Http2Connection} does not exist until Phase 8).
+ */
+@Slf4j
+public final class ConnectionRunner {
+
+ private final ExecutorService executorService;
+ private final Set activeSockets;
+ private final ScratchPool scratchPool;
+ private final AbstractRouter router;
+ private final AbstractWsRouter wsRouter;
+ private final FlashConfiguration configuration;
+ private final ConnectionProtocol http1Protocol;
+
+ public ConnectionRunner(ExecutorService executorService, Set activeSockets, ScratchPool scratchPool,
+ AbstractRouter router, AbstractWsRouter wsRouter, FlashConfiguration configuration,
+ ConnectionProtocol http1Protocol) {
+ this.executorService = executorService;
+ this.activeSockets = activeSockets;
+ this.scratchPool = scratchPool;
+ this.router = router;
+ this.wsRouter = wsRouter;
+ this.configuration = configuration;
+ this.http1Protocol = http1Protocol;
+ }
+
+ /** Submits {@code socket} to the virtual-thread executor for full connection handling.
+ * {@code stopped} is threaded through to the eventual {@link ConnectionContext} so the
+ * protocol implementation can observe an in-progress graceful shutdown. */
+ public void accept(Socket socket, BooleanSupplier stopped) {
+ try {
+ executorService.submit(() -> handle(socket, stopped));
+ } catch (RejectedExecutionException ignored) {
+ try { socket.close(); } catch (IOException e) { log.debug("Error closing socket on shutdown", e); }
+ }
+ }
+
+ private void handle(Socket socket, BooleanSupplier stopped) {
+ activeSockets.add(socket);
+ ConnectionScratch scratch = scratchPool.acquire();
+ try (socket;
+ OutputStream out = new BufferedOutputStream(socket.getOutputStream())) {
+
+ // TCP_NODELAY: disable Nagle's algorithm. Small WS frames (< MSS) are sent
+ // immediately rather than waiting up to 200 ms for more data to coalesce.
+ socket.setTcpNoDelay(true);
+ socket.setSendBufferSize(TransportTuning.SOCKET_BUF_SIZE);
+
+ SSLSocket sslSocket = socket instanceof SSLSocket ssl ? ssl : null;
+ if (sslSocket != null) {
+ // EX-30: force the handshake explicitly, under a bounded timeout, before any
+ // protocol decision — SSLSocket#getApplicationProtocol() (which
+ // ProtocolNegotiator relies on) returns null until the handshake has run.
+ socket.setSoTimeout(configuration.getHeaderReadTimeoutMs());
+ sslSocket.startHandshake();
+ socket.setSoTimeout(0); // BufferedByteSource's own deadline takes over below
+ }
+
+ // rawOut is the unbuffered socket stream — passed to WebSocketSession directly.
+ // WS writes are already bulk; HTTP responses use the buffered `out` because
+ // Http1ResponseWriter does several small writes that benefit from coalescing.
+ OutputStream rawOut = socket.getOutputStream();
+ BufferedByteSource in = new BufferedByteSource(socket.getInputStream(), socket);
+
+ NegotiatedProtocol negotiated = negotiateProtocol(socket, in);
+ if (negotiated == NegotiatedProtocol.H2) {
+ // No Http2Connection exists yet (lands in Phase 8) — close cleanly rather than
+ // attempt to speak a protocol this version cannot yet serve.
+ return;
+ }
+
+ ConnectionContext ctx = new ConnectionContext(
+ socket, sslSocket, in, out, rawOut,
+ (InetSocketAddress) socket.getRemoteSocketAddress(),
+ scratch, router, wsRouter, configuration, stopped);
+ http1Protocol.run(ctx);
+
+ } catch (IOException e) {
+ if (!stopped.getAsBoolean()) {
+ if (e instanceof SocketException) log.debug("Connection closed: {}", e.getMessage());
+ else log.error("I/O error handling request", e);
+ }
+ } catch (Exception e) {
+ // Anything not an IOException here means a collaborator misbehaved on the TLS
+ // handshake path — most likely a custom TlsConfig#ofContext KeyManager/TrustManager
+ // throwing. That failure is isolated to this one virtual thread/connection.
+ if (!stopped.getAsBoolean()) log.error("Unexpected error handling connection", e);
+ } finally {
+ activeSockets.remove(socket);
+ scratchPool.release(scratch);
+ }
+ }
+
+ /**
+ * Decides h1 vs h2 for this connection, applying {@link FlashConfiguration#isHttp2Enabled()}
+ * to the plaintext (h2c) path — see {@link ProtocolNegotiator}'s Javadoc for why the flag is
+ * applied here rather than inside the negotiator itself.
+ */
+ private NegotiatedProtocol negotiateProtocol(Socket socket, BufferedByteSource in) throws IOException {
+ if (socket instanceof SSLSocket) {
+ return ProtocolNegotiator.negotiate(socket, in); // ALPN — already resolved, no I/O
+ }
+ if (!configuration.isHttp2Enabled()) {
+ return NegotiatedProtocol.HTTP_1_1; // skip the h2c peek entirely when disabled
+ }
+ in.setDeadline(System.nanoTime() + configuration.getIdleKeepAliveTimeoutMs() * 1_000_000L);
+ try {
+ return ProtocolNegotiator.negotiate(socket, in);
+ } finally {
+ in.clearDeadline();
+ }
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/ConnectionScratch.java b/flash/src/main/java/dev/relism/flash/transport/ConnectionScratch.java
new file mode 100644
index 0000000..4866d07
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/ConnectionScratch.java
@@ -0,0 +1,65 @@
+package dev.relism.flash.transport;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+
+/**
+ * The {@code EX-06} fix. Owns every per-connection reusable buffer that used to live in a
+ * {@link ThreadLocal} on {@code HttpServer}: the decimal-formatting scratch, the streaming
+ * relay buffer, and the WebSocket-handshake {@link MessageDigest}.
+ *
+ * Why not {@code ThreadLocal}
+ * {@code ThreadLocal} is the right idiom for a bounded platform-thread pool, where "one per
+ * thread" means "one per core". Flash runs one virtual thread per connection
+ * ({@code Executors.newVirtualThreadPerTaskExecutor()}), so a {@code ThreadLocal} here means
+ * one per connection, not one per core — with no upper bound. At 100 000 concurrent
+ * connections, an 8 KB relay buffer alone is ~800 MB of memory that a bounded pool would
+ * instead cap. {@code ConnectionScratch} is therefore explicit and pooled ({@link ScratchPool}),
+ * not thread-local.
+ *
+ * Lifetime and thread-safety contract
+ * Allocated once per connection (or reused from {@link ScratchPool}), owned exclusively by the
+ * single virtual thread driving that connection for its whole lifetime, and returned to the
+ * pool when the connection closes. Never shared between two connections at once — there is no
+ * synchronization here because none is needed.
+ *
+ * Extended in Phase 4 with the router's reusable {@code MatchResult}/path-view fields
+ * (currently still {@code ThreadLocal} in {@code FastPathRouterImpl}, per {@code EX-06}'s own
+ * multi-phase assignment — see {@code DECISIONS.md} for why Phase 2 does not also absorb that
+ * part of the fix) and in later phases with HTTP/2 write/HPACK scratch.
+ */
+public final class ConnectionScratch {
+
+ /** Matches the relay-buffer size the {@code ThreadLocal} it replaces used. */
+ public static final int RELAY_BUFFER_SIZE = 8192;
+
+ /** Large enough for the decimal digits of any {@code long}, including a sign. */
+ public static final int DECIMAL_BUFFER_SIZE = 20;
+
+ /** Scratch for {@code Http1ResponseWriter}'s decimal (status code / Content-Length) encoding. */
+ public final byte[] decimalBuffer = new byte[DECIMAL_BUFFER_SIZE];
+
+ /** Scratch for relaying a streaming or chunked response body without allocating per response. */
+ public final byte[] relayBuffer = new byte[RELAY_BUFFER_SIZE];
+
+ /** Scratch for the WebSocket handshake's {@code Sec-WebSocket-Accept} SHA-1 digest. */
+ public final MessageDigest sha1;
+
+ ConnectionScratch() {
+ try {
+ this.sha1 = MessageDigest.getInstance("SHA-1");
+ } catch (NoSuchAlgorithmException e) {
+ // Every JDK ships SHA-1 — this is a broken-runtime condition, not a request-time one.
+ throw new IllegalStateException("SHA-1 MessageDigest unavailable", e);
+ }
+ }
+
+ /** Called by {@link ScratchPool} before handing a reused instance to a new connection. */
+ void reset() {
+ sha1.reset();
+ // decimalBuffer/relayBuffer need no clearing: every reader of either only ever reads
+ // back exactly the region the immediately preceding writer just wrote (writeLong fills
+ // from the end backward and reports its own start position; relay() reports its own
+ // fill length), so stale bytes from a previous connection are never observed.
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/ListenerBinder.java b/flash/src/main/java/dev/relism/flash/transport/ListenerBinder.java
new file mode 100644
index 0000000..5aed7ba
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/ListenerBinder.java
@@ -0,0 +1,43 @@
+package dev.relism.flash.transport;
+
+import dev.relism.flash.extension.FlashConfiguration;
+import dev.relism.flash.tls.TlsConfig;
+
+import javax.net.ssl.SSLServerSocket;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+
+/**
+ * Turns a {@link FlashConfiguration.Listener} into a bound {@link ServerSocket}. Sole
+ * responsibility: binding — not accepting, not connection handling.
+ *
+ *
A TLS listener gets its {@link ServerSocket} from {@link TlsConfig#serverSocketFactory()}
+ * instead of {@code new ServerSocket()}, and its protocol/client-auth/cipher parameters from
+ * {@link TlsConfig#applyTo}; reuse-address, receive buffer size, backlog and the bind call
+ * itself are identical either way. TLS only changes which bytes come out of {@code accept()};
+ * it never changes how the accept loop, or anything downstream of it, treats them.
+ */
+public final class ListenerBinder {
+
+ private ListenerBinder() {
+ }
+
+ public static BoundListener bind(FlashConfiguration.Listener spec) throws IOException {
+ TlsConfig tls = spec.tls();
+
+ ServerSocket socket = tls != null ? tls.serverSocketFactory().createServerSocket() : new ServerSocket();
+ // setReuseAddress(true) must be called BEFORE bind().
+ socket.setReuseAddress(true);
+ socket.setReceiveBufferSize(TransportTuning.SOCKET_BUF_SIZE);
+ if (tls != null) tls.applyTo((SSLServerSocket) socket);
+
+ InetSocketAddress addr = spec.host() != null
+ ? new InetSocketAddress(spec.host(), spec.port())
+ : new InetSocketAddress(spec.port());
+ socket.bind(addr, TransportTuning.ACCEPT_BACKLOG);
+
+ return new BoundListener(socket, tls != null);
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/ScratchPool.java b/flash/src/main/java/dev/relism/flash/transport/ScratchPool.java
new file mode 100644
index 0000000..df2ed27
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/ScratchPool.java
@@ -0,0 +1,61 @@
+package dev.relism.flash.transport;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A bounded cache of {@link ConnectionScratch} instances, reused across connections instead of
+ * being allocated and garbage-collected per connection.
+ *
+ *
This is a cache, not a leak-free arena: a burst of 100 000 concurrent connections
+ * still allocates 100 000 {@link ConnectionScratch} instances (one per connection, since each
+ * connection needs its own for as long as it is open), but only {@link #bound} of them survive
+ * being released back to the pool afterward — the rest are simply dropped for the garbage
+ * collector, exactly as they would have been without this class. What the pool buys is avoiding
+ * repeated allocation for the common case of many short-lived or sequential connections sharing
+ * a bounded set of scratch objects.
+ *
+ *
Thread-safety
+ * {@link #acquire()} and {@link #release} are safe to call concurrently from any number of
+ * threads — the underlying queue and size guard are lock-free.
+ */
+public final class ScratchPool {
+
+ /** Default bound: generous enough that a real workload rarely misses, small enough that it
+ * is not itself a meaningful memory commitment (a few hundred KB at most). */
+ public static final int DEFAULT_BOUND = Math.min(Runtime.getRuntime().availableProcessors() * 64, 4096);
+
+ private final ConcurrentLinkedQueue pool = new ConcurrentLinkedQueue<>();
+ private final AtomicInteger size = new AtomicInteger();
+ private final int bound;
+
+ public ScratchPool() {
+ this(DEFAULT_BOUND);
+ }
+
+ public ScratchPool(int bound) {
+ this.bound = bound;
+ }
+
+ /** Returns a reset, ready-to-use scratch — either reused from the pool or freshly allocated. */
+ public ConnectionScratch acquire() {
+ ConnectionScratch scratch = pool.poll();
+ if (scratch != null) {
+ size.decrementAndGet();
+ scratch.reset();
+ return scratch;
+ }
+ return new ConnectionScratch();
+ }
+
+ /**
+ * Returns {@code scratch} to the pool for reuse, unless the pool is already at its bound —
+ * in which case it is simply dropped, for the garbage collector, so an unusually large burst
+ * of connections cannot grow this cache without limit.
+ */
+ public void release(ConnectionScratch scratch) {
+ if (size.get() >= bound) return;
+ size.incrementAndGet();
+ pool.offer(scratch);
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/ServerLifecycle.java b/flash/src/main/java/dev/relism/flash/transport/ServerLifecycle.java
new file mode 100644
index 0000000..9235876
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/ServerLifecycle.java
@@ -0,0 +1,115 @@
+package dev.relism.flash.transport;
+
+import dev.relism.flash.ServerHandle;
+import dev.relism.flash.extension.FlashConfiguration;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Owns the server's lifecycle: the accept threads (one per listener ×
+ * {@code TransportTuning.ACCEPT_THREADS}), the active-socket registry, and the two-stage
+ * graceful shutdown ({@code EX-32}) — stop accepting, let in-flight connections drain up to
+ * {@code shutdownDrainTimeoutMs} (during which {@code Http1Connection} forces
+ * {@code Connection: close} on the next response once it observes {@link #isStopped()}), then
+ * force-close whatever remains.
+ *
+ * Implements {@link ServerHandle} directly — its three methods already match that contract
+ * exactly, so no separate wrapper class is needed.
+ */
+@Slf4j
+public final class ServerLifecycle implements ServerHandle {
+
+ private final List listeners;
+ private final ConnectionRunner runner;
+ private final FlashConfiguration configuration;
+ private final ExecutorService executorService;
+ private final Set activeSockets;
+ private final CountDownLatch acceptLatch;
+ private volatile boolean stopped = false;
+
+ public ServerLifecycle(List listeners, ConnectionRunner runner,
+ FlashConfiguration configuration, ExecutorService executorService,
+ Set activeSockets) {
+ this.listeners = listeners;
+ this.runner = runner;
+ this.configuration = configuration;
+ this.executorService = executorService;
+ this.activeSockets = activeSockets;
+ this.acceptLatch = new CountDownLatch(TransportTuning.ACCEPT_THREADS * listeners.size());
+ }
+
+ /** Whether the server has begun shutting down. Passed down to every connection as a
+ * {@link java.util.function.BooleanSupplier} so in-flight request loops can drain
+ * promptly instead of waiting for their next keep-alive request. */
+ public boolean isStopped() {
+ return stopped;
+ }
+
+ @Override
+ public void start() {
+ for (int li = 0; li < listeners.size(); li++) {
+ BoundListener listener = listeners.get(li);
+ for (int i = 0; i < TransportTuning.ACCEPT_THREADS; i++) {
+ Thread.ofPlatform()
+ .name("flash-accept-" + li + "-" + i)
+ .daemon(false)
+ .start(() -> {
+ try {
+ AcceptLoop.run(listener, runner, this::isStopped);
+ } finally {
+ acceptLatch.countDown();
+ }
+ });
+ }
+ }
+ }
+
+ @Override
+ public void startAndBlock() {
+ start();
+ try { acceptLatch.await(); }
+ catch (InterruptedException e) { Thread.currentThread().interrupt(); }
+ }
+
+ @Override
+ public CompletableFuture stop() {
+ return CompletableFuture.runAsync(() -> {
+ stopped = true;
+ for (BoundListener bl : listeners) {
+ try { bl.socket().close(); } catch (IOException e) { log.error("Error closing server socket", e); }
+ }
+
+ // EX-32: give in-flight connections a chance to finish their current response and
+ // exit (Http1Connection forces Connection: close once it observes isStopped())
+ // before force-closing whatever is still open.
+ long deadlineNanos = System.nanoTime() + configuration.getShutdownDrainTimeoutMs() * 1_000_000L;
+ while (!activeSockets.isEmpty() && System.nanoTime() < deadlineNanos) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ break;
+ }
+ }
+
+ activeSockets.forEach(s -> { try { s.close(); } catch (IOException ignored) { } });
+ executorService.shutdown();
+ try {
+ if (!executorService.awaitTermination(30, TimeUnit.SECONDS))
+ executorService.shutdownNow();
+ } catch (InterruptedException e) {
+ executorService.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ });
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/TransportFactory.java b/flash/src/main/java/dev/relism/flash/transport/TransportFactory.java
new file mode 100644
index 0000000..eb2587b
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/TransportFactory.java
@@ -0,0 +1,63 @@
+package dev.relism.flash.transport;
+
+import dev.relism.flash.ServerHandle;
+import dev.relism.flash.extension.FlashConfiguration;
+import dev.relism.flash.http1.Http1Connection;
+import dev.relism.flash.routing.AbstractRouter;
+import dev.relism.flash.routing.AbstractWsRouter;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * Composes the whole transport: binds every configured listener, wires the connection runner
+ * and the h1 protocol, and returns the {@link ServerHandle} implementation
+ * ({@link ServerLifecycle}) that {@link dev.relism.flash.ServerHandle#create} exposes publicly.
+ *
+ * {@code EX-34}: this is the "composed transport rather than a god object" the registry
+ * asked for — {@code ServerHandle.create} used to construct {@code HttpServer} directly, which
+ * no longer exists. Package-private-in-spirit (public only because {@code ServerHandle} lives
+ * in a different package and must call it) — user code has no reason to call this directly.
+ */
+@Slf4j
+public final class TransportFactory {
+
+ private TransportFactory() {
+ }
+
+ public static ServerHandle create(FlashConfiguration configuration,
+ AbstractRouter router, AbstractWsRouter wsRouter) throws IOException {
+ List specs = configuration.getListeners().isEmpty()
+ ? List.of(new FlashConfiguration.Listener(
+ configuration.getPort(), configuration.getHost(), configuration.getTls()))
+ : configuration.getListeners();
+
+ List bound = new ArrayList<>(specs.size());
+ for (FlashConfiguration.Listener spec : specs) bound.add(ListenerBinder.bind(spec));
+ List boundListeners = List.copyOf(bound);
+
+ for (BoundListener bl : boundListeners) {
+ log.info("HTTP server bound on {}:{} (tls={}, backlog={}, acceptThreads={})",
+ bl.socket().getInetAddress(), bl.socket().getLocalPort(), bl.secure(),
+ TransportTuning.ACCEPT_BACKLOG, TransportTuning.ACCEPT_THREADS);
+ }
+
+ ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor();
+ Set activeSockets = ConcurrentHashMap.newKeySet();
+ ScratchPool scratchPool = new ScratchPool();
+
+ ConnectionRunner runner = new ConnectionRunner(
+ executorService, activeSockets, scratchPool, router, wsRouter, configuration,
+ new Http1Connection());
+
+ return new ServerLifecycle(boundListeners, runner, configuration, executorService, activeSockets);
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/transport/TransportTuning.java b/flash/src/main/java/dev/relism/flash/transport/TransportTuning.java
new file mode 100644
index 0000000..51e1d8a
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/transport/TransportTuning.java
@@ -0,0 +1,39 @@
+package dev.relism.flash.transport;
+
+/** Tuning constants shared by {@link ListenerBinder}, {@link ServerLifecycle}, and
+ * {@link ConnectionRunner} — grouped here so the accept-side and connection-side constants
+ * that must stay consistent with each other (e.g. the socket buffer size applied at bind time
+ * and at accept time) are declared exactly once. */
+final class TransportTuning {
+
+ private TransportTuning() {
+ }
+
+ /**
+ * Number of platform threads competing on {@code serverSocket.accept()}.
+ * Rule of thumb: number of available CPU cores, capped at 8.
+ * More than this rarely helps — accept is cheap; the bottleneck is usually
+ * the virtual-thread executor dispatching the connection handler.
+ */
+ static final int ACCEPT_THREADS = Math.min(Runtime.getRuntime().availableProcessors(), 8);
+
+ /**
+ * TCP listen backlog. The kernel holds up to this many fully-established
+ * (SYN+ACK sent, ACK received) connections waiting for accept().
+ * 4096 is safe on Linux; /proc/sys/net/core/somaxconn must be >= this value,
+ * or the kernel silently caps it. Raise somaxconn if needed:
+ * sysctl -w net.core.somaxconn=4096
+ */
+ static final int ACCEPT_BACKLOG = 4096;
+
+ /**
+ * Socket send/receive buffer sizes. Matched to the WS frame read buffer
+ * ({@code FlashConfiguration#getWsFrameBufferSize()}) so the kernel never
+ * needs to fragment a full frame into multiple TCP segments on the receive
+ * side, and never blocks a write waiting for the send buffer to drain.
+ *
+ * Linux default is 87380 bytes (rmem) / 16384 bytes (wmem). We raise both
+ * to 256 KB — a good fit for up to ~250 KB WS frames with no partial reads.
+ */
+ static final int SOCKET_BUF_SIZE = 256 * 1024;
+}
diff --git a/flash/src/main/java/dev/relism/flash/websocket/WebSocketLoop.java b/flash/src/main/java/dev/relism/flash/websocket/WebSocketLoop.java
new file mode 100644
index 0000000..56e8349
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/websocket/WebSocketLoop.java
@@ -0,0 +1,43 @@
+package dev.relism.flash.websocket;
+
+import java.io.IOException;
+
+/**
+ * Drives one {@link WebSocketSession}'s read loop until the session closes, dispatching frames
+ * to the user's {@link WebSocketHandler}. Extracted from {@code HttpServer} (Phase 2) — its only
+ * responsibility is this loop; the handshake and upgrade detection live in
+ * {@link WebSocketUpgrade}.
+ */
+public final class 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) {
+ // EX-12: tell the peer why, with the correct close code, before tearing down.
+ 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();
+ }
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/websocket/WebSocketProtocolException.java b/flash/src/main/java/dev/relism/flash/websocket/WebSocketProtocolException.java
new file mode 100644
index 0000000..0ce014b
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/websocket/WebSocketProtocolException.java
@@ -0,0 +1,23 @@
+package dev.relism.flash.websocket;
+
+import java.io.IOException;
+
+/**
+ * A WebSocket frame violated RFC 6455 (bad opcode, wrong masking direction, oversized control
+ * frame, fragmented control frame, or a message exceeding the session's buffer). Carries the
+ * close code ({@code 1002} protocol error, {@code 1009} message too big) the session must send
+ * before closing — see {@code WebSocketLoop}, the single site that catches this.
+ */
+public final class WebSocketProtocolException extends IOException {
+
+ private final int closeCode;
+
+ public WebSocketProtocolException(int closeCode, String message) {
+ super(message);
+ this.closeCode = closeCode;
+ }
+
+ public int closeCode() {
+ return closeCode;
+ }
+}
diff --git a/flash/src/main/java/dev/relism/flash/websocket/WebSocketSession.java b/flash/src/main/java/dev/relism/flash/websocket/WebSocketSession.java
index 9e47442..1ca094f 100644
--- a/flash/src/main/java/dev/relism/flash/websocket/WebSocketSession.java
+++ b/flash/src/main/java/dev/relism/flash/websocket/WebSocketSession.java
@@ -10,13 +10,14 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantLock;
/**
* Per-connection WebSocket I/O state. One instance per virtual thread.
*
*
* -
- *
With {@code TCP_NODELAY} enabled on the socket (set in {@code HttpServer}),
+ *
With {@code TCP_NODELAY} enabled on the socket (set by the connection runner),
* Nagle's algorithm is disabled: the kernel sends data as soon as it lands in
* the send buffer, without waiting. {@link java.io.BufferedOutputStream} will
* still batch multiple small writes into one syscall when they happen in the
@@ -26,7 +27,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
*
The only place an explicit flush is still needed is after the WS
* handshake (one-time, not on the hot path) and after the CLOSE frame
* (end of session). Both are handled in {@link #close} and in
- * {@code HttpServer#performHandshake}.
+ * {@code WebSocketUpgrade#performHandshake}.
*
* - Flush on CLOSE frame: {@link #close} still flushes explicitly
* because the CLOSE frame is the last thing written before the stream is
@@ -34,10 +35,23 @@ import java.util.concurrent.atomic.AtomicBoolean;
*
*
* Thread safety
- * {@link #sendText}, {@link #send}, and {@link #close} are synchronized on
- * {@code out} and safe to call from threads other than the session loop.
- * {@link #close} uses a CAS on {@code open} to guarantee exactly-once CLOSE
+ * {@link #sendText}, {@link #send}, and {@link #close} are serialized on a
+ * {@link ReentrantLock} (never {@code synchronized} — see {@code EX-01}: a virtual thread
+ * blocking inside {@code synchronized} pins its carrier platform thread on Java 21, and a
+ * blocking socket write is exactly the kind of call that can block. {@link ReentrantLock}
+ * unmounts the blocked virtual thread instead) and are safe to call from threads other than the
+ * session loop. {@link #close} uses a CAS on {@code open} to guarantee exactly-once CLOSE
* frame emission under concurrent calls.
+ *
+ * Fragmentation, masking, and control frames (RFC 6455 §5)
+ * {@link #readFrame} reassembles continuation frames into one logical message (bounded by the
+ * read buffer's capacity — the same bound a single unfragmented frame already had), enforces
+ * that incoming frames are masked exactly when this session's role requires it (server sessions
+ * require masked frames from the client; client-mode sessions require unmasked frames from the
+ * server), validates the opcode against RFC 6455's defined set, and enforces the control-frame
+ * constraints (FIN must be set, payload ≤ 125 bytes). A violation throws
+ * {@link WebSocketProtocolException} carrying the correct close code (1002 protocol error, 1009
+ * message too big) for the caller to send before closing.
*/
public final class WebSocketSession {
@@ -47,12 +61,27 @@ public final class WebSocketSession {
private final Request request;
private final boolean maskOutgoing;
+ /** Server sessions (the common case) require every incoming frame to be masked, per RFC
+ * 6455 §5.1 ("a server MUST close the connection upon receiving a frame that is not
+ * masked"). A client-mode session ({@link #maskOutgoing} true) requires the opposite. */
+ private final boolean requireMaskedIncoming;
+
private final AtomicBoolean open = new AtomicBoolean(true);
private int closeCode = 1000;
+ private final ReentrantLock writeLock = new ReentrantLock();
/** 1 opcode byte + up to 8 extended-length bytes + up to 4 mask-key bytes (masked mode only). */
private final byte[] hdrScratch = new byte[14];
+ /** Scratch for control-frame payloads (RFC 6455 §5.5: at most 125 bytes), kept separate
+ * from {@link #readBuf} so a control frame arriving mid-fragmentation (RFC 6455 §5.4
+ * permits this) never disturbs the data message being reassembled there. */
+ private final byte[] controlBuf = new byte[125];
+
+ // Fragmentation state (RFC 6455 §5.4). fragmentLength == 0 means "no message in progress".
+ private byte fragmentOpcode;
+ private int fragmentLength;
+
public WebSocketSession(InputStream in, OutputStream out, int bufferSize) {
this(in, out, bufferSize, null, false);
}
@@ -64,14 +93,17 @@ public final class WebSocketSession {
* @param maskOutgoing {@code true} if this session is acting as a WS client — RFC 6455
* requires client-to-server frames to be masked, unlike the server-to-client
* direction {@link #writeFrame} originally only supported. See {@link
- * #writeFrame} for how masking is applied without allocating.
+ * #writeFrame} for how masking is applied without allocating. Also
+ * determines the expected masking of *incoming* frames — see
+ * {@link #requireMaskedIncoming}.
*/
public WebSocketSession(InputStream in, OutputStream out, int bufferSize, Request request, boolean maskOutgoing) {
- this.in = in;
- this.out = out;
- this.readBuf = new byte[bufferSize];
- this.request = request;
- this.maskOutgoing = maskOutgoing;
+ this.in = in;
+ this.out = out;
+ this.readBuf = new byte[bufferSize];
+ this.request = request;
+ this.maskOutgoing = maskOutgoing;
+ this.requireMaskedIncoming = !maskOutgoing;
}
public boolean isOpen() { return open.get(); }
@@ -109,50 +141,128 @@ public final class WebSocketSession {
*/
public void close(int code) throws IOException {
if (!open.compareAndSet(true, false)) return;
- synchronized (out) {
+ writeLock.lock();
+ try {
out.write(0x88);
out.write(0x02);
out.write((code >> 8) & 0xFF);
out.write(code & 0xFF);
+ } finally {
+ writeLock.unlock();
}
}
// ── Session loop internals ─────────────────────────────────────────────
+ /**
+ * Reads the next complete message, reassembling continuation frames and delivering control
+ * frames (CLOSE/PING/PONG) as soon as they arrive — RFC 6455 §5.4 explicitly permits a
+ * control frame to interleave with a fragmented data message, and this must not disturb the
+ * data message's in-progress reassembly.
+ *
+ * @return {@code false} only on a clean EOF between messages (the peer closed the TCP
+ * connection without sending a CLOSE frame); an EOF in the middle of a frame is a
+ * protocol violation and throws, it is not reported as {@code false}.
+ * @throws WebSocketProtocolException on any RFC 6455 violation (bad opcode, unmasked/masked
+ * frame when the opposite was required, oversized control frame, fragmented control
+ * frame, message exceeding the buffer) — carries the correct close code.
+ */
public boolean readFrame(WebSocketFrame frame) throws IOException {
- int b0 = in.read();
- if (b0 < 0) return false;
- int b1 = in.read();
- if (b1 < 0) return false;
+ while (true) {
+ int b0 = in.read();
+ if (b0 < 0) return false; // clean EOF between messages
+ int b1 = in.read();
+ if (b1 < 0) throw new EOFException("WebSocket stream closed mid-frame");
- boolean fin = (b0 & 0x80) != 0;
- byte opcode = (byte) (b0 & 0x0F);
- boolean masked = (b1 & 0x80) != 0;
- long payLen = (b1 & 0x7F);
+ boolean fin = (b0 & 0x80) != 0;
+ byte opcode = (byte) (b0 & 0x0F);
+ boolean masked = (b1 & 0x80) != 0;
+ int lenBits = b1 & 0x7F;
- if (payLen == 126) {
- payLen = ((in.read() & 0xFF) << 8) | (in.read() & 0xFF);
- } else if (payLen == 127) {
- payLen = 0;
- for (int i = 0; i < 8; i++) payLen = (payLen << 8) | (in.read() & 0xFF);
+ validateOpcode(opcode);
+
+ if (masked != requireMaskedIncoming) {
+ throw new WebSocketProtocolException(1002,
+ requireMaskedIncoming ? "client frame must be masked" : "server frame must not be masked");
+ }
+
+ int extLenBytes = lenBits == 127 ? 8 : lenBits == 126 ? 2 : 0;
+ int maskBytes = masked ? 4 : 0;
+ int extraLen = extLenBytes + maskBytes;
+ if (extraLen > 0) readFullyHeader(extraLen);
+
+ long payLen;
+ int pos;
+ if (extLenBytes == 2) {
+ payLen = ((hdrScratch[0] & 0xFFL) << 8) | (hdrScratch[1] & 0xFFL);
+ pos = 2;
+ } else if (extLenBytes == 8) {
+ // RFC 6455 §5.2: the most significant bit of the 64-bit length MUST be 0.
+ if ((hdrScratch[0] & 0x80) != 0) {
+ throw new WebSocketProtocolException(1002, "extended payload length MSB must be 0");
+ }
+ payLen = 0;
+ for (int i = 0; i < 8; i++) payLen = (payLen << 8) | (hdrScratch[i] & 0xFFL);
+ pos = 8;
+ } else {
+ payLen = lenBits;
+ pos = 0;
+ }
+
+ boolean isControl = opcode == WebSocketFrame.OP_CLOSE
+ || opcode == WebSocketFrame.OP_PING || opcode == WebSocketFrame.OP_PONG;
+
+ if (isControl) {
+ if (!fin) throw new WebSocketProtocolException(1002, "control frame must not be fragmented");
+ if (payLen > controlBuf.length) throw new WebSocketProtocolException(1002, "control frame payload exceeds 125 bytes");
+ } else if (opcode == WebSocketFrame.OP_CONTINUATION) {
+ if (fragmentLength == 0) throw new WebSocketProtocolException(1002, "continuation frame without an initiated message");
+ } else { // TEXT or BINARY
+ if (fragmentLength != 0) throw new WebSocketProtocolException(1002, "new data frame while a fragmented message is in progress");
+ }
+
+ byte m0 = 0, m1 = 0, m2 = 0, m3 = 0;
+ if (masked) {
+ m0 = hdrScratch[pos]; m1 = hdrScratch[pos + 1]; m2 = hdrScratch[pos + 2]; m3 = hdrScratch[pos + 3];
+ }
+
+ int len = (int) payLen;
+
+ if (isControl) {
+ readFully(controlBuf, 0, len);
+ if (masked) unmaskInPlace(controlBuf, 0, len, m0, m1, m2, m3);
+ frame.reset(controlBuf, 0, len, opcode, true);
+ return true;
+ }
+
+ // Data frame (fresh TEXT/BINARY, or a CONTINUATION of one already in progress):
+ // accumulate into readBuf, bounded by its capacity — the same bound a single
+ // unfragmented frame already had before this fix.
+ if (fragmentLength + (long) len > readBuf.length) {
+ throw new WebSocketProtocolException(1009, "message exceeds " + readBuf.length + " bytes");
+ }
+ readFully(readBuf, fragmentLength, len);
+ if (masked) unmaskInPlace(readBuf, fragmentLength, len, m0, m1, m2, m3);
+
+ byte messageOpcode = opcode == WebSocketFrame.OP_CONTINUATION ? fragmentOpcode : opcode;
+ if (opcode != WebSocketFrame.OP_CONTINUATION) fragmentOpcode = opcode;
+ fragmentLength += len;
+
+ if (fin) {
+ frame.reset(readBuf, 0, fragmentLength, messageOpcode, true);
+ fragmentLength = 0;
+ return true;
+ }
+ // Not FIN: loop to read the next continuation frame (or an interleaved control frame).
}
+ }
- if (payLen > readBuf.length) throw new IOException(
- "WS frame payload " + payLen + " bytes exceeds buffer " + readBuf.length);
-
- byte m0 = 0, m1 = 0, m2 = 0, m3 = 0;
- if (masked) {
- m0 = (byte) in.read(); m1 = (byte) in.read();
- m2 = (byte) in.read(); m3 = (byte) in.read();
+ private static void validateOpcode(byte opcode) throws WebSocketProtocolException {
+ switch (opcode) {
+ case WebSocketFrame.OP_CONTINUATION, WebSocketFrame.OP_TEXT, WebSocketFrame.OP_BINARY,
+ WebSocketFrame.OP_CLOSE, WebSocketFrame.OP_PING, WebSocketFrame.OP_PONG -> { /* valid */ }
+ default -> throw new WebSocketProtocolException(1002, "reserved/invalid opcode " + opcode);
}
-
- int len = (int) payLen;
- readFully(readBuf, 0, len);
-
- if (masked) unmaskInPlace(readBuf, 0, len, m0, m1, m2, m3);
-
- frame.reset(readBuf, 0, len, opcode, fin);
- return true;
}
public void sendPong(WebSocketFrame ping) throws IOException {
@@ -187,13 +297,12 @@ public final class WebSocketSession {
* extended-length + up to 4 mask-key), then writes header + payload in two bulk calls to the
* raw socket stream.
*
- * No {@code flush()} — {@code out} is the unbuffered socket {@link OutputStream}
- * (see {@code HttpServer#process}). Each {@code write()} lands directly in the
- * kernel send buffer. With {@code TCP_NODELAY} set on the socket, the kernel
- * transmits the segment immediately without Nagle coalescing. The two writes
- * (header then payload) will be merged into a single TCP segment by the kernel
- * because they arrive faster than the ACK from the peer — exactly the coalescing
- * we want, at zero cost.
+ *
No {@code flush()} — {@code out} is the unbuffered socket {@link OutputStream}.
+ * Each {@code write()} lands directly in the kernel send buffer. With {@code TCP_NODELAY}
+ * set on the socket, the kernel transmits the segment immediately without Nagle coalescing.
+ * The two writes (header then payload) will be merged into a single TCP segment by the
+ * kernel because they arrive faster than the ACK from the peer — exactly the coalescing we
+ * want, at zero cost.
*
*
{@link #maskOutgoing} (client mode): RFC 6455 requires every client-to-server frame
* to be masked. The mask key is generated into {@link #hdrScratch} (no new allocation — same
@@ -204,7 +313,8 @@ public final class WebSocketSession {
* masked mode must not reuse that buffer expecting it unchanged after the call.
*/
private void writeFrame(byte opcode, byte[] payload, int off, int len) throws IOException {
- synchronized (out) {
+ writeLock.lock();
+ try {
int hlen = 0;
hdrScratch[hlen++] = (byte) (0x80 | opcode);
int maskBit = maskOutgoing ? 0x80 : 0x00;
@@ -237,6 +347,19 @@ public final class WebSocketSession {
out.write(hdrScratch, 0, hlen);
out.write(payload, off, len);
// No flush — TCP_NODELAY handles delivery. See Javadoc above.
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ /** Bulk-reads {@code len} bytes into {@link #hdrScratch} starting at offset 0 — the {@code
+ * EX-11} fix: the extended-length and mask-key bytes used to be read one at a time. */
+ private void readFullyHeader(int len) throws IOException {
+ int remaining = len;
+ while (remaining > 0) {
+ int n = in.read(hdrScratch, len - remaining, remaining);
+ if (n < 0) throw new EOFException("WebSocket stream closed mid-frame");
+ remaining -= n;
}
}
@@ -265,4 +388,4 @@ public final class WebSocketSession {
if (i < end) { buf[i++] ^= m1; }
if (i < end) { buf[i] ^= m2; }
}
-}
\ No newline at end of file
+}
diff --git a/flash/src/main/java/dev/relism/flash/websocket/WebSocketUpgrade.java b/flash/src/main/java/dev/relism/flash/websocket/WebSocketUpgrade.java
new file mode 100644
index 0000000..6938c5b
--- /dev/null
+++ b/flash/src/main/java/dev/relism/flash/websocket/WebSocketUpgrade.java
@@ -0,0 +1,71 @@
+package dev.relism.flash.websocket;
+
+import dev.relism.flash.http1.Http1KeepAlive;
+import dev.relism.flash.models.Request;
+import dev.relism.flash.transport.ConnectionScratch;
+import dev.relism.fpr.core.ByteView;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.Base64;
+
+/**
+ * WebSocket upgrade detection (RFC 6455 §4.2.1) and handshake response. Extracted from
+ * {@code HttpServer} (Phase 2) — its only responsibility is deciding whether a request is an
+ * upgrade request and, if so, answering the {@code 101 Switching Protocols} handshake. The
+ * session loop itself lives in {@link WebSocketLoop}.
+ */
+public final class WebSocketUpgrade {
+
+ private WebSocketUpgrade() {
+ }
+
+ private static final byte[] WS_HANDSHAKE_PREFIX =
+ ("HTTP/1.1 101 Switching Protocols\r\n" +
+ "Upgrade: websocket\r\n" +
+ "Connection: Upgrade\r\n" +
+ "Sec-WebSocket-Accept: ")
+ .getBytes(StandardCharsets.ISO_8859_1);
+ private static final byte[] WS_HANDSHAKE_SUFFIX =
+ "\r\n\r\n".getBytes(StandardCharsets.ISO_8859_1);
+
+ public static final byte[] REJECT_400 =
+ "HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
+ .getBytes(StandardCharsets.ISO_8859_1);
+
+ private static final byte[] WS_GUID_BYTES =
+ "258EAFA5-E914-47DA-95CA-C5AB0DC85B11".getBytes(StandardCharsets.ISO_8859_1);
+
+ /**
+ * Whether {@code request} is a WebSocket upgrade request: {@code Upgrade: websocket} and a
+ * {@code Connection} header whose token list includes {@code upgrade} ({@code EX-13} — the
+ * shared token-list scanner in {@link Http1KeepAlive} is what fixed the whole-value compare
+ * bug this check used to have too).
+ */
+ public static boolean isWebSocketUpgrade(Request request) {
+ ByteView upgrade = request.getRequestLine().getHeaders().view("Upgrade");
+ if (upgrade == null) return false;
+ if (!Http1KeepAlive.tokenEqualsIgnoreCase(upgrade, 0, upgrade.length(), "websocket")) return false;
+ return Http1KeepAlive.connectionContainsToken(request, "upgrade");
+ }
+
+ /** Writes and flushes the {@code 101 Switching Protocols} handshake response. */
+ public static void performHandshake(OutputStream out, Request request, ConnectionScratch scratch) throws IOException {
+ ByteView keyView = request.getRequestLine().getHeaders().view("Sec-WebSocket-Key");
+ if (keyView == null) throw new IOException("Missing Sec-WebSocket-Key header");
+
+ MessageDigest sha1 = scratch.sha1;
+ sha1.reset();
+ for (int i = 0, len = keyView.length(); i < len; i++) sha1.update(keyView.byteAt(i));
+ sha1.update(WS_GUID_BYTES);
+
+ byte[] accept = Base64.getEncoder().encode(sha1.digest());
+
+ out.write(WS_HANDSHAKE_PREFIX);
+ out.write(accept);
+ out.write(WS_HANDSHAKE_SUFFIX);
+ out.flush();
+ }
+}
diff --git a/flash/src/test/java/dev/relism/flash/RequestParserSecurityTest.java b/flash/src/test/java/dev/relism/flash/RequestParserSecurityTest.java
index 4410a8d..045bac1 100644
--- a/flash/src/test/java/dev/relism/flash/RequestParserSecurityTest.java
+++ b/flash/src/test/java/dev/relism/flash/RequestParserSecurityTest.java
@@ -15,7 +15,7 @@ import static org.junit.jupiter.api.Assertions.*;
/**
* One test per rejection rule added in HTTP/2 plan Phase 1 (EX-02, EX-03, EX-08, EX-18, EX-35,
* EX-36), each asserting the specific status code {@link MalformedRequestException} carries —
- * not merely that some exception was thrown. {@code HttpServer} always closes the connection
+ * not merely that some exception was thrown. {@code Http1Connection}/{@code ConnectionRunner} always closes the connection
* after any of these (never keep-alive); that behaviour is exercised at the integration level
* by {@code HttpServerTest}.
*/
diff --git a/flash/src/test/java/dev/relism/flash/architecture/PackageBoundaryTest.java b/flash/src/test/java/dev/relism/flash/architecture/PackageBoundaryTest.java
new file mode 100644
index 0000000..665e94b
--- /dev/null
+++ b/flash/src/test/java/dev/relism/flash/architecture/PackageBoundaryTest.java
@@ -0,0 +1,62 @@
+package dev.relism.flash.architecture;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * {@code R1}/{@code DEC-02}: HTTP/1.1 and HTTP/2 are peers behind the {@code ConnectionProtocol}
+ * seam, never coupled to each other directly. A lightweight source-scan rather than ArchUnit —
+ * this project has no bytecode-analysis test dependency yet, and one import-statement check per
+ * package pair does not need one; record the choice here rather than in {@code DECISIONS.md}
+ * since it is this test's own implementation detail, not a design decision affecting shipped
+ * code.
+ */
+class PackageBoundaryTest {
+
+ @Test
+ void http1DoesNotImportH2() throws IOException {
+ assertNoImportOfPackage("dev/relism/flash/http1", "dev.relism.flash.h2");
+ }
+
+ @Test
+ void h2DoesNotImportHttp1() throws IOException {
+ assertNoImportOfPackage("dev/relism/flash/h2", "dev.relism.flash.http1");
+ }
+
+ private static void assertNoImportOfPackage(String sourceDirRelative, String forbiddenImportPrefix) throws IOException {
+ Path root = findSourceRoot(sourceDirRelative);
+ // Neither package boundary can be meaningfully checked before both packages exist; once
+ // dev.relism.flash.h2 gains real classes (Phase 3+) this stops being a no-op for the
+ // h2-side test.
+ if (root == null) return;
+
+ try (Stream files = Files.walk(root)) {
+ List javaFiles = files.filter(p -> p.toString().endsWith(".java")).toList();
+ for (Path file : javaFiles) {
+ for (String line : Files.readAllLines(file)) {
+ String trimmed = line.strip();
+ if (trimmed.startsWith("import " + forbiddenImportPrefix + ".")
+ || trimmed.startsWith("import " + forbiddenImportPrefix + ";")) {
+ fail(file + " imports " + forbiddenImportPrefix
+ + " — violates the h1/h2 package boundary (R1/DEC-02): " + trimmed);
+ }
+ }
+ }
+ }
+ }
+
+ private static Path findSourceRoot(String packageRelativePath) {
+ for (String base : List.of("flash/src/main/java", "src/main/java")) {
+ Path candidate = Path.of(base).resolve(packageRelativePath);
+ if (Files.isDirectory(candidate)) return candidate;
+ }
+ return null;
+ }
+}
diff --git a/flash/src/test/java/dev/relism/flash/http1/Http1ResponseWriterTest.java b/flash/src/test/java/dev/relism/flash/http1/Http1ResponseWriterTest.java
new file mode 100644
index 0000000..a2f4233
--- /dev/null
+++ b/flash/src/test/java/dev/relism/flash/http1/Http1ResponseWriterTest.java
@@ -0,0 +1,136 @@
+package dev.relism.flash.http1;
+
+import dev.relism.flash.http.ContentType;
+import dev.relism.flash.http.HttpMethod;
+import dev.relism.flash.models.Response;
+import dev.relism.flash.transport.ConnectionScratch;
+import dev.relism.flash.transport.ScratchPool;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class Http1ResponseWriterTest {
+
+ private static ConnectionScratch scratch() {
+ return new ScratchPool().acquire();
+ }
+
+ private static String write(Response response, HttpMethod method, boolean keepAlive, boolean sendDate) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ Http1ResponseWriter.writeResponse(out, response, method, keepAlive, sendDate, scratch());
+ return out.toString(StandardCharsets.UTF_8);
+ }
+
+ // --- EX-14: HEAD ------------------------------------------------------------
+
+ @Test
+ void head_reportsContentLengthButWritesNoBody() throws IOException {
+ Response response = new Response(200, "hello world", ContentType.TEXT_PLAIN);
+ String raw = write(response, HttpMethod.HEAD, true, false);
+
+ assertTrue(raw.contains("Content-Length: 11\r\n"), raw);
+ assertFalse(raw.contains("hello world"), raw);
+ }
+
+ @Test
+ void get_writesTheBody_forComparison() throws IOException {
+ Response response = new Response(200, "hello world", ContentType.TEXT_PLAIN);
+ String raw = write(response, HttpMethod.GET, true, false);
+
+ assertTrue(raw.contains("Content-Length: 11\r\n"), raw);
+ assertTrue(raw.endsWith("hello world"), raw);
+ }
+
+ // --- EX-15: 204 / 304 / 1xx never carry Content-Length or a body ------------
+
+ @Test
+ void status204_omitsContentLengthAndBody() throws IOException {
+ Response response = new Response(204, ContentType.NONE);
+ response.setBody("should never appear");
+ String raw = write(response, HttpMethod.GET, true, false);
+
+ assertFalse(raw.contains("Content-Length"), raw);
+ assertFalse(raw.contains("should never appear"), raw);
+ }
+
+ @Test
+ void status304_omitsContentLengthAndBody() throws IOException {
+ Response response = new Response(304, ContentType.NONE);
+ response.setBody("should never appear");
+ String raw = write(response, HttpMethod.GET, true, false);
+
+ assertFalse(raw.contains("Content-Length"), raw);
+ assertFalse(raw.contains("should never appear"), raw);
+ }
+
+ @Test
+ void status1xx_omitsContentLengthAndBody() throws IOException {
+ Response response = new Response(103, ContentType.NONE);
+ response.setBody("should never appear");
+ String raw = write(response, HttpMethod.GET, true, false);
+
+ assertFalse(raw.contains("Content-Length"), raw);
+ assertFalse(raw.contains("should never appear"), raw);
+ }
+
+ @Test
+ void status200_stillCarriesContentLength_forComparison() throws IOException {
+ Response response = new Response(200, "x", ContentType.TEXT_PLAIN);
+ String raw = write(response, HttpMethod.GET, true, false);
+ assertTrue(raw.contains("Content-Length: 1\r\n"), raw);
+ }
+
+ // --- EX-15: ContentType.NONE omits the Content-Type line entirely -----------
+
+ @Test
+ void contentTypeNone_omitsContentTypeLine() throws IOException {
+ Response response = new Response(200, ContentType.NONE);
+ String raw = write(response, HttpMethod.GET, true, false);
+ assertFalse(raw.contains("Content-Type"), raw);
+ }
+
+ @Test
+ void contentTypeTextPlain_includesContentTypeLine() throws IOException {
+ Response response = new Response(200, "x", ContentType.TEXT_PLAIN);
+ String raw = write(response, HttpMethod.GET, true, false);
+ assertTrue(raw.contains("Content-Type: text/plain\r\n"), raw);
+ }
+
+ // --- EX-16: Date header -------------------------------------------------------
+
+ @Test
+ void sendDateTrue_includesDateHeader() throws IOException {
+ Response response = new Response(200, "x", ContentType.TEXT_PLAIN);
+ String raw = write(response, HttpMethod.GET, true, true);
+ assertTrue(raw.contains("Date: "), raw);
+ // RFC 9110 IMF-fixdate, e.g. "Date: Tue, 03 Jun 2008 11:05:30 GMT\r\n"
+ assertTrue(raw.matches("(?s).*Date: [A-Za-z]{3}, \\d{2} [A-Za-z]{3} \\d{4} \\d{2}:\\d{2}:\\d{2} GMT\\r\\n.*"), raw);
+ }
+
+ @Test
+ void sendDateFalse_omitsDateHeader() throws IOException {
+ Response response = new Response(200, "x", ContentType.TEXT_PLAIN);
+ String raw = write(response, HttpMethod.GET, true, false);
+ assertFalse(raw.contains("Date: "), raw);
+ }
+
+ // --- Connection header --------------------------------------------------------
+
+ @Test
+ void keepAlive_writesKeepAliveConnectionHeader() throws IOException {
+ Response response = new Response(200, "x", ContentType.TEXT_PLAIN);
+ String raw = write(response, HttpMethod.GET, true, false);
+ assertTrue(raw.contains("Connection: keep-alive\r\n"), raw);
+ }
+
+ @Test
+ void notKeepAlive_writesCloseConnectionHeader() throws IOException {
+ Response response = new Response(200, "x", ContentType.TEXT_PLAIN);
+ String raw = write(response, HttpMethod.GET, false, false);
+ assertTrue(raw.contains("Connection: close\r\n"), raw);
+ }
+}
diff --git a/flash/src/test/java/dev/relism/flash/transport/ConnectionRunnerTest.java b/flash/src/test/java/dev/relism/flash/transport/ConnectionRunnerTest.java
new file mode 100644
index 0000000..bdb0f4d
--- /dev/null
+++ b/flash/src/test/java/dev/relism/flash/transport/ConnectionRunnerTest.java
@@ -0,0 +1,71 @@
+package dev.relism.flash.transport;
+
+import dev.relism.flash.extension.FlashConfiguration;
+import dev.relism.flash.routing.AbstractRouter;
+import dev.relism.flash.routing.AbstractWsRouter;
+import dev.relism.flash.routing.routers.fastpathrouter.FastPathRouterImpl;
+import dev.relism.flash.routing.routers.fastpathrouter.FastPathWsRouterImpl;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * A scratch is always released — including on an exception path — and a socket is always
+ * removed from {@code activeSockets}, regardless of how the dispatched
+ * {@link ConnectionProtocol} exits. This is a resource-leak safety property (Phase 2's Safety
+ * checks list), verified here with a protocol implementation that deliberately throws.
+ */
+class ConnectionRunnerTest {
+
+ @Test
+ void scratchAndActiveSocketEntry_alwaysReleased_evenWhenTheProtocolThrows() throws Exception {
+ ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
+ Set activeSockets = ConcurrentHashMap.newKeySet();
+ ScratchPool scratchPool = new ScratchPool();
+ AbstractRouter router = new FastPathRouterImpl();
+ AbstractWsRouter wsRouter = new FastPathWsRouterImpl();
+ FlashConfiguration configuration = FlashConfiguration.builder().port(0).build();
+
+ ConnectionProtocol throwingProtocol = ctx -> {
+ throw new IOException("simulated protocol failure");
+ };
+
+ ConnectionRunner runner = new ConnectionRunner(
+ executor, activeSockets, scratchPool, router, wsRouter, configuration, throwingProtocol);
+
+ try (ServerSocket serverSocket = new ServerSocket(0)) {
+ int port = serverSocket.getLocalPort();
+ CountDownLatch accepted = new CountDownLatch(1);
+
+ Thread acceptThread = new Thread(() -> {
+ try (Socket serverSide = serverSocket.accept()) {
+ runner.accept(serverSide, () -> false);
+ accepted.countDown();
+ Thread.sleep(300); // give the submitted virtual-thread task time to run
+ } catch (Exception ignored) {
+ }
+ });
+ acceptThread.start();
+
+ try (Socket client = new Socket("127.0.0.1", port)) {
+ assertTrue(accepted.await(2, TimeUnit.SECONDS));
+ Thread.sleep(300); // let ConnectionRunner's virtual thread finish
+
+ assertTrue(activeSockets.isEmpty(), "socket must be removed from activeSockets on every exit path");
+ }
+ acceptThread.join(2000);
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+}
diff --git a/flash/src/test/java/dev/relism/flash/transport/ProtocolNegotiatorTest.java b/flash/src/test/java/dev/relism/flash/transport/ProtocolNegotiatorTest.java
index 4f9762b..404834f 100644
--- a/flash/src/test/java/dev/relism/flash/transport/ProtocolNegotiatorTest.java
+++ b/flash/src/test/java/dev/relism/flash/transport/ProtocolNegotiatorTest.java
@@ -22,7 +22,7 @@ import static org.junit.jupiter.api.Assertions.*;
/**
* {@link ProtocolNegotiator#negotiate} is a pure, directly-testable detector (see its Javadoc
* for why it does not itself consult {@code FlashConfiguration.http2Enabled}) — every case here
- * calls it directly rather than through {@code HttpServer}.
+ * calls it directly rather than through {@code Http1Connection}/{@code ConnectionRunner}.
*/
class ProtocolNegotiatorTest {
@@ -84,7 +84,7 @@ class ProtocolNegotiatorTest {
/**
* Binds a real TLS listener offering {@code serverAlpn}, connects a client offering
- * {@code clientAlpn}, forces the handshake on both sides (mirroring {@code HttpServer}'s
+ * {@code clientAlpn}, forces the handshake on both sides (mirroring {@code Http1Connection}/{@code ConnectionRunner}'s
* EX-30 fix), and hands the accepted server-side socket to {@code assertion}.
*/
private static void withNegotiatedAlpn(Path dir, String[] serverAlpn, String[] clientAlpn,
diff --git a/flash/src/test/java/dev/relism/flash/transport/ScratchPoolTest.java b/flash/src/test/java/dev/relism/flash/transport/ScratchPoolTest.java
new file mode 100644
index 0000000..ac3b77a
--- /dev/null
+++ b/flash/src/test/java/dev/relism/flash/transport/ScratchPoolTest.java
@@ -0,0 +1,72 @@
+package dev.relism.flash.transport;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class ScratchPoolTest {
+
+ @Test
+ void acquire_withEmptyPool_returnsFreshInstance() {
+ ScratchPool pool = new ScratchPool();
+ ConnectionScratch scratch = pool.acquire();
+ assertNotNull(scratch);
+ assertNotNull(scratch.sha1);
+ assertEquals(ConnectionScratch.DECIMAL_BUFFER_SIZE, scratch.decimalBuffer.length);
+ assertEquals(ConnectionScratch.RELAY_BUFFER_SIZE, scratch.relayBuffer.length);
+ }
+
+ @Test
+ void release_thenAcquire_reusesTheSameInstance() {
+ ScratchPool pool = new ScratchPool();
+ ConnectionScratch first = pool.acquire();
+ pool.release(first);
+ ConnectionScratch second = pool.acquire();
+ assertSame(first, second);
+ }
+
+ @Test
+ void bound_isRespected_excessReleasesAreDropped() {
+ ScratchPool pool = new ScratchPool(2);
+ ConnectionScratch a = pool.acquire();
+ ConnectionScratch b = pool.acquire();
+ ConnectionScratch c = pool.acquire();
+ pool.release(a);
+ pool.release(b);
+ pool.release(c); // pool already has 2 -- this one is dropped, not queued
+
+ Set reacquired = new HashSet<>();
+ reacquired.add(pool.acquire());
+ reacquired.add(pool.acquire());
+ ConnectionScratch third = pool.acquire(); // freshly allocated, pool was exhausted at 2
+ assertFalse(reacquired.contains(third));
+ assertEquals(2, reacquired.size());
+ }
+
+ @Test
+ void reset_clearsTheMessageDigestState() {
+ // A dirty digest (mid-update, not yet digested) must not leak into the next connection
+ // that reuses this scratch -- the classic cross-connection-leak hazard for pooled state.
+ ScratchPool pool = new ScratchPool();
+ ConnectionScratch scratch = pool.acquire();
+ scratch.sha1.update((byte) 'x');
+ pool.release(scratch);
+
+ ConnectionScratch reused = pool.acquire();
+ assertSame(scratch, reused);
+ // If reset() had not run, digesting an empty input now would still reflect the earlier
+ // update. A byte array is not the actual assertion here (MessageDigest doesn't expose
+ // "reset happened") - the practical proof is that digest() with no further updates
+ // matches the well-known empty-input SHA-1 digest.
+ byte[] emptyDigest = reused.sha1.digest();
+ byte[] expected = {
+ (byte) 0xda, 0x39, (byte) 0xa3, (byte) 0xee, 0x5e, 0x6b, 0x4b, 0x0d,
+ 0x32, 0x55, (byte) 0xbf, (byte) 0xef, (byte) 0x95, 0x60, 0x18, (byte) 0x90,
+ (byte) 0xaf, (byte) 0xd8, 0x07, 0x09
+ };
+ assertArrayEquals(expected, emptyDigest);
+ }
+}
diff --git a/flash/src/test/java/dev/relism/flash/transport/ServerLifecycleGracefulShutdownTest.java b/flash/src/test/java/dev/relism/flash/transport/ServerLifecycleGracefulShutdownTest.java
new file mode 100644
index 0000000..99e4173
--- /dev/null
+++ b/flash/src/test/java/dev/relism/flash/transport/ServerLifecycleGracefulShutdownTest.java
@@ -0,0 +1,116 @@
+package dev.relism.flash.transport;
+
+import dev.relism.flash.extension.FlashApp;
+import dev.relism.flash.extension.FlashConfiguration;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * {@code EX-32}: the two-stage graceful shutdown — stop accepting, let an in-flight request
+ * finish (forced to {@code Connection: close}), then force-close whatever remains after
+ * {@code shutdownDrainTimeoutMs}.
+ */
+class ServerLifecycleGracefulShutdownTest {
+
+ private FlashApp app;
+
+ @AfterEach
+ void tearDown() {
+ if (app != null) app.stop();
+ }
+
+ private static int freePort() throws Exception {
+ try (ServerSocket s = new ServerSocket(0)) {
+ return s.getLocalPort();
+ }
+ }
+
+ @Test
+ void inFlightRequest_completesWithConnectionClose_duringShutdown() throws Exception {
+ int port = freePort();
+ CountDownLatch handlerStarted = new CountDownLatch(1);
+ CountDownLatch releaseHandler = new CountDownLatch(1);
+
+ app = FlashApp.create(FlashConfiguration.builder()
+ .port(port).host("127.0.0.1")
+ .shutdownDrainTimeoutMs(5_000)
+ .build());
+ app.get("/slow", (req, res) -> {
+ handlerStarted.countDown();
+ assertTrue(releaseHandler.await(5, TimeUnit.SECONDS));
+ return "done";
+ });
+ app.start();
+
+ try (Socket socket = new Socket("127.0.0.1", port)) {
+ socket.setSoTimeout(5_000);
+ OutputStream out = socket.getOutputStream();
+ out.write("GET /slow HTTP/1.1\r\nHost: localhost\r\n\r\n".getBytes(StandardCharsets.UTF_8));
+ out.flush();
+
+ assertTrue(handlerStarted.await(2, TimeUnit.SECONDS));
+
+ // Begin shutdown while the handler is still running.
+ CompletableFuture stopping = app.stop();
+ // Give stop() a moment to mark the server as stopping and close the listener.
+ Thread.sleep(100);
+ releaseHandler.countDown();
+
+ byte[] buf = new byte[4096];
+ int n = socket.getInputStream().read(buf);
+ String response = new String(buf, 0, n, StandardCharsets.UTF_8);
+
+ assertTrue(response.startsWith("HTTP/1.1 200 OK"), response);
+ assertTrue(response.contains("done"), response);
+ // EX-32: the in-flight request is forced to close rather than keep-alive, even
+ // though the client asked for HTTP/1.1's default keep-alive.
+ assertTrue(response.contains("Connection: close"), response);
+
+ stopping.get(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ void stop_closesListener_soNewConnectionsAreRefused() throws Exception {
+ int port = freePort();
+ app = FlashApp.create(FlashConfiguration.builder()
+ .port(port).host("127.0.0.1")
+ .shutdownDrainTimeoutMs(500)
+ .build());
+ app.get("/ping", (req, res) -> "pong");
+ app.start();
+
+ // Confirm the server actually answers before stopping it.
+ try (Socket probe = new Socket("127.0.0.1", port)) {
+ probe.setSoTimeout(2_000);
+ probe.getOutputStream().write("GET /ping HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"
+ .getBytes(StandardCharsets.UTF_8));
+ probe.getOutputStream().flush();
+ assertTrue(probe.getInputStream().read() != -1);
+ }
+
+ app.stop().get(5, TimeUnit.SECONDS);
+
+ assertThrows(Exception.class, () -> {
+ try (Socket socket = new Socket()) {
+ socket.connect(new java.net.InetSocketAddress("127.0.0.1", port), 500);
+ socket.setSoTimeout(500);
+ socket.getOutputStream().write("GET /ping HTTP/1.1\r\nHost: localhost\r\n\r\n"
+ .getBytes(StandardCharsets.UTF_8));
+ socket.getOutputStream().flush();
+ int result = socket.getInputStream().read();
+ if (result == -1) throw new java.io.IOException("connection refused/closed, as expected");
+ }
+ });
+ }
+}
diff --git a/flash/src/test/java/dev/relism/flash/websocket/WebSocketFragmentationAndValidationTest.java b/flash/src/test/java/dev/relism/flash/websocket/WebSocketFragmentationAndValidationTest.java
new file mode 100644
index 0000000..2bda2dc
--- /dev/null
+++ b/flash/src/test/java/dev/relism/flash/websocket/WebSocketFragmentationAndValidationTest.java
@@ -0,0 +1,220 @@
+package dev.relism.flash.websocket;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * {@code EX-11} (bulk header read) and {@code EX-12} (continuation reassembly, mandatory
+ * masking, opcode validation, control-frame constraints, correct close codes) coverage for
+ * {@link WebSocketSession#readFrame}.
+ */
+class WebSocketFragmentationAndValidationTest {
+
+ private static final byte[] MASK = {1, 2, 3, 4};
+
+ private static byte[] frame(int opcode, boolean fin, boolean masked, byte[] payload) throws IOException {
+ ByteArrayOutputStream buf = new ByteArrayOutputStream();
+ buf.write((fin ? 0x80 : 0) | opcode);
+ int len = payload.length;
+ int maskBit = masked ? 0x80 : 0x00;
+ if (len <= 125) {
+ buf.write(maskBit | len);
+ } else {
+ buf.write(maskBit | 126);
+ buf.write((len >> 8) & 0xFF);
+ buf.write(len & 0xFF);
+ }
+ byte[] out = payload;
+ if (masked) {
+ buf.write(MASK);
+ out = payload.clone();
+ for (int i = 0; i < out.length; i++) out[i] ^= MASK[i % 4];
+ }
+ buf.write(out);
+ return buf.toByteArray();
+ }
+
+ private static byte[] concat(byte[]... arrays) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ for (byte[] a : arrays) out.write(a);
+ return out.toByteArray();
+ }
+
+ /** Server-mode session (masked incoming required) over the given raw bytes. */
+ private static WebSocketSession serverSession(byte[] raw, int bufferSize) {
+ return new WebSocketSession(new ByteArrayInputStream(raw), new ByteArrayOutputStream(), bufferSize);
+ }
+
+ // --- EX-12: continuation reassembly ------------------------------------------
+
+ @Test
+ void continuationFrames_reassembleIntoOneMessage() throws IOException {
+ byte[] raw = concat(
+ frame(WebSocketFrame.OP_TEXT, false, true, "hel".getBytes(StandardCharsets.UTF_8)),
+ frame(WebSocketFrame.OP_CONTINUATION, false, true, "lo ".getBytes(StandardCharsets.UTF_8)),
+ frame(WebSocketFrame.OP_CONTINUATION, true, true, "world".getBytes(StandardCharsets.UTF_8)));
+ WebSocketSession session = serverSession(raw, 64);
+ WebSocketFrame frame = new WebSocketFrame();
+
+ assertTrue(session.readFrame(frame));
+ assertEquals(WebSocketFrame.OP_TEXT, frame.opcode());
+ assertTrue(frame.isFin());
+ assertEquals("hello world", new String(frame.copyPayload(), StandardCharsets.UTF_8));
+ }
+
+ @Test
+ void controlFrame_interleavedDuringFragmentation_deliveredWithoutDisturbingReassembly() throws IOException {
+ byte[] raw = concat(
+ frame(WebSocketFrame.OP_TEXT, false, true, "AB".getBytes(StandardCharsets.UTF_8)),
+ frame(WebSocketFrame.OP_PING, true, true, "ping".getBytes(StandardCharsets.UTF_8)),
+ frame(WebSocketFrame.OP_CONTINUATION, true, true, "CD".getBytes(StandardCharsets.UTF_8)));
+ WebSocketSession session = serverSession(raw, 64);
+ WebSocketFrame frame = new WebSocketFrame();
+
+ assertTrue(session.readFrame(frame));
+ assertEquals(WebSocketFrame.OP_PING, frame.opcode());
+ assertEquals("ping", new String(frame.copyPayload(), StandardCharsets.UTF_8));
+
+ assertTrue(session.readFrame(frame));
+ assertEquals(WebSocketFrame.OP_TEXT, frame.opcode());
+ assertEquals("ABCD", new String(frame.copyPayload(), StandardCharsets.UTF_8));
+ }
+
+ @Test
+ void continuationWithoutInitiatedMessage_rejected1002() throws IOException {
+ byte[] raw = frame(WebSocketFrame.OP_CONTINUATION, true, true, "x".getBytes(StandardCharsets.UTF_8));
+ WebSocketSession session = serverSession(raw, 64);
+ WebSocketProtocolException e = assertThrows(WebSocketProtocolException.class,
+ () -> session.readFrame(new WebSocketFrame()));
+ assertEquals(1002, e.closeCode());
+ }
+
+ @Test
+ void newDataFrameWhileFragmenting_rejected1002() throws IOException {
+ byte[] raw = concat(
+ frame(WebSocketFrame.OP_TEXT, false, true, "a".getBytes(StandardCharsets.UTF_8)),
+ frame(WebSocketFrame.OP_TEXT, true, true, "b".getBytes(StandardCharsets.UTF_8)));
+ WebSocketSession session = serverSession(raw, 64);
+ WebSocketProtocolException e = assertThrows(WebSocketProtocolException.class,
+ () -> session.readFrame(new WebSocketFrame()));
+ assertEquals(1002, e.closeCode());
+ }
+
+ @Test
+ void reassembledMessageExceedingBuffer_rejected1009() throws IOException {
+ byte[] raw = concat(
+ frame(WebSocketFrame.OP_TEXT, false, true, new byte[5]),
+ frame(WebSocketFrame.OP_CONTINUATION, true, true, new byte[5]));
+ WebSocketSession session = serverSession(raw, 8); // 5 + 5 = 10 > 8
+ WebSocketProtocolException e = assertThrows(WebSocketProtocolException.class,
+ () -> session.readFrame(new WebSocketFrame()));
+ assertEquals(1009, e.closeCode());
+ }
+
+ // --- EX-12: mandatory masking direction ---------------------------------------
+
+ @Test
+ void serverSession_unmaskedIncomingFrame_rejected1002() throws IOException {
+ byte[] raw = frame(WebSocketFrame.OP_TEXT, true, false, "hi".getBytes(StandardCharsets.UTF_8));
+ WebSocketSession session = serverSession(raw, 64);
+ WebSocketProtocolException e = assertThrows(WebSocketProtocolException.class,
+ () -> session.readFrame(new WebSocketFrame()));
+ assertEquals(1002, e.closeCode());
+ }
+
+ @Test
+ void clientSession_maskedIncomingFrame_rejected1002() throws IOException {
+ byte[] raw = frame(WebSocketFrame.OP_TEXT, true, true, "hi".getBytes(StandardCharsets.UTF_8));
+ WebSocketSession session = new WebSocketSession(
+ new ByteArrayInputStream(raw), new ByteArrayOutputStream(), 64, null, true);
+ WebSocketProtocolException e = assertThrows(WebSocketProtocolException.class,
+ () -> session.readFrame(new WebSocketFrame()));
+ assertEquals(1002, e.closeCode());
+ }
+
+ @Test
+ void clientSession_unmaskedIncomingFrame_accepted() throws IOException {
+ byte[] raw = frame(WebSocketFrame.OP_TEXT, true, false, "hi".getBytes(StandardCharsets.UTF_8));
+ WebSocketSession session = new WebSocketSession(
+ new ByteArrayInputStream(raw), new ByteArrayOutputStream(), 64, null, true);
+ WebSocketFrame frame = new WebSocketFrame();
+ assertTrue(session.readFrame(frame));
+ assertEquals("hi", new String(frame.copyPayload(), StandardCharsets.UTF_8));
+ }
+
+ // --- EX-12: opcode validation --------------------------------------------------
+
+ @Test
+ void reservedOpcode_rejected1002() throws IOException {
+ byte[] raw = frame(0x3, true, true, new byte[0]); // 0x3 is reserved
+ WebSocketSession session = serverSession(raw, 64);
+ WebSocketProtocolException e = assertThrows(WebSocketProtocolException.class,
+ () -> session.readFrame(new WebSocketFrame()));
+ assertEquals(1002, e.closeCode());
+ }
+
+ // --- EX-12: control-frame constraints ------------------------------------------
+
+ @Test
+ void fragmentedControlFrame_rejected1002() throws IOException {
+ byte[] raw = frame(WebSocketFrame.OP_PING, false, true, "x".getBytes(StandardCharsets.UTF_8));
+ WebSocketSession session = serverSession(raw, 64);
+ WebSocketProtocolException e = assertThrows(WebSocketProtocolException.class,
+ () -> session.readFrame(new WebSocketFrame()));
+ assertEquals(1002, e.closeCode());
+ }
+
+ @Test
+ void oversizedControlFramePayload_rejected1002() throws IOException {
+ byte[] raw = frame(WebSocketFrame.OP_PING, true, true, new byte[126]);
+ WebSocketSession session = serverSession(raw, 200);
+ WebSocketProtocolException e = assertThrows(WebSocketProtocolException.class,
+ () -> session.readFrame(new WebSocketFrame()));
+ assertEquals(1002, e.closeCode());
+ }
+
+ @Test
+ void controlFrameAtTheMaxAllowedSize_accepted() throws IOException {
+ byte[] raw = frame(WebSocketFrame.OP_PING, true, true, new byte[125]);
+ WebSocketSession session = serverSession(raw, 200);
+ WebSocketFrame frame = new WebSocketFrame();
+ assertTrue(session.readFrame(frame));
+ assertEquals(125, frame.payloadLength());
+ }
+
+ // --- EX-11: bulk header read, not one syscall per byte -------------------------
+
+ private static final class CountingInputStream extends InputStream {
+ private final InputStream delegate;
+ int reads = 0;
+ CountingInputStream(InputStream delegate) { this.delegate = delegate; }
+ @Override public int read() throws IOException { reads++; return delegate.read(); }
+ @Override public int read(byte[] b, int off, int len) throws IOException { reads++; return delegate.read(b, off, len); }
+ }
+
+ @Test
+ void readFrame_withExtendedLengthAndMask_doesNotReadOneByteAtATime() throws IOException {
+ // 200-byte payload forces the 16-bit extended length; masked, so 4 mask bytes too.
+ // Pre-fix: 1 (b0) + 1 (b1) + 2 (extended length, read one at a time originally via two
+ // separate in.read() calls — already bulk-free in that part) + 4 (mask, one at a time)
+ // = several individual reads for the header alone, on top of one per payload byte if
+ // the underlying stream were unbuffered. Post-fix: the header's variable remainder
+ // (length + mask) is exactly one readFully call.
+ byte[] raw = frame(WebSocketFrame.OP_BINARY, true, true, new byte[200]);
+ CountingInputStream counting = new CountingInputStream(new ByteArrayInputStream(raw));
+ WebSocketSession session = new WebSocketSession(counting, new ByteArrayOutputStream(), 256);
+
+ assertTrue(session.readFrame(new WebSocketFrame()));
+
+ // b0, b1, one bulk read for (2 extended-length + 4 mask) bytes, one bulk read for the
+ // 200-byte payload: 4 total, independent of the payload size.
+ assertTrue(counting.reads <= 4, "expected at most 4 underlying reads, was " + counting.reads);
+ }
+}