Relism/Flash5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core): HTTP/2 Phase 2 — transport decomposition

Browse Source 
Breaks HttpServer (563 lines, eleven responsibilities) into named,
single-purpose components and introduces the ConnectionProtocol seam
HTTP/2 plugs into starting Phase 8, per
flash/docs/http2/IMPLEMENTATION-PLAN.md Phase 2.

New packages:
- dev.relism.flash.transport: TransportFactory (composition root, EX-34),
  ListenerBinder, BoundListener, TransportTuning, AcceptLoop,
  ConnectionRunner (per-connection setup/teardown), ConnectionProtocol
  (the h1/h2 seam), ConnectionContext, ConnectionScratch + ScratchPool
  (EX-06), ServerLifecycle (implements ServerHandle; start/stop/graceful
  shutdown, EX-32).
- dev.relism.flash.http1: Http1Connection (the keep-alive request loop,
  implements ConnectionProtocol), Http1ResponseWriter, Http1KeepAlive
  (the shared Connection-header token-list scanner, EX-13).
- dev.relism.flash.websocket additions: WebSocketUpgrade (detection +
  handshake), WebSocketLoop (session loop), WebSocketProtocolException.

Existing-code defects fixed (EX-nn):
- EX-01: WebSocketSession's two blocking-write sites use ReentrantLock
  instead of synchronized (out) -- a virtual thread blocking inside
  synchronized pins its carrier platform thread on Java 21.
- EX-06: HttpServer's three ThreadLocals (SHA1, LONG_BUF,
  STREAM_RELAY_BUFFER) replaced by ConnectionScratch, pooled via
  ScratchPool instead of one-per-virtual-thread (i.e. one-per-connection)
  growth. The router's ThreadLocals are deliberately deferred to Phase 4
  per this EX item's own phasing -- see DEC-15 for the plan-wording fix.
- EX-11: WebSocketSession.readFrame's extended-length and mask-key bytes
  are now read in a single bounded readFully instead of one at a time.
- EX-12: full RFC 6455 frame validation -- continuation-frame
  reassembly, mandatory masking-direction enforcement, opcode
  validation, control-frame constraints (not fragmented, <=125 bytes),
  and WebSocketProtocolException carrying the correct close code (1002
  protocol error, 1009 message too big).
- EX-13: Connection header token-list scanning shared between the
  keep-alive decision and the WebSocket upgrade check.
- EX-14: HEAD responses report Content-Length but write no body.
- EX-15: Content-Type omitted when empty; Content-Length and the body
  omitted entirely for 204/304/1xx responses.
- EX-16: Date header (dev.relism.flash.http.DateHeader), refreshed once
  per second by a shared daemon thread; FlashConfiguration.sendDate.
- EX-32: two-stage graceful shutdown -- stop accepting, force
  Connection: close on the response an in-flight handler is still
  producing (re-checked after the handler runs, not just before
  dispatch, so a shutdown beginning mid-handler is still honoured),
  drain up to shutdownDrainTimeoutMs, then force-close.
- EX-34: ServerHandle.create delegates to TransportFactory instead of
  constructing HttpServer directly.

Two plan corrections recorded: DEC-15 (Phase 2's "no ThreadLocal
anywhere" DoD line contradicted EX-06's own multi-phase assignment --
corrected to match the registry) and DEC-16 (no separate
WebSocketFrameCodec class this phase; the EX-11/EX-12 fixes stay inside
WebSocketSession, which is one cohesive state machine under R6's own
carve-out -- revisit at Phase 15 if RFC 8441 needs the decoupling for
real).

HttpServer.java deleted.

311/311 tests green (flash module), run three times for stability of
the wall-clock-based timeout/shutdown tests. Whole-repo build green.
h1 benchmark regression check remains unverified in the plan's DoD (no
JMH harness until Phase 3, same caveat as Phase 1).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Zakaria El Orche
2026-08-13 12:03:44 +00:00
co-authored by Claude Sonnet 5
parent 5a2aaf5a07
commit a315e1df8b
35 changed files with 2332 additions and 757 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flash/src/main/java/dev/relism/flash/HttpServer.java
-667
View File
@@ -1,667 +0,0 @@
package dev.relism.flash;
import dev.relism.flash.exceptions.MalformedRequestException;
import dev.relism.flash.extension.FlashApp;
import dev.relism.flash.extension.FlashConfiguration;
import dev.relism.flash.http.ContentType;
import dev.relism.flash.http.HttpMethod;
import dev.relism.flash.http.HttpStatus;
import dev.relism.flash.models.Request;
import dev.relism.flash.models.RequestHandler;
import dev.relism.flash.models.Response;
import dev.relism.flash.routing.AbstractRouter;
import dev.relism.flash.routing.AbstractWsRouter;
import dev.relism.flash.tls.TlsConfig;
import dev.relism.flash.transport.BufferedByteSource;
import dev.relism.flash.transport.NegotiatedProtocol;
import dev.relism.flash.transport.ProtocolNegotiator;
import dev.relism.flash.websocket.WebSocketFrame;
import dev.relism.flash.websocket.WebSocketHandler;
import dev.relism.flash.websocket.WebSocketSession;
import dev.relism.fpr.core.ByteView;
import lombok.extern.slf4j.Slf4j;
import javax.net.ssl.SSLServerSocket;
import javax.net.ssl.SSLSocket;
import java.io.*;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
/**
* Pure I/O transport layer. Owns one {@link ServerSocket} per configured listener (plain or
* TLS), the virtual-thread executor, and the keep-alive accept loop. Routing is delegated to
* HTTP and WS routers — identically, regardless of which listener accepted the connection.
*
* <p>TLS is a transport-level concern only: once a {@link BoundListener} is bound, an accepted
* {@link Socket} is either plain or an {@code SSLSocket} indistinguishably from here on —
* {@link #process} never branches on it. This is also why WSS needs no separate code path from
* WS: the WebSocket upgrade happens over whatever transport {@link #process} was handed.
*
* <h3>Allocation model</h3>
* <ul>
* <li>{@code LONG_BUF} (20 bytes) and {@code STREAM_RELAY_BUFFER} (8 KB, for a streaming
* {@link Response} body — see {@link #writeStreamingBody}) are the only {@link ThreadLocal}s
* kept here. Both are per-connection, not per-request: one virtual thread runs a
* connection's whole keep-alive request loop (see {@link #process}), so a handler that
* streams a large response on every request allocates its relay buffer once per
* connection, not once per request.</li>
* <li>WS handshake SHA-1: {@link ThreadLocal}&lt;{@link MessageDigest}&gt; — one per
* accept thread (there are now {@code ACCEPT_THREADS} of them, not one).</li>
* </ul>
*/
@Slf4j
class HttpServer implements ServerHandle {
// ── Tuning constants ──────────────────────────────────────────────────────
/**
* 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.
*/
private 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
*/
private static final int ACCEPT_BACKLOG = 4096;
/**
* Socket send/receive buffer sizes. Matched to the WS frame read buffer
* ({@link 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.
*/
private static final int SOCKET_BUF_SIZE = 256 * 1024;
// ── Instance fields ───────────────────────────────────────────────────────
private final FlashConfiguration configuration;
private final List<BoundListener> boundListeners;
private final AbstractRouter router;
private final AbstractWsRouter wsRouter;
private final ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor();
private final Set<Socket> activeSockets = ConcurrentHashMap.newKeySet();
private volatile boolean stopped = false;
/** Latch that reaches 0 when all accept threads, across all listeners, have exited. */
private final CountDownLatch acceptLatch;
/** One bound listener socket (plain or TLS) plus whether it is TLS, for logging only. */
private record BoundListener(ServerSocket socket, boolean secure) {}
// ── Static byte constants (written once, read-only on hot path) ──────────
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);
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);
private static final byte[] WS_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);
private static final ThreadLocal<MessageDigest> SHA1 =
ThreadLocal.withInitial(() -> {
try { return MessageDigest.getInstance("SHA-1"); }
catch (NoSuchAlgorithmException e) { throw new RuntimeException(e); }
});
private static final ThreadLocal<byte[]> LONG_BUF = ThreadLocal.withInitial(() -> new byte[20]);
/**
* Relay buffer for copying a streaming {@link Response} body to the client — shared by
* {@link #writeStreamingBody}'s non-chunked path and {@link #writeChunked}, so both draw
* from the same reused array instead of each allocating its own {@code byte[8192]} (the
* non-chunked path previously relied on {@link InputStream#transferTo}, which allocates
* internally on every call). Sized to match the pre-existing behavior this replaces, not
* newly tuned — not exposed as a {@link FlashConfiguration} tunable since nothing here
* needed one before.
*/
private static final int STREAM_RELAY_BUFFER_SIZE = 8192;
private static final ThreadLocal<byte[]> STREAM_RELAY_BUFFER =
ThreadLocal.withInitial(() -> new byte[STREAM_RELAY_BUFFER_SIZE]);
private static final int SHA1_LEN = 20;
private static final int WS_ACCEPT_LEN = 28;
// ── Constructor ───────────────────────────────────────────────────────────
HttpServer(FlashConfiguration configuration, AbstractRouter router, AbstractWsRouter wsRouter) throws IOException {
this.configuration = configuration;
this.router = router;
this.wsRouter = wsRouter;
List<FlashConfiguration.Listener> specs = configuration.getListeners().isEmpty()
? List.of(new FlashConfiguration.Listener(
configuration.getPort(), configuration.getHost(), configuration.getTls()))
: configuration.getListeners();
List<BoundListener> bound = new ArrayList<>(specs.size());
for (FlashConfiguration.Listener spec : specs) bound.add(bind(spec));
this.boundListeners = List.copyOf(bound);
this.acceptLatch = new CountDownLatch(ACCEPT_THREADS * boundListeners.size());
for (BoundListener bl : boundListeners) {
log.info("HttpServer bound on {}:{} (tls={}, backlog={}, acceptThreads={})",
bl.socket().getInetAddress(), bl.socket().getLocalPort(), bl.secure(),
ACCEPT_BACKLOG, ACCEPT_THREADS);
}
}
/**
* Binds one listener. A TLS listener gets its {@link ServerSocket} from
* {@link TlsConfig#serverSocketFactory()} instead of {@code new ServerSocket()}, and its
* protocol/client-auth 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.
*/
private 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(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, ACCEPT_BACKLOG);
return new BoundListener(socket, tls != null);
}
// ── Lifecycle ─────────────────────────────────────────────────────────────
@Override
public void start() {
for (int li = 0; li < boundListeners.size(); li++) {
BoundListener listener = boundListeners.get(li);
for (int i = 0; i < ACCEPT_THREADS; i++) {
Thread.ofPlatform()
.name("flash-accept-" + li + "-" + i)
.daemon(false)
.start(() -> acceptLoop(listener));
}
}
}
@Override
public void startAndBlock() {
start();
try { acceptLatch.await(); }
catch (InterruptedException e) { Thread.currentThread().interrupt(); }
}
/**
* Single accept loop body — runs on each of the {@code ACCEPT_THREADS}
* platform threads bound to one {@code listener}. All threads for that listener block on
* the same {@link ServerSocket}; the JVM ensures only one wakes per incoming connection
* (no thundering herd). Other listeners' accept threads are entirely independent.
*/
private void acceptLoop(BoundListener listener) {
try {
while (!stopped) {
try {
process(listener.socket().accept());
} catch (IOException e) {
if (!stopped) log.error("Accept error", e);
}
}
} finally {
acceptLatch.countDown();
}
}
@Override
public CompletableFuture<Void> 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');
}
}
flash/src/main/java/dev/relism/flash/RequestParser.java
+1 -1
View File
@@ -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;
}
flash/src/main/java/dev/relism/flash/ServerHandle.java
+4 -2
View File
@@ -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);
}
}
flash/src/main/java/dev/relism/flash/extension/FlashConfiguration.java
+8
View File
@@ -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); }
flash/src/main/java/dev/relism/flash/http/DateHeader.java
+56
View File
@@ -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).
*
* <p>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;
}
}
flash/src/main/java/dev/relism/flash/http1/Http1Connection.java
+131
View File
@@ -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;
}
}
}
flash/src/main/java/dev/relism/flash/http1/Http1KeepAlive.java
+71
View File
@@ -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.
*
* <p>{@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;
}
}
flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java
+170
View File
@@ -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}.
*
* <p>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');
}
}
flash/src/main/java/dev/relism/flash/transport/AcceptLoop.java
+29
View File
@@ -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);
}
}
}
}
flash/src/main/java/dev/relism/flash/transport/BoundListener.java
+7
View File
@@ -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) {
}
flash/src/main/java/dev/relism/flash/transport/ConnectionContext.java
+49
View File
@@ -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
) {
}
flash/src/main/java/dev/relism/flash/transport/ConnectionProtocol.java
+14
View File
@@ -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;
}
flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java
+142
View File
@@ -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.
*
* <p>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<Socket> 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<Socket> 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();
}
}
}
flash/src/main/java/dev/relism/flash/transport/ConnectionScratch.java
+65
View File
@@ -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}.
*
* <h3>Why not {@code ThreadLocal}</h3>
* {@code ThreadLocal} is the right idiom for a bounded platform-thread pool, where "one per
* thread" means "one per core". Flash runs one <em>virtual</em> 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.
*
* <h3>Lifetime and thread-safety contract</h3>
* 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.
*
* <p>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.
}
}
flash/src/main/java/dev/relism/flash/transport/ListenerBinder.java
+43
View File
@@ -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.
*
* <p>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);
}
}
flash/src/main/java/dev/relism/flash/transport/ScratchPool.java
+61
View File
@@ -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.
*
* <p>This is a <em>cache</em>, 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.
*
* <h3>Thread-safety</h3>
* {@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<ConnectionScratch> 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);
}
}
flash/src/main/java/dev/relism/flash/transport/ServerLifecycle.java
+115
View File
@@ -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.
*
* <p>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<BoundListener> listeners;
private final ConnectionRunner runner;
private final FlashConfiguration configuration;
private final ExecutorService executorService;
private final Set<Socket> activeSockets;
private final CountDownLatch acceptLatch;
private volatile boolean stopped = false;
public ServerLifecycle(List<BoundListener> listeners, ConnectionRunner runner,
FlashConfiguration configuration, ExecutorService executorService,
Set<Socket> 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<Void> 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();
}
});
}
}
flash/src/main/java/dev/relism/flash/transport/TransportFactory.java
+63
View File
@@ -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.
*
* <p>{@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<FlashConfiguration.Listener> specs = configuration.getListeners().isEmpty()
? List.of(new FlashConfiguration.Listener(
configuration.getPort(), configuration.getHost(), configuration.getTls()))
: configuration.getListeners();
List<BoundListener> bound = new ArrayList<>(specs.size());
for (FlashConfiguration.Listener spec : specs) bound.add(ListenerBinder.bind(spec));
List<BoundListener> 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<Socket> 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);
}
}
flash/src/main/java/dev/relism/flash/transport/TransportTuning.java
+39
View File
@@ -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;
}
flash/src/main/java/dev/relism/flash/websocket/WebSocketLoop.java
+43
View File
@@ -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();
}
}
}
flash/src/main/java/dev/relism/flash/websocket/WebSocketProtocolException.java
+23
View File
@@ -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;
}
}
flash/src/main/java/dev/relism/flash/websocket/WebSocketSession.java
+172 -49
View File
@@ -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.
*
* <ul>
* <li>
* <p>With {@code TCP_NODELAY} enabled on the socket (set in {@code HttpServer}),
* <p>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;
* <p>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}.</li>
* {@code WebSocketUpgrade#performHandshake}.</li>
*
* <li><b>Flush on CLOSE frame</b>: {@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;
* </ul>
*
* <h3>Thread safety</h3>
* {@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.
*
* <h3>Fragmentation, masking, and control frames (RFC 6455 §5)</h3>
* {@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 <em>client</em> — 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.
*
* <p>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.
* <p>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.
*
* <p><b>{@link #maskOutgoing} (client mode):</b> 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; }
}
}
}
flash/src/main/java/dev/relism/flash/websocket/WebSocketUpgrade.java
+71
View File
@@ -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();
}
}