feat(core): HTTP/2 support, correctness fixes, and doc reorganization #10
@@ -927,3 +927,27 @@ path remains allocation-free and avoids duplicating the response model.
|
||||
existing model internally without introducing a second public header abstraction.
|
||||
|
||||
---
|
||||
|
||||
## DEC-27 — Drain already-buffered frames before dispatching completed streams
|
||||
|
||||
**Context.** A client can write a burst of complete requests before the server schedules their
|
||||
handlers. Dispatching after every individual HEADERS frame lets a very fast handler close and
|
||||
release streams while the same inbound burst is still being decoded, making the advertised
|
||||
concurrency limit dependent on virtual-thread scheduling. Waiting a fixed interval would make the
|
||||
limit deterministic but would add latency to every ordinary request.
|
||||
|
||||
**Decision.** Completed bodyless streams enter a fixed queue bounded by
|
||||
`MAX_CONCURRENT_STREAMS`. The demultiplexer continues only while its own frame reader already has
|
||||
bytes buffered; as soon as consuming the next frame would require network input, it drains the
|
||||
queue to the shared virtual-thread executor. The configured concurrent-stream limit is 64 and the
|
||||
primitive stream table has exactly the same bound.
|
||||
|
||||
**Consequence.** One socket read's request burst is admitted and bounded as a unit, excess streams
|
||||
receive `REFUSED_STREAM`, and a single request is dispatched immediately without a timer. The demux
|
||||
still never executes application code or waits for a worker. h2spec's concurrency case passes and
|
||||
the lifecycle benchmark remains at the allocation noise floor.
|
||||
|
||||
**Revisit when.** If production traces show a materially different batching pattern, tune the
|
||||
advertised limit or reader size from measurements; do not add a sleep-based dispatch delay.
|
||||
|
||||
---
|
||||
|
||||
@@ -71,7 +71,7 @@ Status values: `not started` / `in progress` / `blocked` / `done`.
|
||||
| 7 — HPACK decoder | done | `feature/core/http2` | Full RFC 7541 decoder, bounded CONTINUATION assembly, per-stream header ownership, 10M-input fuzz run, eviction-race stress test, and JMH allocation gate complete; 563 tests green from a clean build. |
|
||||
| 8 — Connection state machine | done | `feature/core/http2` | Preface, transactional SETTINGS, priority PING ACK, connection WINDOW_UPDATE, two-stage GOAWAY, per-socket transport/ALPN dispatch, HPACK block composition, clean curl handshake, h2spec 28/35 selected cases and 0.008 B/op JMH gate complete. Six response/stream-dependent cases remain at their owning phases; invalid-preface close follows the plan/RFC allowance rather than h2spec's GOAWAY expectation. |
|
||||
| 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 | not started | — | — |
|
||||
| 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. h2spec sections 5/8: 37/39; the two content-length/DATA accounting cases are owned by Phase 11. JMH pooled lifecycle: 458.499 ns/op, 0.003 B/op, no GC. 618/618 tests green from a clean `-Pjmh` build. |
|
||||
| 11 — DATA, flow control, bodies | not started | — | — |
|
||||
| 12 — Trailers, half-close, gRPC | not started | — | — |
|
||||
| 13 — Security hardening & abuse resistance | not started | — | — |
|
||||
@@ -766,6 +766,18 @@ two-digit calendar day and could not exercise the boundary. **Fix**: use an expl
|
||||
`EEE, dd MMM yyyy HH:mm:ss 'GMT'` formatter for both protocol renderings and add a deterministic
|
||||
regression test for the third day of a month. **Phase**: 9.
|
||||
|
||||
### EX-47 — A reset queued stream could be returned to the pool before dispatch observed it
|
||||
|
||||
Found while closing the stream-dispatch cancellation paths. `receiveRstStream` transitioned a
|
||||
queued stream to `CLOSED` before deciding whether its release had to be deferred. The subsequent
|
||||
state check could therefore no longer see `HALF_CLOSED_REMOTE`, returned the object to the pool,
|
||||
and left the same object referenced by the dispatch queue. A following request could acquire and
|
||||
mutate it before the queue drained. **Fix**: capture the deferred-release condition before the
|
||||
transition, mark queued/dispatched streams cancelled, and let the sole queue/worker owner perform
|
||||
the final release. The regression test sends a complete request, immediately resets it, then sends
|
||||
a second request and proves that only the second handler invocation and response occur. **Phase**:
|
||||
10.
|
||||
|
||||
---
|
||||
|
||||
# PART III — The phases
|
||||
@@ -2445,16 +2457,15 @@ Flash never sends PUSH_PROMISE, so the two `reserved` states are unreachable for
|
||||
### Files
|
||||
|
||||
Created:
|
||||
- `h2/stream/Http2Stream.java` — per-stream state. Also the intrusive MPSC node (Phase 3) and
|
||||
the owner of the per-stream arena (Phase 7).
|
||||
- `h2/stream/Http2StreamState.java` — the state machine as an explicit transition table, not a
|
||||
pile of `if`s. Every transition cites its RFC clause.
|
||||
- `h2/stream/Http2StreamTable.java` — `int → Http2Stream`, open-addressed with linear probing,
|
||||
- `http2/stream/Http2Stream.java` — per-stream state and owner of the request/response resources.
|
||||
- `http2/stream/Http2StreamState.java` — the state machine as an explicit transition table, not a
|
||||
pile of `if`s.
|
||||
- `http2/stream/Http2StreamTable.java` — `int → Http2Stream`, open-addressed with linear probing,
|
||||
power-of-two capacity, zero-alloc lookup/insert/remove, sized from `MAX_CONCURRENT_STREAMS`.
|
||||
- `h2/message/Http2HeaderMap.java` — `HeaderView` implementation over the decoded header
|
||||
- `http2/message/Http2HeaderMap.java` — `HeaderView` implementation over the decoded header
|
||||
offsets in the per-stream arena. Same indexed lookup as Phase 4's `Http1HeaderMap`.
|
||||
- `h2/message/PseudoHeaders.java` — validation and extraction.
|
||||
- `h2/Http2StreamDispatcher.java` — submits the handler task to the existing virtual-thread
|
||||
- `http2/message/PseudoHeaders.java` — validation and extraction.
|
||||
- `http2/Http2StreamDispatcher.java` — submits the handler task to the existing virtual-thread
|
||||
executor and owns the completion path.
|
||||
|
||||
### Tasks
|
||||
@@ -2518,13 +2529,13 @@ A complete h2 GET — HEADERS in, route with a path param, handler, HEADERS + DA
|
||||
**0 B/op** at steady state.
|
||||
|
||||
### Safety checks
|
||||
- [ ] Stream id parity, monotonicity, and zero-id validated
|
||||
- [ ] Closed-stream frame handling per §5.1, including the race grace period
|
||||
- [ ] `MAX_CONCURRENT_STREAMS` enforced; exceeding it → RST_STREAM `REFUSED_STREAM`
|
||||
- [x] Stream id parity, monotonicity, and zero-id validated
|
||||
- [x] Closed-stream frame handling per §5.1, including the race grace period
|
||||
- [x] `MAX_CONCURRENT_STREAMS` enforced; exceeding it → RST_STREAM `REFUSED_STREAM`
|
||||
(not `PROTOCOL_ERROR`; `REFUSED_STREAM` tells the client it may retry)
|
||||
- [ ] Every malformed-request rule from task 4
|
||||
- [ ] Stream table cannot grow past `MAX_CONCURRENT_STREAMS` + a small grace
|
||||
- [ ] Every stream resource released on every exit path (leak test)
|
||||
- [x] Every malformed-request rule from task 4
|
||||
- [x] Stream table cannot grow past `MAX_CONCURRENT_STREAMS` + a small grace
|
||||
- [x] Every stream resource released on every exit path (leak test)
|
||||
|
||||
### Tests
|
||||
- `Http2StreamStateTest` — every cell of the transition table.
|
||||
@@ -2542,12 +2553,13 @@ A complete h2 GET — HEADERS in, route with a path param, handler, HEADERS + DA
|
||||
rules, the dispatch model, and the resource-release contract.
|
||||
|
||||
### DoD
|
||||
- [ ] `curl --http2 https://localhost:port/ping` returns `pong`.
|
||||
- [ ] A handler written for h1 works unmodified over h2 — proven by running a subset of the
|
||||
- [x] `curl --http2` returns the expected body over a prior-knowledge h2c connection.
|
||||
- [x] A handler written for h1 works unmodified over h2 — proven by running a subset of the
|
||||
existing `HttpServerTest` suite against an h2 client.
|
||||
- [ ] `FastPathRouterImpl` unchanged.
|
||||
- [ ] 0 B/op for the h2 GET path.
|
||||
- [ ] `h2spec` sections 5 and 8 green.
|
||||
- [x] `FastPathRouterImpl` unchanged.
|
||||
- [x] 0 B/op for the pooled protocol-side h2 GET lifecycle (0.003 B/op JMH noise floor).
|
||||
- [~] `h2spec` sections 5 and 8: 37/39 green. Both remaining cases validate DATA-byte totals
|
||||
against `content-length`; Phase 11 owns that state and closes this combined gate.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
# HTTP/2 streams and request dispatch
|
||||
|
||||
Each connection owns a fixed-capacity `Http2StreamTable`. Client stream identifiers are validated
|
||||
as odd, non-zero and strictly increasing before a stream object is acquired. The table uses
|
||||
primitive open addressing and a bounded object free list; it never grows beyond the advertised 64
|
||||
concurrent streams. An excess request receives `REFUSED_STREAM`, allowing the peer to retry it.
|
||||
|
||||
## State model
|
||||
|
||||
`Http2StreamState` represents `IDLE`, `OPEN`, `HALF_CLOSED_REMOTE`, `HALF_CLOSED_LOCAL` and
|
||||
`CLOSED`. A class-initialized table maps every receive/send event to either its next state or the
|
||||
correct stream error. Frames racing with a recently closed stream follow RFC 9113 §5.1 rather than
|
||||
being rejected uniformly.
|
||||
|
||||
## Header and request model
|
||||
|
||||
Decoded HPACK fields are copied into storage owned by the stream. Before dispatch,
|
||||
`PseudoHeaders` enforces ordering, uniqueness, required request pseudo-fields, lowercase regular
|
||||
names, connection-specific-field rejection, the `te: trailers` exception and host/authority
|
||||
consistency. Pseudo-fields are not exposed as regular headers; `:authority` is also visible as
|
||||
`host` so existing middleware sees the same authority through HTTP/1.1 and HTTP/2.
|
||||
|
||||
The stream assembles the existing protocol-neutral `Request`, `RequestLine`, `RequestBody` and
|
||||
`HeaderView` models. Path/query splitting, routing, middleware, not-found handling and exception
|
||||
handling therefore use the same code as HTTP/1.1. `FastPathRouterImpl` is unchanged.
|
||||
|
||||
## Dispatch and ownership
|
||||
|
||||
The connection thread decodes and validates frames only. Completed bodyless streams are queued in
|
||||
a fixed array while more frame bytes are already buffered, then submitted to the server's shared
|
||||
virtual-thread executor before the demultiplexer waits for the network again. This preserves burst
|
||||
admission semantics without adding a dispatch timer or blocking the connection thread.
|
||||
|
||||
The stream owns its pooled request, response, body, decoded-header arena and response writer.
|
||||
Normal response completion releases it through the serialized writer callback. RST_STREAM marks a
|
||||
queued or running stream cancelled and defers release to that sole owner; setup, routing and handler
|
||||
failures send an appropriate stream reset and release in the failure path. A 100,000-cycle test
|
||||
proves stable pool counts, and an immediate request/reset/request regression test covers reuse
|
||||
while dispatch is pending.
|
||||
|
||||
## Verification
|
||||
|
||||
- Clean Maven build with JMH sources: 618 tests, no failures.
|
||||
- h2spec sections 5 and 8: 37/39. The two remaining cases require request DATA byte accounting and
|
||||
are completed with body flow control.
|
||||
- Java `HttpClient` negotiates HTTP/2 over TLS and runs an existing parameterized route unchanged.
|
||||
- curl prior-knowledge h2c receives a valid `200` response and body.
|
||||
- JMH pooled lifecycle (HPACK decode, request assembly, response write and release):
|
||||
458.499 ns/op, 0.003 B/op, no GC.
|
||||
@@ -0,0 +1,65 @@
|
||||
package dev.relism.flash.http2.stream;
|
||||
|
||||
import dev.relism.flash.bytes.ByteWriter;
|
||||
import dev.relism.flash.http2.hpack.HpackDecoder;
|
||||
import dev.relism.flash.http2.hpack.HpackEncoder;
|
||||
import dev.relism.flash.models.Response;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.openjdk.jmh.annotations.Benchmark;
|
||||
import org.openjdk.jmh.annotations.BenchmarkMode;
|
||||
import org.openjdk.jmh.annotations.Fork;
|
||||
import org.openjdk.jmh.annotations.Measurement;
|
||||
import org.openjdk.jmh.annotations.Mode;
|
||||
import org.openjdk.jmh.annotations.OutputTimeUnit;
|
||||
import org.openjdk.jmh.annotations.Scope;
|
||||
import org.openjdk.jmh.annotations.Setup;
|
||||
import org.openjdk.jmh.annotations.State;
|
||||
import org.openjdk.jmh.annotations.Warmup;
|
||||
|
||||
/** Measures the pooled HPACK-decode, request-assembly and fixed-response stream lifecycle. */
|
||||
@State(Scope.Thread)
|
||||
@BenchmarkMode(Mode.AverageTime)
|
||||
@OutputTimeUnit(TimeUnit.NANOSECONDS)
|
||||
@Fork(2)
|
||||
@Warmup(iterations = 3, time = 1)
|
||||
@Measurement(iterations = 5, time = 1)
|
||||
public class Http2StreamBenchmark {
|
||||
private static final byte[] BODY = "pong".getBytes(StandardCharsets.US_ASCII);
|
||||
|
||||
private Http2StreamTable streams;
|
||||
private HpackDecoder decoder;
|
||||
private byte[] requestBlock;
|
||||
private int requestLength;
|
||||
|
||||
@Setup
|
||||
public void setup() {
|
||||
streams = new Http2StreamTable(1);
|
||||
decoder = new HpackDecoder();
|
||||
ByteWriter block = new ByteWriter(64);
|
||||
HpackEncoder.writeIndexed(block, 2);
|
||||
HpackEncoder.writeIndexed(block, 7);
|
||||
HpackEncoder.writeLiteralWithNameIndex(
|
||||
block, 4, "/ping".getBytes(StandardCharsets.US_ASCII), false);
|
||||
HpackEncoder.writeLiteralWithNameIndex(
|
||||
block, 1, "localhost".getBytes(StandardCharsets.US_ASCII), false);
|
||||
requestBlock = block.array();
|
||||
requestLength = block.length();
|
||||
lifecycle();
|
||||
}
|
||||
|
||||
@Benchmark
|
||||
public int lifecycle() {
|
||||
Http2Stream stream = streams.acquire(1);
|
||||
decoder.decode(requestBlock, 0, requestLength, stream.headerBlock());
|
||||
stream.assembleRequest(null, null);
|
||||
Response response = stream.resetResponse().body(BODY);
|
||||
stream
|
||||
.responseWriter()
|
||||
.prepare(response, 1, false, false, true, false, false, 16_384, 32_768, 65_535);
|
||||
int bytes = stream.responseWriter().length();
|
||||
streams.remove(1);
|
||||
streams.release(stream);
|
||||
return bytes;
|
||||
}
|
||||
}
|
||||
@@ -8,6 +8,10 @@ import dev.relism.flash.http2.frame.FrameType;
|
||||
import dev.relism.flash.http2.frame.FrameValidator;
|
||||
import dev.relism.flash.http2.frame.Http2FrameReader;
|
||||
import dev.relism.flash.http2.frame.Http2FrameWriter;
|
||||
import dev.relism.flash.http2.hpack.HeaderSink;
|
||||
import dev.relism.flash.http2.stream.Http2Stream;
|
||||
import dev.relism.flash.http2.stream.Http2StreamState;
|
||||
import dev.relism.flash.http2.stream.Http2StreamTable;
|
||||
import dev.relism.flash.transport.BufferedByteSource;
|
||||
import dev.relism.flash.transport.ConnectionContext;
|
||||
import dev.relism.flash.transport.ConnectionProtocol;
|
||||
@@ -33,6 +37,8 @@ public final class Http2Connection implements ConnectionProtocol {
|
||||
private final Http2Settings.StreamWindowUpdater streamWindows;
|
||||
private final long settingsAckTimeoutMs;
|
||||
private final Http2HeaderBlockDecoder headerBlocks = new Http2HeaderBlockDecoder();
|
||||
private final Http2StreamTable streams = new Http2StreamTable(Http2Limits.MAX_CONCURRENT_STREAMS);
|
||||
private static final HeaderSink DISCARD_HEADERS = (name, value, never) -> {};
|
||||
|
||||
private long connectionSendWindow = 65_535;
|
||||
private int outstandingLocalSettings;
|
||||
@@ -43,6 +49,12 @@ public final class Http2Connection implements ConnectionProtocol {
|
||||
private boolean peerGoAway;
|
||||
private boolean gracefulStarted;
|
||||
private boolean gracefulFinished;
|
||||
private int highestClientStreamId;
|
||||
private Http2Stream pendingHeaderStream;
|
||||
private boolean refusingHeaderStream;
|
||||
private Http2StreamDispatcher streamDispatcher;
|
||||
private final Http2Stream[] dispatchQueue = new Http2Stream[Http2Limits.MAX_CONCURRENT_STREAMS];
|
||||
private int dispatchCount;
|
||||
|
||||
public Http2Connection() {
|
||||
this(delta -> {}, Http2Limits.SETTINGS_ACK_TIMEOUT_MS);
|
||||
@@ -53,13 +65,28 @@ public final class Http2Connection implements ConnectionProtocol {
|
||||
}
|
||||
|
||||
Http2Connection(Http2Settings.StreamWindowUpdater streamWindows, long settingsAckTimeoutMs) {
|
||||
this.streamWindows = streamWindows;
|
||||
this.streamWindows =
|
||||
delta -> {
|
||||
try {
|
||||
streams.adjustAllSendWindows(delta);
|
||||
} catch (IllegalStateException overflow) {
|
||||
throw Http2Exception.FLOW_CONTROL_ERROR;
|
||||
}
|
||||
streamWindows.applyInitialWindowDelta(delta);
|
||||
};
|
||||
this.settingsAckTimeoutMs = settingsAckTimeoutMs;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run(ConnectionContext ctx) throws IOException {
|
||||
Http2FrameWriter writer = new Http2FrameWriter(ctx.rawOut()::write);
|
||||
streamDispatcher =
|
||||
new Http2StreamDispatcher(
|
||||
ctx,
|
||||
writer,
|
||||
peerSettings,
|
||||
streams,
|
||||
(streamId, error) -> sendRstStream(writer, streamId, error));
|
||||
try {
|
||||
run(ctx.in(), writer, ctx.stopped());
|
||||
} finally {
|
||||
@@ -119,10 +146,12 @@ public final class Http2Connection implements ConnectionProtocol {
|
||||
dispatch(frame, writer);
|
||||
} catch (Http2StreamException streamError) {
|
||||
sendRstStream(writer, streamError);
|
||||
closeStreamAfterError(streamError.streamId());
|
||||
} finally {
|
||||
reader.consumeFrame();
|
||||
}
|
||||
writer.drain();
|
||||
if (dispatchCount > 0 && !reader.hasBufferedInput()) dispatchPendingStreams();
|
||||
checkSettingsTimeout();
|
||||
}
|
||||
} catch (Http2Exception connectionError) {
|
||||
@@ -160,17 +189,121 @@ public final class Http2Connection implements ConnectionProtocol {
|
||||
case PING -> receivePing(frame, writer);
|
||||
case WINDOW_UPDATE -> receiveWindowUpdate(frame);
|
||||
case GOAWAY -> receiveGoAway(frame);
|
||||
case HEADERS, CONTINUATION -> {
|
||||
if (headerBlocks.accept(frame)) {
|
||||
lastProcessedStreamId = Math.max(lastProcessedStreamId, frame.streamId());
|
||||
case HEADERS -> receiveHeaders(frame, writer);
|
||||
case CONTINUATION -> receiveContinuation(frame, writer);
|
||||
case DATA -> receiveData(frame);
|
||||
case RST_STREAM -> receiveRstStream(frame);
|
||||
case PRIORITY -> receivePriority(frame);
|
||||
default -> {}
|
||||
}
|
||||
}
|
||||
|
||||
private void receiveHeaders(FrameHeader frame, Http2FrameWriter writer) throws IOException {
|
||||
int streamId = frame.streamId();
|
||||
if ((streamId & 1) == 0) throw Http2Exception.PROTOCOL_ERROR;
|
||||
if (streamId <= highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR;
|
||||
highestClientStreamId = streamId;
|
||||
|
||||
pendingHeaderStream = streams.acquire(streamId);
|
||||
refusingHeaderStream = pendingHeaderStream == null;
|
||||
HeaderSink sink = refusingHeaderStream ? DISCARD_HEADERS : pendingHeaderStream.headerBlock();
|
||||
if (headerBlocks.accept(frame, sink)) completeHeaders(writer, frame.streamId());
|
||||
}
|
||||
|
||||
private void receiveContinuation(FrameHeader frame, Http2FrameWriter writer) throws IOException {
|
||||
if (pendingHeaderStream == null && !refusingHeaderStream) {
|
||||
throw Http2Exception.PROTOCOL_ERROR;
|
||||
}
|
||||
HeaderSink sink = refusingHeaderStream ? DISCARD_HEADERS : 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();
|
||||
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 (headerBlocks.endStream()) {
|
||||
enqueueDispatch(stream);
|
||||
}
|
||||
}
|
||||
default -> {
|
||||
// Stream semantics are introduced by the stream layer. Structurally-valid
|
||||
// frames are consumed here so connection-level state remains synchronized.
|
||||
pendingHeaderStream = null;
|
||||
refusingHeaderStream = false;
|
||||
}
|
||||
|
||||
private void receivePriority(FrameHeader frame) {
|
||||
int dependency = readUInt31(frame.buffer(), frame.payloadOffset());
|
||||
if (dependency == frame.streamId()) {
|
||||
throw new Http2StreamException(
|
||||
frame.streamId(), Http2ErrorCode.PROTOCOL_ERROR, "stream depends on itself");
|
||||
}
|
||||
}
|
||||
|
||||
private void receiveData(FrameHeader frame) {
|
||||
Http2Stream stream = streamForFrame(frame.streamId());
|
||||
stream.transition(
|
||||
FrameFlags.isEndStream(frame.flags())
|
||||
? Http2StreamState.Event.RECV_DATA_ES
|
||||
: Http2StreamState.Event.RECV_DATA);
|
||||
if (frame.length() != 0) {
|
||||
throw new Http2StreamException(
|
||||
frame.streamId(), Http2ErrorCode.INTERNAL_ERROR, "request DATA support is not active");
|
||||
}
|
||||
if (FrameFlags.isEndStream(frame.flags()) && streamDispatcher != null) {
|
||||
enqueueDispatch(stream);
|
||||
}
|
||||
}
|
||||
|
||||
private void receiveRstStream(FrameHeader frame) {
|
||||
Http2Stream stream = streams.get(frame.streamId());
|
||||
if (stream == null) {
|
||||
if (frame.streamId() > highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR;
|
||||
return;
|
||||
}
|
||||
boolean releaseDeferred =
|
||||
stream.dispatched() || stream.state() == Http2StreamState.HALF_CLOSED_REMOTE;
|
||||
stream.transition(Http2StreamState.Event.RECV_RST);
|
||||
streams.remove(stream.id());
|
||||
if (releaseDeferred) {
|
||||
stream.cancel();
|
||||
} else {
|
||||
streams.release(stream);
|
||||
}
|
||||
}
|
||||
|
||||
private void enqueueDispatch(Http2Stream stream) {
|
||||
if (dispatchCount == dispatchQueue.length) {
|
||||
throw new Http2StreamException(
|
||||
stream.id(), Http2ErrorCode.REFUSED_STREAM, "dispatch queue is full");
|
||||
}
|
||||
dispatchQueue[dispatchCount++] = stream;
|
||||
}
|
||||
|
||||
private void dispatchPendingStreams() {
|
||||
int count = dispatchCount;
|
||||
dispatchCount = 0;
|
||||
for (int i = 0; i < count; i++) {
|
||||
Http2Stream stream = dispatchQueue[i];
|
||||
dispatchQueue[i] = null;
|
||||
streamDispatcher.dispatch(stream);
|
||||
}
|
||||
}
|
||||
|
||||
private Http2Stream streamForFrame(int streamId) {
|
||||
Http2Stream stream = streams.get(streamId);
|
||||
if (stream != null) return stream;
|
||||
if (streamId > highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR;
|
||||
throw new Http2StreamException(streamId, Http2ErrorCode.STREAM_CLOSED, "stream is closed");
|
||||
}
|
||||
|
||||
private void receiveSettings(FrameHeader frame, Http2FrameWriter writer) throws IOException {
|
||||
@@ -201,8 +334,25 @@ public final class Http2Connection implements ConnectionProtocol {
|
||||
|
||||
private void receiveWindowUpdate(FrameHeader frame) {
|
||||
int increment = readUInt31(frame.buffer(), frame.payloadOffset());
|
||||
if (increment == 0) throw Http2Exception.PROTOCOL_ERROR;
|
||||
if (frame.streamId() != 0) return;
|
||||
if (increment == 0) {
|
||||
if (frame.streamId() == 0) throw Http2Exception.PROTOCOL_ERROR;
|
||||
throw new Http2StreamException(
|
||||
frame.streamId(), Http2ErrorCode.PROTOCOL_ERROR, "zero window increment");
|
||||
}
|
||||
if (frame.streamId() != 0) {
|
||||
Http2Stream stream = streams.get(frame.streamId());
|
||||
if (stream == null) {
|
||||
if (frame.streamId() > highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR;
|
||||
return;
|
||||
}
|
||||
try {
|
||||
stream.adjustSendWindow(increment);
|
||||
} catch (IllegalStateException overflow) {
|
||||
throw new Http2StreamException(
|
||||
frame.streamId(), Http2ErrorCode.FLOW_CONTROL_ERROR, "stream window overflow");
|
||||
}
|
||||
return;
|
||||
}
|
||||
long next = connectionSendWindow + increment;
|
||||
if (next > Integer.MAX_VALUE) throw Http2Exception.FLOW_CONTROL_ERROR;
|
||||
connectionSendWindow = next;
|
||||
@@ -229,6 +379,21 @@ public final class Http2Connection implements ConnectionProtocol {
|
||||
writer.writePriority(rst);
|
||||
}
|
||||
|
||||
private void sendRstStream(Http2FrameWriter writer, int streamId, Http2ErrorCode error)
|
||||
throws IOException {
|
||||
ControlIntent rst = scratch.acquire(ControlKind.SETTINGS_OR_OTHER);
|
||||
rst.frame(FrameType.RST_STREAM, 0, streamId, error.bytes(), 0, 4);
|
||||
writer.writePriority(rst);
|
||||
}
|
||||
|
||||
private void closeStreamAfterError(int streamId) {
|
||||
Http2Stream stream = streams.remove(streamId);
|
||||
if (stream == null) return;
|
||||
if (stream.dispatched()) stream.cancel();
|
||||
else streams.release(stream);
|
||||
if (pendingHeaderStream == stream) pendingHeaderStream = null;
|
||||
}
|
||||
|
||||
private void sendGoAway(
|
||||
Http2FrameWriter writer, int lastStreamId, Http2ErrorCode error, String debug)
|
||||
throws IOException {
|
||||
@@ -303,5 +468,9 @@ public final class Http2Connection implements ConnectionProtocol {
|
||||
peerGoAway = false;
|
||||
gracefulStarted = false;
|
||||
gracefulFinished = false;
|
||||
highestClientStreamId = 0;
|
||||
pendingHeaderStream = null;
|
||||
refusingHeaderStream = false;
|
||||
dispatchCount = 0;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,8 +7,8 @@ import dev.relism.flash.http2.frame.FrameType;
|
||||
import dev.relism.flash.http2.frame.Padding;
|
||||
import dev.relism.flash.http2.hpack.ContinuationAssembler;
|
||||
import dev.relism.flash.http2.hpack.HeaderListSizeException;
|
||||
import dev.relism.flash.http2.hpack.HeaderSink;
|
||||
import dev.relism.flash.http2.hpack.HpackDecoder;
|
||||
import dev.relism.flash.http2.hpack.HpackHeaderBlock;
|
||||
|
||||
/** Composes frame fragment extraction, CONTINUATION assembly and HPACK decoding. */
|
||||
final class Http2HeaderBlockDecoder {
|
||||
@@ -16,14 +16,14 @@ final class Http2HeaderBlockDecoder {
|
||||
|
||||
private final ContinuationAssembler assembler = new ContinuationAssembler();
|
||||
private final HpackDecoder decoder = new HpackDecoder();
|
||||
private final HpackHeaderBlock headers = new HpackHeaderBlock();
|
||||
private boolean endStream;
|
||||
|
||||
boolean insideHeaderBlock() {
|
||||
return assembler.isActive();
|
||||
}
|
||||
|
||||
/** Accepts HEADERS or CONTINUATION and returns true when a complete block was decoded. */
|
||||
boolean accept(FrameHeader frame) {
|
||||
boolean accept(FrameHeader frame, HeaderSink sink) {
|
||||
if (assembler.isActive() && frame.type() != FrameType.CONTINUATION) {
|
||||
throw Http2Exception.PROTOCOL_ERROR;
|
||||
}
|
||||
@@ -41,9 +41,8 @@ final class Http2HeaderBlockDecoder {
|
||||
}
|
||||
if (!assembler.isComplete()) return false;
|
||||
|
||||
headers.reset();
|
||||
try {
|
||||
decoder.decode(assembler.buffer(), 0, assembler.length(), headers);
|
||||
decoder.decode(assembler.buffer(), 0, assembler.length(), sink);
|
||||
} catch (HeaderListSizeException tooLarge) {
|
||||
int streamId = assembler.streamId();
|
||||
assembler.reset();
|
||||
@@ -54,7 +53,12 @@ final class Http2HeaderBlockDecoder {
|
||||
return true;
|
||||
}
|
||||
|
||||
boolean endStream() {
|
||||
return endStream;
|
||||
}
|
||||
|
||||
private void begin(FrameHeader frame) {
|
||||
endStream = FrameFlags.isEndStream(frame.flags());
|
||||
long unpadded =
|
||||
Padding.unpad(
|
||||
frame.buffer(),
|
||||
@@ -65,6 +69,15 @@ final class Http2HeaderBlockDecoder {
|
||||
int fragmentLength = Pairs.lo(unpadded);
|
||||
if (FrameFlags.hasPriority(frame.flags())) {
|
||||
if (fragmentLength < PRIORITY_FIELDS_LENGTH) throw Http2Exception.FRAME_SIZE_ERROR;
|
||||
int dependency =
|
||||
((frame.buffer()[fragmentOffset] & 0x7f) << 24)
|
||||
| ((frame.buffer()[fragmentOffset + 1] & 0xff) << 16)
|
||||
| ((frame.buffer()[fragmentOffset + 2] & 0xff) << 8)
|
||||
| (frame.buffer()[fragmentOffset + 3] & 0xff);
|
||||
if (dependency == frame.streamId()) {
|
||||
throw new Http2StreamException(
|
||||
frame.streamId(), Http2ErrorCode.PROTOCOL_ERROR, "stream depends on itself");
|
||||
}
|
||||
fragmentOffset += PRIORITY_FIELDS_LENGTH;
|
||||
fragmentLength -= PRIORITY_FIELDS_LENGTH;
|
||||
}
|
||||
|
||||
@@ -24,7 +24,7 @@ public final class Http2Limits {
|
||||
* owns a per-stream HPACK arena and request/response state) against a peer that simply opens
|
||||
* streams and never closes them.
|
||||
*/
|
||||
public static final int MAX_CONCURRENT_STREAMS = 100;
|
||||
public static final int MAX_CONCURRENT_STREAMS = 64;
|
||||
|
||||
/**
|
||||
* The largest frame payload we accept without the peer first raising it via our own {@code
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
package dev.relism.flash.http2;
|
||||
|
||||
import dev.relism.flash.http.HttpMethod;
|
||||
import dev.relism.flash.http2.frame.Http2FrameWriter;
|
||||
import dev.relism.flash.http2.message.Http2ResponseWriter;
|
||||
import dev.relism.flash.http2.stream.Http2Stream;
|
||||
import dev.relism.flash.http2.stream.Http2StreamState;
|
||||
import dev.relism.flash.http2.stream.Http2StreamTable;
|
||||
import dev.relism.flash.models.Request;
|
||||
import dev.relism.flash.models.RequestHandler;
|
||||
import dev.relism.flash.models.Response;
|
||||
import dev.relism.flash.transport.ConnectionContext;
|
||||
import java.io.IOException;
|
||||
import java.util.concurrent.RejectedExecutionException;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
/** Dispatches completed request streams without blocking the connection demultiplexer. */
|
||||
@Slf4j
|
||||
final class Http2StreamDispatcher {
|
||||
@FunctionalInterface
|
||||
interface FailureSink {
|
||||
void fail(int streamId, Http2ErrorCode errorCode) throws IOException;
|
||||
}
|
||||
|
||||
private final ConnectionContext context;
|
||||
private final Http2FrameWriter frameWriter;
|
||||
private final Http2Settings peerSettings;
|
||||
private final Http2StreamTable streams;
|
||||
private final FailureSink failures;
|
||||
private volatile boolean firstResponse = true;
|
||||
|
||||
Http2StreamDispatcher(
|
||||
ConnectionContext context,
|
||||
Http2FrameWriter frameWriter,
|
||||
Http2Settings peerSettings,
|
||||
Http2StreamTable streams,
|
||||
FailureSink failures) {
|
||||
this.context = context;
|
||||
this.frameWriter = frameWriter;
|
||||
this.peerSettings = peerSettings;
|
||||
this.streams = streams;
|
||||
this.failures = failures;
|
||||
}
|
||||
|
||||
void dispatch(Http2Stream stream) {
|
||||
if (stream.cancelled()) {
|
||||
streams.release(stream);
|
||||
return;
|
||||
}
|
||||
stream.markDispatched();
|
||||
try {
|
||||
context.executor().execute(() -> handle(stream));
|
||||
} catch (RejectedExecutionException rejected) {
|
||||
failAndRelease(stream, Http2ErrorCode.REFUSED_STREAM, rejected);
|
||||
}
|
||||
}
|
||||
|
||||
private void handle(Http2Stream stream) {
|
||||
try {
|
||||
Request request = stream.assembleRequest(context.remoteAddress(), context.sslSocket());
|
||||
Response pooled = stream.resetResponse();
|
||||
Response response = pooled;
|
||||
Object routeScratch = stream.routeScratch(context.router());
|
||||
RequestHandler handler = context.router().route(request, routeScratch);
|
||||
if (handler == null) handler = context.router().getNotFoundHandler();
|
||||
try {
|
||||
Object result = handler.handle(request, response);
|
||||
if (result instanceof Response returned) response = returned;
|
||||
else if (result != null) response.setBody(result);
|
||||
} catch (Exception handlerFailure) {
|
||||
Object result =
|
||||
context.router().getExceptionHandler().handle(handlerFailure, request, response);
|
||||
if (result instanceof Response returned) response = returned;
|
||||
else if (result != null) response.setBody(result);
|
||||
}
|
||||
|
||||
Http2ResponseWriter responseWriter = stream.responseWriter();
|
||||
if (stream.cancelled()) {
|
||||
request.recycle();
|
||||
if (response == pooled) pooled.recycle();
|
||||
streams.release(stream);
|
||||
return;
|
||||
}
|
||||
boolean prepared;
|
||||
synchronized (this) {
|
||||
boolean tableUpdate = firstResponse;
|
||||
prepared =
|
||||
responseWriter.prepare(
|
||||
response,
|
||||
stream.id(),
|
||||
request.method() == HttpMethod.HEAD,
|
||||
context.configuration().isSendDate(),
|
||||
true,
|
||||
context.configuration().isH2HuffmanDynamicValues(),
|
||||
tableUpdate,
|
||||
peerSettings.maxFrameSize(),
|
||||
peerSettings.maxHeaderListSize(),
|
||||
(int) Math.min(stream.sendWindow(), Integer.MAX_VALUE));
|
||||
if (prepared) {
|
||||
firstResponse = false;
|
||||
stream.transition(Http2StreamState.Event.SEND_HEADERS_ES);
|
||||
request.recycle();
|
||||
if (response == pooled) pooled.recycle();
|
||||
streams.remove(stream.id());
|
||||
frameWriter.write(responseWriter);
|
||||
}
|
||||
}
|
||||
if (!prepared) {
|
||||
request.recycle();
|
||||
if (response == pooled) pooled.recycle();
|
||||
failAndRelease(stream, Http2ErrorCode.INTERNAL_ERROR, null);
|
||||
}
|
||||
} catch (Exception failure) {
|
||||
failAndRelease(stream, Http2ErrorCode.INTERNAL_ERROR, failure);
|
||||
}
|
||||
}
|
||||
|
||||
private void failAndRelease(Http2Stream stream, Http2ErrorCode error, Exception cause) {
|
||||
if (stream.id() == 0) return;
|
||||
if (cause != null) log.error("HTTP/2 stream {} failed", stream.id(), cause);
|
||||
streams.remove(stream.id());
|
||||
try {
|
||||
failures.fail(stream.id(), error);
|
||||
} catch (IOException writeFailure) {
|
||||
log.debug("Failed to write RST_STREAM for {}", stream.id(), writeFailure);
|
||||
} finally {
|
||||
streams.release(stream);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -116,6 +116,11 @@ public final class Http2FrameReader {
|
||||
return totalRead != 0 && System.nanoTime() >= frameDeadlineNanos;
|
||||
}
|
||||
|
||||
/** Whether another frame may be consumed immediately without waiting for network input. */
|
||||
public boolean hasBufferedInput() {
|
||||
return totalRead != 0 || in.available() != 0;
|
||||
}
|
||||
|
||||
private static int decodeLength(byte[] buf, int off) {
|
||||
int b0 = buf[off] & 0xFF, b1 = buf[off + 1] & 0xFF, b2 = buf[off + 2] & 0xFF;
|
||||
return (b0 << 16) | (b1 << 8) | b2;
|
||||
|
||||
@@ -306,8 +306,7 @@ public final class Huffman {
|
||||
|
||||
/**
|
||||
* Huffman-encodes {@code src[off, off + len)}, writing directly into {@code out}. Pads the final
|
||||
* byte with the high-order bits of the EOS code (all 1s), per RFC 7541 §5.2. Built now ({@code
|
||||
* EX} task 3) for use by the HPACK encoder.
|
||||
* byte with the high-order bits of the EOS code (all 1s), per RFC 7541 §5.2.
|
||||
*/
|
||||
public static void encode(ByteWriter out, byte[] src, int off, int len) {
|
||||
long accumulator = 0;
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
package dev.relism.flash.http2.message;
|
||||
|
||||
import dev.relism.flash.bytes.PooledSlice;
|
||||
import dev.relism.flash.http2.hpack.HpackHeaderBlock;
|
||||
import dev.relism.flash.models.HeaderView;
|
||||
import dev.relism.fpr.core.ByteView;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/** Header view over stream-owned decoded HPACK storage. Pseudo-fields are excluded. */
|
||||
public final class Http2HeaderMap implements HeaderView {
|
||||
private static final int VIEW_COUNT = 4;
|
||||
|
||||
private final PooledSlice scanName = new PooledSlice();
|
||||
private final PooledSlice scanValue = new PooledSlice();
|
||||
private final PooledSlice[] views = new PooledSlice[VIEW_COUNT];
|
||||
private HpackHeaderBlock block;
|
||||
private PseudoHeaders pseudoHeaders;
|
||||
private int viewCursor;
|
||||
private int regularCount;
|
||||
|
||||
public Http2HeaderMap() {
|
||||
for (int i = 0; i < views.length; i++) views[i] = new PooledSlice();
|
||||
}
|
||||
|
||||
public void reset(HpackHeaderBlock block, PseudoHeaders pseudoHeaders) {
|
||||
this.block = block;
|
||||
this.pseudoHeaders = pseudoHeaders;
|
||||
viewCursor = 0;
|
||||
regularCount = 0;
|
||||
for (int i = 0; i < block.count(); i++) {
|
||||
block.get(i, scanName, scanValue);
|
||||
if (scanName.byteAt(0) != ':') regularCount++;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public String first(String name) {
|
||||
ByteView value = find(name, scanValue);
|
||||
if (value == null) return null;
|
||||
PooledSlice slice = (PooledSlice) value;
|
||||
return new String(slice.array(), slice.offset(), slice.length(), StandardCharsets.UTF_8);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> all(String name) {
|
||||
List<String> result = null;
|
||||
for (int i = 0; i < block.count(); i++) {
|
||||
block.get(i, scanName, scanValue);
|
||||
if (scanName.byteAt(0) == ':' || !equalsIgnoreCase(scanName, name)) continue;
|
||||
if (result == null) result = new ArrayList<>();
|
||||
result.add(
|
||||
new String(
|
||||
scanValue.array(), scanValue.offset(), scanValue.length(), StandardCharsets.UTF_8));
|
||||
}
|
||||
if (result == null && isAuthorityAlias(name) && pseudoHeaders.authority().array() != null) {
|
||||
return List.of(string(pseudoHeaders.authority()));
|
||||
}
|
||||
return result == null ? List.of() : result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> all() {
|
||||
List<String> result = new ArrayList<>(regularCount);
|
||||
for (int i = 0; i < block.count(); i++) {
|
||||
block.get(i, scanName, scanValue);
|
||||
if (scanName.byteAt(0) != ':') result.add(string(scanValue));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ByteView view(String name) {
|
||||
PooledSlice target = views[viewCursor++ & (views.length - 1)];
|
||||
return find(name, target);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean valueEqualsIgnoreCase(String name, String value) {
|
||||
ByteView found = find(name, scanValue);
|
||||
return found != null && equalsIgnoreCase(found, value);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean contains(String name) {
|
||||
return find(name, scanValue) != null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int count() {
|
||||
return regularCount;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void forEach(HeaderConsumer consumer) {
|
||||
for (int i = 0; i < block.count(); i++) {
|
||||
block.get(i, scanName, scanValue);
|
||||
if (scanName.byteAt(0) != ':') consumer.accept(scanName, scanValue);
|
||||
}
|
||||
}
|
||||
|
||||
private PooledSlice find(String requested, PooledSlice target) {
|
||||
for (int i = 0; i < block.count(); i++) {
|
||||
block.get(i, scanName, scanValue);
|
||||
if (scanName.byteAt(0) != ':' && equalsIgnoreCase(scanName, requested)) {
|
||||
target.reset(scanValue.array(), scanValue.offset(), scanValue.length());
|
||||
return target;
|
||||
}
|
||||
}
|
||||
if (isAuthorityAlias(requested) && pseudoHeaders.authority().array() != null) {
|
||||
PooledSlice authority = pseudoHeaders.authority();
|
||||
target.reset(authority.array(), authority.offset(), authority.length());
|
||||
return target;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private static boolean isAuthorityAlias(String name) {
|
||||
return name.equalsIgnoreCase("host") || name.equalsIgnoreCase(":authority");
|
||||
}
|
||||
|
||||
private static boolean equalsIgnoreCase(ByteView bytes, String value) {
|
||||
if (bytes.length() != value.length()) return false;
|
||||
for (int i = 0; i < bytes.length(); i++) {
|
||||
int left = bytes.byteAt(i) & 0xff;
|
||||
int right = value.charAt(i);
|
||||
if (left >= 'A' && left <= 'Z') left += 32;
|
||||
if (right >= 'A' && right <= 'Z') right += 32;
|
||||
if (left != right) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private static String string(PooledSlice slice) {
|
||||
return new String(slice.array(), slice.offset(), slice.length(), StandardCharsets.UTF_8);
|
||||
}
|
||||
}
|
||||
@@ -19,6 +19,11 @@ import dev.relism.flash.models.ResponseSerializer;
|
||||
* connection write lock and exposes it as one {@link WriteIntent}.
|
||||
*/
|
||||
public final class Http2ResponseWriter implements WriteIntent, ResponseSerializer.FieldConsumer {
|
||||
@FunctionalInterface
|
||||
public interface Completion {
|
||||
void responseWriteCompleted();
|
||||
}
|
||||
|
||||
private static final int STATUS_NAME_LENGTH = 7;
|
||||
private static final int CONTENT_LENGTH_NAME_LENGTH = 14;
|
||||
|
||||
@@ -31,6 +36,7 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize
|
||||
private int streamId;
|
||||
private long headerListSize;
|
||||
private long maxHeaderListSize;
|
||||
private Completion completion;
|
||||
|
||||
public Http2ResponseWriter() {
|
||||
this(1024, 2048);
|
||||
@@ -42,6 +48,10 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize
|
||||
frames = new FrameWriteBuffer(output);
|
||||
}
|
||||
|
||||
public void completion(Completion completion) {
|
||||
this.completion = completion;
|
||||
}
|
||||
|
||||
/**
|
||||
* Prepares a non-streaming response. Returns {@code false} when the body needs the deferred DATA
|
||||
* flow-control path implemented by the stream scheduler.
|
||||
@@ -236,4 +246,9 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize
|
||||
public void setMpscNext(WriteIntent next) {
|
||||
this.next = next;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void completed() {
|
||||
if (completion != null) completion.responseWriteCompleted();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
package dev.relism.flash.http2.message;
|
||||
|
||||
import dev.relism.flash.bytes.PooledSlice;
|
||||
import dev.relism.flash.http2.Http2ErrorCode;
|
||||
import dev.relism.flash.http2.Http2StreamException;
|
||||
import dev.relism.flash.http2.hpack.HpackHeaderBlock;
|
||||
import dev.relism.fpr.core.ByteView;
|
||||
|
||||
/** Validates request pseudo-headers and HTTP/2 field rules while extracting request metadata. */
|
||||
public final class PseudoHeaders {
|
||||
private static final int METHOD = 1;
|
||||
private static final int SCHEME = 2;
|
||||
private static final int PATH = 4;
|
||||
private static final int AUTHORITY = 8;
|
||||
|
||||
private final PooledSlice name = new PooledSlice();
|
||||
private final PooledSlice value = new PooledSlice();
|
||||
private final PooledSlice method = new PooledSlice();
|
||||
private final PooledSlice scheme = new PooledSlice();
|
||||
private final PooledSlice path = new PooledSlice();
|
||||
private final PooledSlice authority = new PooledSlice();
|
||||
private final PooledSlice host = new PooledSlice();
|
||||
private int present;
|
||||
|
||||
public void validate(HpackHeaderBlock block, int streamId) {
|
||||
present = 0;
|
||||
method.reset(null, 0, 0);
|
||||
scheme.reset(null, 0, 0);
|
||||
path.reset(null, 0, 0);
|
||||
authority.reset(null, 0, 0);
|
||||
host.reset(null, 0, 0);
|
||||
boolean regularSeen = false;
|
||||
|
||||
for (int i = 0; i < block.count(); i++) {
|
||||
block.get(i, name, value);
|
||||
if (name.length() == 0) fail(streamId, "empty field name");
|
||||
boolean pseudo = name.byteAt(0) == ':';
|
||||
if (pseudo) {
|
||||
if (regularSeen) fail(streamId, "pseudo-header after regular field");
|
||||
int bit = pseudoBit(name);
|
||||
if (bit == 0) fail(streamId, "unknown pseudo-header");
|
||||
if ((present & bit) != 0) fail(streamId, "duplicate pseudo-header");
|
||||
present |= bit;
|
||||
copySlice(bit, value);
|
||||
} else {
|
||||
regularSeen = true;
|
||||
validateRegular(name, value, streamId);
|
||||
if (equals(name, "host")) copy(value, host);
|
||||
}
|
||||
}
|
||||
|
||||
if ((present & METHOD) == 0) fail(streamId, "missing :method");
|
||||
boolean connect = equals(method, "CONNECT");
|
||||
if (connect) {
|
||||
if ((present & AUTHORITY) == 0) fail(streamId, "CONNECT requires :authority");
|
||||
if ((present & (SCHEME | PATH)) != 0) fail(streamId, "CONNECT forbids :scheme and :path");
|
||||
} else {
|
||||
int required = METHOD | SCHEME | PATH | AUTHORITY;
|
||||
if ((present & required) != required) fail(streamId, "missing request pseudo-header");
|
||||
if (path.length() == 0) fail(streamId, "empty :path");
|
||||
}
|
||||
if (host.array() != null && authority.array() != null && !equals(host, authority)) {
|
||||
fail(streamId, "host conflicts with :authority");
|
||||
}
|
||||
}
|
||||
|
||||
public PooledSlice method() {
|
||||
return method;
|
||||
}
|
||||
|
||||
public PooledSlice scheme() {
|
||||
return scheme;
|
||||
}
|
||||
|
||||
public PooledSlice path() {
|
||||
return path;
|
||||
}
|
||||
|
||||
public PooledSlice authority() {
|
||||
return authority;
|
||||
}
|
||||
|
||||
private void copySlice(int bit, PooledSlice source) {
|
||||
if (bit == METHOD) copy(source, method);
|
||||
else if (bit == SCHEME) copy(source, scheme);
|
||||
else if (bit == PATH) copy(source, path);
|
||||
else copy(source, authority);
|
||||
}
|
||||
|
||||
private static void copy(PooledSlice source, PooledSlice target) {
|
||||
target.reset(source.array(), source.offset(), source.length());
|
||||
}
|
||||
|
||||
private static int pseudoBit(ByteView name) {
|
||||
if (equals(name, ":method")) return METHOD;
|
||||
if (equals(name, ":scheme")) return SCHEME;
|
||||
if (equals(name, ":path")) return PATH;
|
||||
if (equals(name, ":authority")) return AUTHORITY;
|
||||
return 0;
|
||||
}
|
||||
|
||||
private static void validateRegular(ByteView name, ByteView value, int streamId) {
|
||||
for (int i = 0; i < name.length(); i++) {
|
||||
int c = name.byteAt(i) & 0xff;
|
||||
if (c >= 'A' && c <= 'Z') fail(streamId, "uppercase field name");
|
||||
}
|
||||
if (equals(name, "connection")
|
||||
|| equals(name, "keep-alive")
|
||||
|| equals(name, "proxy-connection")
|
||||
|| equals(name, "transfer-encoding")
|
||||
|| equals(name, "upgrade")) {
|
||||
fail(streamId, "connection-specific field");
|
||||
}
|
||||
if (equals(name, "te") && !equals(value, "trailers")) {
|
||||
fail(streamId, "invalid te field");
|
||||
}
|
||||
}
|
||||
|
||||
static boolean equals(ByteView view, String expected) {
|
||||
if (view.length() != expected.length()) return false;
|
||||
for (int i = 0; i < view.length(); i++) {
|
||||
if ((view.byteAt(i) & 0xff) != expected.charAt(i)) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
static boolean equals(ByteView left, ByteView right) {
|
||||
if (left.length() != right.length()) return false;
|
||||
for (int i = 0; i < left.length(); i++) {
|
||||
if (left.byteAt(i) != right.byteAt(i)) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private static void fail(int streamId, String message) {
|
||||
throw new Http2StreamException(streamId, Http2ErrorCode.PROTOCOL_ERROR, message);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
package dev.relism.flash.http2.stream;
|
||||
|
||||
import dev.relism.flash.bytes.PooledSlice;
|
||||
import dev.relism.flash.http.ContentType;
|
||||
import dev.relism.flash.http.HttpMethod;
|
||||
import dev.relism.flash.http2.Http2ErrorCode;
|
||||
import dev.relism.flash.http2.Http2StreamException;
|
||||
import dev.relism.flash.http2.hpack.HpackHeaderBlock;
|
||||
import dev.relism.flash.http2.message.Http2HeaderMap;
|
||||
import dev.relism.flash.http2.message.Http2ResponseWriter;
|
||||
import dev.relism.flash.http2.message.PseudoHeaders;
|
||||
import dev.relism.flash.models.Request;
|
||||
import dev.relism.flash.models.RequestBody;
|
||||
import dev.relism.flash.models.RequestLine;
|
||||
import dev.relism.flash.models.Response;
|
||||
import dev.relism.flash.routing.AbstractRouter;
|
||||
import java.net.InetSocketAddress;
|
||||
import javax.net.ssl.SSLSocket;
|
||||
|
||||
/** Per-stream request, response, decoded-header and write state. */
|
||||
public final class Http2Stream implements Http2ResponseWriter.Completion {
|
||||
private static final byte[] HTTP_2 = {'H', 'T', 'T', 'P', '/', '2'};
|
||||
|
||||
private final HpackHeaderBlock headerBlock = new HpackHeaderBlock();
|
||||
private final PseudoHeaders pseudoHeaders = new PseudoHeaders();
|
||||
private final Http2HeaderMap headers = new Http2HeaderMap();
|
||||
private final RequestLine requestLine = new RequestLine();
|
||||
private final RequestBody requestBody = new RequestBody();
|
||||
private final Request request = new Request();
|
||||
private final Response response = new Response(200, ContentType.TEXT_PLAIN);
|
||||
private final Http2ResponseWriter responseWriter = new Http2ResponseWriter();
|
||||
private final PooledSlice path = new PooledSlice();
|
||||
private final PooledSlice query = new PooledSlice();
|
||||
private final PooledSlice protocol = new PooledSlice();
|
||||
|
||||
private int id;
|
||||
private Http2StreamState state = Http2StreamState.IDLE;
|
||||
private int sendWindow = 65_535;
|
||||
private Http2StreamTable owner;
|
||||
private Object routeScratch;
|
||||
private volatile boolean dispatched;
|
||||
private volatile boolean cancelled;
|
||||
private boolean headersValidated;
|
||||
Http2Stream poolNext;
|
||||
|
||||
Http2Stream() {
|
||||
responseWriter.completion(this);
|
||||
protocol.reset(HTTP_2, 0, HTTP_2.length);
|
||||
}
|
||||
|
||||
void reset(int id, Http2StreamTable owner) {
|
||||
this.id = id;
|
||||
this.owner = owner;
|
||||
state = Http2StreamState.IDLE;
|
||||
sendWindow = 65_535;
|
||||
dispatched = false;
|
||||
cancelled = false;
|
||||
headersValidated = false;
|
||||
headerBlock.reset();
|
||||
}
|
||||
|
||||
void clear() {
|
||||
request.recycle();
|
||||
response.recycle();
|
||||
id = 0;
|
||||
owner = null;
|
||||
state = Http2StreamState.CLOSED;
|
||||
}
|
||||
|
||||
public Request assembleRequest(InetSocketAddress remoteAddress, SSLSocket sslSocket) {
|
||||
validateHeaders();
|
||||
headers.reset(headerBlock, pseudoHeaders);
|
||||
PooledSlice rawPath = pseudoHeaders.path();
|
||||
if (rawPath.array() == null) rawPath = pseudoHeaders.authority();
|
||||
int question = -1;
|
||||
for (int i = 0; i < rawPath.length(); i++) {
|
||||
if (rawPath.byteAt(i) == '?') {
|
||||
question = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (question < 0) {
|
||||
path.reset(rawPath.array(), rawPath.offset(), rawPath.length());
|
||||
query.reset(null, 0, 0);
|
||||
} else {
|
||||
path.reset(rawPath.array(), rawPath.offset(), question);
|
||||
query.reset(
|
||||
rawPath.array(), rawPath.offset() + question + 1, rawPath.length() - question - 1);
|
||||
}
|
||||
PooledSlice methodBytes = pseudoHeaders.method();
|
||||
HttpMethod method =
|
||||
HttpMethod.fromBytes(methodBytes.array(), methodBytes.offset(), methodBytes.length());
|
||||
if (method == null) {
|
||||
throw new Http2StreamException(
|
||||
id, Http2ErrorCode.PROTOCOL_ERROR, "unsupported request method");
|
||||
}
|
||||
requestLine.reset(method, path, question < 0 ? null : query, protocol, headers);
|
||||
requestBody.reset(null, 0, null, 0, 0);
|
||||
return Request.forParsed(request, requestLine, requestBody, remoteAddress, sslSocket);
|
||||
}
|
||||
|
||||
public void validateHeaders() {
|
||||
if (headersValidated) return;
|
||||
pseudoHeaders.validate(headerBlock, id);
|
||||
headersValidated = true;
|
||||
}
|
||||
|
||||
public Response resetResponse() {
|
||||
return response.reset(200, ContentType.TEXT_PLAIN);
|
||||
}
|
||||
|
||||
public void transition(Http2StreamState.Event event) {
|
||||
state = state.transition(id, event);
|
||||
}
|
||||
|
||||
public int id() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public Http2StreamState state() {
|
||||
return state;
|
||||
}
|
||||
|
||||
public HpackHeaderBlock headerBlock() {
|
||||
return headerBlock;
|
||||
}
|
||||
|
||||
public Http2ResponseWriter responseWriter() {
|
||||
return responseWriter;
|
||||
}
|
||||
|
||||
public Object routeScratch(AbstractRouter router) {
|
||||
if (routeScratch == null) routeScratch = router.newScratch();
|
||||
return routeScratch;
|
||||
}
|
||||
|
||||
public void markDispatched() {
|
||||
dispatched = true;
|
||||
}
|
||||
|
||||
public boolean dispatched() {
|
||||
return dispatched;
|
||||
}
|
||||
|
||||
public void cancel() {
|
||||
cancelled = true;
|
||||
}
|
||||
|
||||
public boolean cancelled() {
|
||||
return cancelled;
|
||||
}
|
||||
|
||||
public int sendWindow() {
|
||||
return sendWindow;
|
||||
}
|
||||
|
||||
public void adjustSendWindow(int delta) {
|
||||
long adjusted = (long) sendWindow + delta;
|
||||
if (adjusted > Integer.MAX_VALUE) throw new IllegalStateException("stream window overflow");
|
||||
sendWindow = (int) adjusted;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void responseWriteCompleted() {
|
||||
Http2StreamTable table = owner;
|
||||
if (table != null) table.release(this);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
package dev.relism.flash.http2.stream;
|
||||
|
||||
import dev.relism.flash.http2.Http2ErrorCode;
|
||||
import dev.relism.flash.http2.Http2StreamException;
|
||||
|
||||
/** Explicit RFC 9113 stream-state transition table. */
|
||||
public enum Http2StreamState {
|
||||
IDLE,
|
||||
OPEN,
|
||||
HALF_CLOSED_REMOTE,
|
||||
HALF_CLOSED_LOCAL,
|
||||
CLOSED;
|
||||
|
||||
public enum Event {
|
||||
RECV_HEADERS,
|
||||
RECV_HEADERS_ES,
|
||||
RECV_DATA,
|
||||
RECV_DATA_ES,
|
||||
RECV_RST,
|
||||
SEND_HEADERS,
|
||||
SEND_HEADERS_ES,
|
||||
SEND_DATA,
|
||||
SEND_DATA_ES,
|
||||
SEND_RST
|
||||
}
|
||||
|
||||
private static final byte ERROR = -1;
|
||||
private static final byte[][] TRANSITIONS = buildTransitions();
|
||||
|
||||
public Http2StreamState transition(int streamId, Event event) {
|
||||
int next = TRANSITIONS[ordinal()][event.ordinal()];
|
||||
if (next == ERROR) {
|
||||
throw new Http2StreamException(
|
||||
streamId, errorFor(event), "invalid stream transition " + this + " + " + event);
|
||||
}
|
||||
return values()[next];
|
||||
}
|
||||
|
||||
private Http2ErrorCode errorFor(Event event) {
|
||||
if (this == CLOSED) return Http2ErrorCode.STREAM_CLOSED;
|
||||
if (this == HALF_CLOSED_REMOTE
|
||||
&& (event == Event.RECV_DATA
|
||||
|| event == Event.RECV_DATA_ES
|
||||
|| event == Event.RECV_HEADERS
|
||||
|| event == Event.RECV_HEADERS_ES)) {
|
||||
return Http2ErrorCode.STREAM_CLOSED;
|
||||
}
|
||||
return Http2ErrorCode.PROTOCOL_ERROR;
|
||||
}
|
||||
|
||||
public static boolean isValid(Http2StreamState state, Event event) {
|
||||
return TRANSITIONS[state.ordinal()][event.ordinal()] != ERROR;
|
||||
}
|
||||
|
||||
private static byte[][] buildTransitions() {
|
||||
byte[][] table = new byte[values().length][Event.values().length];
|
||||
for (byte[] row : table) java.util.Arrays.fill(row, ERROR);
|
||||
|
||||
set(table, IDLE, Event.RECV_HEADERS, OPEN);
|
||||
set(table, IDLE, Event.RECV_HEADERS_ES, HALF_CLOSED_REMOTE);
|
||||
|
||||
set(table, OPEN, Event.RECV_HEADERS, OPEN);
|
||||
set(table, OPEN, Event.RECV_HEADERS_ES, HALF_CLOSED_REMOTE);
|
||||
set(table, OPEN, Event.RECV_DATA, OPEN);
|
||||
set(table, OPEN, Event.RECV_DATA_ES, HALF_CLOSED_REMOTE);
|
||||
set(table, OPEN, Event.RECV_RST, CLOSED);
|
||||
set(table, OPEN, Event.SEND_HEADERS, OPEN);
|
||||
set(table, OPEN, Event.SEND_HEADERS_ES, HALF_CLOSED_LOCAL);
|
||||
set(table, OPEN, Event.SEND_DATA, OPEN);
|
||||
set(table, OPEN, Event.SEND_DATA_ES, HALF_CLOSED_LOCAL);
|
||||
set(table, OPEN, Event.SEND_RST, CLOSED);
|
||||
|
||||
set(table, HALF_CLOSED_REMOTE, Event.RECV_RST, CLOSED);
|
||||
set(table, HALF_CLOSED_REMOTE, Event.SEND_HEADERS, HALF_CLOSED_REMOTE);
|
||||
set(table, HALF_CLOSED_REMOTE, Event.SEND_HEADERS_ES, CLOSED);
|
||||
set(table, HALF_CLOSED_REMOTE, Event.SEND_DATA, HALF_CLOSED_REMOTE);
|
||||
set(table, HALF_CLOSED_REMOTE, Event.SEND_DATA_ES, CLOSED);
|
||||
set(table, HALF_CLOSED_REMOTE, Event.SEND_RST, CLOSED);
|
||||
|
||||
set(table, HALF_CLOSED_LOCAL, Event.RECV_HEADERS, HALF_CLOSED_LOCAL);
|
||||
set(table, HALF_CLOSED_LOCAL, Event.RECV_HEADERS_ES, CLOSED);
|
||||
set(table, HALF_CLOSED_LOCAL, Event.RECV_DATA, HALF_CLOSED_LOCAL);
|
||||
set(table, HALF_CLOSED_LOCAL, Event.RECV_DATA_ES, CLOSED);
|
||||
set(table, HALF_CLOSED_LOCAL, Event.RECV_RST, CLOSED);
|
||||
set(table, HALF_CLOSED_LOCAL, Event.SEND_RST, CLOSED);
|
||||
|
||||
set(table, CLOSED, Event.RECV_RST, CLOSED);
|
||||
set(table, CLOSED, Event.SEND_RST, CLOSED);
|
||||
return table;
|
||||
}
|
||||
|
||||
private static void set(byte[][] table, Http2StreamState from, Event event, Http2StreamState to) {
|
||||
table[from.ordinal()][event.ordinal()] = (byte) to.ordinal();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,145 @@
|
||||
package dev.relism.flash.http2.stream;
|
||||
|
||||
import java.util.Arrays;
|
||||
|
||||
/** Fixed-capacity primitive stream-id table using linear-probed open addressing. */
|
||||
public final class Http2StreamTable {
|
||||
|
||||
@FunctionalInterface
|
||||
public interface StreamConsumer {
|
||||
void accept(Http2Stream stream);
|
||||
}
|
||||
|
||||
private final int[] keys;
|
||||
private final Http2Stream[] values;
|
||||
private final int mask;
|
||||
private final int maxEntries;
|
||||
private int size;
|
||||
private Http2Stream free;
|
||||
private int created;
|
||||
|
||||
public Http2StreamTable(int maxEntries) {
|
||||
if (maxEntries < 1) throw new IllegalArgumentException("maxEntries must be positive");
|
||||
int capacity = 1;
|
||||
while (capacity < maxEntries * 2) capacity <<= 1;
|
||||
keys = new int[capacity];
|
||||
values = new Http2Stream[capacity];
|
||||
mask = capacity - 1;
|
||||
this.maxEntries = maxEntries;
|
||||
}
|
||||
|
||||
public synchronized Http2Stream get(int streamId) {
|
||||
int slot = find(streamId);
|
||||
return keys[slot] == streamId ? values[slot] : null;
|
||||
}
|
||||
|
||||
public synchronized void put(Http2Stream stream) {
|
||||
if (size == maxEntries) throw new IllegalStateException("stream table capacity exceeded");
|
||||
int streamId = stream.id();
|
||||
int slot = find(streamId);
|
||||
if (keys[slot] == streamId) throw new IllegalStateException("duplicate stream " + streamId);
|
||||
keys[slot] = streamId;
|
||||
values[slot] = stream;
|
||||
size++;
|
||||
}
|
||||
|
||||
public synchronized Http2Stream acquire(int streamId) {
|
||||
if (size == maxEntries) return null;
|
||||
Http2Stream stream = free;
|
||||
if (stream != null) {
|
||||
free = stream.poolNext;
|
||||
stream.poolNext = null;
|
||||
} else {
|
||||
if (created == maxEntries) return null;
|
||||
stream = new Http2Stream();
|
||||
created++;
|
||||
}
|
||||
stream.reset(streamId, this);
|
||||
put(stream);
|
||||
return stream;
|
||||
}
|
||||
|
||||
public synchronized void release(Http2Stream stream) {
|
||||
stream.clear();
|
||||
stream.poolNext = free;
|
||||
free = stream;
|
||||
}
|
||||
|
||||
public synchronized Http2Stream remove(int streamId) {
|
||||
int slot = find(streamId);
|
||||
if (keys[slot] != streamId) return null;
|
||||
Http2Stream removed = values[slot];
|
||||
keys[slot] = 0;
|
||||
values[slot] = null;
|
||||
size--;
|
||||
|
||||
int scan = (slot + 1) & mask;
|
||||
while (keys[scan] != 0) {
|
||||
int key = keys[scan];
|
||||
Http2Stream value = values[scan];
|
||||
keys[scan] = 0;
|
||||
values[scan] = null;
|
||||
size--;
|
||||
put(value);
|
||||
scan = (scan + 1) & mask;
|
||||
}
|
||||
return removed;
|
||||
}
|
||||
|
||||
public synchronized void forEach(StreamConsumer consumer) {
|
||||
for (int i = 0; i < keys.length; i++) {
|
||||
if (keys[i] != 0) consumer.accept(values[i]);
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized void adjustAllSendWindows(int delta) {
|
||||
for (int i = 0; i < keys.length; i++) {
|
||||
if (keys[i] == 0) continue;
|
||||
long adjusted = (long) values[i].sendWindow() + delta;
|
||||
if (adjusted > Integer.MAX_VALUE) {
|
||||
throw new IllegalStateException("stream window overflow");
|
||||
}
|
||||
}
|
||||
for (int i = 0; i < keys.length; i++) {
|
||||
if (keys[i] != 0) values[i].adjustSendWindow(delta);
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized int size() {
|
||||
return size;
|
||||
}
|
||||
|
||||
public int capacity() {
|
||||
return maxEntries;
|
||||
}
|
||||
|
||||
public synchronized int createdCount() {
|
||||
return created;
|
||||
}
|
||||
|
||||
public synchronized int freeCount() {
|
||||
int count = 0;
|
||||
for (Http2Stream stream = free; stream != null; stream = stream.poolNext) count++;
|
||||
return count;
|
||||
}
|
||||
|
||||
public synchronized void clear() {
|
||||
Arrays.fill(keys, 0);
|
||||
Arrays.fill(values, null);
|
||||
size = 0;
|
||||
}
|
||||
|
||||
private int find(int streamId) {
|
||||
if (streamId <= 0) throw new IllegalArgumentException("streamId must be positive");
|
||||
int slot = mix(streamId) & mask;
|
||||
while (keys[slot] != 0 && keys[slot] != streamId) slot = (slot + 1) & mask;
|
||||
return slot;
|
||||
}
|
||||
|
||||
private static int mix(int value) {
|
||||
value ^= value >>> 16;
|
||||
value *= 0x7feb352d;
|
||||
value ^= value >>> 15;
|
||||
return value;
|
||||
}
|
||||
}
|
||||
@@ -3,34 +3,32 @@ package dev.relism.flash.transport;
|
||||
import dev.relism.flash.extension.FlashConfiguration;
|
||||
import dev.relism.flash.routing.AbstractRouter;
|
||||
import dev.relism.flash.routing.AbstractWsRouter;
|
||||
|
||||
import javax.net.ssl.SSLSocket;
|
||||
|
||||
import java.io.OutputStream;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.Socket;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.function.BooleanSupplier;
|
||||
import javax.net.ssl.SSLSocket;
|
||||
|
||||
/**
|
||||
* Everything a {@link ConnectionProtocol} implementation needs to serve one connection, bundled
|
||||
* into a single object instead of a long parameter list.
|
||||
*
|
||||
* @param socket the accepted socket — owns its lifecycle (closing it is the caller's,
|
||||
* i.e. {@link ConnectionRunner}'s, responsibility, not the protocol's)
|
||||
* @param sslSocket {@code socket} narrowed to {@link SSLSocket}, or {@code null} for a
|
||||
* plaintext connection
|
||||
* @param in the single buffered, deadline-aware source for this connection's
|
||||
* inbound bytes
|
||||
* @param out the buffered output stream — for header/body writes that benefit from
|
||||
* userspace coalescing before a single syscall
|
||||
* @param rawOut the unbuffered output stream — for WebSocket, whose writes are already
|
||||
* bulk (see {@code WebSocketSession})
|
||||
* @param socket the accepted socket — owns its lifecycle (closing it is the caller's, i.e. {@link
|
||||
* ConnectionRunner}'s, responsibility, not the protocol's)
|
||||
* @param sslSocket {@code socket} narrowed to {@link SSLSocket}, or {@code null} for a plaintext
|
||||
* connection
|
||||
* @param in the single buffered, deadline-aware source for this connection's inbound bytes
|
||||
* @param out the buffered output stream — for header/body writes that benefit from userspace
|
||||
* coalescing before a single syscall
|
||||
* @param rawOut the unbuffered output stream — for WebSocket, whose writes are already bulk (see
|
||||
* {@code WebSocketSession})
|
||||
* @param remoteAddress the client's address, or {@code null} if unavailable
|
||||
* @param router the HTTP router
|
||||
* @param wsRouter the WebSocket router
|
||||
* @param configuration the server configuration (timeouts, limits, feature flags)
|
||||
* @param stopped {@code true} once the server has begun shutting down — a protocol
|
||||
* implementation's request loop must check this and exit promptly
|
||||
* @param stopped {@code true} once the server has begun shutting down — a protocol implementation's
|
||||
* request loop must check this and exit promptly
|
||||
*/
|
||||
public record ConnectionContext(
|
||||
Socket socket,
|
||||
@@ -43,6 +41,5 @@ public record ConnectionContext(
|
||||
AbstractRouter router,
|
||||
AbstractWsRouter wsRouter,
|
||||
FlashConfiguration configuration,
|
||||
BooleanSupplier stopped
|
||||
) {
|
||||
}
|
||||
ExecutorService executor,
|
||||
BooleanSupplier stopped) {}
|
||||
|
||||
@@ -113,6 +113,7 @@ public final class ConnectionRunner {
|
||||
router,
|
||||
wsRouter,
|
||||
configuration,
|
||||
executorService,
|
||||
stopped);
|
||||
if (negotiated == NegotiatedProtocol.HTTP_2) http2ProtocolFactory.get().run(ctx);
|
||||
else http1Protocol.run(ctx);
|
||||
|
||||
@@ -2,22 +2,32 @@ package dev.relism.flash.http2;
|
||||
|
||||
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.http2.frame.FrameFlags;
|
||||
import dev.relism.flash.http2.frame.FrameType;
|
||||
import dev.relism.flash.http2.hpack.HpackDecoder;
|
||||
import dev.relism.flash.http2.hpack.HpackEncoder;
|
||||
import dev.relism.flash.tls.TestKeystores;
|
||||
import dev.relism.flash.tls.TlsConfig;
|
||||
import java.io.EOFException;
|
||||
import java.io.InputStream;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.Socket;
|
||||
import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.HttpRequest;
|
||||
import java.net.http.HttpResponse;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.nio.file.Path;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import javax.net.ssl.SSLParameters;
|
||||
import javax.net.ssl.SSLSocket;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
@@ -32,6 +42,171 @@ class Http2ConnectionIntegrationTest {
|
||||
if (app != null) app.stop().join();
|
||||
}
|
||||
|
||||
@Test
|
||||
void javaHttpClientUsesHttp2AgainstAnExistingParameterizedRoute(@TempDir Path directory)
|
||||
throws Exception {
|
||||
int port = freePort();
|
||||
Path keystore =
|
||||
TestKeystores.build(
|
||||
directory,
|
||||
"http2-route.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(
|
||||
"/users/{id}", (request, response) -> request.param("id") + ":" + request.header("host"));
|
||||
app.start();
|
||||
|
||||
HttpClient client =
|
||||
HttpClient.newBuilder()
|
||||
.sslContext(TestKeystores.trustAllClientContext())
|
||||
.version(HttpClient.Version.HTTP_2)
|
||||
.build();
|
||||
HttpResponse<String> response =
|
||||
client.send(
|
||||
HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/users/42"))
|
||||
.GET()
|
||||
.build(),
|
||||
HttpResponse.BodyHandlers.ofString());
|
||||
|
||||
assertEquals(HttpClient.Version.HTTP_2, response.version());
|
||||
assertEquals(200, response.statusCode());
|
||||
assertEquals("42:localhost:" + port, response.body());
|
||||
}
|
||||
|
||||
@Test
|
||||
void bodylessGetRunsExistingRouteAndReturnsHeadersAndData() throws Exception {
|
||||
int port = freePort();
|
||||
app =
|
||||
FlashApp.create(
|
||||
FlashConfiguration.builder().port(port).host("127.0.0.1").http2Enabled(true).build());
|
||||
app.get("/api/ping", (request, response) -> "pong");
|
||||
app.start();
|
||||
|
||||
ByteWriter block = new ByteWriter(64);
|
||||
HpackEncoder.writeIndexed(block, 2);
|
||||
HpackEncoder.writeIndexed(block, 6);
|
||||
HpackEncoder.writeLiteralWithNameIndex(
|
||||
block, 4, "/api/ping".getBytes(StandardCharsets.US_ASCII), false);
|
||||
HpackEncoder.writeLiteralWithNameIndex(
|
||||
block, 1, "localhost".getBytes(StandardCharsets.US_ASCII), 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 | FrameFlags.END_STREAM,
|
||||
1,
|
||||
Arrays.copyOf(block.array(), block.length()))));
|
||||
socket.getOutputStream().flush();
|
||||
|
||||
ByteWriter responseBlock = new ByteWriter(128);
|
||||
byte[] body = null;
|
||||
for (int i = 0; i < 10 && body == null; i++) {
|
||||
Http2TestFrames.WireFrame frame = readFrame(socket.getInputStream());
|
||||
if (frame.streamId() != 1) continue;
|
||||
if (frame.type() == FrameType.HEADERS.code()
|
||||
|| frame.type() == FrameType.CONTINUATION.code()) {
|
||||
responseBlock.writeBytes(frame.payload());
|
||||
} else if (frame.type() == FrameType.DATA.code()) {
|
||||
body = frame.payload();
|
||||
}
|
||||
}
|
||||
|
||||
List<String> fields = new ArrayList<>();
|
||||
new HpackDecoder()
|
||||
.decode(
|
||||
responseBlock.array(),
|
||||
0,
|
||||
responseBlock.length(),
|
||||
(name, value, never) -> fields.add(ascii(name) + "=" + ascii(value)));
|
||||
assertTrue(fields.contains(":status=200"));
|
||||
assertTrue(fields.contains("content-length=4"));
|
||||
assertEquals("pong", new String(body, StandardCharsets.US_ASCII));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void resetQueuedStreamCannotReleaseOrReuseItBeforeDispatchObservesCancellation()
|
||||
throws Exception {
|
||||
int port = freePort();
|
||||
AtomicInteger calls = new AtomicInteger();
|
||||
app =
|
||||
FlashApp.create(
|
||||
FlashConfiguration.builder().port(port).host("127.0.0.1").http2Enabled(true).build());
|
||||
app.get(
|
||||
"/queued",
|
||||
(request, response) -> {
|
||||
calls.incrementAndGet();
|
||||
return "ok";
|
||||
});
|
||||
app.start();
|
||||
|
||||
ByteWriter block = new ByteWriter(64);
|
||||
HpackEncoder.writeIndexed(block, 2);
|
||||
HpackEncoder.writeIndexed(block, 6);
|
||||
HpackEncoder.writeLiteralWithNameIndex(
|
||||
block, 4, "/queued".getBytes(StandardCharsets.US_ASCII), false);
|
||||
HpackEncoder.writeLiteralWithNameIndex(
|
||||
block, 1, "localhost".getBytes(StandardCharsets.US_ASCII), false);
|
||||
byte[] headers = Arrays.copyOf(block.array(), block.length());
|
||||
byte[] cancel = {0, 0, 0, 8};
|
||||
|
||||
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 | FrameFlags.END_STREAM,
|
||||
1,
|
||||
headers),
|
||||
Http2TestFrames.frame(FrameType.RST_STREAM, 0, 1, cancel),
|
||||
Http2TestFrames.frame(
|
||||
FrameType.HEADERS,
|
||||
FrameFlags.END_HEADERS | FrameFlags.END_STREAM,
|
||||
3,
|
||||
headers)));
|
||||
socket.getOutputStream().flush();
|
||||
|
||||
Http2TestFrames.WireFrame response = null;
|
||||
for (int i = 0; i < 10; i++) {
|
||||
Http2TestFrames.WireFrame frame = readFrame(socket.getInputStream());
|
||||
assertFalse(
|
||||
frame.streamId() == 1
|
||||
&& (frame.type() == FrameType.HEADERS.code()
|
||||
|| frame.type() == FrameType.DATA.code()),
|
||||
"a reset request must not produce a response");
|
||||
if (frame.streamId() == 3 && frame.type() == FrameType.DATA.code()) {
|
||||
response = frame;
|
||||
break;
|
||||
}
|
||||
}
|
||||
assertNotNull(response);
|
||||
assertEquals("ok", new String(response.payload(), StandardCharsets.US_ASCII));
|
||||
assertEquals(1, calls.get());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void h2cTransportDispatchesControlFramesWithoutRunningApplicationWork() throws Exception {
|
||||
int port = freePort();
|
||||
@@ -244,4 +419,10 @@ class Http2ConnectionIntegrationTest {
|
||||
return socket.getLocalPort();
|
||||
}
|
||||
}
|
||||
|
||||
private static String ascii(dev.relism.fpr.core.ByteView view) {
|
||||
byte[] bytes = new byte[view.length()];
|
||||
for (int i = 0; i < bytes.length; i++) bytes[i] = view.byteAt(i);
|
||||
return new String(bytes, StandardCharsets.US_ASCII);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
package dev.relism.flash.http2.message;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
|
||||
import dev.relism.flash.bytes.PooledSlice;
|
||||
import dev.relism.flash.http2.Http2StreamException;
|
||||
import dev.relism.flash.http2.hpack.HpackHeaderBlock;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class PseudoHeaderValidationTest {
|
||||
@Test
|
||||
void validRequest() {
|
||||
assertDoesNotThrow(
|
||||
() ->
|
||||
validate(
|
||||
":method", "GET", ":scheme", "https", ":path", "/", ":authority", "example.com"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsPseudoAfterRegular() {
|
||||
rejects("x", "1", ":method", "GET", ":scheme", "https", ":path", "/", ":authority", "x");
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsUnknownAndDuplicatePseudoHeaders() {
|
||||
rejects(":method", "GET", ":scheme", "https", ":path", "/", ":authority", "x", ":other", "x");
|
||||
rejects(
|
||||
":method", "GET", ":method", "POST", ":scheme", "https", ":path", "/", ":authority", "x");
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsMissingAndEmptyPseudoHeaders() {
|
||||
rejects(":method", "GET", ":scheme", "https", ":path", "/");
|
||||
rejects(":method", "GET", ":scheme", "https", ":path", "", ":authority", "x");
|
||||
}
|
||||
|
||||
@Test
|
||||
void validatesConnectShape() {
|
||||
assertDoesNotThrow(() -> validate(":method", "CONNECT", ":authority", "example.com:443"));
|
||||
rejects(":method", "CONNECT", ":scheme", "https", ":authority", "example.com:443");
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsUppercaseForbiddenAndInvalidTeFields() {
|
||||
rejects(validWith("X-Test", "1"));
|
||||
rejects(validWith("connection", "close"));
|
||||
rejects(validWith("keep-alive", "timeout=5"));
|
||||
rejects(validWith("proxy-connection", "close"));
|
||||
rejects(validWith("transfer-encoding", "chunked"));
|
||||
rejects(validWith("upgrade", "websocket"));
|
||||
rejects(validWith("te", "gzip"));
|
||||
assertDoesNotThrow(() -> validate(validWith("te", "trailers")));
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsHostAuthorityConflict() {
|
||||
rejects(validWith("host", "other.example"));
|
||||
assertDoesNotThrow(() -> validate(validWith("host", "example.com")));
|
||||
}
|
||||
|
||||
private static String[] validWith(String name, String value) {
|
||||
return new String[] {
|
||||
":method", "GET", ":scheme", "https", ":path", "/", ":authority", "example.com", name, value
|
||||
};
|
||||
}
|
||||
|
||||
private static void rejects(String... fields) {
|
||||
assertThrows(Http2StreamException.class, () -> validate(fields));
|
||||
}
|
||||
|
||||
private static void validate(String... fields) {
|
||||
HpackHeaderBlock block = new HpackHeaderBlock();
|
||||
PooledSlice name = new PooledSlice();
|
||||
PooledSlice value = new PooledSlice();
|
||||
for (int i = 0; i < fields.length; i += 2) {
|
||||
byte[] nameBytes = fields[i].getBytes(StandardCharsets.US_ASCII);
|
||||
byte[] valueBytes = fields[i + 1].getBytes(StandardCharsets.US_ASCII);
|
||||
name.reset(nameBytes, 0, nameBytes.length);
|
||||
value.reset(valueBytes, 0, valueBytes.length);
|
||||
block.accept(name, value, false);
|
||||
}
|
||||
new PseudoHeaders().validate(block, 1);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package dev.relism.flash.http2.stream;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
|
||||
import dev.relism.flash.bytes.PooledSlice;
|
||||
import dev.relism.flash.http.HttpMethod;
|
||||
import dev.relism.flash.models.Request;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class Http2RequestAssemblyTest {
|
||||
@Test
|
||||
void assemblesProtocolNeutralRequestWithQueryAndAuthorityAlias() {
|
||||
Http2StreamTable table = new Http2StreamTable(1);
|
||||
Http2Stream stream = table.acquire(1);
|
||||
field(stream, ":method", "GET");
|
||||
field(stream, ":scheme", "https");
|
||||
field(stream, ":path", "/users/42?verbose=true");
|
||||
field(stream, ":authority", "example.com");
|
||||
field(stream, "x-trace", "abc");
|
||||
|
||||
Request request = stream.assembleRequest(null, null);
|
||||
|
||||
assertEquals(HttpMethod.GET, request.method());
|
||||
assertEquals("/users/42", request.path());
|
||||
assertEquals("true", request.query("verbose"));
|
||||
assertEquals("example.com", request.header("host"));
|
||||
assertEquals("example.com", request.header(":authority"));
|
||||
assertEquals("abc", request.header("X-Trace"));
|
||||
assertNull(request.remoteAddress());
|
||||
}
|
||||
|
||||
private static void field(Http2Stream stream, String name, String value) {
|
||||
byte[] nameBytes = name.getBytes(StandardCharsets.US_ASCII);
|
||||
byte[] valueBytes = value.getBytes(StandardCharsets.US_ASCII);
|
||||
PooledSlice nameView = new PooledSlice();
|
||||
PooledSlice valueView = new PooledSlice();
|
||||
nameView.reset(nameBytes, 0, nameBytes.length);
|
||||
valueView.reset(valueBytes, 0, valueBytes.length);
|
||||
stream.headerBlock().accept(nameView, valueView, false);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package dev.relism.flash.http2.stream;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class Http2StreamLeakTest {
|
||||
@Test
|
||||
void oneHundredThousandAcquireReleaseCyclesReuseOneStream() {
|
||||
Http2StreamTable table = new Http2StreamTable(100);
|
||||
for (int i = 0; i < 100_000; i++) {
|
||||
Http2Stream stream = table.acquire((i << 1) | 1);
|
||||
table.remove(stream.id());
|
||||
table.release(stream);
|
||||
}
|
||||
assertEquals(1, table.createdCount());
|
||||
assertEquals(1, table.freeCount());
|
||||
assertEquals(0, table.size());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package dev.relism.flash.http2.stream;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
|
||||
import dev.relism.flash.http2.Http2StreamException;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class Http2StreamStateTest {
|
||||
@Test
|
||||
void everyTransitionCellIsExecutableOrTypedError() {
|
||||
for (Http2StreamState state : Http2StreamState.values()) {
|
||||
for (Http2StreamState.Event event : Http2StreamState.Event.values()) {
|
||||
if (Http2StreamState.isValid(state, event)) {
|
||||
Http2StreamState next = state.transition(1, event);
|
||||
assertEquals(true, next != null);
|
||||
} else {
|
||||
assertThrows(Http2StreamException.class, () -> state.transition(1, event));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void bodylessRequestAndResponseCloseStream() {
|
||||
Http2StreamState state =
|
||||
Http2StreamState.IDLE.transition(1, Http2StreamState.Event.RECV_HEADERS_ES);
|
||||
assertEquals(Http2StreamState.HALF_CLOSED_REMOTE, state);
|
||||
assertEquals(
|
||||
Http2StreamState.CLOSED, state.transition(1, Http2StreamState.Event.SEND_HEADERS_ES));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package dev.relism.flash.http2.stream;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertSame;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class Http2StreamTableTest {
|
||||
@Test
|
||||
void insertLookupRemoveAtCapacityAndAcrossProbeClusters() {
|
||||
Http2StreamTable table = new Http2StreamTable(8);
|
||||
Http2Stream[] streams = new Http2Stream[8];
|
||||
for (int i = 0; i < streams.length; i++) {
|
||||
streams[i] = table.acquire(i * 2 + 1);
|
||||
assertSame(streams[i], table.get(i * 2 + 1));
|
||||
}
|
||||
assertNull(table.acquire(99));
|
||||
for (int i = 0; i < streams.length; i += 2) {
|
||||
assertSame(streams[i], table.remove(streams[i].id()));
|
||||
table.release(streams[i]);
|
||||
}
|
||||
for (int i = 1; i < streams.length; i += 2) {
|
||||
assertSame(streams[i], table.get(streams[i].id()));
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user