Ensure 100-Continue is handled when collecting client's metrics (#3895)

Fixes #3883

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

Author: violetagg

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

@@ -23,6 +23,7 @@
 import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.LastHttpContent;
 import reactor.netty.channel.ChannelOperations;
 import reactor.util.annotation.Nullable;
@@ -73,6 +74,8 @@ abstract class AbstractHttpClientMetricsHandler extends ChannelDuplexHandler {
 
 	int lastWriteSeq;
 
+	boolean isNot100Continue;
+
 	protected AbstractHttpClientMetricsHandler(SocketAddress remoteAddress, @Nullable SocketAddress proxyAddress, @Nullable Function<String, String> uriTagValue) {
 		this.proxyAddress = proxyAddress;
 		this.remoteAddress = remoteAddress;
@@ -93,6 +96,7 @@ protected AbstractHttpClientMetricsHandler(AbstractHttpClientMetricsHandler copy
 		this.uriTagValue = copy.uriTagValue;
 		this.lastWriteSeq = copy.lastWriteSeq;
 		this.lastReadSeq = copy.lastReadSeq;
+		this.isNot100Continue = copy.isNot100Continue;
 	}
 
 	@Override
@@ -139,14 +143,18 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
 	public void channelRead(ChannelHandlerContext ctx, Object msg) {
 		try {
 			if (msg instanceof HttpResponse) {
-				status = ((HttpResponse) msg).status().codeAsText().toString();
+				HttpResponseStatus httpResponseStatus = ((HttpResponse) msg).status();
+				isNot100Continue = httpResponseStatus.code() != HttpResponseStatus.CONTINUE.code();
+				if (isNot100Continue) {
+					status = httpResponseStatus.codeAsText().toString();
 
-				startRead((HttpResponse) msg);
+					startRead((HttpResponse) msg);
+				}
 			}
 
 			dataReceived += extractProcessedDataFromBuffer(msg);
 
-			if (msg instanceof LastHttpContent) {
+			if (isNot100Continue && msg instanceof LastHttpContent) {
 				// Detect if we have received an early response before the request has been fully flushed.
 				// In this case, invoke #recordWrite now (because next we will reset all class fields).
 				lastReadSeq = (lastReadSeq + 1) & 0x7F_FF_FF_FF;

reactor-netty-http/src/test/java/reactor/netty/http/HttpMetricsHandlerTests.java

@@ -30,6 +30,8 @@
 import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.channel.unix.DomainSocketAddress;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.handler.codec.http2.Http2StreamChannel;
@@ -169,13 +171,14 @@ static void createSelfSignedCertificate() throws CertificateException {
 	 * <ul>
 	 *  <li> /1 is used by testExistingEndpoint test</li>
 	 *  <li> /2 is used by testExistingEndpoint, and testUriTagValueFunctionNotSharedForClient tests</li>
-	 *  <li> /3 does not exists but is used by testNonExistingEndpoint, checkExpectationsNonExisting tests</li>
+	 *  <li> /3 does not exist but is used by testNonExistingEndpoint, checkExpectationsNonExisting tests</li>
 	 *  <li> /4 is used by testServerConnectionsMicrometer test</li>
 	 *  <li> /5 is used by testServerConnectionsRecorder test</li>
 	 *  <li> /6 is used by testServerConnectionsMicrometerConnectionClose test</li>
 	 *  <li> /7 is used by testServerConnectionsRecorderConnectionClose test</li>
 	 *  <li> /8 is used by testServerConnectionsWebsocketMicrometer test</li>
 	 *  <li> /9 is used by testServerConnectionsWebsocketRecorder test</li>
+	 *  <li> /10 is used by testIssue3883 for testing 100-Continue</li>
 	 * </ul>
 	 */
 	@BeforeEach
@@ -210,6 +213,10 @@ void setUp() {
 				                     out.sendString(Mono.just("Hello World!").doOnNext(b -> checkServerConnectionsMicrometer(req)))))
 				             .get("/9", (req, res) -> res.sendWebsocket((in, out) ->
 				                     out.sendString(Mono.just("Hello World!").doOnNext(b -> checkServerConnectionsRecorder(req)))))
+				             .post("/10", (req, res) -> req.receive()
+				                                           .aggregate()
+				                                           .asString()
+				                                           .flatMap(s -> res.header("Connection", "close").sendString(Mono.just(s)).then()))
 				);
 
 		provider = ConnectionProvider.create("HttpMetricsHandlerTests", 1);
@@ -1076,6 +1083,58 @@ void testIssue3060ConnectTimeoutException(@SuppressWarnings("unused") HttpProtoc
 		assertTimer(registry, CLIENT_CONNECT_TIME, summaryTags).hasCountEqualTo(1);
 	}
 
+	@ParameterizedTest
+	@MethodSource("httpCompatibleProtocols")
+	void testIssue3883(HttpProtocol[] serverProtocols, HttpProtocol[] clientProtocols,
+			@Nullable ProtocolSslContextSpec serverCtx, @Nullable ProtocolSslContextSpec clientCtx) throws Exception {
+		CountDownLatch responseSent = new CountDownLatch(1); // response fully sent by the server
+		AtomicReference<CountDownLatch> responseSentRef = new AtomicReference<>(responseSent);
+		ResponseSentHandler responseSentHandler = ResponseSentHandler.INSTANCE;
+		disposableServer = customizeServerOptions(httpServer, serverCtx, serverProtocols)
+				.doOnConnection(cnx -> responseSentHandler.register(responseSentRef, cnx.channel().pipeline()))
+				.bindNow();
+
+		AtomicReference<SocketAddress> serverAddress = new AtomicReference<>();
+		CountDownLatch clientCompleted = new CountDownLatch(1); // client received full response
+		AtomicReference<CountDownLatch> clientCompletedRef = new AtomicReference<>(clientCompleted);
+		httpClient = customizeClientOptions(httpClient, clientCtx, clientProtocols)
+				.doAfterRequest((req, conn) -> serverAddress.set(conn.channel().remoteAddress()))
+				.doAfterResponseSuccess((resp, conn) -> clientCompletedRef.get().countDown());
+
+		httpClient.headers(h -> h.add(HttpHeaderNames.EXPECT, HttpHeaderValues.CONTINUE))
+		          .post()
+		          .uri("/10")
+		          .send(body)
+		          .responseContent()
+		          .aggregate()
+		          .asString()
+		          .as(StepVerifier::create)
+		          .expectNext("Hello World!")
+		          .expectComplete()
+		          .verify(Duration.ofSeconds(5));
+
+		assertThat(responseSentRef.get().await(30, TimeUnit.SECONDS)).as("responseSentRef latch await").isTrue();
+		assertThat(clientCompletedRef.get().await(30, TimeUnit.SECONDS)).as("clientCompletedRef latch await").isTrue();
+
+		InetSocketAddress sa = (InetSocketAddress) serverAddress.get();
+
+		int[] numWrites = new int[]{14, 25};
+		int[] bytesWrite = new int[]{160, 243};
+		if ((serverProtocols.length == 1 && serverProtocols[0] == HttpProtocol.HTTP11) ||
+				(clientProtocols.length == 1 && clientProtocols[0] == HttpProtocol.HTTP11)) {
+			numWrites = new int[]{14, 28};
+			bytesWrite = new int[]{151, 310};
+		}
+		else if (clientProtocols.length == 2 &&
+				Arrays.equals(clientProtocols, new HttpProtocol[]{HttpProtocol.H2C, HttpProtocol.HTTP11})) {
+			numWrites = new int[]{17, 28};
+			bytesWrite = new int[]{315, 435};
+		}
+
+		checkExpectationsExisting("/10", sa.getHostString() + ":" + sa.getPort(), 1, serverCtx != null,
+				numWrites[0], bytesWrite[0]);
+	}
+
 	static Stream<Arguments> combinationsIssue2956() {
 		return Stream.of(
 				// isCustomRecorder, isHttp2