Add HTTP/2 connection multiplexing with dedicated registry, semaphore bypass, and GOAWAY handling

Author: hyperxpro

client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java

@@ -282,18 +282,39 @@ public interface AsyncHttpClientConfig {
      */
     boolean isHttp2Enabled();
 
+    /**
+     * @return the HTTP/2 initial window size in bytes, defaults to 65535
+     */
     int getHttp2InitialWindowSize();
 
+    /**
+     * @return the HTTP/2 max frame size in bytes, must be between 16384 and 16777215 per RFC 7540 §4.2
+     */
     int getHttp2MaxFrameSize();
 
+    /**
+     * @return the HTTP/2 HPACK header table size in bytes, defaults to 4096
+     */
     int getHttp2HeaderTableSize();
 
+    /**
+     * @return the HTTP/2 max header list size in bytes, defaults to 8192
+     */
     int getHttp2MaxHeaderListSize();
 
+    /**
+     * @return the HTTP/2 max concurrent streams per connection, -1 means unlimited (server-controlled)
+     */
     int getHttp2MaxConcurrentStreams();
 
+    /**
+     * @return the interval between HTTP/2 PING keepalive frames, {@link Duration#ZERO} disables pinging
+     */
     Duration getHttp2PingInterval();
 
+    /**
+     * @return true if cleartext HTTP/2 (h2c) via prior knowledge is enabled for non-TLS connections
+     */
     boolean isHttp2CleartextEnabled();
 
     /**

client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java

@@ -413,6 +413,14 @@ private DefaultAsyncHttpClientConfig(// http
             throw new IllegalArgumentException("Native Transport must be enabled to use Epoll Native Transport only");
         }
 
+        if (http2MaxFrameSize < 16384 || http2MaxFrameSize > 16777215) {
+            throw new IllegalArgumentException("HTTP/2 max frame size must be between 16384 and 16777215 per RFC 7540 §4.2");
+        }
+
+        if (http2InitialWindowSize < 0) {
+            throw new IllegalArgumentException("HTTP/2 initial window size must be non-negative");
+        }
+
         this.allocator = allocator;
         this.nettyTimer = nettyTimer;
         this.threadFactory = threadFactory;

client/src/main/java/org/asynchttpclient/HttpProtocol.java

@@ -22,7 +22,7 @@ public enum HttpProtocol {
 
     HTTP_1_0("HTTP/1.0"),
     HTTP_1_1("HTTP/1.1"),
-    HTTP_2("HTTP/2.0");
+    HTTP_2("HTTP/2");
 
     private final String text;
 
@@ -31,7 +31,7 @@ public enum HttpProtocol {
     }
 
     /**
-     * @return the protocol version string (e.g. "HTTP/1.1", "HTTP/2.0")
+     * @return the protocol version string (e.g. "HTTP/1.1", "HTTP/2")
      */
     public String getText() {
         return text;

client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -20,6 +20,7 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFactory;
 import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
@@ -41,16 +42,17 @@
 import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
 import io.netty.handler.codec.http2.Http2MultiplexHandler;
 import io.netty.handler.codec.http2.Http2Settings;
+import io.netty.handler.codec.http2.Http2GoAwayFrame;
 import io.netty.handler.codec.http2.Http2SettingsFrame;
 import io.netty.handler.codec.http2.Http2StreamChannel;
 import io.netty.handler.logging.LogLevel;
 import io.netty.handler.logging.LoggingHandler;
-import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.handler.proxy.ProxyHandler;
 import io.netty.handler.proxy.Socks4ProxyHandler;
 import io.netty.handler.proxy.Socks5ProxyHandler;
 import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.stream.ChunkedWriteHandler;
+import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.resolver.NameResolver;
 import io.netty.util.Timer;
 import io.netty.util.concurrent.DefaultThreadFactory;
@@ -89,6 +91,7 @@
 import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -122,6 +125,7 @@ public class ChannelManager {
 
     private final ChannelPool channelPool;
     private final ChannelGroup openChannels;
+    private final ConcurrentHashMap<Object, Channel> http2Connections = new ConcurrentHashMap<>();
 
     private AsyncHttpClientHandler wsHandler;
     private Http2Handler http2Handler;
@@ -338,6 +342,59 @@ public final void tryToOfferChannelToPool(Channel channel, AsyncHandler<?> async
         }
     }
 
+    /**
+     * Registers an HTTP/2 connection in the registry for the given partition key.
+     * The connection stays in the registry (not the regular pool) to allow multiplexing —
+     * multiple requests can share the same connection concurrently.
+     */
+    public void registerHttp2Connection(Object partitionKey, Channel channel) {
+        Http2ConnectionState state = channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get();
+        if (state != null) {
+            state.setPartitionKey(partitionKey);
+        }
+        http2Connections.put(partitionKey, channel);
+        // Auto-remove from registry when the connection closes
+        channel.closeFuture().addListener(future -> removeHttp2Connection(partitionKey, channel));
+    }
+
+    /**
+     * Removes an HTTP/2 connection from the registry, but only if it's the currently registered
+     * connection for that partition key (avoids removing a replacement connection).
+     */
+    public void removeHttp2Connection(Object partitionKey, Channel channel) {
+        http2Connections.remove(partitionKey, channel);
+    }
+
+    /**
+     * Returns an active, non-draining HTTP/2 connection for the given partition key, or {@code null}.
+     * Unlike the regular pool, this does NOT remove the connection — it remains available for
+     * concurrent multiplexed requests.
+     */
+    public Channel pollHttp2Connection(Object partitionKey) {
+        Channel channel = http2Connections.get(partitionKey);
+        if (channel == null) {
+            return null;
+        }
+        if (!channel.isActive()) {
+            http2Connections.remove(partitionKey, channel);
+            return null;
+        }
+        Http2ConnectionState state = channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get();
+        if (state != null && state.isDraining()) {
+            return null;
+        }
+        return channel;
+    }
+
+    /**
+     * Polls for an HTTP/2 connection by URI/virtualHost/proxy, using the same partition key logic
+     * as the regular pool. Returns the connection without removing it from the registry.
+     */
+    public Channel pollHttp2(Uri uri, String virtualHost, ProxyServer proxy, ChannelPoolPartitioning connectionPoolPartitioning) {
+        Object partitionKey = connectionPoolPartitioning.getPartitionKey(uri, virtualHost, proxy);
+        return pollHttp2Connection(partitionKey);
+    }
+
     public Channel poll(Uri uri, String virtualHost, ProxyServer proxy, ChannelPoolPartitioning connectionPoolPartitioning) {
         Object partitionKey = connectionPoolPartitioning.getPartitionKey(uri, virtualHost, proxy);
         return channelPool.poll(partitionKey);
@@ -348,6 +405,7 @@ public void removeAll(Channel connection) {
     }
 
     private void doClose() {
+        http2Connections.clear();
         ChannelGroupFuture groupFuture = openChannels.close();
         channelPool.destroy();
         groupFuture.addListener(future -> sslEngineFactory.destroy());
@@ -644,7 +702,7 @@ protected void initChannel(Channel ch) {
         // Install SETTINGS listener to update MAX_CONCURRENT_STREAMS from server
         pipeline.addLast("http2-settings-listener", new ChannelInboundHandlerAdapter() {
             @Override
-            public void channelRead(io.netty.channel.ChannelHandlerContext ctx, Object msg) throws Exception {
+            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                 if (msg instanceof Http2SettingsFrame) {
                     Http2SettingsFrame settingsFrame = (Http2SettingsFrame) msg;
                     Long maxStreams = settingsFrame.settings().maxConcurrentStreams();
@@ -659,10 +717,38 @@ public void channelRead(io.netty.channel.ChannelHandlerContext ctx, Object msg)
             }
         });
 
+        // Install GOAWAY handler on the parent channel to mark the connection as draining
+        // and remove it from the HTTP/2 registry. GOAWAY is a connection-level frame that
+        // arrives on the parent channel, not on stream child channels.
+        pipeline.addLast("http2-goaway-listener", new ChannelInboundHandlerAdapter() {
+            @Override
+            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+                if (msg instanceof Http2GoAwayFrame) {
+                    Http2GoAwayFrame goAwayFrame = (Http2GoAwayFrame) msg;
+                    int lastStreamId = goAwayFrame.lastStreamId();
+                    Http2ConnectionState connState = ctx.channel().attr(Http2ConnectionState.HTTP2_STATE_KEY).get();
+                    if (connState != null) {
+                        connState.setDraining(lastStreamId);
+                        Object pk = connState.getPartitionKey();
+                        if (pk != null) {
+                            removeHttp2Connection(pk, ctx.channel());
+                        }
+                    }
+                    LOGGER.debug("HTTP/2 GOAWAY received on {}, lastStreamId={}, errorCode={}",
+                            ctx.channel(), lastStreamId, goAwayFrame.errorCode());
+                    // Close the connection when no more active streams
+                    if (connState != null && connState.getActiveStreams() <= 0) {
+                        closeChannel(ctx.channel());
+                    }
+                }
+                ctx.fireChannelRead(msg);
+            }
+        });
+
         // Install PING handler for keepalive if configured
         long pingIntervalMs = config.getHttp2PingInterval().toMillis();
         if (pingIntervalMs > 0) {
-            pipeline.addLast("http2-idle-state", new IdleStateHandler(0, 0, pingIntervalMs, java.util.concurrent.TimeUnit.MILLISECONDS));
+            pipeline.addLast("http2-idle-state", new IdleStateHandler(0, 0, pingIntervalMs, TimeUnit.MILLISECONDS));
             pipeline.addLast("http2-ping", new Http2PingHandler());
         }
     }
@@ -732,4 +818,4 @@ public boolean isOpen() {
     public boolean isHttp2CleartextEnabled() {
         return config.isHttp2Enabled() && config.isHttp2CleartextEnabled();
     }
-}
+}

client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java

@@ -35,6 +35,7 @@ public class Http2ConnectionState {
     private final AtomicBoolean draining = new AtomicBoolean(false);
     private volatile int lastGoAwayStreamId = Integer.MAX_VALUE;
     private final ConcurrentLinkedQueue<Runnable> pendingOpeners = new ConcurrentLinkedQueue<>();
+    private volatile Object partitionKey;
 
     public boolean tryAcquireStream() {
         if (draining.get()) {
@@ -97,4 +98,12 @@ public void setDraining(int lastStreamId) {
     public int getLastGoAwayStreamId() {
         return lastGoAwayStreamId;
     }
+
+    public void setPartitionKey(Object partitionKey) {
+        this.partitionKey = partitionKey;
+    }
+
+    public Object getPartitionKey() {
+        return partitionKey;
+    }
 }

client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java

@@ -24,6 +24,7 @@
 import org.asynchttpclient.netty.NettyResponseFuture;
 import org.asynchttpclient.netty.SimpleFutureListener;
 import org.asynchttpclient.netty.future.StackTraceInspector;
+import org.asynchttpclient.channel.ChannelPoolPartitioning;
 import org.asynchttpclient.netty.request.NettyRequestSender;
 import org.asynchttpclient.netty.timeout.TimeoutsHolder;
 import org.asynchttpclient.proxy.ProxyServer;
@@ -81,14 +82,9 @@ private void writeRequest(Channel channel) {
     }
 
     public void onSuccess(Channel channel, InetSocketAddress remoteAddress) {
-        if (connectionSemaphore != null) {
-            // transfer lock from future to channel
-            Object partitionKeyLock = future.takePartitionKeyLock();
-
-            if (partitionKeyLock != null) {
-                channel.closeFuture().addListener(future -> connectionSemaphore.releaseChannelLock(partitionKeyLock));
-            }
-        }
+        // Take the semaphore lock from the future. For HTTP/1.1, we'll transfer it to channel.closeFuture().
+        // For HTTP/2, we release it immediately after ALPN negotiation since the connection is multiplexed.
+        final Object partitionKeyLock = (connectionSemaphore != null) ? future.takePartitionKeyLock() : null;
 
         Channels.setActiveToken(channel);
         TimeoutsHolder timeoutsHolder = future.getTimeoutsHolder();
@@ -140,6 +136,7 @@ protected void onSuccess(Channel value) {
                         return;
                     }
                     // After SSL handshake to proxy, continue with normal proxy request
+                    attachSemaphoreToChannelClose(channel, partitionKeyLock);
                     writeRequest(channel);
                 }
 
@@ -190,6 +187,10 @@ protected void onSuccess(Channel value) {
                     String alpnProtocol = sslHandler.applicationProtocol();
                     if (ApplicationProtocolNames.HTTP_2.equals(alpnProtocol)) {
                         channelManager.upgradePipelineToHttp2(channel.pipeline());
+                        registerHttp2AndReleaseSemaphore(channel);
+                        releaseSemaphoreImmediately(partitionKeyLock);
+                    } else {
+                        attachSemaphoreToChannelClose(channel, partitionKeyLock);
                     }
                     writeRequest(channel);
                 }
@@ -211,11 +212,48 @@ protected void onFailure(Throwable cause) {
             // h2c (cleartext HTTP/2 prior knowledge): upgrade to HTTP/2 without TLS
             if (!uri.isSecured() && channelManager.isHttp2CleartextEnabled()) {
                 channelManager.upgradePipelineToHttp2(channel.pipeline());
+                registerHttp2AndReleaseSemaphore(channel);
+                releaseSemaphoreImmediately(partitionKeyLock);
+            } else {
+                attachSemaphoreToChannelClose(channel, partitionKeyLock);
             }
             writeRequest(channel);
         }
     }
 
+    /**
+     * Attaches the semaphore lock to the channel's close future (HTTP/1.1 behavior).
+     * The semaphore slot is released when the connection closes.
+     */
+    private void attachSemaphoreToChannelClose(Channel channel, Object partitionKeyLock) {
+        if (connectionSemaphore != null && partitionKeyLock != null) {
+            channel.closeFuture().addListener(f -> connectionSemaphore.releaseChannelLock(partitionKeyLock));
+        }
+    }
+
+    /**
+     * Releases the semaphore lock immediately (HTTP/2 behavior).
+     * HTTP/2 connections are multiplexed, so the semaphore should not be held
+     * for the lifetime of the connection.
+     */
+    private void releaseSemaphoreImmediately(Object partitionKeyLock) {
+        if (connectionSemaphore != null && partitionKeyLock != null) {
+            connectionSemaphore.releaseChannelLock(partitionKeyLock);
+        }
+    }
+
+    /**
+     * Registers the HTTP/2 connection in the channel manager's H2 registry.
+     */
+    private void registerHttp2AndReleaseSemaphore(Channel channel) {
+        Request request = future.getTargetRequest();
+        Uri uri = request.getUri();
+        ProxyServer proxy = future.getProxyServer();
+        ChannelPoolPartitioning partitioning = request.getChannelPoolPartitioning();
+        Object partitionKey = partitioning.getPartitionKey(uri, request.getVirtualHost(), proxy);
+        channelManager.registerHttp2Connection(partitionKey, channel);
+    }
+
     public void onFailure(Channel channel, Throwable cause) {
 
         // beware, channel can be null