diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/HttpClientConnection.java b/vertx-core/src/main/java/io/vertx/core/http/impl/HttpClientConnection.java index ffc7e8f99f5..d7e118f28ce 100644 --- a/vertx-core/src/main/java/io/vertx/core/http/impl/HttpClientConnection.java +++ b/vertx-core/src/main/java/io/vertx/core/http/impl/HttpClientConnection.java @@ -15,8 +15,8 @@ import io.vertx.core.Future; import io.vertx.core.Handler; import io.vertx.core.MultiMap; -import io.vertx.core.http.HttpConnection; import io.vertx.core.internal.ContextInternal; +import io.vertx.core.internal.http.HttpConnectionInternal; import io.vertx.core.internal.logging.Logger; import io.vertx.core.internal.logging.LoggerFactory; import io.vertx.core.net.HostAndPort; @@ -24,7 +24,7 @@ /** * @author Julien Viet */ -public interface HttpClientConnection extends HttpConnection { +public interface HttpClientConnection extends HttpConnectionInternal { Logger log = LoggerFactory.getLogger(HttpClientConnection.class); diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/HttpServerConnection.java b/vertx-core/src/main/java/io/vertx/core/http/impl/HttpServerConnection.java index 960ceb7f322..cb9774622c4 100644 --- a/vertx-core/src/main/java/io/vertx/core/http/impl/HttpServerConnection.java +++ b/vertx-core/src/main/java/io/vertx/core/http/impl/HttpServerConnection.java @@ -13,13 +13,13 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.Headers; import io.vertx.core.Handler; -import io.vertx.core.http.HttpConnection; import io.vertx.core.internal.ContextInternal; +import io.vertx.core.internal.http.HttpConnectionInternal; /** * @author Julien Viet */ -public interface HttpServerConnection extends HttpConnection { +public interface HttpServerConnection extends HttpConnectionInternal { HttpServerConnection streamHandler(Handler handler); diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1Connection.java b/vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1Connection.java index c06b2394fce..f35485f8f38 100644 --- a/vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1Connection.java +++ b/vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1Connection.java @@ -29,17 +29,20 @@ import io.vertx.core.net.impl.VertxConnection; import java.time.Duration; +import java.util.HashMap; +import java.util.Map; /** * @author Julien Viet */ -abstract class Http1Connection extends VertxConnection implements io.vertx.core.http.HttpConnection { +abstract class Http1Connection extends VertxConnection implements HttpConnection { protected boolean closeInitiated; protected Duration shutdownInitiated; protected ChannelPromise closePromise; private Handler shutdownHandler; + private Map attachments; Http1Connection(ContextInternal context, ChannelHandlerContext chctx) { this(context, chctx, false); @@ -142,6 +145,26 @@ public Future ping(Buffer data) { throw new UnsupportedOperationException("HTTP/1.x connections don't support PING"); } + @Override + protected void handleClosed() { + synchronized (this) { + attachments = null; + } + super.handleClosed(); + } + + @SuppressWarnings("unchecked") + public synchronized T attachment(Object key) { + return attachments != null ? (T) attachments.get(key) : null; + } + + public synchronized void attach(Object key, Object value) { + if (attachments == null) { + attachments = new HashMap<>(); + } + attachments.put(key, value); + } + @Override protected long sizeof(Object obj) { // https://github.com/netty/netty/issues/12708 diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/http2/codec/Http2ConnectionImpl.java b/vertx-core/src/main/java/io/vertx/core/http/impl/http2/codec/Http2ConnectionImpl.java index 796f51ff5f5..8fd4d5e743a 100644 --- a/vertx-core/src/main/java/io/vertx/core/http/impl/http2/codec/Http2ConnectionImpl.java +++ b/vertx-core/src/main/java/io/vertx/core/http/impl/http2/codec/Http2ConnectionImpl.java @@ -43,6 +43,7 @@ import java.time.Duration; import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.concurrent.ConcurrentHashMap; import java.util.Map; import java.util.Objects; @@ -74,6 +75,7 @@ private static ByteBuf safeBuffer(ByteBuf buf) { private GoAway goAwayStatus; private int windowSize; private long maxConcurrentStreams; + private final Map attachments = new ConcurrentHashMap<>(); public Http2ConnectionImpl(ContextInternal context, VertxHttp2ConnectionHandler handler) { super(context, handler.context()); @@ -87,9 +89,19 @@ public Http2ConnectionImpl(ContextInternal context, VertxHttp2ConnectionHandler @Override public void handleClosed() { + attachments.clear(); super.handleClosed(); } + @SuppressWarnings("unchecked") + public T attachment(Object key) { + return (T) attachments.get(key); + } + + public void attach(Object key, Object value) { + attachments.put(key, value); + } + protected void handleIdle(IdleStateEvent event) { log.debug("The connection will be closed due to timeout"); chctx.close(); diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/http2/multiplex/Http2MultiplexConnection.java b/vertx-core/src/main/java/io/vertx/core/http/impl/http2/multiplex/Http2MultiplexConnection.java index c51f3c090d5..efa2444a69d 100644 --- a/vertx-core/src/main/java/io/vertx/core/http/impl/http2/multiplex/Http2MultiplexConnection.java +++ b/vertx-core/src/main/java/io/vertx/core/http/impl/http2/multiplex/Http2MultiplexConnection.java @@ -55,6 +55,8 @@ import java.time.Duration; import java.util.ArrayDeque; import java.util.Deque; +import java.util.concurrent.ConcurrentHashMap; +import java.util.Map; import java.util.Objects; public abstract class Http2MultiplexConnection extends ConnectionBase implements HttpConnection { @@ -69,6 +71,7 @@ public abstract class Http2MultiplexConnection extends Co private Handler shutdownHandler; private Handler goAwayHandler; private Handler pingHandler; + private final Map attachments = new ConcurrentHashMap<>(); public Http2MultiplexConnection(Http2MultiplexHandler handler, TransportMetrics transportMetrics, ChannelHandlerContext chctx, ContextInternal context) { super(context, chctx); @@ -418,6 +421,21 @@ void onClose() { handleClosed(); } + @Override + protected void handleClosed() { + attachments.clear(); + super.handleClosed(); + } + + @SuppressWarnings("unchecked") + public T attachment(Object key) { + return (T) attachments.get(key); + } + + public void attach(Object key, Object value) { + attachments.put(key, value); + } + void onException(Throwable err) { handleException(err); if (err instanceof Http2Exception) { diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/http3/Http3Connection.java b/vertx-core/src/main/java/io/vertx/core/http/impl/http3/Http3Connection.java index 49d9b13321b..7ec198fd7fc 100644 --- a/vertx-core/src/main/java/io/vertx/core/http/impl/http3/Http3Connection.java +++ b/vertx-core/src/main/java/io/vertx/core/http/impl/http3/Http3Connection.java @@ -33,6 +33,7 @@ import javax.net.ssl.SSLSession; import java.time.Duration; import java.util.ArrayList; +import java.util.concurrent.ConcurrentHashMap; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -55,6 +56,7 @@ public abstract class Http3Connection implements HttpConnection { private Http3Settings localSettings; private Http3Settings remoteSettings; private Handler remoteSettingsHandler; + private final Map attachments = new ConcurrentHashMap<>(); public Http3Connection(QuicConnectionInternal connection, Http3Settings localSettings, Http3FrameLogger frameLogger) { this.streams = new LongObjectHashMap<>(); @@ -291,12 +293,22 @@ private void handleGrace(QuicStreamChannel localControlStream) { } protected void handleClosed() { + attachments.clear(); Handler handler = closeHandler; if (handler != null) { context.emit(null, handler); } } + @SuppressWarnings("unchecked") + public T attachment(Object key) { + return (T) attachments.get(key); + } + + public void attach(Object key, Object value) { + attachments.put(key, value); + } + private void sendGoAway(QuicStreamChannel controlStream, long streamId, PromiseInternal promise) { Iterator>> iterator = streams.entries().iterator(); List> toCancel = new ArrayList<>(); diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/tcp/Http2UpgradeClientConnection.java b/vertx-core/src/main/java/io/vertx/core/http/impl/tcp/Http2UpgradeClientConnection.java index 7eeed6ccabe..0211fc8420e 100644 --- a/vertx-core/src/main/java/io/vertx/core/http/impl/tcp/Http2UpgradeClientConnection.java +++ b/vertx-core/src/main/java/io/vertx/core/http/impl/tcp/Http2UpgradeClientConnection.java @@ -38,6 +38,7 @@ import javax.net.ssl.SSLSession; import java.time.Duration; +import java.util.concurrent.ConcurrentHashMap; import java.util.List; import java.util.Map; @@ -56,6 +57,7 @@ public class Http2UpgradeClientConnection implements io.vertx.core.http.impl.Htt private final ClientMetrics clientMetrics; private io.vertx.core.http.impl.HttpClientConnection current; private boolean upgradeProcessed; + private final Map attachments = new ConcurrentHashMap<>(); private Handler closeHandler; private Handler shutdownHandler; @@ -73,6 +75,7 @@ public Http2UpgradeClientConnection(Http1ClientConnection connection, ClientMetr this.current = connection; this.upgrade = upgrade; this.clientMetrics = clientMetrics; + connection.closeHandler(this::handleClosed); } public io.vertx.core.http.impl.HttpClientConnection unwrap() { @@ -334,7 +337,7 @@ void handleUpgrade(io.vertx.core.http.impl.HttpClientConnection conn, HttpClient pushHandler = null; closeHandler = null; upgradedConnection.current = conn; - conn.closeHandler(upgradedConnection.closeHandler); + conn.closeHandler(upgradedConnection::handleClosed); conn.exceptionHandler(upgradedConnection.exceptionHandler); conn.pingHandler(upgradedConnection.pingHandler); conn.goAwayHandler(upgradedConnection.goAwayHandler); @@ -776,13 +779,22 @@ public HttpConnection shutdownHandler(@Nullable Handler handler) { @Override public HttpConnection closeHandler(Handler handler) { - if (current instanceof Http1ClientConnection) { - closeHandler = handler; - } - current.closeHandler(handler); + closeHandler = handler; + current.closeHandler(this::handleClosed); return this; } + private void handleClosed(Void event) { + attachments.clear(); + Handler handler; + synchronized (this) { + handler = closeHandler; + } + if (handler != null) { + handler.handle(event); + } + } + @Override public HttpConnection exceptionHandler(Handler handler) { if (current instanceof Http1ClientConnection) { @@ -913,6 +925,17 @@ public String toString() { return getClass().getSimpleName() + "[current=" + current.getClass().getSimpleName() + "]"; } + @Override + @SuppressWarnings("unchecked") + public T attachment(Object key) { + return (T) attachments.get(key); + } + + @Override + public void attach(Object key, Object value) { + attachments.put(key, value); + } + /** * The outcome of the upgrade signalled by the upgrade. */ diff --git a/vertx-core/src/main/java/io/vertx/core/internal/http/HttpConnectionInternal.java b/vertx-core/src/main/java/io/vertx/core/internal/http/HttpConnectionInternal.java new file mode 100644 index 00000000000..bb72b29c79f --- /dev/null +++ b/vertx-core/src/main/java/io/vertx/core/internal/http/HttpConnectionInternal.java @@ -0,0 +1,39 @@ +/* + * Copyright (c) 2011-2026 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License 2.0 which is available at + * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0 + * which is available at https://www.apache.org/licenses/LICENSE-2.0. + * + * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0 + */ +package io.vertx.core.internal.http; + +import io.vertx.core.http.HttpConnection; + +/** + * Extends to expose internal methods that are necessary for integration. + */ +public interface HttpConnectionInternal extends HttpConnection { + + /** + * Get an attachment previously stored with {@link #attach(Object, Object)}. + *

+ * Attachments are scoped to the lifecycle of the connection: they are discarded when the + * connection closes and do not need to be cleaned up manually. + * + * @param key the attachment key + * @return the attachment value, or {@code null} if absent + */ + T attachment(Object key); + + /** + * Store an attachment on this connection, associated with {@code key}. + * + * @param key the attachment key + * @param value the attachment value + */ + void attach(Object key, Object value); + +} diff --git a/vertx-core/src/test/java/io/vertx/tests/http/Http2ClientTest.java b/vertx-core/src/test/java/io/vertx/tests/http/Http2ClientTest.java index bb7a54f0c85..3500c83e365 100644 --- a/vertx-core/src/test/java/io/vertx/tests/http/Http2ClientTest.java +++ b/vertx-core/src/test/java/io/vertx/tests/http/Http2ClientTest.java @@ -30,6 +30,7 @@ import io.vertx.core.http.Http2Settings; import io.vertx.core.internal.ContextInternal; import io.vertx.core.internal.buffer.BufferInternal; +import io.vertx.core.internal.http.HttpConnectionInternal; import io.vertx.core.http.impl.tcp.Http2UpgradeClientConnection; import io.vertx.core.http.impl.HttpClientConnection; import io.vertx.core.net.NetServer; @@ -1729,6 +1730,41 @@ public void testClearTextUpgrade() throws Exception { Assert.assertEquals(Arrays.asList("GET", "GET"), requests); } + @Test + public void testClearTextUpgradePreservesConnectionAttachment() throws Exception { + Assume.assumeTrue(testAddress.isInetSocket()); + Object key = new Object(); + Object value = new Object(); + ServerBootstrap bootstrap = createH2CServer((dec, enc) -> new Http2EventAdapter() { + @Override + public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency, short weight, boolean exclusive, int padding, boolean endStream) throws Http2Exception { + enc.writeHeaders(ctx, streamId, new DefaultHttp2Headers().status("200"), 0, true, ctx.newPromise()); + ctx.flush(); + } + }, upgrade -> { + }, true); + ChannelFuture channel = bootstrap.bind(DEFAULT_HTTPS_HOST, DEFAULT_HTTPS_PORT).sync(); + try { + client.close(); + client = vertx.createHttpClient(clientOptions + .setUseAlpn(false) + .setSsl(false) + .setHttp2ClearTextUpgrade(true)); + client.request(requestOptions).onComplete(TestUtils.onSuccess(request -> { + HttpConnectionInternal connection = (HttpConnectionInternal) request.connection(); + connection.attach(key, value); + request.send().onComplete(TestUtils.onSuccess(response -> { + Assert.assertSame(connection, response.request().connection()); + Assert.assertSame(value, connection.attachment(key)); + testComplete(); + })); + })); + await(); + } finally { + channel.channel().close().sync(); + } + } + @Test public void testClearTextUpgradeWithPreflightRequest() throws Exception { List requests = testClearText(true, true);