Add proper initialisation of max concurrent streams when successful HTTP/1.1 to HTTP/2 upgrade (#3832)

- Reset maxConcurrentStreams to 0 when invalidating a connection
- Add logging when an upgrade stream is opened

Signed-off-by: Violeta Georgieva <696661+violetagg@users.noreply.github.com>

Author: violetagg

reactor-netty-http/src/main/java/reactor/netty/http/client/Http2Pool.java

@@ -972,7 +972,6 @@ static class Slot extends AtomicBoolean implements PooledRefMetadata {
 				this.applicationProtocol = null;
 			}
 			initMaxConcurrentStreams();
-			TOTAL_MAX_CONCURRENT_STREAMS.addAndGet(this.pool, this.maxConcurrentStreams);
 		}
 
 		void initMaxConcurrentStreams() {
@@ -982,6 +981,7 @@ void initMaxConcurrentStreams() {
 				this.maxConcurrentStreams = pool.maxConcurrentStreams == -1 ? maxConcurrentStreams :
 						Math.min(pool.maxConcurrentStreams, maxConcurrentStreams);
 			}
+			TOTAL_MAX_CONCURRENT_STREAMS.addAndGet(this.pool, this.maxConcurrentStreams);
 		}
 
 		boolean canOpenStream() {
@@ -1072,6 +1072,7 @@ void invalidate() {
 				}
 				pool.poolConfig.allocationStrategy().returnPermits(1);
 				TOTAL_MAX_CONCURRENT_STREAMS.addAndGet(this.pool, -maxConcurrentStreams);
+				maxConcurrentStreams = 0;
 			}
 		}
 

reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClientConfig.java

@@ -99,6 +99,8 @@
 import static reactor.netty.ReactorNetty.format;
 import static reactor.netty.ReactorNetty.setChannelContext;
 import static reactor.netty.http.client.Http2ConnectionProvider.OWNER;
+import static reactor.netty.http.client.Http2ConnectionProvider.http2PooledRef;
+import static reactor.netty.http.client.Http2ConnectionProvider.logStreamsState;
 import static reactor.netty.http.client.Http3Codec.newHttp3ClientConnectionHandler;
 
 /**
@@ -892,9 +894,11 @@ public void handlerAdded(ChannelHandlerContext ctx) {
 			ConnectionObserver channelOwner = ctx.channel().attr(OWNER).get();
 			Http2ConnectionProvider.DisposableAcquire owner = null;
 			ConnectionObserver obs = null;
+			Http2Pool.Http2PooledRef http2PooledRef = null;
 			if (channelOwner instanceof Http2ConnectionProvider.DisposableAcquire) {
 				owner = (Http2ConnectionProvider.DisposableAcquire) channelOwner;
 				obs = owner.obs;
+				http2PooledRef = http2PooledRef(owner.pooledRef);
 			}
 			if (responseTimeoutHandler != null) {
 				pipeline.remove(NettyPipeline.ResponseTimeoutHandler);
@@ -913,6 +917,10 @@ public void handlerAdded(ChannelHandlerContext ctx) {
 			}
 			pipeline.remove(NettyPipeline.ReactiveBridge);
 			pipeline.remove(this);
+
+			if (http2PooledRef != null) {
+				http2PooledRef.slot.initMaxConcurrentStreams();
+			}
 		}
 	}
 
@@ -972,6 +980,9 @@ protected void initChannel(Channel ch) {
 				}
 				addStreamHandlers(ch, observer.then(new StreamConnectionObserver(owner.currentContext())), opsFactory,
 						acceptGzip, metricsRecorder, proxyAddress, remoteAddress, responseTimeoutMillis, uriTagValue);
+				if (log.isDebugEnabled()) {
+					logStreamsState(ch, http2PooledRef(owner.pooledRef).slot, "Stream opened");
+				}
 			}
 			else {
 				// Handle server pushes (inbound streams)