feat: add per-connection attachment storage to HttpConnectionInternal

vertx-core/src/main/java/io/vertx/core/http/impl/HttpClientConnection.java

@@ -15,16 +15,16 @@
 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;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-public interface HttpClientConnection extends HttpConnection {
+public interface HttpClientConnection extends HttpConnectionInternal {
 
   Logger log = LoggerFactory.getLogger(HttpClientConnection.class);
 

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 <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-public interface HttpServerConnection extends HttpConnection {
+public interface HttpServerConnection extends HttpConnectionInternal {
 
   HttpServerConnection streamHandler(Handler<HttpServerStream> handler);
 

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 <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-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<Void> shutdownHandler;
+  private Map<Object, Object> attachments;
 
   Http1Connection(ContextInternal context, ChannelHandlerContext chctx) {
     this(context, chctx, false);
@@ -142,6 +145,26 @@ public Future<Buffer> 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> 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

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<Object, Object> 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> 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();

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<S extends Http2Stream> extends ConnectionBase implements HttpConnection {
@@ -69,6 +71,7 @@ public abstract class Http2MultiplexConnection<S extends Http2Stream> extends Co
   private Handler<Void> shutdownHandler;
   private Handler<GoAway> goAwayHandler;
   private Handler<Buffer> pingHandler;
+  private final Map<Object, Object> 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> 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) {

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<HttpSettings> remoteSettingsHandler;
+  private final Map<Object, Object> 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<Void> handler = closeHandler;
     if (handler != null) {
       context.emit(null, handler);
     }
   }
 
+  @SuppressWarnings("unchecked")
+  public <T> 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<Void> promise) {
     Iterator<LongObjectMap.PrimitiveEntry<Http3Stream<?, ?>>> iterator = streams.entries().iterator();
     List<Http3Stream<?, ?>> toCancel = new ArrayList<>();

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<Object, Object> attachments = new ConcurrentHashMap<>();
 
   private Handler<Void> closeHandler;
   private Handler<Void> 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<Void> handler) {
 
   @Override
   public HttpConnection closeHandler(Handler<Void> 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<Void> handler;
+    synchronized (this) {
+      handler = closeHandler;
+    }
+    if (handler != null) {
+      handler.handle(event);
+    }
+  }
+
   @Override
   public HttpConnection exceptionHandler(Handler<Throwable> handler) {
     if (current instanceof Http1ClientConnection) {
@@ -913,6 +925,17 @@ public String toString() {
     return getClass().getSimpleName() + "[current=" + current.getClass().getSimpleName() + "]";
   }
 
+  @Override
+  @SuppressWarnings("unchecked")
+  public <T> 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.
    */

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)}.
+   * <p/>
+   * 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> 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);
+
+}

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<String> requests = testClearText(true, true);