refactor(core): remove out-of-scope HTTP/2 client/proxy, reorganize docs, refresh README
HttpProxy and Http2Client (719 LOC) shipped a reverse-proxy adapter and outbound HTTP/2 client from flash core with zero callers anywhere in the server itself — only each other and their own tests. An HTTP/1.1+2 server framework has no business bundling an outbound client; that capability belongs in its own flash-extensions/flash-ext-* module if/when it's needed. Removed, along with the now-dead src/bench load driver that depended on Http2Client (no replacement client written here — flagged as follow-up work, not silently dropped). docs/http2/ had accumulated core, cross-protocol documentation alongside genuine HTTP/2-protocol internals: HTTP1-HARDENING, TRANSPORT, MESSAGE-MODEL, TRAILERS-AND-STREAMING and BYTES all describe machinery HTTP/1.1 and HTTP/2 share, not HTTP/2 specifically. Moved to a new docs/core/, leaving docs/http2/ to the protocol layers, wire internals and operational docs that are actually HTTP/2-specific. CLEARTEXT-AND-PROXY.md renamed to CLEARTEXT.md and its now-removed upstream-client section cut, matching the source removal above. README.md: removed the "HTTP/2 upstream proxy" section (documented the deleted HttpProxy/Http2Client), the flash-bench module row and build command (not a module that exists in this repo), and fixed every doc link to the new docs/core/ paths. Added the new FlashConfiguration.maxConnections field to the configuration reference. src/bench/ (a load-test harness distinct from the JMH suite, not wired into any Maven profile or CI) is committed here for the first time.
This commit is contained in:
Zakaria El Orche
@@ -0,0 +1,69 @@
|
||||
package dev.relism.flash.bench;
|
||||
|
||||
import dev.relism.flash.extension.FlashApp;
|
||||
import dev.relism.flash.extension.FlashConfiguration;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.URI;
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Real-server, real-network throughput and latency benchmark: boots one live Flash server on
|
||||
* loopback exposing {@code GET /hello}, then drives it end to end — real sockets, real accept
|
||||
* loop, real routing and response serialization — with independent HTTP clients across protocols
|
||||
* and concurrency levels. This is not a component-scoped JMH microbenchmark; it is the same shape
|
||||
* of measurement a tool like {@code h2load} or {@code wrk} gives any other server.
|
||||
*
|
||||
* <p>Never wired into the build or CI — run manually with: {@code mvn -pl flash -Pbench
|
||||
* exec:java}. Override scenario length with {@code -Dflash.bench.warmupSeconds} / {@code
|
||||
* -Dflash.bench.measureSeconds} (defaults: 2 / 5).
|
||||
*/
|
||||
public final class BenchmarkMain {
|
||||
|
||||
private static final int[] CONCURRENCY_LEVELS = {1, 8, 32, 128};
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
Duration warmup = seconds("flash.bench.warmupSeconds", 2);
|
||||
Duration measurement = seconds("flash.bench.measureSeconds", 5);
|
||||
|
||||
int port = freePort();
|
||||
FlashApp app =
|
||||
FlashApp.create(
|
||||
FlashConfiguration.builder()
|
||||
.host("127.0.0.1")
|
||||
.port(port)
|
||||
.http2CleartextEnabled(true)
|
||||
.build());
|
||||
app.get("/hello", (request, response) -> "hello");
|
||||
app.start();
|
||||
|
||||
try {
|
||||
URI target = URI.create("http://127.0.0.1:" + port + "/hello");
|
||||
Report.print(runAllScenarios(target, warmup, measurement));
|
||||
} finally {
|
||||
app.stop().join();
|
||||
}
|
||||
}
|
||||
|
||||
private static List<LoadResult> runAllScenarios(URI target, Duration warmup, Duration measurement)
|
||||
throws InterruptedException {
|
||||
List<LoadResult> results = new ArrayList<>();
|
||||
for (int concurrency : CONCURRENCY_LEVELS) {
|
||||
results.add(
|
||||
new Http1Driver()
|
||||
.run("http/1.1 c=" + concurrency, target, concurrency, warmup, measurement));
|
||||
}
|
||||
return results;
|
||||
}
|
||||
|
||||
private static Duration seconds(String property, int fallback) {
|
||||
return Duration.ofSeconds(Long.getLong(property, fallback));
|
||||
}
|
||||
|
||||
private static int freePort() throws Exception {
|
||||
try (ServerSocket socket = new ServerSocket(0)) {
|
||||
return socket.getLocalPort();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package dev.relism.flash.bench;
|
||||
|
||||
import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.HttpRequest;
|
||||
import java.net.http.HttpResponse;
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* HTTP/1.1 keep-alive load driver backed by the JDK's own {@link HttpClient} — an independent
|
||||
* client implementation, not Flash's own code, measuring the server end to end.
|
||||
*/
|
||||
final class Http1Driver implements LoadDriver {
|
||||
|
||||
@Override
|
||||
public LoadResult run(
|
||||
String scenarioLabel, URI target, int concurrency, Duration warmup, Duration measurement)
|
||||
throws InterruptedException {
|
||||
HttpRequest request = HttpRequest.newBuilder(target).timeout(Duration.ofSeconds(5)).GET().build();
|
||||
return LoadRunner.execute(
|
||||
scenarioLabel,
|
||||
concurrency,
|
||||
warmup,
|
||||
measurement,
|
||||
() -> {
|
||||
// One HttpClient per worker: its own connection pool, reused keep-alive across requests.
|
||||
HttpClient client = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build();
|
||||
return () -> {
|
||||
HttpResponse<Void> response =
|
||||
client.send(request, HttpResponse.BodyHandlers.discarding());
|
||||
if (response.statusCode() != 200) {
|
||||
throw new IllegalStateException("status " + response.statusCode());
|
||||
}
|
||||
};
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package dev.relism.flash.bench;
|
||||
|
||||
import java.util.Arrays;
|
||||
|
||||
/** One worker's latency samples, in nanoseconds. Grows without boxing on the request loop. */
|
||||
final class LatencyRecorder {
|
||||
private long[] samples = new long[1024];
|
||||
private int count;
|
||||
|
||||
void record(long nanos) {
|
||||
if (count == samples.length) samples = Arrays.copyOf(samples, samples.length * 2);
|
||||
samples[count++] = nanos;
|
||||
}
|
||||
|
||||
int count() {
|
||||
return count;
|
||||
}
|
||||
|
||||
long[] toArray() {
|
||||
return Arrays.copyOf(samples, count);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package dev.relism.flash.bench;
|
||||
|
||||
import java.net.URI;
|
||||
import java.time.Duration;
|
||||
|
||||
/** Runs one scenario (a protocol at a fixed concurrency) against a live target and returns its stats. */
|
||||
interface LoadDriver {
|
||||
LoadResult run(
|
||||
String scenarioLabel, URI target, int concurrency, Duration warmup, Duration measurement)
|
||||
throws InterruptedException;
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package dev.relism.flash.bench;
|
||||
|
||||
/** One scenario's outcome: throughput and latency distribution over the measured phase only. */
|
||||
record LoadResult(
|
||||
String scenario,
|
||||
long requests,
|
||||
long errors,
|
||||
double seconds,
|
||||
double meanLatencyMicros,
|
||||
double p50Micros,
|
||||
double p99Micros,
|
||||
double p999Micros) {
|
||||
|
||||
double requestsPerSecond() {
|
||||
return seconds == 0 ? 0 : requests / seconds;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package dev.relism.flash.bench;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.atomic.LongAdder;
|
||||
|
||||
/**
|
||||
* Drives a fixed number of concurrent virtual-thread workers against one {@link WorkerFactory},
|
||||
* each worker looping its own {@link WorkUnit#request()} until a wall-clock deadline. A discarded
|
||||
* warmup phase runs first so JIT warmup and connection setup don't skew the measured phase.
|
||||
*/
|
||||
final class LoadRunner {
|
||||
|
||||
private LoadRunner() {}
|
||||
|
||||
static LoadResult execute(
|
||||
String scenarioLabel,
|
||||
int concurrency,
|
||||
Duration warmup,
|
||||
Duration measurement,
|
||||
WorkerFactory factory)
|
||||
throws InterruptedException {
|
||||
runUntil(concurrency, System.nanoTime() + warmup.toNanos(), factory, null, null);
|
||||
|
||||
LongAdder errors = new LongAdder();
|
||||
List<LatencyRecorder> perWorker = new ArrayList<>(concurrency);
|
||||
for (int i = 0; i < concurrency; i++) perWorker.add(new LatencyRecorder());
|
||||
|
||||
long measureStart = System.nanoTime();
|
||||
runUntil(concurrency, measureStart + measurement.toNanos(), factory, errors, perWorker);
|
||||
|
||||
return Stats.summarize(scenarioLabel, perWorker, errors.sum(), System.nanoTime() - measureStart);
|
||||
}
|
||||
|
||||
private static void runUntil(
|
||||
int concurrency,
|
||||
long deadlineNanos,
|
||||
WorkerFactory factory,
|
||||
LongAdder errors,
|
||||
List<LatencyRecorder> perWorker)
|
||||
throws InterruptedException {
|
||||
try (ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor()) {
|
||||
for (int i = 0; i < concurrency; i++) {
|
||||
LatencyRecorder recorder = perWorker == null ? null : perWorker.get(i);
|
||||
pool.execute(() -> worker(deadlineNanos, factory, errors, recorder));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static void worker(
|
||||
long deadlineNanos, WorkerFactory factory, LongAdder errors, LatencyRecorder recorder) {
|
||||
try (WorkUnit unit = factory.create()) {
|
||||
while (System.nanoTime() < deadlineNanos) {
|
||||
long start = System.nanoTime();
|
||||
try {
|
||||
unit.request();
|
||||
if (recorder != null) recorder.record(System.nanoTime() - start);
|
||||
} catch (Exception requestFailure) {
|
||||
if (errors != null) errors.increment();
|
||||
}
|
||||
}
|
||||
} catch (Exception setupFailure) {
|
||||
if (errors != null) errors.increment();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
package dev.relism.flash.bench;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/** Prints results as a fixed-width table on stdout — no file output, this is a manual tool. */
|
||||
final class Report {
|
||||
|
||||
private Report() {}
|
||||
|
||||
static void print(List<LoadResult> results) {
|
||||
System.out.printf(
|
||||
"%-16s %10s %8s %12s %10s %10s %10s %10s%n",
|
||||
"scenario", "requests", "errors", "req/s", "mean(us)", "p50(us)", "p99(us)", "p999(us)");
|
||||
for (LoadResult result : results) {
|
||||
System.out.printf(
|
||||
"%-16s %10d %8d %12.1f %10.1f %10.1f %10.1f %10.1f%n",
|
||||
result.scenario(),
|
||||
result.requests(),
|
||||
result.errors(),
|
||||
result.requestsPerSecond(),
|
||||
result.meanLatencyMicros(),
|
||||
result.p50Micros(),
|
||||
result.p99Micros(),
|
||||
result.p999Micros());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package dev.relism.flash.bench;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
/** Merges every worker's samples and reduces them to one {@link LoadResult}. */
|
||||
final class Stats {
|
||||
|
||||
private Stats() {}
|
||||
|
||||
static LoadResult summarize(
|
||||
String scenarioLabel, List<LatencyRecorder> perWorker, long errors, long elapsedNanos) {
|
||||
int total = 0;
|
||||
for (LatencyRecorder recorder : perWorker) total += recorder.count();
|
||||
|
||||
long[] merged = new long[total];
|
||||
int offset = 0;
|
||||
for (LatencyRecorder recorder : perWorker) {
|
||||
long[] samples = recorder.toArray();
|
||||
System.arraycopy(samples, 0, merged, offset, samples.length);
|
||||
offset += samples.length;
|
||||
}
|
||||
Arrays.sort(merged);
|
||||
|
||||
return new LoadResult(
|
||||
scenarioLabel,
|
||||
merged.length,
|
||||
errors,
|
||||
elapsedNanos / 1_000_000_000.0,
|
||||
microsOf(mean(merged)),
|
||||
microsOf(percentile(merged, 0.50)),
|
||||
microsOf(percentile(merged, 0.99)),
|
||||
microsOf(percentile(merged, 0.999)));
|
||||
}
|
||||
|
||||
private static double mean(long[] sorted) {
|
||||
if (sorted.length == 0) return 0;
|
||||
long sum = 0;
|
||||
for (long value : sorted) sum += value;
|
||||
return (double) sum / sorted.length;
|
||||
}
|
||||
|
||||
private static long percentile(long[] sorted, double fraction) {
|
||||
if (sorted.length == 0) return 0;
|
||||
int index = (int) Math.min(sorted.length - 1, Math.floor(fraction * sorted.length));
|
||||
return sorted[index];
|
||||
}
|
||||
|
||||
private static double microsOf(double nanos) {
|
||||
return nanos / 1000.0;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
package dev.relism.flash.bench;
|
||||
|
||||
/** One worker's request loop body. {@link #close()} releases whatever {@link WorkerFactory} opened. */
|
||||
interface WorkUnit extends AutoCloseable {
|
||||
void request() throws Exception;
|
||||
|
||||
@Override
|
||||
default void close() throws Exception {}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package dev.relism.flash.bench;
|
||||
|
||||
/** Builds one worker's {@link WorkUnit} — its own connection/client, isolated per virtual thread. */
|
||||
@FunctionalInterface
|
||||
interface WorkerFactory {
|
||||
WorkUnit create() throws Exception;
|
||||
}
|
||||
@@ -1,92 +0,0 @@
|
||||
package dev.relism.flash.http.proxy;
|
||||
|
||||
import dev.relism.flash.http.HopByHopHeaders;
|
||||
import dev.relism.flash.http.HopByHopHeaders.Protocol;
|
||||
import dev.relism.flash.http2.client.Http2Client;
|
||||
import dev.relism.flash.http2.client.Http2ClientResponse;
|
||||
import dev.relism.flash.models.HeaderView;
|
||||
import dev.relism.flash.models.Request;
|
||||
import dev.relism.flash.models.Response;
|
||||
import dev.relism.flash.models.SimpleHandler;
|
||||
import dev.relism.fpr.core.ByteView;
|
||||
import java.net.URI;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Objects;
|
||||
|
||||
/** Protocol-neutral reverse-proxy adapter backed by Flash's HTTP/2 upstream client. */
|
||||
public final class HttpProxy {
|
||||
private HttpProxy() {}
|
||||
|
||||
/** Creates a handler that preserves the incoming path, query, fields, body and trailers. */
|
||||
public static SimpleHandler.FunctionalHandler toHttp2(URI upstreamOrigin, Http2Client client) {
|
||||
Objects.requireNonNull(upstreamOrigin, "upstreamOrigin");
|
||||
Objects.requireNonNull(client, "client");
|
||||
return (request, response) -> relay(upstreamOrigin, client, request, response);
|
||||
}
|
||||
|
||||
private static Response relay(
|
||||
URI upstreamOrigin, Http2Client client, Request request, Response response) throws Exception {
|
||||
byte[] body = request.body().bytes();
|
||||
Protocol downstream =
|
||||
request.getRequestLine().getProtocol() == null ? Protocol.HTTP_2 : Protocol.HTTP_1_1;
|
||||
URI target = upstreamOrigin.resolve(rawTarget(request));
|
||||
Http2ClientResponse upstream =
|
||||
client.exchange(
|
||||
target,
|
||||
request.method(),
|
||||
request.getRequestLine().getHeaders(),
|
||||
body,
|
||||
request.trailers());
|
||||
|
||||
response.status(upstream.statusCode()).body(upstream.body());
|
||||
copyHeaders(upstream.headers(), Protocol.HTTP_2, downstream, response, false);
|
||||
copyHeaders(upstream.trailers(), Protocol.HTTP_2, downstream, response, true);
|
||||
return response;
|
||||
}
|
||||
|
||||
private static String rawTarget(Request request) {
|
||||
String path = request.path();
|
||||
ByteView query = request.getRequestLine().getQuery();
|
||||
if (query == null || query.length() == 0) return path;
|
||||
byte[] bytes = new byte[query.length()];
|
||||
for (int i = 0; i < bytes.length; i++) bytes[i] = query.byteAt(i);
|
||||
return path + "?" + new String(bytes, StandardCharsets.US_ASCII);
|
||||
}
|
||||
|
||||
private static void copyHeaders(
|
||||
HeaderView source,
|
||||
Protocol sourceProtocol,
|
||||
Protocol targetProtocol,
|
||||
Response response,
|
||||
boolean trailers) {
|
||||
source.forEach(
|
||||
(name, value) -> {
|
||||
if (!HopByHopHeaders.shouldForward(
|
||||
source, name, value, sourceProtocol, targetProtocol)) return;
|
||||
if (!trailers && (equalsAscii(name, "content-length") || equalsAscii(name, "content-type"))) {
|
||||
if (equalsAscii(name, "content-type")) response.type(string(value));
|
||||
return;
|
||||
}
|
||||
if (trailers) response.trailer(string(name), string(value));
|
||||
else response.header(string(name), string(value));
|
||||
});
|
||||
}
|
||||
|
||||
private static String string(ByteView value) {
|
||||
byte[] bytes = new byte[value.length()];
|
||||
for (int i = 0; i < bytes.length; i++) bytes[i] = value.byteAt(i);
|
||||
return new String(bytes, StandardCharsets.UTF_8);
|
||||
}
|
||||
|
||||
private static boolean equalsAscii(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 += 'a' - 'A';
|
||||
if (right >= 'A' && right <= 'Z') right += 'a' - 'A';
|
||||
if (left != right) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
@@ -1,627 +0,0 @@
|
||||
package dev.relism.flash.http2.client;
|
||||
|
||||
import dev.relism.flash.bytes.ByteWriter;
|
||||
import dev.relism.flash.bytes.Pairs;
|
||||
import dev.relism.flash.http.HopByHopHeaders;
|
||||
import dev.relism.flash.http.HopByHopHeaders.Protocol;
|
||||
import dev.relism.flash.http.HttpMethod;
|
||||
import dev.relism.flash.http2.Http2Exception;
|
||||
import dev.relism.flash.http2.Http2Limits;
|
||||
import dev.relism.flash.http2.Http2Preface;
|
||||
import dev.relism.flash.http2.Http2Settings;
|
||||
import dev.relism.flash.http2.frame.FrameFlags;
|
||||
import dev.relism.flash.http2.frame.FrameHeader;
|
||||
import dev.relism.flash.http2.frame.FrameType;
|
||||
import dev.relism.flash.http2.frame.FrameWriteBuffer;
|
||||
import dev.relism.flash.http2.frame.Http2FrameReader;
|
||||
import dev.relism.flash.http2.frame.Http2FrameWriter;
|
||||
import dev.relism.flash.http2.frame.Padding;
|
||||
import dev.relism.flash.http2.frame.WriteIntent;
|
||||
import dev.relism.flash.http2.hpack.ContinuationAssembler;
|
||||
import dev.relism.flash.http2.hpack.HpackDecoder;
|
||||
import dev.relism.flash.http2.hpack.HpackEncoder;
|
||||
import dev.relism.flash.models.EmptyHeaderView;
|
||||
import dev.relism.flash.models.HeaderView;
|
||||
import dev.relism.flash.models.MutableHeaderMap;
|
||||
import dev.relism.flash.transport.BufferedByteSource;
|
||||
import dev.relism.fpr.core.ByteView;
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.io.OutputStream;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.Socket;
|
||||
import java.net.URI;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import javax.net.ssl.SSLContext;
|
||||
import javax.net.ssl.SSLParameters;
|
||||
import javax.net.ssl.SSLSocket;
|
||||
|
||||
/**
|
||||
* Small pooled HTTP/2 client for Flash proxy handlers. It intentionally exposes synchronous
|
||||
* request/response exchange rather than trying to be a general-purpose client API.
|
||||
*/
|
||||
public final class Http2Client implements Closeable {
|
||||
private static final int CONNECT_TIMEOUT_MS = 10_000;
|
||||
private static final int MAX_RESPONSE_BODY_SIZE = Http2Limits.MAX_REQUEST_BODY_SIZE;
|
||||
|
||||
private final ConcurrentHashMap<Origin, Connection> connections = new ConcurrentHashMap<>();
|
||||
private final SSLContext sslContext;
|
||||
|
||||
public Http2Client() {
|
||||
this(null);
|
||||
}
|
||||
|
||||
public Http2Client(SSLContext sslContext) {
|
||||
this.sslContext = sslContext;
|
||||
}
|
||||
|
||||
public Http2ClientResponse get(URI uri) throws IOException {
|
||||
return exchange(
|
||||
uri,
|
||||
HttpMethod.GET,
|
||||
EmptyHeaderView.INSTANCE,
|
||||
new byte[0],
|
||||
EmptyHeaderView.INSTANCE);
|
||||
}
|
||||
|
||||
public Http2ClientResponse exchange(
|
||||
URI uri, HttpMethod method, HeaderView headers, byte[] body, HeaderView trailers)
|
||||
throws IOException {
|
||||
Objects.requireNonNull(uri, "uri");
|
||||
Objects.requireNonNull(method, "method");
|
||||
Objects.requireNonNull(headers, "headers");
|
||||
Objects.requireNonNull(body, "body");
|
||||
Objects.requireNonNull(trailers, "trailers");
|
||||
Origin origin = Origin.from(uri);
|
||||
Connection connection;
|
||||
try {
|
||||
connection = connections.computeIfAbsent(origin, this::openUnchecked);
|
||||
} catch (OpenFailure failure) {
|
||||
throw failure.io;
|
||||
}
|
||||
try {
|
||||
return connection.exchange(uri, method, headers, body, trailers);
|
||||
} catch (IOException | RuntimeException failure) {
|
||||
connections.remove(origin, connection);
|
||||
connection.close();
|
||||
throw failure;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
for (Connection connection : connections.values()) connection.close();
|
||||
connections.clear();
|
||||
}
|
||||
|
||||
/** Number of currently pooled origin connections. */
|
||||
public int pooledConnectionCount() {
|
||||
return connections.size();
|
||||
}
|
||||
|
||||
private Connection openUnchecked(Origin origin) {
|
||||
try {
|
||||
return new Connection(origin, sslContext);
|
||||
} catch (IOException failure) {
|
||||
throw new OpenFailure(failure);
|
||||
}
|
||||
}
|
||||
|
||||
private static final class Connection implements Closeable {
|
||||
private final Socket socket;
|
||||
private final OutputStream output;
|
||||
private final Http2FrameReader reader;
|
||||
private final Http2FrameWriter writer;
|
||||
private final Http2Settings peerSettings = new Http2Settings();
|
||||
private final HpackDecoder decoder = new HpackDecoder();
|
||||
private final ContinuationAssembler headers = new ContinuationAssembler();
|
||||
private final ByteWriter outgoing = new ByteWriter(16 * 1024);
|
||||
private final FrameWriteBuffer frames = new FrameWriteBuffer(outgoing);
|
||||
private final BufferIntent intent = new BufferIntent();
|
||||
private int nextStreamId = 1;
|
||||
private int connectionSendWindow = Http2Settings.DEFAULT_INITIAL_WINDOW_SIZE;
|
||||
private int streamSendWindow;
|
||||
private boolean headerEndStream;
|
||||
private boolean closed;
|
||||
|
||||
Connection(Origin origin, SSLContext sslContext) throws IOException {
|
||||
socket = connect(origin, sslContext);
|
||||
output = socket.getOutputStream();
|
||||
reader =
|
||||
new Http2FrameReader(new BufferedByteSource(socket.getInputStream(), socket));
|
||||
writer = new Http2FrameWriter(output::write);
|
||||
writePreface();
|
||||
awaitServerSettings();
|
||||
}
|
||||
|
||||
synchronized Http2ClientResponse exchange(
|
||||
URI uri, HttpMethod method, HeaderView requestHeaders, byte[] body, HeaderView trailers)
|
||||
throws IOException {
|
||||
if (closed) throw new IOException("HTTP/2 connection is closed");
|
||||
if (nextStreamId <= 0) throw new IOException("HTTP/2 stream id space exhausted");
|
||||
int streamId = nextStreamId;
|
||||
nextStreamId += 2;
|
||||
streamSendWindow = peerSettings.initialWindowSize();
|
||||
Exchange exchange = new Exchange(streamId);
|
||||
|
||||
writeRequestHeaders(uri, method, requestHeaders, body.length == 0 && trailers.count() == 0,
|
||||
streamId);
|
||||
if (body.length != 0) writeRequestBody(exchange, body, trailers.count() == 0);
|
||||
if (trailers.count() != 0) writeRequestTrailers(trailers, streamId);
|
||||
while (!exchange.complete) readFrame(exchange);
|
||||
return exchange.response();
|
||||
}
|
||||
|
||||
private void writePreface() throws IOException {
|
||||
output.write(Http2Preface.clientPreface());
|
||||
outgoing.reset();
|
||||
frames.beginFrame(FrameType.SETTINGS, 0, 0);
|
||||
outgoing.writeUInt16(Http2Settings.ENABLE_PUSH);
|
||||
outgoing.writeUInt32(0);
|
||||
outgoing.writeUInt16(Http2Settings.INITIAL_WINDOW_SIZE);
|
||||
outgoing.writeUInt32(Http2Limits.INITIAL_WINDOW_SIZE_LOCAL);
|
||||
frames.endFrame();
|
||||
frames.beginFrame(FrameType.WINDOW_UPDATE, 0, 0);
|
||||
outgoing.writeUInt31(
|
||||
Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL - Http2Settings.DEFAULT_INITIAL_WINDOW_SIZE);
|
||||
frames.endFrame();
|
||||
writeOutgoing();
|
||||
}
|
||||
|
||||
private void awaitServerSettings() throws IOException {
|
||||
boolean received = false;
|
||||
while (!received) {
|
||||
FrameHeader frame = reader.readFrame();
|
||||
if (frame == null) throw new IOException("server closed before SETTINGS");
|
||||
try {
|
||||
if (frame.type() == FrameType.SETTINGS && !FrameFlags.isAck(frame.flags())) {
|
||||
applySettings(frame);
|
||||
sendEmpty(FrameType.SETTINGS, FrameFlags.ACK, 0);
|
||||
received = true;
|
||||
} else if (frame.type() == FrameType.WINDOW_UPDATE) {
|
||||
applyWindowUpdate(frame, 0);
|
||||
} else if (frame.type() == FrameType.GOAWAY) {
|
||||
throw new IOException("server sent GOAWAY during HTTP/2 setup");
|
||||
}
|
||||
} finally {
|
||||
reader.consumeFrame();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void writeRequestHeaders(
|
||||
URI uri, HttpMethod method, HeaderView source, boolean endStream, int streamId)
|
||||
throws IOException {
|
||||
outgoing.reset();
|
||||
frames.beginFrame(
|
||||
FrameType.HEADERS,
|
||||
FrameFlags.END_HEADERS | (endStream ? FrameFlags.END_STREAM : 0),
|
||||
streamId);
|
||||
writeMethod(method);
|
||||
HpackEncoder.writeIndexed(outgoing, "https".equalsIgnoreCase(uri.getScheme()) ? 7 : 6);
|
||||
writeAuthority(uri);
|
||||
writePath(uri);
|
||||
source.forEach(
|
||||
(name, value) -> {
|
||||
if (HopByHopHeaders.shouldForward(
|
||||
source, name, value, Protocol.HTTP_1_1, Protocol.HTTP_2)
|
||||
&& !equalsAscii(name, "host")) {
|
||||
HpackEncoder.writeLiteral(outgoing, name, value);
|
||||
}
|
||||
});
|
||||
frames.endFrame();
|
||||
writeOutgoing();
|
||||
}
|
||||
|
||||
private void writeRequestBody(Exchange exchange, byte[] body, boolean endStream)
|
||||
throws IOException {
|
||||
int offset = 0;
|
||||
while (offset < body.length) {
|
||||
while (connectionSendWindow <= 0 || streamSendWindow <= 0) readFrame(exchange);
|
||||
int count =
|
||||
Math.min(
|
||||
body.length - offset,
|
||||
Math.min(
|
||||
peerSettings.maxFrameSize(),
|
||||
Math.min(connectionSendWindow, streamSendWindow)));
|
||||
outgoing.reset();
|
||||
frames.beginFrame(
|
||||
FrameType.DATA,
|
||||
endStream && offset + count == body.length ? FrameFlags.END_STREAM : 0,
|
||||
exchange.streamId);
|
||||
outgoing.writeBytes(body, offset, count);
|
||||
frames.endFrame();
|
||||
writeOutgoing();
|
||||
connectionSendWindow -= count;
|
||||
streamSendWindow -= count;
|
||||
offset += count;
|
||||
}
|
||||
}
|
||||
|
||||
private void writeRequestTrailers(HeaderView trailers, int streamId) throws IOException {
|
||||
outgoing.reset();
|
||||
frames.beginFrame(
|
||||
FrameType.HEADERS, FrameFlags.END_HEADERS | FrameFlags.END_STREAM, streamId);
|
||||
trailers.forEach(
|
||||
(name, value) -> {
|
||||
if (HopByHopHeaders.shouldForward(
|
||||
trailers, name, value, Protocol.HTTP_1_1, Protocol.HTTP_2)) {
|
||||
HpackEncoder.writeLiteral(outgoing, name, value);
|
||||
}
|
||||
});
|
||||
frames.endFrame();
|
||||
writeOutgoing();
|
||||
}
|
||||
|
||||
private void readFrame(Exchange exchange) throws IOException {
|
||||
FrameHeader frame = reader.readFrame();
|
||||
if (frame == null) throw new IOException("server closed an active HTTP/2 exchange");
|
||||
try {
|
||||
FrameType type = frame.type();
|
||||
if (type == null) return;
|
||||
switch (type) {
|
||||
case SETTINGS -> {
|
||||
if (!FrameFlags.isAck(frame.flags())) {
|
||||
applySettings(frame);
|
||||
sendEmpty(FrameType.SETTINGS, FrameFlags.ACK, 0);
|
||||
}
|
||||
}
|
||||
case WINDOW_UPDATE -> applyWindowUpdate(frame, exchange.streamId);
|
||||
case PING -> {
|
||||
if (!FrameFlags.isAck(frame.flags())) sendPingAck(frame);
|
||||
}
|
||||
case HEADERS, CONTINUATION -> receiveHeaders(frame, exchange);
|
||||
case DATA -> receiveData(frame, exchange);
|
||||
case RST_STREAM -> receiveReset(frame, exchange);
|
||||
case GOAWAY -> throw receiveGoAway(frame);
|
||||
case PUSH_PROMISE -> throw new IOException("server sent PUSH_PROMISE after ENABLE_PUSH=0");
|
||||
default -> {
|
||||
// PRIORITY and unknown extension semantics do not affect this single exchange.
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
reader.consumeFrame();
|
||||
}
|
||||
}
|
||||
|
||||
private void receiveHeaders(FrameHeader frame, Exchange exchange) throws IOException {
|
||||
if (frame.streamId() != exchange.streamId) {
|
||||
throw new IOException("unexpected response stream " + frame.streamId());
|
||||
}
|
||||
if (frame.type() == FrameType.HEADERS) {
|
||||
if (headers.isActive()) throw new IOException("interleaved response header block");
|
||||
headerEndStream = FrameFlags.isEndStream(frame.flags());
|
||||
long unpadded =
|
||||
Padding.unpad(
|
||||
frame.buffer(),
|
||||
frame.payloadOffset(),
|
||||
frame.length(),
|
||||
FrameFlags.isPadded(frame.flags()));
|
||||
int offset = Pairs.hi(unpadded);
|
||||
int length = Pairs.lo(unpadded);
|
||||
if (FrameFlags.hasPriority(frame.flags())) {
|
||||
if (length < 5) throw new IOException("truncated response priority fields");
|
||||
offset += 5;
|
||||
length -= 5;
|
||||
}
|
||||
headers.begin(
|
||||
frame.streamId(),
|
||||
frame.buffer(),
|
||||
offset,
|
||||
length,
|
||||
FrameFlags.isEndHeaders(frame.flags()));
|
||||
} else {
|
||||
headers.continuation(
|
||||
frame.streamId(),
|
||||
frame.buffer(),
|
||||
frame.payloadOffset(),
|
||||
frame.length(),
|
||||
FrameFlags.isEndHeaders(frame.flags()));
|
||||
}
|
||||
if (!headers.isComplete()) return;
|
||||
|
||||
boolean trailers = exchange.statusCode != 0;
|
||||
ResponseHeaderSink sink = new ResponseHeaderSink(exchange, trailers);
|
||||
decoder.decode(headers.buffer(), 0, headers.length(), sink);
|
||||
headers.reset();
|
||||
sink.validate();
|
||||
if (!trailers && exchange.statusCode >= 100 && exchange.statusCode < 200) {
|
||||
if (headerEndStream) throw new IOException("informational response ended the stream");
|
||||
exchange.statusCode = 0;
|
||||
exchange.headers.reset();
|
||||
return;
|
||||
}
|
||||
if (trailers && !headerEndStream) {
|
||||
throw new IOException("response trailers did not end the stream");
|
||||
}
|
||||
if (headerEndStream) exchange.complete = true;
|
||||
}
|
||||
|
||||
private void receiveData(FrameHeader frame, Exchange exchange) throws IOException {
|
||||
if (frame.streamId() != exchange.streamId || exchange.statusCode == 0) {
|
||||
throw new IOException("DATA received before response headers");
|
||||
}
|
||||
long unpadded =
|
||||
Padding.unpad(
|
||||
frame.buffer(),
|
||||
frame.payloadOffset(),
|
||||
frame.length(),
|
||||
FrameFlags.isPadded(frame.flags()));
|
||||
int dataOffset = Pairs.hi(unpadded);
|
||||
int dataLength = Pairs.lo(unpadded);
|
||||
if (exchange.body.size() > MAX_RESPONSE_BODY_SIZE - dataLength) {
|
||||
throw new IOException("proxied HTTP/2 response body exceeds limit");
|
||||
}
|
||||
exchange.body.write(frame.buffer(), dataOffset, dataLength);
|
||||
if (frame.length() != 0) {
|
||||
sendWindowUpdate(0, frame.length());
|
||||
sendWindowUpdate(exchange.streamId, frame.length());
|
||||
}
|
||||
if (FrameFlags.isEndStream(frame.flags())) exchange.complete = true;
|
||||
}
|
||||
|
||||
private void receiveReset(FrameHeader frame, Exchange exchange) throws IOException {
|
||||
if (frame.streamId() != exchange.streamId || frame.length() != 4) return;
|
||||
int code = readInt(frame.buffer(), frame.payloadOffset());
|
||||
throw new IOException("upstream reset HTTP/2 stream with error " + code);
|
||||
}
|
||||
|
||||
private IOException receiveGoAway(FrameHeader frame) {
|
||||
closed = true;
|
||||
int code = frame.length() >= 8 ? readInt(frame.buffer(), frame.payloadOffset() + 4) : -1;
|
||||
return new IOException("upstream sent GOAWAY with error " + code);
|
||||
}
|
||||
|
||||
private void applySettings(FrameHeader frame) {
|
||||
int oldWindow = peerSettings.initialWindowSize();
|
||||
peerSettings.apply(frame.buffer(), frame.payloadOffset(), frame.length(), delta -> {});
|
||||
streamSendWindow += peerSettings.initialWindowSize() - oldWindow;
|
||||
}
|
||||
|
||||
private void applyWindowUpdate(FrameHeader frame, int activeStreamId) throws IOException {
|
||||
if (frame.length() != 4) throw new IOException("invalid WINDOW_UPDATE length");
|
||||
int increment = readInt(frame.buffer(), frame.payloadOffset()) & 0x7fff_ffff;
|
||||
if (increment == 0) throw new IOException("zero WINDOW_UPDATE increment");
|
||||
if (frame.streamId() == 0) connectionSendWindow = addWindow(connectionSendWindow, increment);
|
||||
else if (frame.streamId() == activeStreamId) streamSendWindow = addWindow(streamSendWindow, increment);
|
||||
}
|
||||
|
||||
private void sendPingAck(FrameHeader frame) throws IOException {
|
||||
outgoing.reset();
|
||||
frames.beginFrame(FrameType.PING, FrameFlags.ACK, 0);
|
||||
outgoing.writeBytes(frame.buffer(), frame.payloadOffset(), frame.length());
|
||||
frames.endFrame();
|
||||
writeOutgoing();
|
||||
}
|
||||
|
||||
private void sendWindowUpdate(int streamId, int increment) throws IOException {
|
||||
outgoing.reset();
|
||||
frames.beginFrame(FrameType.WINDOW_UPDATE, 0, streamId);
|
||||
outgoing.writeUInt31(increment);
|
||||
frames.endFrame();
|
||||
writeOutgoing();
|
||||
}
|
||||
|
||||
private void sendEmpty(FrameType type, int flags, int streamId) throws IOException {
|
||||
outgoing.reset();
|
||||
frames.beginFrame(type, flags, streamId);
|
||||
frames.endFrame();
|
||||
writeOutgoing();
|
||||
}
|
||||
|
||||
private void writeOutgoing() throws IOException {
|
||||
intent.reset(outgoing.array(), outgoing.length());
|
||||
writer.write(intent);
|
||||
}
|
||||
|
||||
private void writeMethod(HttpMethod method) {
|
||||
if (method == HttpMethod.GET) HpackEncoder.writeIndexed(outgoing, 2);
|
||||
else if (method == HttpMethod.POST) HpackEncoder.writeIndexed(outgoing, 3);
|
||||
else {
|
||||
byte[] value = method.name().getBytes(StandardCharsets.US_ASCII);
|
||||
HpackEncoder.writeLiteralWithNameIndex(outgoing, 2, value, false);
|
||||
}
|
||||
}
|
||||
|
||||
private void writeAuthority(URI uri) {
|
||||
String authority = uri.getRawAuthority();
|
||||
if (authority == null || authority.isEmpty()) {
|
||||
throw new IllegalArgumentException("HTTP/2 URI requires an authority");
|
||||
}
|
||||
HpackEncoder.writeLiteralWithNameIndex(
|
||||
outgoing, 1, authority.getBytes(StandardCharsets.US_ASCII), false);
|
||||
}
|
||||
|
||||
private void writePath(URI uri) {
|
||||
String path = uri.getRawPath();
|
||||
if (path == null || path.isEmpty()) path = "/";
|
||||
if (uri.getRawQuery() != null) path += "?" + uri.getRawQuery();
|
||||
if ("/".equals(path)) HpackEncoder.writeIndexed(outgoing, 4);
|
||||
else if ("/index.html".equals(path)) HpackEncoder.writeIndexed(outgoing, 5);
|
||||
else {
|
||||
HpackEncoder.writeLiteralWithNameIndex(
|
||||
outgoing, 4, path.getBytes(StandardCharsets.US_ASCII), false);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void close() {
|
||||
if (closed) return;
|
||||
closed = true;
|
||||
writer.close();
|
||||
try {
|
||||
socket.close();
|
||||
} catch (IOException ignored) {
|
||||
// Closing a broken pooled connection is best-effort.
|
||||
}
|
||||
}
|
||||
|
||||
private static Socket connect(Origin origin, SSLContext sslContext) throws IOException {
|
||||
if (!origin.secure) {
|
||||
Socket socket = new Socket();
|
||||
socket.connect(new InetSocketAddress(origin.host, origin.port), CONNECT_TIMEOUT_MS);
|
||||
configureLowLatency(socket);
|
||||
return socket;
|
||||
}
|
||||
SSLContext context;
|
||||
try {
|
||||
context = sslContext == null ? SSLContext.getDefault() : sslContext;
|
||||
} catch (Exception failure) {
|
||||
throw new IOException("cannot initialize TLS context", failure);
|
||||
}
|
||||
SSLSocket socket =
|
||||
(SSLSocket) context.getSocketFactory().createSocket(origin.host, origin.port);
|
||||
configureLowLatency(socket);
|
||||
SSLParameters parameters = socket.getSSLParameters();
|
||||
parameters.setApplicationProtocols(new String[] {"h2"});
|
||||
parameters.setEndpointIdentificationAlgorithm("HTTPS");
|
||||
socket.setSSLParameters(parameters);
|
||||
socket.startHandshake();
|
||||
if (!"h2".equals(socket.getApplicationProtocol())) {
|
||||
socket.close();
|
||||
throw new IOException("upstream did not negotiate HTTP/2 through ALPN");
|
||||
}
|
||||
return socket;
|
||||
}
|
||||
}
|
||||
|
||||
static void configureLowLatency(Socket socket) throws IOException {
|
||||
socket.setTcpNoDelay(true);
|
||||
}
|
||||
|
||||
private static final class Exchange {
|
||||
private final int streamId;
|
||||
private final MutableHeaderMap headers = new MutableHeaderMap();
|
||||
private final MutableHeaderMap trailers = new MutableHeaderMap();
|
||||
private final ByteArrayOutputStream body = new ByteArrayOutputStream();
|
||||
private int statusCode;
|
||||
private boolean complete;
|
||||
|
||||
private Exchange(int streamId) {
|
||||
this.streamId = streamId;
|
||||
}
|
||||
|
||||
private Http2ClientResponse response() {
|
||||
return new Http2ClientResponse(statusCode, headers, body.toByteArray(), trailers);
|
||||
}
|
||||
}
|
||||
|
||||
private static final class ResponseHeaderSink
|
||||
implements dev.relism.flash.http2.hpack.HeaderSink {
|
||||
private final Exchange exchange;
|
||||
private final boolean trailers;
|
||||
private boolean regular;
|
||||
private boolean status;
|
||||
|
||||
private ResponseHeaderSink(Exchange exchange, boolean trailers) {
|
||||
this.exchange = exchange;
|
||||
this.trailers = trailers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void accept(ByteView name, ByteView value, boolean neverIndexed) {
|
||||
if (name.length() != 0 && name.byteAt(0) == ':') {
|
||||
if (trailers || regular || status || !equalsAscii(name, ":status")) {
|
||||
throw Http2Exception.PROTOCOL_ERROR;
|
||||
}
|
||||
exchange.statusCode = parseStatus(value);
|
||||
status = true;
|
||||
return;
|
||||
}
|
||||
regular = true;
|
||||
MutableHeaderMap target = trailers ? exchange.trailers : exchange.headers;
|
||||
byte[] nameBytes = copy(name);
|
||||
byte[] valueBytes = copy(value);
|
||||
target.add(nameBytes, 0, nameBytes.length, valueBytes, 0, valueBytes.length);
|
||||
}
|
||||
|
||||
private void validate() throws IOException {
|
||||
if (!trailers && !status) throw new IOException("HTTP/2 response omitted :status");
|
||||
}
|
||||
|
||||
private static int parseStatus(ByteView value) {
|
||||
if (value.length() != 3) throw Http2Exception.PROTOCOL_ERROR;
|
||||
int code = 0;
|
||||
for (int i = 0; i < 3; i++) {
|
||||
int digit = (value.byteAt(i) & 0xff) - '0';
|
||||
if (digit < 0 || digit > 9) throw Http2Exception.PROTOCOL_ERROR;
|
||||
code = code * 10 + digit;
|
||||
}
|
||||
return code;
|
||||
}
|
||||
}
|
||||
|
||||
private static final class BufferIntent implements WriteIntent {
|
||||
private byte[] bytes;
|
||||
private int length;
|
||||
private WriteIntent next;
|
||||
|
||||
private void reset(byte[] bytes, int length) {
|
||||
this.bytes = bytes;
|
||||
this.length = length;
|
||||
this.next = null;
|
||||
}
|
||||
|
||||
@Override public byte[] buffer() { return bytes; }
|
||||
@Override public int offset() { return 0; }
|
||||
@Override public int length() { return length; }
|
||||
@Override public WriteIntent mpscNext() { return next; }
|
||||
@Override public void setMpscNext(WriteIntent next) { this.next = next; }
|
||||
}
|
||||
|
||||
private record Origin(String scheme, String host, int port, boolean secure) {
|
||||
private static Origin from(URI uri) {
|
||||
String scheme = uri.getScheme();
|
||||
boolean secure;
|
||||
if ("https".equalsIgnoreCase(scheme)) secure = true;
|
||||
else if ("http".equalsIgnoreCase(scheme)) secure = false;
|
||||
else throw new IllegalArgumentException("HTTP/2 URI scheme must be http or https");
|
||||
if (uri.getHost() == null) throw new IllegalArgumentException("HTTP/2 URI requires a host");
|
||||
int port = uri.getPort() >= 0 ? uri.getPort() : secure ? 443 : 80;
|
||||
return new Origin(scheme.toLowerCase(), uri.getHost(), port, secure);
|
||||
}
|
||||
}
|
||||
|
||||
private static final class OpenFailure extends RuntimeException {
|
||||
private final IOException io;
|
||||
|
||||
private OpenFailure(IOException io) {
|
||||
super(io);
|
||||
this.io = io;
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean equalsAscii(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 += 'a' - 'A';
|
||||
if (right >= 'A' && right <= 'Z') right += 'a' - 'A';
|
||||
if (left != right) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private static byte[] copy(ByteView view) {
|
||||
byte[] result = new byte[view.length()];
|
||||
for (int i = 0; i < result.length; i++) result[i] = view.byteAt(i);
|
||||
return result;
|
||||
}
|
||||
|
||||
private static int addWindow(int current, int increment) throws IOException {
|
||||
long next = (long) current + increment;
|
||||
if (next > Integer.MAX_VALUE) throw new IOException("HTTP/2 flow-control window overflow");
|
||||
return (int) next;
|
||||
}
|
||||
|
||||
private static int readInt(byte[] bytes, int offset) {
|
||||
return ((bytes[offset] & 0xff) << 24)
|
||||
| ((bytes[offset + 1] & 0xff) << 16)
|
||||
| ((bytes[offset + 2] & 0xff) << 8)
|
||||
| (bytes[offset + 3] & 0xff);
|
||||
}
|
||||
}
|
||||
@@ -1,7 +0,0 @@
|
||||
package dev.relism.flash.http2.client;
|
||||
|
||||
import dev.relism.flash.models.HeaderView;
|
||||
|
||||
/** Complete response returned by Flash's proxy-oriented HTTP/2 client. */
|
||||
public record Http2ClientResponse(
|
||||
int statusCode, HeaderView headers, byte[] body, HeaderView trailers) {}
|
||||
@@ -1,131 +0,0 @@
|
||||
package dev.relism.flash.http2;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import dev.relism.flash.extension.FlashApp;
|
||||
import dev.relism.flash.extension.FlashConfiguration;
|
||||
import dev.relism.flash.http.HttpMethod;
|
||||
import dev.relism.flash.http.proxy.HttpProxy;
|
||||
import dev.relism.flash.http2.client.Http2Client;
|
||||
import dev.relism.flash.http2.client.Http2ClientResponse;
|
||||
import dev.relism.flash.models.MutableHeaderMap;
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.Socket;
|
||||
import java.net.URI;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class ProxyTrailerRelayTest {
|
||||
private FlashApp upstream;
|
||||
private FlashApp proxy;
|
||||
private Http2Client proxyUpstream;
|
||||
|
||||
@AfterEach
|
||||
void stop() {
|
||||
if (proxyUpstream != null) proxyUpstream.close();
|
||||
if (proxy != null) proxy.stop().join();
|
||||
if (upstream != null) upstream.stop().join();
|
||||
}
|
||||
|
||||
@Test
|
||||
void requestAndResponseTrailersSurviveH2AndH1DownstreamProxyHops() throws Exception {
|
||||
int upstreamPort = freePort();
|
||||
upstream =
|
||||
FlashApp.create(
|
||||
FlashConfiguration.builder()
|
||||
.host("127.0.0.1")
|
||||
.port(upstreamPort)
|
||||
.http2CleartextEnabled(true)
|
||||
.build());
|
||||
upstream.post(
|
||||
"/relay",
|
||||
(request, response) ->
|
||||
response
|
||||
.header("x-query", request.query("mode"))
|
||||
.header("x-private-seen", String.valueOf(request.header("x-private") != null))
|
||||
.body(request.body().bytes())
|
||||
.trailer("x-relayed-trailer", request.trailers().first("x-request-trailer")));
|
||||
upstream.start();
|
||||
|
||||
int proxyPort = freePort();
|
||||
proxyUpstream = new Http2Client();
|
||||
proxy =
|
||||
FlashApp.create(
|
||||
FlashConfiguration.builder()
|
||||
.host("127.0.0.1")
|
||||
.port(proxyPort)
|
||||
.http2CleartextEnabled(true)
|
||||
.build());
|
||||
proxy.post(
|
||||
"/relay",
|
||||
HttpProxy.toHttp2(URI.create("http://127.0.0.1:" + upstreamPort), proxyUpstream));
|
||||
proxy.start();
|
||||
|
||||
MutableHeaderMap h2Headers = fields("connection", "x-private");
|
||||
add(h2Headers, "x-private", "must-not-cross");
|
||||
MutableHeaderMap h2Trailers = fields("x-request-trailer", "from-h2");
|
||||
try (Http2Client downstream = new Http2Client()) {
|
||||
Http2ClientResponse response =
|
||||
downstream.exchange(
|
||||
URI.create("http://127.0.0.1:" + proxyPort + "/relay?mode=h2"),
|
||||
HttpMethod.POST,
|
||||
h2Headers,
|
||||
"hello-h2".getBytes(StandardCharsets.UTF_8),
|
||||
h2Trailers);
|
||||
assertEquals("hello-h2", new String(response.body(), StandardCharsets.UTF_8));
|
||||
assertEquals("h2", response.headers().first("x-query"));
|
||||
assertEquals("false", response.headers().first("x-private-seen"));
|
||||
assertEquals("from-h2", response.trailers().first("x-relayed-trailer"));
|
||||
}
|
||||
|
||||
String h1 = h1Exchange(proxyPort);
|
||||
assertTrue(h1.contains("hello-h1"), h1);
|
||||
assertTrue(h1.toLowerCase().contains("x-query: h1"), h1);
|
||||
assertTrue(h1.toLowerCase().contains("x-private-seen: false"), h1);
|
||||
assertTrue(h1.toLowerCase().contains("x-relayed-trailer: from-h1"), h1);
|
||||
assertFalse(h1.contains("must-not-cross"), h1);
|
||||
}
|
||||
|
||||
private static String h1Exchange(int port) throws Exception {
|
||||
try (Socket socket = new Socket("127.0.0.1", port)) {
|
||||
socket.setSoTimeout(2_000);
|
||||
socket
|
||||
.getOutputStream()
|
||||
.write(
|
||||
("POST /relay?mode=h1 HTTP/1.1\r\n"
|
||||
+ "Host: 127.0.0.1\r\n"
|
||||
+ "Connection: x-private, close\r\n"
|
||||
+ "X-Private: must-not-cross\r\n"
|
||||
+ "Transfer-Encoding: chunked\r\n"
|
||||
+ "Trailer: x-request-trailer\r\n\r\n"
|
||||
+ "8\r\nhello-h1\r\n"
|
||||
+ "0\r\nX-Request-Trailer: from-h1\r\n\r\n")
|
||||
.getBytes(StandardCharsets.US_ASCII));
|
||||
ByteArrayOutputStream bytes = new ByteArrayOutputStream();
|
||||
socket.getInputStream().transferTo(bytes);
|
||||
return bytes.toString(StandardCharsets.UTF_8);
|
||||
}
|
||||
}
|
||||
|
||||
private static MutableHeaderMap fields(String name, String value) {
|
||||
MutableHeaderMap headers = new MutableHeaderMap();
|
||||
add(headers, name, value);
|
||||
return headers;
|
||||
}
|
||||
|
||||
private static void add(MutableHeaderMap headers, String name, String value) {
|
||||
byte[] nameBytes = name.getBytes(StandardCharsets.US_ASCII);
|
||||
byte[] valueBytes = value.getBytes(StandardCharsets.UTF_8);
|
||||
headers.add(nameBytes, 0, nameBytes.length, valueBytes, 0, valueBytes.length);
|
||||
}
|
||||
|
||||
private static int freePort() throws Exception {
|
||||
try (ServerSocket socket = new ServerSocket(0)) {
|
||||
return socket.getLocalPort();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,129 +0,0 @@
|
||||
package dev.relism.flash.http2.client;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import dev.relism.flash.extension.FlashApp;
|
||||
import dev.relism.flash.extension.FlashConfiguration;
|
||||
import dev.relism.flash.http.HttpMethod;
|
||||
import dev.relism.flash.models.MutableHeaderMap;
|
||||
import dev.relism.flash.tls.TestKeystores;
|
||||
import dev.relism.flash.tls.TlsConfig;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.Socket;
|
||||
import java.net.URI;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.nio.file.Path;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
|
||||
class Http2ClientTest {
|
||||
private FlashApp app;
|
||||
|
||||
@AfterEach
|
||||
void stop() {
|
||||
if (app != null) app.stop().join();
|
||||
}
|
||||
|
||||
@Test
|
||||
void reusesOriginConnectionAndExchangesFlowControlledBodiesAndTrailers() throws Exception {
|
||||
int port = freePort();
|
||||
app =
|
||||
FlashApp.create(
|
||||
FlashConfiguration.builder()
|
||||
.host("127.0.0.1")
|
||||
.port(port)
|
||||
.http2CleartextEnabled(true)
|
||||
.build());
|
||||
app.post(
|
||||
"/relay",
|
||||
(request, response) -> {
|
||||
byte[] body = request.body().bytes();
|
||||
String checksum = request.trailers().first("x-request-checksum");
|
||||
return response
|
||||
.header("x-upstream", request.header("x-forwarded-test"))
|
||||
.body(body)
|
||||
.trailer("x-response-checksum", checksum);
|
||||
});
|
||||
app.start();
|
||||
|
||||
byte[] body = new byte[2 * 1024 * 1024 + 31];
|
||||
for (int i = 0; i < body.length; i++) body[i] = (byte) (i * 29);
|
||||
MutableHeaderMap requestHeaders = fields("x-forwarded-test", "yes");
|
||||
MutableHeaderMap requestTrailers = fields("x-request-checksum", "valid");
|
||||
|
||||
try (Http2Client client = new Http2Client()) {
|
||||
URI uri = URI.create("http://127.0.0.1:" + port + "/relay");
|
||||
Http2ClientResponse first =
|
||||
client.exchange(uri, HttpMethod.POST, requestHeaders, body, requestTrailers);
|
||||
Http2ClientResponse second =
|
||||
client.exchange(
|
||||
uri,
|
||||
HttpMethod.POST,
|
||||
requestHeaders,
|
||||
"again".getBytes(StandardCharsets.UTF_8),
|
||||
requestTrailers);
|
||||
|
||||
assertEquals(200, first.statusCode());
|
||||
assertEquals("yes", first.headers().first("x-upstream"));
|
||||
assertArrayEquals(body, first.body());
|
||||
assertEquals("valid", first.trailers().first("x-response-checksum"));
|
||||
assertArrayEquals("again".getBytes(StandardCharsets.UTF_8), second.body());
|
||||
assertEquals(1, client.pooledConnectionCount());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void negotiatesTlsAlpnAndVerifiesTheUpstreamHostname(@TempDir Path directory) throws Exception {
|
||||
int port = freePort();
|
||||
Path keystore =
|
||||
TestKeystores.build(
|
||||
directory,
|
||||
"http2-client.p12",
|
||||
"changeit",
|
||||
TestKeystores.Entry.of("server", "localhost", "localhost"));
|
||||
app =
|
||||
FlashApp.create(
|
||||
FlashConfiguration.builder()
|
||||
.host("127.0.0.1")
|
||||
.port(port)
|
||||
.tls(TlsConfig.keystore(keystore, "changeit"))
|
||||
.http2Enabled(true)
|
||||
.build());
|
||||
app.get("/secure", (request, response) -> "tls-h2");
|
||||
app.start();
|
||||
|
||||
try (Http2Client client = new Http2Client(TestKeystores.trustAllClientContext())) {
|
||||
Http2ClientResponse response =
|
||||
client.get(URI.create("https://localhost:" + port + "/secure"));
|
||||
assertEquals(200, response.statusCode());
|
||||
assertEquals("tls-h2", new String(response.body(), StandardCharsets.UTF_8));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void configuresConnectionsForRequestResponseLatency() throws Exception {
|
||||
try (Socket socket = new Socket()) {
|
||||
assertFalse(socket.getTcpNoDelay());
|
||||
Http2Client.configureLowLatency(socket);
|
||||
assertTrue(socket.getTcpNoDelay());
|
||||
}
|
||||
}
|
||||
|
||||
private static MutableHeaderMap fields(String name, String value) {
|
||||
MutableHeaderMap headers = new MutableHeaderMap();
|
||||
byte[] nameBytes = name.getBytes(StandardCharsets.US_ASCII);
|
||||
byte[] valueBytes = value.getBytes(StandardCharsets.UTF_8);
|
||||
headers.add(nameBytes, 0, nameBytes.length, valueBytes, 0, valueBytes.length);
|
||||
return headers;
|
||||
}
|
||||
|
||||
private static int freePort() throws Exception {
|
||||
try (ServerSocket socket = new ServerSocket(0)) {
|
||||
return socket.getLocalPort();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user