From ee90ac44ff02901da915990bfdc5c1d1d1058245 Mon Sep 17 00:00:00 2001 From: Zakaria El Orche Date: Thu, 13 Aug 2026 19:23:26 +0000 Subject: [PATCH] feat(core): add HTTP trailers and push streaming --- README.md | 30 +++- flash/docs/http2/DECISIONS.md | 19 +++ flash/docs/http2/IMPLEMENTATION-PLAN.md | 37 +++-- flash/docs/http2/TRAILERS-AND-STREAMING.md | 36 +++++ .../dev/relism/flash/ChunkedInputStream.java | 83 +++++++++- .../java/dev/relism/flash/RequestParser.java | 11 +- .../flash/http1/Http1ResponseWriter.java | 41 ++++- .../relism/flash/http2/Http2Connection.java | 89 ++++++++--- .../flash/http2/Http2StreamDispatcher.java | 43 +++-- .../flash/http2/message/Http2HeaderMap.java | 10 +- .../flash/http2/message/Http2RequestBody.java | 20 ++- .../http2/message/Http2ResponseWriter.java | 86 +++++++--- .../flash/http2/message/PseudoHeaders.java | 14 ++ .../flash/http2/stream/Http2Stream.java | 17 +- .../relism/flash/models/BodyCompletion.java | 6 + .../relism/flash/models/EmptyHeaderView.java | 18 +++ .../relism/flash/models/MutableHeaderMap.java | 138 ++++++++++++++++ .../flash/models/ProducerInputStream.java | 103 ++++++++++++ .../java/dev/relism/flash/models/Request.java | 20 +++ .../dev/relism/flash/models/RequestBody.java | 13 ++ .../dev/relism/flash/models/Response.java | 81 ++++++++++ .../flash/models/ResponseSerializer.java | 5 + .../relism/flash/models/ResponseStream.java | 11 ++ .../relism/flash/routing/AbstractRouter.java | 5 +- .../dev/relism/flash/routing/PathUtils.java | 8 + .../relism/flash/ChunkedInputStreamTest.java | 2 +- .../dev/relism/flash/Http1TrailersTest.java | 68 ++++++++ .../relism/flash/http2/GrpcInteropTest.java | 114 +++++++++++++ .../relism/flash/http2/Http2ConnectTest.java | 103 ++++++++++++ .../http2/Http2ConnectionIntegrationTest.java | 57 +++++++ .../flash/http2/Http2HalfCloseTest.java | 50 ++++++ .../relism/flash/http2/Http2TrailersTest.java | 150 ++++++++++++++++++ .../message/Http2ResponseWriterTest.java | 17 ++ .../flash/models/ResponseStreamTest.java | 50 ++++++ 34 files changed, 1468 insertions(+), 87 deletions(-) create mode 100644 flash/docs/http2/TRAILERS-AND-STREAMING.md create mode 100644 flash/src/main/java/dev/relism/flash/models/BodyCompletion.java create mode 100644 flash/src/main/java/dev/relism/flash/models/EmptyHeaderView.java create mode 100644 flash/src/main/java/dev/relism/flash/models/MutableHeaderMap.java create mode 100644 flash/src/main/java/dev/relism/flash/models/ProducerInputStream.java create mode 100644 flash/src/main/java/dev/relism/flash/models/ResponseStream.java create mode 100644 flash/src/test/java/dev/relism/flash/Http1TrailersTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/GrpcInteropTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/Http2ConnectTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/Http2HalfCloseTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/Http2TrailersTest.java create mode 100644 flash/src/test/java/dev/relism/flash/models/ResponseStreamTest.java diff --git a/README.md b/README.md index 134f89b..5c0dec8 100644 --- a/README.md +++ b/README.md @@ -172,7 +172,7 @@ app.onException((ex, req, res) -> { | `idleKeepAliveTimeoutMs` | `60000` | How long a keep-alive connection may sit idle waiting for its next request. | | `bodyReadTimeoutMs` | `30000` | How long reading a request body (handler or automatic drain) may take. | | `shutdownDrainTimeoutMs` | `15000` | How long graceful shutdown waits for in-flight requests before force-closing. | -| `http2Enabled` | `false` | Whether this server will ever negotiate HTTP/2. Off by default until the HTTP/2 connection state machine lands (see `flash/docs/http2/IMPLEMENTATION-PLAN.md`). | +| `http2Enabled` | `false` | Whether the server negotiates HTTP/2 through ALPN or accepts h2c prior knowledge. The conservative default keeps protocol rollout explicit. | ## TLS @@ -316,6 +316,34 @@ app.get("/health", (req, res) -> res.header(NO_STORE).body("ok")); HTTP/1-only; HPACK needs the name and value as separate fields. Prefer `PreEncodedHeader` for shared application and middleware code. +### Trailers and push streaming + +Request trailers become available after the body reaches EOF: + +```java +byte[] payload = req.body().bytes(); +String status = req.trailers().first("grpc-status"); +``` + +For a producer-driven response, `Response.streaming` provides a blocking `ResponseStream`. Its +bounded buffer and HTTP/2 flow-control windows apply backpressure directly to the producer's +virtual thread: + +```java +return res.streaming(stream -> { + try { + stream.write(payload, 0, payload.length); + stream.trailer("result", "complete"); + } catch (IOException failure) { + throw new UncheckedIOException(failure); + } +}); +``` + +The API renders as chunked data and trailers on HTTP/1.1, and DATA plus trailing HEADERS on +HTTP/2. Flash core supplies these transport primitives; a higher-level gRPC codec belongs in a +separate extension. + ## Architecture ``` diff --git a/flash/docs/http2/DECISIONS.md b/flash/docs/http2/DECISIONS.md index c073a40..5a59eb0 100644 --- a/flash/docs/http2/DECISIONS.md +++ b/flash/docs/http2/DECISIONS.md @@ -981,3 +981,22 @@ the allocation noise floor. window and pool byte capacity together; never raise credit independently of bounded storage. --- + +## DEC-29 — Keep HTTP/2 opt-in until the adversarial phase is complete + +**Context.** Trailers and push streaming make the protocol feature-complete for ordinary and gRPC- +shaped traffic, but the dedicated rate-based and composite abuse controls are deliberately owned +by the following security phase. + +**Decision.** Keep `FlashConfiguration.http2Enabled` defaulting to `false` during this phase. +Applications can enable the complete path explicitly; the default changes only after the hostile- +peer suite and its limits are green. + +**Consequence.** Existing deployments do not silently expose a newly completed protocol before its +adversarial gate. This is rollout sequencing, not an architectural separation: both protocols use +the same public request/response, header, trailer and streaming APIs. + +**Revisit when.** At Phase 13 closure; either flip the default with evidence or record why it must +remain opt-in. + +--- diff --git a/flash/docs/http2/IMPLEMENTATION-PLAN.md b/flash/docs/http2/IMPLEMENTATION-PLAN.md index 653f174..4d7b03d 100644 --- a/flash/docs/http2/IMPLEMENTATION-PLAN.md +++ b/flash/docs/http2/IMPLEMENTATION-PLAN.md @@ -73,7 +73,7 @@ Status values: `not started` / `in progress` / `blocked` / `done`. | 9 — HPACK encoder + h2 response path | done | `feature/core/http2` | Stateless static-table HPACK encoder; precompiled status/content-type/date fields; reusable response writer with header filtering, bounds, CONTINUATION splitting and fixed DATA happy path; HTTP/1/2 serializer parity test. EX-46 fixed the one-digit Date day-of-month bug. JMH: 174.309 ns/op, 0.001 B/op (noise floor), no GC. 603/603 tests green from a clean `-Pjmh` build. | | 10 — Stream state machine + dispatch | done | `feature/core/http2` | Explicit stream transition table, bounded primitive stream table and pool, pseudo-header/message validation, protocol-neutral `Request` assembly, virtual-thread dispatch and exception path, cancellation-safe release, raw h2c + Java HTTP/2 integration. Phase 11 closed the two deferred content-length/DATA cases; h2spec sections 5/8 are now 39/39. JMH pooled lifecycle: 458.499 ns/op, 0.003 B/op, no GC. 618/618 tests green at phase closure. | | 11 — DATA, flow control, bodies | done | `feature/core/http2` | Two-level receive/send flow control, consumption-driven WINDOW_UPDATE hysteresis, bounded/coalescing DATA pool, inline and blocking streaming request bodies through the existing `RequestBody`, resumable fixed/known/unknown response streams, content-length and empty-DATA validation. Real TLS HTTP/2 transfer: 100 MiB upload + 100 MiB download verified byte-for-byte. h2spec combined sections 5, 6.1, 6.9 and 8: 50 passed, 1 tool-skipped, 0 failed. JMH: inline materialization exactly one 1,040-byte array; request streaming 0.001 B/op; response streaming 0.002 B/op; full pooled lifecycle 0.003 B/op. 633/633 tests green from a clean `-Pjmh` build. | -| 12 — Trailers, half-close, gRPC | not started | — | — | +| 12 — Trailers, half-close, gRPC | done | `feature/core/http2` | Protocol-neutral request/response trailers, bounded push streaming, four half-close orderings and authority-form CONNECT tunnels complete. Real grpcurl 1.9.3 unary/server-streaming/error interop passes. EX-48/49 fixed. HTTP/2 remains opt-in until the Phase 13 hostile-peer gate (DEC-29). 649/649 tests green from a clean `-Pjmh` build. | | 13 — Security hardening & abuse resistance | not started | — | — | | 14 — h2c prior knowledge + proxy support | not started | — | — | | 15 — RFC 8441 extended CONNECT (WS over h2) | not started | — | — | @@ -778,6 +778,22 @@ the final release. The regression test sends a complete request, immediately res a second request and proves that only the second handler invocation and response occur. **Phase**: 10. +### EX-48 — HTTP/1.1 request trailers were parsed and discarded + +Found while exposing the protocol-neutral request trailer API. `ChunkedInputStream` consumed and +bounded the final trailer section but discarded every field, so no honest API could provide the +same semantics on HTTP/1.1 and HTTP/2. **Fix**: parse the bounded section into a connection-owned +`MutableHeaderMap`, expose it through `Request.trailers()` only after body EOF, reject malformed and +framing-sensitive fields, and add HTTP/1 parity/regression tests. **Phase**: 12. + +### EX-49 — CONNECT routes were registered as origin-form paths + +Found while exercising an HTTP/2 tunnel. The public `connect("authority", handler)` API passed +through the ordinary path sanitizer, which prepended `/`; both HTTP/1.1 authority-form request +targets and HTTP/2 `:authority` arrive without that prefix, so the existing CONNECT API could +never match its documented target. **Fix**: normalize CONNECT authority targets separately in the +shared router registration path and verify a live bidirectional HTTP/2 tunnel. **Phase**: 12. + --- # PART III — The phases @@ -2730,11 +2746,11 @@ error. boundary in `DECISIONS.md` as `DEC-08`. ### Safety checks -- [ ] Trailers without `END_STREAM` rejected -- [ ] Pseudo-headers in trailers rejected -- [ ] Trailer count and size bounded (they go through the same HPACK limits) -- [ ] `ResponseStream.write` after `close` throws, does not corrupt the stream -- [ ] CONNECT tunnels are bounded by the same timeouts and flow control as normal streams +- [x] Trailers without `END_STREAM` rejected +- [x] Pseudo-headers in trailers rejected +- [x] Trailer count and size bounded (they go through the same HPACK limits) +- [x] `ResponseStream.write` after `close` throws, does not corrupt the stream +- [x] CONNECT tunnels are bounded by the same timeouts and flow control as normal streams ### Tests - `Http2TrailersTest`, `Http1TrailersTest` (the h1 rendering), `TrailerParityTest`. @@ -2748,11 +2764,12 @@ error. - `README.md` — the `ResponseStream` API, with a gRPC-shaped example. ### DoD -- [ ] `grpcurl` completes a unary and a server-streaming call against a Flash handler. -- [ ] Trailers work on both protocols through one API. -- [ ] `FlashConfiguration.http2Enabled` flips to default `true` (the feature is now complete +- [x] `grpcurl` completes a unary and a server-streaming call against a Flash handler. +- [x] Trailers work on both protocols through one API. +- [x] `FlashConfiguration.http2Enabled` flips to default `true` (the feature is now complete enough to be on by default) — or, if the team prefers a conservative rollout, stays - `false` with the decision recorded. + `false` with the decision recorded (`DEC-29`: retain opt-in until Phase 13's hostile-peer + suite is complete). --- diff --git a/flash/docs/http2/TRAILERS-AND-STREAMING.md b/flash/docs/http2/TRAILERS-AND-STREAMING.md new file mode 100644 index 0000000..2043f80 --- /dev/null +++ b/flash/docs/http2/TRAILERS-AND-STREAMING.md @@ -0,0 +1,36 @@ +# Trailers and streaming + +Flash exposes the same request and response model on HTTP/1.1 and HTTP/2. Request trailers are +available through `Request.trailers()` after the body has reached EOF. Calling it earlier throws +`IllegalStateException`; this prevents handlers from observing an incomplete trailer section. +HTTP/1.1 reads trailers from the final chunk, while HTTP/2 decodes the trailing HEADERS block in +the connection's existing HPACK context. + +Response trailers are added with `Response.trailer(name, value)` or a `PreEncodedHeader`. HTTP/1.1 +uses chunked framing and writes the fields after the zero chunk. HTTP/2 writes a trailing HEADERS +block with `END_STREAM`; the final DATA frame deliberately does not carry `END_STREAM`. + +`Response.streaming(producer)` is the push alternative to `stream(InputStream, length)` and +`chunked(InputStream)`. Its `ResponseStream` is a bounded blocking bridge. A producer runs on a +virtual thread and blocks when the protocol writer or the HTTP/2 flow-control windows cannot make +progress. This keeps backpressure explicit without callbacks or reactive types: + +```java +return response.type("application/grpc").streaming(stream -> { + try { + for (byte[] message : messages) stream.write(message, 0, message.length); + stream.trailer("grpc-status", "0"); + } catch (IOException failure) { + throw new UncheckedIOException(failure); + } +}); +``` + +The transport supports the primitives required by gRPC, but the core does not provide protobuf +codecs, generated stubs, service descriptors, or a gRPC service API. Those belong in a future +`flash-ext-grpc` module. `GrpcInteropTest` verifies the boundary with the external `grpcurl` client +and a hand-written wire-format handler. + +CONNECT requests follow RFC 9113 request pseudo-header rules: `:authority` is required and +`:scheme`/`:path` are forbidden. Their DATA remains subject to the ordinary request limits, +timeouts and two-level flow control. diff --git a/flash/src/main/java/dev/relism/flash/ChunkedInputStream.java b/flash/src/main/java/dev/relism/flash/ChunkedInputStream.java index d3d6ec9..182c8a1 100644 --- a/flash/src/main/java/dev/relism/flash/ChunkedInputStream.java +++ b/flash/src/main/java/dev/relism/flash/ChunkedInputStream.java @@ -2,6 +2,8 @@ package dev.relism.flash; import dev.relism.flash.exceptions.MalformedRequestException; import dev.relism.flash.http.Http1Limits; +import dev.relism.flash.models.BodyCompletion; +import dev.relism.flash.models.MutableHeaderMap; import dev.relism.flash.transport.BufferedByteSource; import java.io.IOException; @@ -20,15 +22,28 @@ import java.io.InputStream; * {@link BufferedByteSource#prependOnce}, replacing the {@code SequenceInputStream}/ * {@code ByteArrayInputStream} pair the previous implementation allocated per chunked request. */ -final class ChunkedInputStream extends InputStream { +final class ChunkedInputStream extends InputStream implements BodyCompletion { private final BufferedByteSource src; private int chunkRemaining = 0; private boolean done = false; private int chunksSeen = 0; + private final MutableHeaderMap trailers; + private final byte[] trailerLine = new byte[Http1Limits.MAX_HEADER_VALUE_LENGTH]; + + ChunkedInputStream(BufferedByteSource src, byte[] preBuf, int preBufOff, int preBufLen, + MutableHeaderMap trailers) { + this.src = src; + this.trailers = trailers; + if (preBufLen > 0) src.prependOnce(preBuf, preBufOff, preBufLen); + } ChunkedInputStream(BufferedByteSource src, byte[] preBuf, int preBufOff, int preBufLen) { - this.src = src; - if (preBufLen > 0) src.prependOnce(preBuf, preBufOff, preBufLen); + this(src, preBuf, preBufOff, preBufLen, new MutableHeaderMap()); + } + + @Override + public boolean fullyRead() { + return done; } @Override @@ -129,7 +144,7 @@ final class ChunkedInputStream extends InputStream { int trailerCount = 0; while (true) { int b = src.read(); - if (b == -1) return; // EOF mid-trailers — nothing left to bound. + if (b == -1) throw new MalformedRequestException(400, "Truncated trailer section"); if (b == '\r') { if (src.read() != '\n') { throw new MalformedRequestException(400, "Malformed trailer section terminator"); @@ -139,12 +154,68 @@ final class ChunkedInputStream extends InputStream { if (++trailerCount > Http1Limits.MAX_TRAILER_COUNT) { throw new MalformedRequestException(431, "Too many trailers"); } - int lineLen = 1; + int lineLen = 0; + trailerLine[lineLen++] = (byte) b; while ((b = src.read()) != -1 && b != '\n') { - if (++lineLen > Http1Limits.MAX_HEADER_VALUE_LENGTH) { + if (lineLen == trailerLine.length) { throw new MalformedRequestException(431, "Trailer line too long"); } + trailerLine[lineLen++] = (byte) b; } + if (b != '\n' || lineLen == 0 || trailerLine[lineLen - 1] != '\r') { + throw new MalformedRequestException(400, "Malformed trailer line"); + } + addTrailer(lineLen - 1); } } + + private void addTrailer(int lineLength) throws MalformedRequestException { + int colon = -1; + for (int i = 0; i < lineLength; i++) { + if (trailerLine[i] == ':') { colon = i; break; } + } + if (colon <= 0) throw new MalformedRequestException(400, "Malformed trailer field"); + for (int i = 0; i < colon; i++) { + int c = trailerLine[i] & 0xff; + if (!isToken(c)) { + throw new MalformedRequestException(400, "Invalid trailer field name"); + } + } + int valueStart = colon + 1; + while (valueStart < lineLength + && (trailerLine[valueStart] == ' ' || trailerLine[valueStart] == '\t')) valueStart++; + int valueEnd = lineLength; + while (valueEnd > valueStart + && (trailerLine[valueEnd - 1] == ' ' || trailerLine[valueEnd - 1] == '\t')) valueEnd--; + if (forbidden(trailerLine, colon)) { + throw new MalformedRequestException(400, "Forbidden trailer field"); + } + trailers.add(trailerLine, 0, colon, trailerLine, valueStart, valueEnd - valueStart); + } + + private static boolean isToken(int c) { + return (c >= '0' && c <= '9') + || (c >= 'A' && c <= 'Z') + || (c >= 'a' && c <= 'z') + || c == '!' || c == '#' || c == '$' || c == '%' || c == '&' || c == '\'' + || c == '*' || c == '+' || c == '-' || c == '.' || c == '^' || c == '_' + || c == '`' || c == '|' || c == '~'; + } + + private static boolean forbidden(byte[] name, int length) { + return asciiEquals(name, length, "content-length") + || asciiEquals(name, length, "transfer-encoding") + || asciiEquals(name, length, "host") + || asciiEquals(name, length, "trailer"); + } + + private static boolean asciiEquals(byte[] bytes, int length, String expected) { + if (length != expected.length()) return false; + for (int i = 0; i < length; i++) { + int c = bytes[i] & 0xff; + if (c >= 'A' && c <= 'Z') c += 32; + if (c != expected.charAt(i)) return false; + } + return true; + } } diff --git a/flash/src/main/java/dev/relism/flash/RequestParser.java b/flash/src/main/java/dev/relism/flash/RequestParser.java index e50f8c9..dc4d435 100644 --- a/flash/src/main/java/dev/relism/flash/RequestParser.java +++ b/flash/src/main/java/dev/relism/flash/RequestParser.java @@ -5,6 +5,7 @@ import dev.relism.flash.exceptions.MalformedRequestException; import dev.relism.flash.http.Http1Limits; import dev.relism.flash.http.HttpMethod; import dev.relism.flash.models.Http1HeaderMap; +import dev.relism.flash.models.MutableHeaderMap; import dev.relism.flash.models.Request; import dev.relism.flash.models.RequestBody; import dev.relism.flash.models.RequestLine; @@ -57,6 +58,7 @@ public class RequestParser { private final InetSocketAddress remoteAddress; private final SSLSocket sslSocket; private final Http1HeaderMap headerMap = new Http1HeaderMap(); + private final MutableHeaderMap trailerMap = new MutableHeaderMap(); // request — same idiom as headerMap above. private final RequestLine requestLine = new RequestLine(); private final Request request = new Request(); @@ -115,6 +117,7 @@ public class RequestParser { * @throws IOException on genuine I/O failure (socket reset, timeout). */ public Request parse(BufferedByteSource in) throws IOException { + trailerMap.reset(); // Snapshot leftover bytes from the previous request, then reset immediately. // Any exception thrown below leaves bufBase/bufLen at 0 — safe state. int base = bufBase; @@ -293,11 +296,15 @@ public class RequestParser { // is handled by the same call: preBufLen is already forced to 0 for it above) or the // chunked case, never reallocated. if (isChunked) { - requestBody.reset(new ChunkedInputStream(in, buffer, bodyStart, preBufLen), -1L, null, 0, 0); + requestBody.reset( + new ChunkedInputStream(in, buffer, bodyStart, preBufLen, trailerMap), + -1L, null, 0, 0); } else { requestBody.reset(in, contentLength, buffer, bodyStart, preBufLen); } - return Request.forParsed(request, requestLine, requestBody, remoteAddress, sslSocket); + Request parsed = Request.forParsed(request, requestLine, requestBody, remoteAddress, sslSocket); + parsed.setTrailers(trailerMap); + return parsed; } /** diff --git a/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java b/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java index c68332a..fe0b68f 100644 --- a/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java +++ b/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java @@ -82,7 +82,9 @@ public final class Http1ResponseWriter { response.writeHeadersInto(head); - if (response.isStreaming()) { + if (response.hasTrailers()) { + writeTrailerBody(out, head, response, keepAlive, suppressBody, scratch); + } else if (response.isStreaming()) { writeStreamingBody(out, head, response, keepAlive, noContentAllowed, suppressBody, scratch); } else { byte[] body = response.getBody(); @@ -109,6 +111,28 @@ public final class Http1ResponseWriter { out.flush(); } + private static void writeTrailerBody(OutputStream out, ByteWriter head, Response response, + boolean keepAlive, boolean suppressBody, + ConnectionScratch scratch) throws IOException { + head.writeBytes(TRANSFER_CHUNKED); + head.writeBytes(keepAlive ? CONNECTION_KEEPALIVE : CONNECTION_CLOSE); + head.writeBytes(CRLF); + out.write(head.array(), 0, head.length()); + if (suppressBody) return; + if (response.isStreaming()) { + writeChunked(out, response.getStream(), response, scratch); + } else { + byte[] body = response.getBody(); + if (body != null && body.length != 0) { + writeHex(out, body.length); + out.write(CRLF); + out.write(body); + out.write(CRLF); + } + writeFinalChunk(out, response); + } + } + private static void writeStreamingBody(OutputStream out, ByteWriter head, Response response, boolean keepAlive, boolean noContentAllowed, boolean suppressBody, ConnectionScratch scratch) throws IOException { @@ -130,7 +154,7 @@ public final class Http1ResponseWriter { // 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); + if (!suppressBody) writeChunked(out, response.getStream(), response, scratch); } } @@ -150,7 +174,8 @@ public final class Http1ResponseWriter { else { head.writeDecimal(statusCode); head.writeBytes(UNKNOWN_STATUS_SUFFIX); } } - private static void writeChunked(OutputStream out, InputStream stream, ConnectionScratch scratch) throws IOException { + private static void writeChunked(OutputStream out, InputStream stream, Response response, + ConnectionScratch scratch) throws IOException { byte[] buf = scratch.relayBuffer; int n; while ((n = stream.read(buf)) > 0) { @@ -159,7 +184,15 @@ public final class Http1ResponseWriter { out.write(buf, 0, n); out.write(CRLF); } - out.write(FINAL_CHUNK); + if (response.hasTrailers()) writeFinalChunk(out, response); + else out.write(FINAL_CHUNK); + } + + private static void writeFinalChunk(OutputStream out, Response response) throws IOException { + out.write('0'); + out.write(CRLF); + response.writeTrailers(out); + out.write(CRLF); } private static void writeHex(OutputStream out, int value) throws IOException { diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java b/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java index 68bc4bd..35d763c 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java @@ -59,6 +59,7 @@ public final class Http2Connection implements ConnectionProtocol { private int highestClientStreamId; private Http2Stream pendingHeaderStream; private boolean refusingHeaderStream; + private boolean pendingTrailers; private Http2StreamDispatcher streamDispatcher; private final Http2Stream[] dispatchQueue = new Http2Stream[Http2Limits.MAX_CONCURRENT_STREAMS]; private int dispatchCount; @@ -218,8 +219,21 @@ public final class Http2Connection implements ConnectionProtocol { private void receiveHeaders(FrameHeader frame, Http2FrameWriter writer) throws IOException { int streamId = frame.streamId(); if ((streamId & 1) == 0) throw Http2Exception.PROTOCOL_ERROR; + Http2Stream existing = streams.get(streamId); + if (existing != null) { + if (!FrameFlags.isEndStream(frame.flags())) { + throw new Http2StreamException( + streamId, Http2ErrorCode.PROTOCOL_ERROR, "trailers require END_STREAM"); + } + pendingHeaderStream = existing; + pendingTrailers = true; + existing.trailerBlock().reset(); + if (headerBlocks.accept(frame, existing.trailerBlock())) completeHeaders(writer, streamId); + return; + } if (streamId <= highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR; highestClientStreamId = streamId; + pendingTrailers = false; pendingHeaderStream = streams.acquire(streamId); refusingHeaderStream = pendingHeaderStream == null; @@ -227,7 +241,12 @@ public final class Http2Connection implements ConnectionProtocol { flowController.initializeStreamSendWindow( pendingHeaderStream, peerSettings.initialWindowSize()); } - HeaderSink sink = refusingHeaderStream ? DISCARD_HEADERS : pendingHeaderStream.headerBlock(); + HeaderSink sink = + refusingHeaderStream + ? DISCARD_HEADERS + : (pendingTrailers + ? pendingHeaderStream.trailerBlock() + : pendingHeaderStream.headerBlock()); if (headerBlocks.accept(frame, sink)) completeHeaders(writer, frame.streamId()); } @@ -235,33 +254,53 @@ public final class Http2Connection implements ConnectionProtocol { if (pendingHeaderStream == null && !refusingHeaderStream) { throw Http2Exception.PROTOCOL_ERROR; } - HeaderSink sink = refusingHeaderStream ? DISCARD_HEADERS : pendingHeaderStream.headerBlock(); + HeaderSink sink = + refusingHeaderStream + ? DISCARD_HEADERS + : (pendingTrailers + ? pendingHeaderStream.trailerBlock() + : pendingHeaderStream.headerBlock()); if (headerBlocks.accept(frame, sink)) completeHeaders(writer, frame.streamId()); } private void completeHeaders(Http2FrameWriter writer, int streamId) throws IOException { - if (refusingHeaderStream) { - sendRstStream(writer, streamId, Http2ErrorCode.REFUSED_STREAM); - } else { - Http2Stream stream = pendingHeaderStream; - if (streamDispatcher != null) stream.validateHeaders(); - boolean dispatch = - stream.prepareRequestBody(flowController, headerBlocks.endStream()); - stream.transition( - headerBlocks.endStream() - ? Http2StreamState.Event.RECV_HEADERS_ES - : Http2StreamState.Event.RECV_HEADERS); - lastProcessedStreamId = Math.max(lastProcessedStreamId, streamId); - if (streamDispatcher == null) { - streams.remove(streamId); - streams.release(stream); - if (!gracefulStarted) startGracefulShutdown(writer); - } else if (dispatch) { - enqueueDispatch(stream); + try { + if (refusingHeaderStream) { + sendRstStream(writer, streamId, Http2ErrorCode.REFUSED_STREAM); + } else { + Http2Stream stream = pendingHeaderStream; + boolean dispatch; + if (pendingTrailers) { + stream.validateTrailers(); + stream.finishRequestBody(); + stream.transition(Http2StreamState.Event.RECV_HEADERS_ES); + dispatch = !stream.dispatched(); + } else { + if (streamDispatcher != null) stream.validateHeaders(); + dispatch = stream.prepareRequestBody(flowController, headerBlocks.endStream()); + stream.transition( + headerBlocks.endStream() + ? Http2StreamState.Event.RECV_HEADERS_ES + : Http2StreamState.Event.RECV_HEADERS); + lastProcessedStreamId = Math.max(lastProcessedStreamId, streamId); + } + if (streamDispatcher == null && !pendingTrailers) { + streams.remove(streamId); + streams.release(stream); + if (!gracefulStarted) startGracefulShutdown(writer); + } else if (dispatch) { + enqueueDispatch(stream); + } else if (stream.responseStarted() && stream.responseWriter().finished() + && !stream.responseInFlight() && stream.state() == Http2StreamState.CLOSED) { + streams.remove(stream.id()); + streams.release(stream); + } } + } finally { + pendingHeaderStream = null; + refusingHeaderStream = false; + pendingTrailers = false; } - pendingHeaderStream = null; - refusingHeaderStream = false; } private void receivePriority(FrameHeader frame) { @@ -310,6 +349,12 @@ public final class Http2Connection implements ConnectionProtocol { if (FrameFlags.isEndStream(frame.flags())) { stream.finishRequestBody(); if (streamDispatcher != null && !stream.dispatched()) enqueueDispatch(stream); + else if (stream.responseStarted() && stream.responseWriter().finished() + && !stream.responseInFlight() + && stream.state() == Http2StreamState.CLOSED) { + streams.remove(stream.id()); + streams.release(stream); + } } } catch (RuntimeException failure) { if (!bodyAccepted) discardConnectionBytes(frame.length()); diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java b/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java index f54babd..473e7c4 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java @@ -115,7 +115,8 @@ final class Http2StreamDispatcher implements Http2Stream.ResponseSink { else if (result != null) response.setBody(result); } - request.drain(); + boolean pushStreaming = response.isPushStreaming(); + if (!pushStreaming) request.drain(); Http2ResponseWriter responseWriter = stream.responseWriter(); if (stream.cancelled()) { request.recycle(); @@ -148,8 +149,7 @@ final class Http2StreamDispatcher implements Http2Stream.ResponseSink { } firstResponse = false; } - request.recycle(); - if (response == pooled) pooled.recycle(); + if (!pushStreaming) request.recycle(); stream.markResponseStarted(); applyBatchTransition(stream, responseWriter); if (!stream.beginResponseBatch()) { @@ -170,8 +170,10 @@ final class Http2StreamDispatcher implements Http2Stream.ResponseSink { Http2ResponseWriter responseWriter = stream.responseWriter(); if (responseWriter.finished()) { stream.endResponseBatch(); - streams.remove(stream.id()); - streams.release(stream); + if (stream.state() == Http2StreamState.CLOSED) { + streams.remove(stream.id()); + streams.release(stream); + } return; } int reserved = flowController.reserveSend(stream, peerSettings.maxFrameSize()); @@ -202,17 +204,25 @@ final class Http2StreamDispatcher implements Http2Stream.ResponseSink { private static void applyBatchTransition( Http2Stream stream, Http2ResponseWriter responseWriter) { if (responseWriter.headersInBatch()) { - if (responseWriter.finished() && responseWriter.dataBytesInBatch() == 0) { + if (responseWriter.finished() && responseWriter.dataBytesInBatch() == 0 + && !responseWriter.trailerHeadersInBatch()) { stream.transition(Http2StreamState.Event.SEND_HEADERS_ES); return; } stream.transition(Http2StreamState.Event.SEND_HEADERS); } if (responseWriter.dataBytesInBatch() != 0 || responseWriter.endStreamInBatch()) { - stream.transition( - responseWriter.endStreamInBatch() - ? Http2StreamState.Event.SEND_DATA_ES - : Http2StreamState.Event.SEND_DATA); + if (responseWriter.dataBytesInBatch() != 0) { + stream.transition( + responseWriter.endStreamInBatch() && !responseWriter.trailerHeadersInBatch() + ? Http2StreamState.Event.SEND_DATA_ES + : Http2StreamState.Event.SEND_DATA); + } + if (responseWriter.trailerHeadersInBatch()) { + stream.transition(Http2StreamState.Event.SEND_HEADERS_ES); + } else if (responseWriter.dataBytesInBatch() == 0 && responseWriter.endStreamInBatch()) { + stream.transition(Http2StreamState.Event.SEND_DATA_ES); + } } } @@ -220,12 +230,19 @@ final class Http2StreamDispatcher implements Http2Stream.ResponseSink { public void responseBatchCompleted(Http2Stream stream) { stream.endResponseBatch(); if (stream.id() == 0) return; - if (stream.cancelled() || stream.responseWriter().finished()) { + if (stream.cancelled()) { streams.remove(stream.id()); streams.release(stream); - } else { - scheduleResume(stream); + return; } + if (stream.responseWriter().finished()) { + if (stream.state() == Http2StreamState.CLOSED) { + streams.remove(stream.id()); + streams.release(stream); + } + return; + } + scheduleResume(stream); } private void failAndRelease(Http2Stream stream, Http2ErrorCode error, Exception cause) { diff --git a/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java b/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java index 1fbad74..645616a 100644 --- a/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java +++ b/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java @@ -35,6 +35,10 @@ public final class Http2HeaderMap implements HeaderView { } } + public void reset(HpackHeaderBlock block) { + reset(block, null); + } + @Override public String first(String name) { ByteView value = find(name, scanValue); @@ -54,7 +58,8 @@ public final class Http2HeaderMap implements HeaderView { new String( scanValue.array(), scanValue.offset(), scanValue.length(), StandardCharsets.UTF_8)); } - if (result == null && isAuthorityAlias(name) && pseudoHeaders.authority().array() != null) { + if (result == null && pseudoHeaders != null + && isAuthorityAlias(name) && pseudoHeaders.authority().array() != null) { return List.of(string(pseudoHeaders.authority())); } return result == null ? List.of() : result; @@ -108,7 +113,8 @@ public final class Http2HeaderMap implements HeaderView { return target; } } - if (isAuthorityAlias(requested) && pseudoHeaders.authority().array() != null) { + if (pseudoHeaders != null + && isAuthorityAlias(requested) && pseudoHeaders.authority().array() != null) { PooledSlice authority = pseudoHeaders.authority(); target.reset(authority.array(), authority.offset(), authority.length()); return target; diff --git a/flash/src/main/java/dev/relism/flash/http2/message/Http2RequestBody.java b/flash/src/main/java/dev/relism/flash/http2/message/Http2RequestBody.java index 9d92d2a..92b64ae 100644 --- a/flash/src/main/java/dev/relism/flash/http2/message/Http2RequestBody.java +++ b/flash/src/main/java/dev/relism/flash/http2/message/Http2RequestBody.java @@ -6,11 +6,12 @@ import dev.relism.flash.http2.Http2StreamException; import dev.relism.flash.http2.message.DataBufferPool.DataBuffer; import java.io.IOException; import java.io.InputStream; +import dev.relism.flash.models.BodyCompletion; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; /** Reusable request-body source fed by the connection demultiplexer. */ -public final class Http2RequestBody extends InputStream { +public final class Http2RequestBody extends InputStream implements BodyCompletion { @FunctionalInterface public interface ConsumptionListener { void consumed(int flowControlledBytes) throws IOException; @@ -30,6 +31,7 @@ public final class Http2RequestBody extends InputStream { private int inlineFlowControlledBytes; private boolean inlineMode; private boolean finished; + private boolean fullyRead; public Http2RequestBody(DataBufferPool pool) { this.pool = pool; @@ -44,6 +46,7 @@ public final class Http2RequestBody extends InputStream { inlinePosition = 0; inlineFlowControlledBytes = 0; finished = false; + fullyRead = false; if (inlineMode && inline == null) inline = new byte[Http2Limits.INLINE_BODY_THRESHOLD]; } @@ -162,7 +165,10 @@ public final class Http2RequestBody extends InputStream { throw new IOException("interrupted while waiting for request DATA", interrupted); } } - if (head == null) return -1; + if (head == null) { + fullyRead = true; + return -1; + } DataBuffer buffer = head; copied = Math.min(length, buffer.length - buffer.position); System.arraycopy(buffer.bytes, buffer.position, target, offset, copied); @@ -187,7 +193,10 @@ public final class Http2RequestBody extends InputStream { if (!finished) { throw new IOException("inline request body is not complete"); } - if (inlinePosition == received) return -1; + if (inlinePosition == received) { + fullyRead = true; + return -1; + } int copied = (int) Math.min(length, received - inlinePosition); System.arraycopy(inline, inlinePosition, target, offset, copied); inlinePosition += copied; @@ -199,6 +208,11 @@ public final class Http2RequestBody extends InputStream { return copied; } + @Override + public boolean fullyRead() { + return fullyRead || (finished && received == 0); + } + private void notifyConsumed(int bytes) { if (bytes == 0 || listener == null) return; try { diff --git a/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java b/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java index 4c7b256..3733c45 100644 --- a/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java +++ b/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java @@ -46,10 +46,13 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize private long bodyRemaining; private int fixedPosition; private boolean unknownLength; + private boolean pushBody; private boolean finished; private boolean headersInBatch; private boolean endStreamInBatch; private int dataBytesInBatch; + private boolean trailerHeadersInBatch; + private Response response; public Http2ResponseWriter() { this(1024, 2048); @@ -104,6 +107,7 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize this.streamId = streamId; this.huffmanDynamicValues = huffmanDynamicValues; this.maxHeaderListSize = maxHeaderListSize; + this.response = response; headerListSize = 0; next = null; @@ -121,12 +125,14 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize } ResponseSerializer.forEachCustomField(response, this); - writeHeaderFrames(maxFrameSize, suppressBody || bodyLength == 0); + boolean hasTrailers = response.hasTrailers() && !suppressBody; + writeHeaderFrames(maxFrameSize, suppressBody || (bodyLength == 0 && !hasTrailers)); if (!suppressBody && bodyLength > 0) { - frames.beginFrame(FrameType.DATA, FrameFlags.END_STREAM, streamId); + frames.beginFrame(FrameType.DATA, hasTrailers ? 0 : FrameFlags.END_STREAM, streamId); output.writeBytes(body); frames.endFrame(); } + if (hasTrailers) appendTrailers(maxFrameSize); return true; } @@ -157,10 +163,13 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize headersInBatch = true; endStreamInBatch = false; dataBytesInBatch = 0; + trailerHeadersInBatch = false; + this.response = response; fixedPosition = 0; fixedBody = response.isStreaming() ? null : response.getBody(); streamBody = response.isStreaming() ? response.getStream() : null; unknownLength = response.isStreaming() && response.isChunked(); + pushBody = response.isPushStreaming(); if (response.isStreaming() && !unknownLength && response.getStreamLength() < 0) { throw new IllegalArgumentException("known response stream length must not be negative"); } @@ -195,8 +204,14 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize ResponseSerializer.forEachCustomField(response, this); boolean hasBody = unknownLength || bodyRemaining > 0; - writeHeaderFrames(maxFrameSize, !hasBody); - finished = !hasBody; + boolean hasTrailers = !headRequest && !bodyForbidden && response.hasTrailers(); + writeHeaderFrames(maxFrameSize, !hasBody && !hasTrailers); + finished = !hasBody && !hasTrailers; + if (!hasBody && hasTrailers) { + appendTrailers(maxFrameSize); + finished = true; + endStreamInBatch = true; + } if (hasBody && availableFlowWindow > 0) { appendData(maxFrameSize, availableFlowWindow); } @@ -211,6 +226,7 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize headersInBatch = false; endStreamInBatch = false; dataBytesInBatch = 0; + trailerHeadersInBatch = false; appendData(maxFrameSize, availableFlowWindow); return dataBytesInBatch; } @@ -221,8 +237,10 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize boolean end; if (fixedBody != null) { count = (int) Math.min(target, bodyRemaining); + boolean finalData = count == bodyRemaining; + boolean trailersFollow = finalData && response.hasTrailers(); frames.beginFrame( - FrameType.DATA, count == bodyRemaining ? FrameFlags.END_STREAM : 0, streamId); + FrameType.DATA, finalData && !trailersFollow ? FrameFlags.END_STREAM : 0, streamId); output.writeBytes(fixedBody, fixedPosition, count); frames.endFrame(); fixedPosition += count; @@ -232,21 +250,27 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize int limit = unknownLength ? target : (int) Math.min(target, bodyRemaining); count = 0; boolean eof = false; - while (count < limit) { - int read = streamBody.read(relay, count, limit - count); - if (read < 0) { - eof = true; - break; - } - if (read == 0) { - int one = streamBody.read(); - if (one < 0) { + if (pushBody) { + int read = streamBody.read(relay, 0, limit); + if (read < 0) eof = true; + else count = read; + } else { + while (count < limit) { + int read = streamBody.read(relay, count, limit - count); + if (read < 0) { eof = true; break; } - relay[count++] = (byte) one; - } else { - count += read; + if (read == 0) { + int one = streamBody.read(); + if (one < 0) { + eof = true; + break; + } + relay[count++] = (byte) one; + } else { + count += read; + } } } if (!unknownLength) { @@ -256,15 +280,31 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize } } end = unknownLength ? eof : bodyRemaining == 0; - frames.beginFrame(FrameType.DATA, end ? FrameFlags.END_STREAM : 0, streamId); - output.writeBytes(relay, 0, count); - frames.endFrame(); + boolean trailersFollow = end && response.hasTrailers(); + if (count != 0 || !trailersFollow) { + frames.beginFrame(FrameType.DATA, end && !trailersFollow ? FrameFlags.END_STREAM : 0, streamId); + output.writeBytes(relay, 0, count); + frames.endFrame(); + } } dataBytesInBatch = count; - endStreamInBatch = end; + if (end && response.hasTrailers()) { + appendTrailers(maxFrameSize); + endStreamInBatch = true; + } else { + endStreamInBatch = end; + } finished = end; } + private void appendTrailers(int maxFrameSize) { + headerBlock.reset(); + headerListSize = 0; + ResponseSerializer.forEachTrailerField(response, this); + writeHeaderFrames(maxFrameSize, true); + trailerHeadersInBatch = true; + } + public boolean finished() { return finished; } @@ -281,6 +321,10 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize return dataBytesInBatch; } + public boolean trailerHeadersInBatch() { + return trailerHeadersInBatch; + } + @Override public void accept( byte[] name, int nameOff, int nameLen, byte[] value, int valueOff, int valueLen) { diff --git a/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java b/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java index 59428b7..bb46284 100644 --- a/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java +++ b/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java @@ -64,6 +64,20 @@ public final class PseudoHeaders { } } + /** Validates a trailing field section, where pseudo-fields are never permitted. */ + public static void validateTrailers(HpackHeaderBlock block, int streamId) { + PooledSlice name = new PooledSlice(); + PooledSlice value = new PooledSlice(); + for (int i = 0; i < block.count(); i++) { + block.get(i, name, value); + if (name.length() == 0 || name.byteAt(0) == ':') fail(streamId, "pseudo-header in trailers"); + validateRegular(name, value, streamId); + if (equals(name, "content-length") || equals(name, "host") || equals(name, "te")) { + fail(streamId, "field is not permitted in trailers"); + } + } + } + public PooledSlice method() { return method; } diff --git a/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java b/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java index c73b9e7..627d5b1 100644 --- a/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java +++ b/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java @@ -36,8 +36,10 @@ public final class Http2Stream private static final byte[] HTTP_2 = {'H', 'T', 'T', 'P', '/', '2'}; private final HpackHeaderBlock headerBlock = new HpackHeaderBlock(); + private final HpackHeaderBlock trailerBlock = new HpackHeaderBlock(); private final PseudoHeaders pseudoHeaders = new PseudoHeaders(); private final Http2HeaderMap headers = new Http2HeaderMap(); + private final Http2HeaderMap trailers = new Http2HeaderMap(); private final RequestLine requestLine = new RequestLine(); private final RequestBody requestBody = new RequestBody(); private final Http2RequestBody http2Body; @@ -92,6 +94,8 @@ public final class Http2Stream resumeTask = false; responseSink = null; headerBlock.reset(); + trailerBlock.reset(); + trailers.reset(trailerBlock); } void clear() { @@ -131,7 +135,9 @@ public final class Http2Stream } requestLine.reset(method, path, question < 0 ? null : query, protocol, headers); requestBody.reset(http2Body, http2Body.declaredLength(), null, 0, 0); - return Request.forParsed(request, requestLine, requestBody, remoteAddress, sslSocket); + Request assembled = Request.forParsed(request, requestLine, requestBody, remoteAddress, sslSocket); + assembled.setTrailers(trailers); + return assembled; } public void validateHeaders() { @@ -158,6 +164,11 @@ public final class Http2Stream http2Body.finish(id); } + public void validateTrailers() { + PseudoHeaders.validateTrailers(trailerBlock, id); + trailers.reset(trailerBlock); + } + private long parseContentLength() { long parsed = -1; for (int i = 0; i < headerBlock.count(); i++) { @@ -217,6 +228,10 @@ public final class Http2Stream return headerBlock; } + public HpackHeaderBlock trailerBlock() { + return trailerBlock; + } + public Http2ResponseWriter responseWriter() { return responseWriter; } diff --git a/flash/src/main/java/dev/relism/flash/models/BodyCompletion.java b/flash/src/main/java/dev/relism/flash/models/BodyCompletion.java new file mode 100644 index 0000000..106cb69 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/models/BodyCompletion.java @@ -0,0 +1,6 @@ +package dev.relism.flash.models; + +/** Internal completion signal used to enforce request-trailer ordering. */ +public interface BodyCompletion { + boolean fullyRead(); +} diff --git a/flash/src/main/java/dev/relism/flash/models/EmptyHeaderView.java b/flash/src/main/java/dev/relism/flash/models/EmptyHeaderView.java new file mode 100644 index 0000000..92afc66 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/models/EmptyHeaderView.java @@ -0,0 +1,18 @@ +package dev.relism.flash.models; + +import dev.relism.fpr.core.ByteView; +import java.util.List; + +/** Immutable empty header collection shared by requests without trailers. */ +public enum EmptyHeaderView implements HeaderView { + INSTANCE; + + @Override public String first(String name) { return null; } + @Override public List all(String name) { return List.of(); } + @Override public List all() { return List.of(); } + @Override public ByteView view(String name) { return null; } + @Override public boolean valueEqualsIgnoreCase(String name, String value) { return false; } + @Override public boolean contains(String name) { return false; } + @Override public int count() { return 0; } + @Override public void forEach(HeaderConsumer consumer) {} +} diff --git a/flash/src/main/java/dev/relism/flash/models/MutableHeaderMap.java b/flash/src/main/java/dev/relism/flash/models/MutableHeaderMap.java new file mode 100644 index 0000000..b008a26 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/models/MutableHeaderMap.java @@ -0,0 +1,138 @@ +package dev.relism.flash.models; + +import dev.relism.flash.bytes.ByteScan; +import dev.relism.flash.bytes.ByteWriter; +import dev.relism.flash.bytes.PooledSlice; +import dev.relism.fpr.core.ByteView; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.io.IOException; +import java.io.OutputStream; + +/** Reusable owned-byte header collection for sections parsed outside the request head buffer. */ +public final class MutableHeaderMap implements HeaderView { + private final ByteWriter bytes = new ByteWriter(128); + private final PooledSlice view = new PooledSlice(); + private final PooledSlice scanName = new PooledSlice(); + private final PooledSlice scanValue = new PooledSlice(); + private int[] fields = new int[16]; + private int count; + + public void reset() { + bytes.reset(); + count = 0; + } + + public void add(byte[] name, int nameOffset, int nameLength, + byte[] value, int valueOffset, int valueLength) { + ensure(count + 1); + int base = count * 4; + fields[base] = bytes.length(); + fields[base + 1] = nameLength; + bytes.writeBytes(name, nameOffset, nameLength); + fields[base + 2] = bytes.length(); + fields[base + 3] = valueLength; + bytes.writeBytes(value, valueOffset, valueLength); + count++; + } + + public void writeLines(OutputStream output) throws IOException { + for (int i = 0; i < count; i++) { + int base = i * 4; + output.write(bytes.array(), fields[base], fields[base + 1]); + output.write(':'); + output.write(' '); + output.write(bytes.array(), fields[base + 2], fields[base + 3]); + output.write('\r'); + output.write('\n'); + } + } + + void forEachStructured(ResponseSerializer.FieldConsumer consumer) { + for (int i = 0; i < count; i++) { + int base = i * 4; + consumer.accept(bytes.array(), fields[base], fields[base + 1], + bytes.array(), fields[base + 2], fields[base + 3]); + } + } + + @Override + public String first(String name) { + int index = indexOf(name, 0); + return index < 0 ? null : value(index); + } + + @Override + public List all(String name) { + List result = null; + int from = 0; + int index; + while ((index = indexOf(name, from)) >= 0) { + if (result == null) result = new ArrayList<>(); + result.add(value(index)); + from = index + 1; + } + return result == null ? List.of() : result; + } + + @Override + public List all() { + if (count == 0) return List.of(); + List result = new ArrayList<>(count); + for (int i = 0; i < count; i++) result.add(value(i)); + return result; + } + + @Override + public ByteView view(String name) { + int index = indexOf(name, 0); + if (index < 0) return null; + int base = index * 4; + view.reset(bytes.array(), fields[base + 2], fields[base + 3]); + return view; + } + + @Override + public boolean valueEqualsIgnoreCase(String name, String value) { + int index = indexOf(name, 0); + if (index < 0) return false; + int base = index * 4; + return ByteScan.equalsIgnoreCaseAscii( + bytes.array(), fields[base + 2], fields[base + 2] + fields[base + 3], value); + } + + @Override public boolean contains(String name) { return indexOf(name, 0) >= 0; } + @Override public int count() { return count; } + + @Override + public void forEach(HeaderConsumer consumer) { + for (int i = 0; i < count; i++) { + int base = i * 4; + scanName.reset(bytes.array(), fields[base], fields[base + 1]); + scanValue.reset(bytes.array(), fields[base + 2], fields[base + 3]); + consumer.accept(scanName, scanValue); + } + } + + private int indexOf(String name, int from) { + for (int i = from; i < count; i++) { + int base = i * 4; + if (ByteScan.equalsIgnoreCaseAscii( + bytes.array(), fields[base], fields[base] + fields[base + 1], name)) return i; + } + return -1; + } + + private String value(int index) { + int base = index * 4; + return new String(bytes.array(), fields[base + 2], fields[base + 3], StandardCharsets.UTF_8); + } + + private void ensure(int needed) { + int ints = needed * 4; + if (ints <= fields.length) return; + fields = Arrays.copyOf(fields, Math.max(ints, fields.length * 2)); + } +} diff --git a/flash/src/main/java/dev/relism/flash/models/ProducerInputStream.java b/flash/src/main/java/dev/relism/flash/models/ProducerInputStream.java new file mode 100644 index 0000000..7f0e89d --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/models/ProducerInputStream.java @@ -0,0 +1,103 @@ +package dev.relism.flash.models; + +import java.io.IOException; +import java.io.InputStream; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; +import java.util.function.Consumer; + +/** Bounded bridge from a push producer to the protocol writers' common pull path. */ +final class ProducerInputStream extends InputStream { + private final PipedInputStream input; + private final ProducerOutput output; + private final Consumer producer; + private volatile Throwable failure; + private boolean started; + + ProducerInputStream(Consumer producer, Response response) { + try { + input = new PipedInputStream(16 * 1024); + output = new ProducerOutput(new PipedOutputStream(input), response); + } catch (IOException impossible) { + throw new IllegalStateException(impossible); + } + this.producer = producer; + } + + @Override + public int read() throws IOException { + start(); + int value = input.read(); + checkFailure(value < 0); + return value; + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + start(); + int count = input.read(bytes, offset, length); + checkFailure(count < 0); + return count; + } + + @Override + public void close() throws IOException { + input.close(); + } + + private synchronized void start() { + if (started) return; + started = true; + Thread.startVirtualThread(() -> { + try (output) { + producer.accept(output); + } catch (Throwable thrown) { + failure = thrown; + try { + output.close(); + } catch (IOException ignored) { + } + } + }); + } + + private void checkFailure(boolean eof) throws IOException { + if (eof && failure != null) throw new IOException("response stream producer failed", failure); + } + + private static final class ProducerOutput implements ResponseStream { + private final PipedOutputStream output; + private final Response response; + private boolean closed; + + ProducerOutput(PipedOutputStream output, Response response) { + this.output = output; + this.response = response; + } + + @Override + public synchronized void write(byte[] data, int offset, int length) throws IOException { + if (closed) throw new IOException("response stream is closed"); + output.write(data, offset, length); + } + + @Override + public synchronized void flush() throws IOException { + if (closed) throw new IOException("response stream is closed"); + output.flush(); + } + + @Override + public synchronized void trailer(String name, String value) { + if (closed) throw new IllegalStateException("response stream is closed"); + response.trailer(name, value); + } + + @Override + public synchronized void close() throws IOException { + if (closed) return; + closed = true; + output.close(); + } + } +} diff --git a/flash/src/main/java/dev/relism/flash/models/Request.java b/flash/src/main/java/dev/relism/flash/models/Request.java index 82ca3f0..b7b0f03 100644 --- a/flash/src/main/java/dev/relism/flash/models/Request.java +++ b/flash/src/main/java/dev/relism/flash/models/Request.java @@ -52,6 +52,7 @@ import java.util.List; public class Request { private RequestBody body; + private HeaderView trailers = EmptyHeaderView.INSTANCE; /** Internal: the parsed request line (method, path, query, protocol, headers). */ private RequestLine requestLine; @@ -95,6 +96,7 @@ public class Request { void reset(RequestLine requestLine, RequestBody body, InetSocketAddress remoteAddress, SSLSocket sslSocket) { this.requestLine = requestLine; this.body = body; + this.trailers = EmptyHeaderView.INSTANCE; this.pathParams = null; this.queryParams = null; this.cachedPath = null; @@ -143,6 +145,11 @@ public class Request { return pooled; } + /** Internal protocol hook that supplies the request's trailer collection. */ + public void setTrailers(HeaderView trailers) { + this.trailers = trailers == null ? EmptyHeaderView.INSTANCE : trailers; + } + // ── Request line ────────────────────────────────────────────────────────── /** HTTP method ({@code GET}, {@code POST}, …). */ @@ -253,6 +260,19 @@ public class Request { */ public RequestBody body() { checkActive(); return body; } + /** + * Returns request trailers after the body has been consumed completely. + * + * @throws IllegalStateException when called before the body reaches EOF + */ + public HeaderView trailers() { + checkActive(); + if (!body.fullyRead()) { + throw new IllegalStateException("request trailers are available only after the body is fully read"); + } + return trailers; + } + /** Discards unread body bytes; called by the server after each request on keep-alive connections. */ public void drain() { body.drain(); } diff --git a/flash/src/main/java/dev/relism/flash/models/RequestBody.java b/flash/src/main/java/dev/relism/flash/models/RequestBody.java index 44895a6..f60a225 100644 --- a/flash/src/main/java/dev/relism/flash/models/RequestBody.java +++ b/flash/src/main/java/dev/relism/flash/models/RequestBody.java @@ -87,6 +87,15 @@ public class RequestBody { */ public long contentLength() { return contentLength; } + /** Whether the complete body has been consumed by the application. */ + public boolean fullyRead() { + if (resolved != null || contentLength == 0) return true; + if (socket instanceof BodyCompletion completion) return completion.fullyRead(); + if (contentLength < 0) return false; + if (boundedStream != null) return boundedStream.complete(); + return preBufLen >= contentLength; + } + /** * Materialises and caches the full body. Suitable for JSON, small form data, and any payload * that must be inspected in full. The result is cached — repeated calls return the same array. @@ -178,6 +187,10 @@ public class RequestBody { this.socketRemaining = socketRemaining; } + boolean complete() { + return preBufRemaining == 0 && socketRemaining == 0; + } + @Override public int read() throws IOException { if (preBufRemaining > 0) { diff --git a/flash/src/main/java/dev/relism/flash/models/Response.java b/flash/src/main/java/dev/relism/flash/models/Response.java index f694276..246c1bd 100644 --- a/flash/src/main/java/dev/relism/flash/models/Response.java +++ b/flash/src/main/java/dev/relism/flash/models/Response.java @@ -12,6 +12,8 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.Objects; +import java.util.function.Consumer; /** * HTTP response. All mutating methods return {@code this} for fluent chaining. @@ -43,7 +45,9 @@ public class Response { private InputStream stream; private long streamLength; // meaningful only when isStreaming() && !chunked private boolean chunked; + private boolean pushStreaming; private byte[] contentType; + private final MutableHeaderMap trailers = new MutableHeaderMap(); // of a List of fully-rendered "Name: Value\r\n" lines (which cost a StringBuilder + // char[] + String + getBytes() chain per header(String,String) call). Two backing stores, @@ -112,9 +116,11 @@ public class Response { this.stream = null; this.streamLength = 0; this.chunked = false; + this.pushStreaming = false; this.contentType = contentType.getBytes(); this.headerQuadCount = 0; this.headerCount = 0; + this.trailers.reset(); if (rawHeaderLines != null) rawHeaderLines.clear(); this.active = true; return this; @@ -175,6 +181,7 @@ public class Response { checkActive(); this.body = bytes; this.stream = null; + this.pushStreaming = false; return this; } @@ -189,6 +196,7 @@ public class Response { this.streamLength = length; this.chunked = false; this.body = null; + this.pushStreaming = false; return this; } @@ -198,9 +206,62 @@ public class Response { this.stream = is; this.chunked = true; this.body = null; + this.pushStreaming = false; return this; } + /** Push-style streaming response with bounded blocking backpressure. */ + public Response streaming(Consumer producer) { + checkActive(); + this.stream = new ProducerInputStream(Objects.requireNonNull(producer), this); + this.streamLength = -1; + this.chunked = true; + this.pushStreaming = true; + this.body = null; + return this; + } + + /** Adds a trailer rendered after the response body on both HTTP versions. */ + public Response trailer(String name, String value) { + checkActive(); + validateTrailer(name, value); + byte[] nameBytes = name.getBytes(StandardCharsets.US_ASCII); + byte[] valueBytes = value.getBytes(StandardCharsets.UTF_8); + trailers.add(nameBytes, 0, nameBytes.length, valueBytes, 0, valueBytes.length); + return this; + } + + /** Adds a pre-encoded structured trailer. */ + public Response trailer(PreEncodedHeader trailer) { + checkActive(); + byte[] name = trailer.nameBytes(); + byte[] value = trailer.valueBytes(); + validateTrailer( + new String(name, StandardCharsets.US_ASCII), + new String(value, StandardCharsets.US_ASCII)); + trailers.add(name, 0, name.length, value, 0, value.length); + return this; + } + + private static void validateTrailer(String name, String value) { + if (name.isEmpty() || name.charAt(0) == ':' || containsLineBreak(name) + || containsLineBreak(value)) { + throw new IllegalArgumentException("invalid response trailer"); + } + if (name.equalsIgnoreCase("content-length") + || name.equalsIgnoreCase("transfer-encoding") + || name.equalsIgnoreCase("connection") + || name.equalsIgnoreCase("host") + || name.equalsIgnoreCase("te") + || name.equalsIgnoreCase("trailer")) { + throw new IllegalArgumentException("field is not permitted in response trailers: " + name); + } + } + + private static boolean containsLineBreak(String value) { + return value.indexOf('\r') >= 0 || value.indexOf('\n') >= 0; + } + /** * 302 Found redirect. Clears the body, sets status and {@code Location} header. Encoded once at * call time; zero-alloc on the write path. @@ -367,6 +428,12 @@ public class Response { return chunked; } + /** Internal distinction between producer-driven and InputStream-driven response bodies. */ + public boolean isPushStreaming() { + checkActive(); + return pushStreaming; + } + public int getStatusCode() { checkActive(); return statusCode; @@ -397,6 +464,20 @@ public class Response { return streamLength; } + public boolean hasTrailers() { + checkActive(); + return trailers.count() != 0; + } + + public void writeTrailers(OutputStream output) throws IOException { + checkActive(); + trailers.writeLines(output); + } + + void forEachTrailerField(ResponseSerializer.FieldConsumer consumer) { + trailers.forEachStructured(consumer); + } + // ------------------------------------------------------------------------- // Internal setters used by HttpServer for handler return values // ------------------------------------------------------------------------- diff --git a/flash/src/main/java/dev/relism/flash/models/ResponseSerializer.java b/flash/src/main/java/dev/relism/flash/models/ResponseSerializer.java index 70488af..47831fe 100644 --- a/flash/src/main/java/dev/relism/flash/models/ResponseSerializer.java +++ b/flash/src/main/java/dev/relism/flash/models/ResponseSerializer.java @@ -54,4 +54,9 @@ public final class ResponseSerializer { public static void forEachCustomField(Response response, FieldConsumer consumer) { response.forEachStructuredField(consumer); } + + /** Enumerates response trailers in declaration order. */ + public static void forEachTrailerField(Response response, FieldConsumer consumer) { + response.forEachTrailerField(consumer); + } } diff --git a/flash/src/main/java/dev/relism/flash/models/ResponseStream.java b/flash/src/main/java/dev/relism/flash/models/ResponseStream.java new file mode 100644 index 0000000..8c531da --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/models/ResponseStream.java @@ -0,0 +1,11 @@ +package dev.relism.flash.models; + +import java.io.IOException; + +/** Blocking, flow-controlled response body used by push-style streaming producers. */ +public interface ResponseStream extends AutoCloseable { + void write(byte[] data, int offset, int length) throws IOException; + void flush() throws IOException; + void trailer(String name, String value); + @Override void close() throws IOException; +} diff --git a/flash/src/main/java/dev/relism/flash/routing/AbstractRouter.java b/flash/src/main/java/dev/relism/flash/routing/AbstractRouter.java index 94959d0..4ad7dcb 100644 --- a/flash/src/main/java/dev/relism/flash/routing/AbstractRouter.java +++ b/flash/src/main/java/dev/relism/flash/routing/AbstractRouter.java @@ -84,7 +84,10 @@ public abstract class AbstractRouter { */ public AbstractRouter doRegister(HttpMethod method, String path, RequestHandler handler, Middleware[] middlewares) { - return addRoute(method, PathUtils.sanitize(path), compile(handler, middlewares)); + String target = method == HttpMethod.CONNECT + ? PathUtils.sanitizeAuthority(path) + : PathUtils.sanitize(path); + return addRoute(method, target, compile(handler, middlewares)); } /** diff --git a/flash/src/main/java/dev/relism/flash/routing/PathUtils.java b/flash/src/main/java/dev/relism/flash/routing/PathUtils.java index a972b49..c43c88c 100644 --- a/flash/src/main/java/dev/relism/flash/routing/PathUtils.java +++ b/flash/src/main/java/dev/relism/flash/routing/PathUtils.java @@ -23,6 +23,14 @@ public class PathUtils { return sanitized; } + /** Normalizes an authority-form CONNECT target without turning it into an origin-form path. */ + public static String sanitizeAuthority(String authority) { + if (authority == null) return ""; + String sanitized = authority.trim(); + while (sanitized.startsWith("/")) sanitized = sanitized.substring(1); + return sanitized; + } + /** * Joins two path segments and ensures the result is sanitized. * Prevents "double namespace" if the path already starts with the base. diff --git a/flash/src/test/java/dev/relism/flash/ChunkedInputStreamTest.java b/flash/src/test/java/dev/relism/flash/ChunkedInputStreamTest.java index c2aa4f0..206312a 100644 --- a/flash/src/test/java/dev/relism/flash/ChunkedInputStreamTest.java +++ b/flash/src/test/java/dev/relism/flash/ChunkedInputStreamTest.java @@ -68,7 +68,7 @@ class ChunkedInputStreamTest { @Test void trailers_consumed() throws IOException { // trailing headers after 0-chunk must be consumed - assertEquals("hi", readAll(wrap("2\r\nhi\r\n0\r\nTrailer: value\r\n\r\n"))); + assertEquals("hi", readAll(wrap("2\r\nhi\r\n0\r\nX-Trailer: value\r\n\r\n"))); } // --- byte-by-byte read --- diff --git a/flash/src/test/java/dev/relism/flash/Http1TrailersTest.java b/flash/src/test/java/dev/relism/flash/Http1TrailersTest.java new file mode 100644 index 0000000..b06e493 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/Http1TrailersTest.java @@ -0,0 +1,68 @@ +package dev.relism.flash; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import dev.relism.flash.http.ContentType; +import dev.relism.flash.http.HttpMethod; +import dev.relism.flash.http1.Http1ResponseWriter; +import dev.relism.flash.models.Request; +import dev.relism.flash.models.Response; +import dev.relism.flash.transport.BufferedByteSource; +import dev.relism.flash.transport.ConnectionScratch; +import dev.relism.flash.transport.ScratchPool; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; +import org.junit.jupiter.api.Test; + +class Http1TrailersTest { + @Test + void requestTrailersBecomeVisibleOnlyAfterBodyEof() throws Exception { + byte[] wire = ("POST / HTTP/1.1\r\nHost: localhost\r\nTransfer-Encoding: chunked\r\n\r\n" + + "3\r\nabc\r\n0\r\nGrpc-Status: 0\r\nX-Trace: done\r\n\r\n") + .getBytes(StandardCharsets.US_ASCII); + Request request = new RequestParser().parse( + new BufferedByteSource(new ByteArrayInputStream(wire), null)); + + assertThrows(IllegalStateException.class, request::trailers); + assertEquals("abc", new String(request.body().bytes(), StandardCharsets.US_ASCII)); + assertEquals("0", request.trailers().first("grpc-status")); + assertEquals("done", request.trailers().first("x-trace")); + } + + @Test + void responseTrailersUseChunkedRendering() throws Exception { + Response response = new Response(200, "hello", ContentType.TEXT_PLAIN) + .trailer("grpc-status", "0"); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + + Http1ResponseWriter.writeResponse( + output, response, HttpMethod.GET, true, false, new ScratchPool().acquire()); + + String wire = output.toString(StandardCharsets.US_ASCII); + assertEquals(true, wire.contains("Transfer-Encoding: chunked\r\n")); + assertEquals(true, wire.endsWith("5\r\nhello\r\n0\r\ngrpc-status: 0\r\n\r\n")); + } + + @Test + void pushStreamingAndTrailersShareTheSameHttp1Writer() throws Exception { + Response response = new Response(200, ContentType.BINARY).streaming(stream -> { + try { + stream.write("one".getBytes(StandardCharsets.US_ASCII), 0, 3); + stream.write("two".getBytes(StandardCharsets.US_ASCII), 0, 3); + stream.trailer("grpc-status", "0"); + } catch (Exception failure) { + throw new RuntimeException(failure); + } + }); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + Http1ResponseWriter.writeResponse( + output, response, HttpMethod.GET, true, false, new ScratchPool().acquire()); + + String wire = output.toString(StandardCharsets.US_ASCII); + assertEquals(true, wire.contains("one")); + assertEquals(true, wire.contains("two")); + assertEquals(true, wire.endsWith("0\r\ngrpc-status: 0\r\n\r\n")); + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/GrpcInteropTest.java b/flash/src/test/java/dev/relism/flash/http2/GrpcInteropTest.java new file mode 100644 index 0000000..2f81287 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/GrpcInteropTest.java @@ -0,0 +1,114 @@ +package dev.relism.flash.http2; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.relism.flash.extension.FlashApp; +import dev.relism.flash.extension.FlashConfiguration; +import java.net.ServerSocket; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; +import org.junit.jupiter.api.io.TempDir; + +@Tag("interop") +@EnabledIfSystemProperty(named = "grpcurl.executable", matches = ".+") +class GrpcInteropTest { + private FlashApp app; + + @AfterEach + void stop() { + if (app != null) app.stop().join(); + } + + @Test + void grpcurlCompletesUnaryStreamingAndErrorCalls(@TempDir Path directory) throws Exception { + int port = freePort(); + app = FlashApp.create(FlashConfiguration.builder() + .host("127.0.0.1").port(port).http2Enabled(true).build()); + app.post("/flash.test.Echo/Unary", (request, response) -> + response.type("application/grpc") + .body(request.body().bytes()) + .trailer("grpc-status", "0")); + app.post("/flash.test.Echo/Stream", (request, response) -> { + byte[] message = request.body().bytes(); + return response.type("application/grpc").streaming(stream -> { + try { + for (int i = 0; i < 3; i++) stream.write(message, 0, message.length); + stream.trailer("grpc-status", "0"); + } catch (Exception failure) { + throw new RuntimeException(failure); + } + }); + }); + app.post("/flash.test.Echo/Fail", (request, response) -> + response.type("application/grpc") + .trailer("grpc-status", "3") + .trailer("grpc-message", "invalid request")); + app.start(); + + Path proto = directory.resolve("echo.proto"); + Files.writeString(proto, """ + syntax = "proto3"; + package flash.test; + service Echo { + rpc Unary (Message) returns (Message); + rpc Stream (Message) returns (stream Message); + rpc Fail (Message) returns (Message); + } + message Message { string value = 1; } + """); + + Result unary = call(directory, port, "Unary"); + assertEquals(0, unary.exitCode); + assertTrue(unary.output.contains("hello"), unary.output); + + Result streaming = call(directory, port, "Stream"); + assertEquals(0, streaming.exitCode); + assertEquals(3, occurrences(streaming.output, "hello"), streaming.output); + + Result error = call(directory, port, "Fail"); + assertTrue(error.exitCode != 0); + assertTrue(error.output.contains("InvalidArgument"), error.output); + assertTrue(error.output.contains("invalid request"), error.output); + } + + private static Result call(Path directory, int port, String method) throws Exception { + Process process = new ProcessBuilder( + System.getProperty("grpcurl.executable"), + "-plaintext", + "-import-path", directory.toString(), + "-proto", "echo.proto", + "-d", "{\"value\":\"hello\"}", + "127.0.0.1:" + port, + "flash.test.Echo/" + method) + .redirectErrorStream(true) + .start(); + assertTrue(process.waitFor(10, TimeUnit.SECONDS), "grpcurl timed out"); + return new Result(process.exitValue(), + new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8)); + } + + private static int occurrences(String text, String needle) { + int count = 0; + int position = 0; + while ((position = text.indexOf(needle, position)) >= 0) { + count++; + position += needle.length(); + } + return count; + } + + private static int freePort() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } + + private record Result(int exitCode, String output) {} +} diff --git a/flash/src/test/java/dev/relism/flash/http2/Http2ConnectTest.java b/flash/src/test/java/dev/relism/flash/http2/Http2ConnectTest.java new file mode 100644 index 0000000..c02c9ab --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/Http2ConnectTest.java @@ -0,0 +1,103 @@ +package dev.relism.flash.http2; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import dev.relism.flash.bytes.ByteWriter; +import dev.relism.flash.extension.FlashApp; +import dev.relism.flash.extension.FlashConfiguration; +import dev.relism.flash.http.ContentType; +import dev.relism.flash.http2.frame.FrameFlags; +import dev.relism.flash.http2.frame.FrameType; +import dev.relism.flash.http2.hpack.HpackEncoder; +import java.io.EOFException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +class Http2ConnectTest { + private FlashApp app; + + @AfterEach + void stop() { + if (app != null) app.stop().join(); + } + + @Test + void connectTunnelCanExchangeDataBeforeEitherSideCloses() throws Exception { + int port = freePort(); + app = FlashApp.create(FlashConfiguration.builder() + .host("127.0.0.1").port(port).http2Enabled(true).build()); + app.connect("tunnel", (request, response) -> + response.type(ContentType.NONE).streaming(output -> { + byte[] bytes = new byte[16]; + try { + int count; + InputStream input = request.body().stream(); + while ((count = input.read(bytes)) >= 0) { + output.write(bytes, 0, count); + output.flush(); + } + } catch (Exception failure) { + throw new RuntimeException(failure); + } + })); + app.start(); + + ByteWriter block = new ByteWriter(32); + HpackEncoder.writeLiteralWithNameIndex(block, 2, ascii("CONNECT"), false); + HpackEncoder.writeLiteralWithNameIndex(block, 1, ascii("tunnel"), false); + + try (Socket socket = new Socket("127.0.0.1", port)) { + socket.setSoTimeout(5_000); + socket.getOutputStream().write(Http2TestFrames.concat( + Http2TestFrames.PREFACE, Http2TestFrames.settings(), + Http2TestFrames.frame(FrameType.SETTINGS, FrameFlags.ACK, 0, new byte[0]), + Http2TestFrames.frame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1, + Arrays.copyOf(block.array(), block.length())), + Http2TestFrames.frame(FrameType.DATA, 0, 1, ascii("one")))); + socket.getOutputStream().flush(); + + assertEquals("one", new String(readData(socket.getInputStream()).payload(), + StandardCharsets.US_ASCII)); + + socket.getOutputStream().write(Http2TestFrames.frame( + FrameType.DATA, FrameFlags.END_STREAM, 1, ascii("two"))); + socket.getOutputStream().flush(); + assertEquals("two", new String(readData(socket.getInputStream()).payload(), + StandardCharsets.US_ASCII)); + } + } + + private static Http2TestFrames.WireFrame readData(InputStream input) throws Exception { + for (int i = 0; i < 12; i++) { + Http2TestFrames.WireFrame frame = readFrame(input); + if (frame.streamId() == 1 && frame.type() == FrameType.DATA.code() + && frame.payload().length != 0) return frame; + } + throw new AssertionError("missing tunnel DATA"); + } + + private static Http2TestFrames.WireFrame readFrame(InputStream input) throws Exception { + byte[] header = input.readNBytes(9); + if (header.length != 9) throw new EOFException(); + int length = ((header[0] & 0xff) << 16) | ((header[1] & 0xff) << 8) | (header[2] & 0xff); + byte[] payload = input.readNBytes(length); + return new Http2TestFrames.WireFrame( + header[3] & 0xff, header[4] & 0xff, Http2TestFrames.readInt(header, 5) & 0x7fff_ffff, + payload); + } + + private static byte[] ascii(String text) { + return text.getBytes(StandardCharsets.US_ASCII); + } + + private static int freePort() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java b/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java index 32a006c..7d51668 100644 --- a/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java +++ b/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java @@ -5,6 +5,7 @@ import static org.junit.jupiter.api.Assertions.*; import dev.relism.flash.bytes.ByteWriter; import dev.relism.flash.extension.FlashApp; import dev.relism.flash.extension.FlashConfiguration; +import dev.relism.flash.http.ContentType; import dev.relism.flash.http2.frame.FrameFlags; import dev.relism.flash.http2.frame.FrameType; import dev.relism.flash.http2.hpack.HpackDecoder; @@ -14,6 +15,7 @@ import dev.relism.flash.tls.TlsConfig; import java.io.ByteArrayInputStream; import java.io.EOFException; import java.io.InputStream; +import java.io.IOException; import java.net.ServerSocket; import java.net.Socket; import java.net.URI; @@ -138,6 +140,61 @@ class Http2ConnectionIntegrationTest { assertTrue(streamed.headers().firstValue("transfer-encoding").isEmpty()); } + @Test + void pushStreamingAppliesBackpressureAcrossMultipleWindows(@TempDir Path directory) + throws Exception { + int port = freePort(); + int length = 2 * 1024 * 1024 + 31; + Path keystore = + TestKeystores.build( + directory, + "http2-push-stream.p12", + "changeit", + TestKeystores.Entry.of("server", "localhost", "localhost")); + app = + FlashApp.create( + FlashConfiguration.builder() + .port(port) + .host("127.0.0.1") + .tls(TlsConfig.keystore(keystore, "changeit")) + .http2Enabled(true) + .build()); + app.get( + "/push", + (request, response) -> + response.type(ContentType.BINARY).streaming(stream -> { + byte[] block = new byte[8192]; + int written = 0; + try { + while (written < length) { + int count = Math.min(block.length, length - written); + for (int i = 0; i < count; i++) block[i] = (byte) ((written + i) * 31); + stream.write(block, 0, count); + written += count; + } + stream.trailer("grpc-status", "0"); + } catch (IOException failure) { + throw new RuntimeException(failure); + } + })); + app.start(); + + HttpClient client = + HttpClient.newBuilder() + .sslContext(TestKeystores.trustAllClientContext()) + .version(HttpClient.Version.HTTP_2) + .build(); + HttpResponse response = + client.send( + HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/push")).GET().build(), + HttpResponse.BodyHandlers.ofInputStream()); + + assertEquals(HttpClient.Version.HTTP_2, response.version()); + try (InputStream body = response.body()) { + assertEquals(length, verifyPattern(body)); + } + } + @Test void hundredMegabyteUploadAndDownloadRemainStreaming(@TempDir Path directory) throws Exception { int port = freePort(); diff --git a/flash/src/test/java/dev/relism/flash/http2/Http2HalfCloseTest.java b/flash/src/test/java/dev/relism/flash/http2/Http2HalfCloseTest.java new file mode 100644 index 0000000..a0be7f6 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/Http2HalfCloseTest.java @@ -0,0 +1,50 @@ +package dev.relism.flash.http2; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import dev.relism.flash.http2.stream.Http2StreamState; +import org.junit.jupiter.api.Test; + +class Http2HalfCloseTest { + @Test + void remoteMayCloseBeforeLocalResponseCompletes() { + Http2StreamState state = Http2StreamState.IDLE; + state = state.transition(1, Http2StreamState.Event.RECV_HEADERS_ES); + assertEquals(Http2StreamState.HALF_CLOSED_REMOTE, state); + state = state.transition(1, Http2StreamState.Event.SEND_HEADERS); + state = state.transition(1, Http2StreamState.Event.SEND_DATA); + state = state.transition(1, Http2StreamState.Event.SEND_DATA_ES); + assertEquals(Http2StreamState.CLOSED, state); + } + + @Test + void localMayCloseWhileRemoteBodyContinues() { + Http2StreamState state = Http2StreamState.IDLE; + state = state.transition(1, Http2StreamState.Event.RECV_HEADERS); + state = state.transition(1, Http2StreamState.Event.SEND_HEADERS_ES); + assertEquals(Http2StreamState.HALF_CLOSED_LOCAL, state); + state = state.transition(1, Http2StreamState.Event.RECV_DATA); + state = state.transition(1, Http2StreamState.Event.RECV_DATA_ES); + assertEquals(Http2StreamState.CLOSED, state); + } + + @Test + void bothSidesRemainOpenDuringBidirectionalData() { + Http2StreamState state = Http2StreamState.IDLE; + state = state.transition(1, Http2StreamState.Event.RECV_HEADERS); + state = state.transition(1, Http2StreamState.Event.SEND_HEADERS); + state = state.transition(1, Http2StreamState.Event.RECV_DATA); + state = state.transition(1, Http2StreamState.Event.SEND_DATA); + assertEquals(Http2StreamState.OPEN, state); + } + + @Test + void trailingHeadersCanCloseRemoteAfterLocalHalfClose() { + Http2StreamState state = Http2StreamState.IDLE; + state = state.transition(1, Http2StreamState.Event.RECV_HEADERS); + state = state.transition(1, Http2StreamState.Event.SEND_HEADERS_ES); + state = state.transition(1, Http2StreamState.Event.RECV_DATA); + state = state.transition(1, Http2StreamState.Event.RECV_HEADERS_ES); + assertEquals(Http2StreamState.CLOSED, state); + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/Http2TrailersTest.java b/flash/src/test/java/dev/relism/flash/http2/Http2TrailersTest.java new file mode 100644 index 0000000..d225443 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/Http2TrailersTest.java @@ -0,0 +1,150 @@ +package dev.relism.flash.http2; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import dev.relism.flash.bytes.ByteWriter; +import dev.relism.flash.extension.FlashApp; +import dev.relism.flash.extension.FlashConfiguration; +import dev.relism.flash.http2.frame.FrameFlags; +import dev.relism.flash.http2.frame.FrameType; +import dev.relism.flash.http2.hpack.HpackEncoder; +import java.io.EOFException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +class Http2TrailersTest { + private FlashApp app; + + @AfterEach + void stop() { + if (app != null) app.stop().join(); + } + + @Test + void requestTrailersReachHandlerAfterBodyEof() throws Exception { + int port = freePort(); + app = FlashApp.create(FlashConfiguration.builder() + .host("127.0.0.1").port(port).http2Enabled(true).build()); + app.post("/trailers", (request, response) -> { + assertEquals("abc", new String(request.body().bytes(), StandardCharsets.US_ASCII)); + return request.trailers().first("grpc-status"); + }); + app.start(); + + byte[] initial = requestHeaders("/trailers"); + ByteWriter trailer = new ByteWriter(32); + HpackEncoder.writeLiteral( + trailer, "grpc-status".getBytes(StandardCharsets.US_ASCII), + "7".getBytes(StandardCharsets.US_ASCII)); + + try (Socket socket = connect(port)) { + socket.getOutputStream().write(Http2TestFrames.concat( + Http2TestFrames.PREFACE, + Http2TestFrames.settings(), + Http2TestFrames.frame(FrameType.SETTINGS, FrameFlags.ACK, 0, new byte[0]), + Http2TestFrames.frame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1, initial), + Http2TestFrames.frame(FrameType.DATA, 0, 1, "abc".getBytes(StandardCharsets.US_ASCII)), + Http2TestFrames.frame(FrameType.HEADERS, + FrameFlags.END_HEADERS | FrameFlags.END_STREAM, 1, + Arrays.copyOf(trailer.array(), trailer.length())))); + socket.getOutputStream().flush(); + + Http2TestFrames.WireFrame data = frameOfType(socket.getInputStream(), 1, FrameType.DATA); + assertEquals("7", new String(data.payload(), StandardCharsets.US_ASCII)); + } + } + + @Test + void trailersWithoutEndStreamAreRejected() throws Exception { + int port = startBlockingRoute(); + ByteWriter trailer = new ByteWriter(32); + HpackEncoder.writeLiteral(trailer, ascii("x-end"), ascii("no")); + try (Socket socket = connect(port)) { + socket.getOutputStream().write(Http2TestFrames.concat( + Http2TestFrames.PREFACE, Http2TestFrames.settings(), + Http2TestFrames.frame(FrameType.SETTINGS, FrameFlags.ACK, 0, new byte[0]), + Http2TestFrames.frame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1, requestHeaders("/trailers")), + Http2TestFrames.frame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1, + Arrays.copyOf(trailer.array(), trailer.length())))); + socket.getOutputStream().flush(); + Http2TestFrames.WireFrame rst = frameOfType(socket.getInputStream(), 1, FrameType.RST_STREAM); + assertEquals(Http2ErrorCode.PROTOCOL_ERROR.code(), Http2TestFrames.readInt(rst.payload(), 0)); + } + } + + @Test + void pseudoHeaderInTrailersIsRejected() throws Exception { + int port = startBlockingRoute(); + try (Socket socket = connect(port)) { + socket.getOutputStream().write(Http2TestFrames.concat( + Http2TestFrames.PREFACE, Http2TestFrames.settings(), + Http2TestFrames.frame(FrameType.SETTINGS, FrameFlags.ACK, 0, new byte[0]), + Http2TestFrames.frame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1, requestHeaders("/trailers")), + Http2TestFrames.frame(FrameType.HEADERS, + FrameFlags.END_HEADERS | FrameFlags.END_STREAM, 1, new byte[] {(byte) 0x88}))); + socket.getOutputStream().flush(); + Http2TestFrames.WireFrame rst = frameOfType(socket.getInputStream(), 1, FrameType.RST_STREAM); + assertEquals(Http2ErrorCode.PROTOCOL_ERROR.code(), Http2TestFrames.readInt(rst.payload(), 0)); + } + } + + private int startBlockingRoute() throws Exception { + int port = freePort(); + app = FlashApp.create(FlashConfiguration.builder() + .host("127.0.0.1").port(port).http2Enabled(true).build()); + app.post("/trailers", (request, response) -> request.body().bytes()); + app.start(); + return port; + } + + private static byte[] requestHeaders(String path) { + ByteWriter block = new ByteWriter(64); + HpackEncoder.writeIndexed(block, 3); + HpackEncoder.writeIndexed(block, 6); + HpackEncoder.writeLiteralWithNameIndex(block, 4, ascii(path), false); + HpackEncoder.writeLiteralWithNameIndex(block, 1, ascii("localhost"), false); + HpackEncoder.writeLiteralWithNameIndex(block, 59, ascii("trailers"), false); + return Arrays.copyOf(block.array(), block.length()); + } + + private static byte[] ascii(String value) { + return value.getBytes(StandardCharsets.US_ASCII); + } + + private static Socket connect(int port) throws Exception { + Socket socket = new Socket("127.0.0.1", port); + socket.setSoTimeout(5_000); + return socket; + } + + private static Http2TestFrames.WireFrame frameOfType( + InputStream input, int streamId, FrameType type) throws Exception { + for (int i = 0; i < 12; i++) { + Http2TestFrames.WireFrame frame = readFrame(input); + if (frame.streamId() == streamId && frame.type() == type.code()) return frame; + } + throw new AssertionError("missing " + type + " frame"); + } + + private static Http2TestFrames.WireFrame readFrame(InputStream input) throws Exception { + byte[] header = input.readNBytes(9); + if (header.length != 9) throw new EOFException(); + int length = ((header[0] & 0xff) << 16) | ((header[1] & 0xff) << 8) | (header[2] & 0xff); + byte[] payload = input.readNBytes(length); + if (payload.length != length) throw new EOFException(); + return new Http2TestFrames.WireFrame( + header[3] & 0xff, header[4] & 0xff, Http2TestFrames.readInt(header, 5) & 0x7fff_ffff, + payload); + } + + private static int freePort() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java b/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java index 633af58..181ef01 100644 --- a/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java +++ b/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java @@ -132,6 +132,23 @@ class Http2ResponseWriterTest { assertEquals(List.of(FrameType.HEADERS, FrameType.DATA), parsed.types); } + @Test + void finalDataDoesNotEndStreamWhenTrailingHeadersFollow() throws Exception { + Response response = new Response(200, "ok", ContentType.TEXT_PLAIN) + .trailer("grpc-status", "0"); + Http2ResponseWriter writer = new Http2ResponseWriter(); + + writer.startFlowControlled( + response, 1, false, false, true, false, false, 16_384, 4096, 16_384); + Parsed parsed = parse(writer); + + assertEquals(List.of(FrameType.HEADERS, FrameType.DATA, FrameType.HEADERS), parsed.types); + assertEquals(0, parsed.flags.get(1) & FrameFlags.END_STREAM); + assertTrue((parsed.flags.get(2) & FrameFlags.END_STREAM) != 0); + assertTrue(decode(parsed.headerBlock).contains("grpc-status=0")); + assertTrue(writer.trailerHeadersInBatch()); + } + private static Parsed parse(Http2ResponseWriter writer) { Parsed parsed = new Parsed(); byte[] wire = writer.buffer(); diff --git a/flash/src/test/java/dev/relism/flash/models/ResponseStreamTest.java b/flash/src/test/java/dev/relism/flash/models/ResponseStreamTest.java new file mode 100644 index 0000000..68d1319 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/models/ResponseStreamTest.java @@ -0,0 +1,50 @@ +package dev.relism.flash.models; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import dev.relism.flash.http.ContentType; +import java.io.IOException; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; + +class ResponseStreamTest { + @Test + void producerWritesBodyAndTrailersThroughBoundedBridge() throws Exception { + Response response = new Response(200, ContentType.BINARY); + response.streaming(stream -> { + try { + stream.write(new byte[] {1, 2, 3}, 0, 3); + stream.trailer("grpc-status", "0"); + } catch (IOException failure) { + throw new RuntimeException(failure); + } + }); + + assertArrayEquals(new byte[] {1, 2, 3}, response.getStream().readAllBytes()); + assertEquals(true, response.hasTrailers()); + } + + @Test + void writeAfterCloseFailsWithoutWritingMoreBytes() throws Exception { + AtomicReference failure = new AtomicReference<>(); + CountDownLatch attempted = new CountDownLatch(1); + Response response = new Response(200, ContentType.BINARY); + response.streaming(stream -> { + try { + stream.close(); + stream.write(new byte[] {1}, 0, 1); + } catch (IOException expected) { + failure.set(expected); + } finally { + attempted.countDown(); + } + }); + + assertArrayEquals(new byte[0], response.getStream().readAllBytes()); + assertEquals(true, attempted.await(1, TimeUnit.SECONDS)); + assertEquals("response stream is closed", failure.get().getMessage()); + } +}