Ensure the HTTP/2 upgrade stream inherits the proper state from the original HTTP/1.1 connection when successful upgrade from HTTP/1.1 to HTTP/2 (#3838)

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

Author: violetagg

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

@@ -437,7 +437,7 @@ public void operationComplete(Future<Http2StreamChannel> future) {
 						setChannelContext(ch, currentContext());
 					}
 					HttpClientConfig.addStreamHandlers(ch, obs.then(new HttpClientConfig.StreamConnectionObserver(currentContext())),
-							opsFactory, acceptGzip, metricsRecorder, proxyAddress, remoteAddress, -1, uriTagValue);
+							opsFactory, acceptGzip, false, metricsRecorder, proxyAddress, remoteAddress, -1, uriTagValue);
 
 					if (log.isDebugEnabled()) {
 						logStreamsState(ch, http2PooledRef.slot, "Stream opened");

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

@@ -600,6 +600,7 @@ static void addStreamHandlers(
 			ConnectionObserver obs,
 			ChannelOperations.OnSetup opsFactory,
 			boolean acceptGzip,
+			boolean copyState,
 			@Nullable ChannelMetricsRecorder metricsRecorder,
 			@Nullable SocketAddress proxyAddress,
 			SocketAddress remoteAddress,
@@ -677,6 +678,9 @@ else if (metricsRecorder instanceof ContextAwareHttpClientMetricsRecorder) {
 		ChannelOperations<?, ?> ops = opsFactory.create(Connection.from(ch), obs, null);
 		if (ops != null) {
 			ops.bind();
+			if (copyState && ops instanceof HttpClientOperations) {
+				HttpClientOperations.copyState(((HttpClientOperations) ops));
+			}
 		}
 	}
 
@@ -979,7 +983,7 @@ protected void initChannel(Channel ch) {
 					setChannelContext(ch, owner.currentContext());
 				}
 				addStreamHandlers(ch, observer.then(new StreamConnectionObserver(owner.currentContext())), opsFactory,
-						acceptGzip, metricsRecorder, proxyAddress, remoteAddress, responseTimeoutMillis, uriTagValue);
+						acceptGzip, true, metricsRecorder, proxyAddress, remoteAddress, responseTimeoutMillis, uriTagValue);
 				if (log.isDebugEnabled()) {
 					logStreamsState(ch, http2PooledRef(owner.pooledRef).slot, "Stream opened");
 				}

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

@@ -951,6 +951,19 @@ final void withWebsocketSupport(WebsocketClientSpec websocketClientSpec) {
 		}
 	}
 
+	static void copyState(HttpClientOperations streamOps) {
+		ChannelOperations<?, ?> ops = ChannelOperations.get(streamOps.channel().parent());
+		if (ops instanceof HttpClientOperations) {
+			HttpClientOperations parentOps = (HttpClientOperations) ops;
+			if (parentOps.hasSentBody()) {
+				streamOps.markSentHeaderAndBody();
+			}
+			else if (parentOps.hasSentHeaders()) {
+				streamOps.markSentHeaders();
+			}
+		}
+	}
+
 	static Throwable addOutboundErrorCause(Throwable exception, @Nullable Throwable cause) {
 		if (cause != null) {
 			cause.setStackTrace(new StackTraceElement[0]);