HttpServerRequest::receiveContent() never emits any value nor completes when HTTP/1.1 TLS Upgrade (RFC-2817) kicks in

reactor-netty-http/src/main/java/reactor/netty/http/server/HttpServerOperations.java

@@ -814,6 +814,10 @@ else if (msg instanceof HttpRequest) {
 				}
 				else {
 					request.release();
+					// HTTP/1.1 TLS Upgrade (RFC-2817) on empty requests (GET/HEAD/OPTIONS)
+					if (!isHttp2() && request.headers().contains(HttpHeaderNames.UPGRADE)) {
+						onInboundComplete();
+					}
 				}
 				if (isHttp2()) {
 					//force auto read to enable more accurate close selection now inbound is done

reactor-netty-http/src/test/java/reactor/netty/http/client/HttpClientTest.java

@@ -73,6 +73,7 @@
 import io.netty.handler.codec.compression.Brotli;
 import io.netty.handler.codec.http.DefaultFullHttpResponse;
 import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpContentDecompressor;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpHeaderValues;
@@ -82,6 +83,7 @@
 import io.netty.handler.codec.http.HttpResponseEncoder;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.handler.logging.LogLevel;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
@@ -101,7 +103,10 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
 import reactor.netty.BaseHttpTest;
@@ -1662,6 +1667,37 @@ void testIssue632() throws Exception {
 		assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
 	}
 
+	@Test
+	void testIssue3538() throws Exception {
+		disposableServer =
+				createServer()
+				          .protocol(HttpProtocol.H2C, HttpProtocol.HTTP11)
+				          .route(r -> r.get("/", (req, res) -> {
+				              final EchoAction action = new EchoAction();
+
+				              req
+				                  .receiveContent().switchIfEmpty(Mono.just(LastHttpContent.EMPTY_LAST_CONTENT))
+				                  .subscribe(action);
+
+				              return res.sendObject(action);
+				          }
+				      ))
+				      .bindNow();
+		assertThat(disposableServer).isNotNull();
+
+		final ByteBuf content = createHttpClientForContextWithPort()
+		        .protocol(HttpProtocol.HTTP11)
+		        .headers(h ->
+		            h.add(HttpHeaderNames.CONNECTION, HttpHeaderValues.UPGRADE)
+		             .add(HttpHeaderNames.UPGRADE, "TLS/1.2"))
+		        .get()
+		        .uri("/")
+		        .responseContent()
+		        .blockLast(Duration.ofSeconds(30));
+
+		assertThat(content).isNull();
+	}
+
 	@Test
 	void testIssue694() {
 		disposableServer =
@@ -3513,4 +3549,29 @@ void testDeleteMethod(boolean chunked) {
 		        .expectComplete()
 		        .verify(Duration.ofSeconds(5));
 	}
+
+    private static class EchoAction implements Publisher<HttpContent>, Consumer<HttpContent> {
+        private final Publisher<HttpContent> sender;
+        private volatile FluxSink<HttpContent> emitter;
+
+        EchoAction() {
+            this.sender = Flux.create(emitter ->  this.emitter = emitter);
+        }
+
+        @Override
+        public void accept(HttpContent message) {
+            if (message instanceof LastHttpContent) {
+                emitter.complete();
+            }
+            else {
+                emitter.next(message.retain());
+            }
+        }
+
+        @Override
+        public void subscribe(Subscriber<? super HttpContent> s) {
+            sender.subscribe(s);
+        }
+
+    }
 }