Re enable HttpMetricsHandlerTests in netty5 branch

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

@@ -20,6 +20,7 @@
 import io.netty5.channel.Channel;
 import io.netty5.channel.ChannelHandlerAdapter;
 import io.netty5.channel.ChannelHandlerContext;
+import io.netty5.handler.codec.http.HttpContent;
 import io.netty5.handler.codec.http.HttpRequest;
 import io.netty5.handler.codec.http.HttpResponse;
 import io.netty5.handler.codec.http.LastHttpContent;
@@ -165,6 +166,9 @@ private long extractProcessedDataFromBuffer(Object msg) {
 		else if (msg instanceof Buffer buffer) {
 			return buffer.readableBytes();
 		}
+		else if (msg instanceof HttpContent<?> httpContent) {
+			return httpContent.payload().readableBytes();
+		}
 		return 0;
 	}
 

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

@@ -82,6 +82,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
 	@Override
 	public void channelInboundEvent(ChannelHandlerContext ctx, Object evt) {
 		Channel channel = ctx.channel();
+		boolean removeThisHandler = false;
 		if (evt == UPGRADE_ISSUED) {
 			if (log.isDebugEnabled()) {
 				log.debug(format(channel, "An upgrade request was sent to the server."));
@@ -92,7 +93,7 @@ else if (evt == UPGRADE_SUCCESSFUL) {
 				log.debug(format(channel, "The upgrade to H2C protocol was successful."));
 			}
 			sendNewState(Connection.from(channel), HttpClientState.UPGRADE_SUCCESSFUL);
-			ctx.pipeline().remove(this);
+			removeThisHandler = true; // we have to remove ourselves from the pipeline after having fired the event below.
 		}
 		else if (evt == UPGRADE_REJECTED) {
 			if (log.isDebugEnabled()) {
@@ -101,6 +102,9 @@ else if (evt == UPGRADE_REJECTED) {
 			sendNewState(Connection.from(channel), HttpClientState.UPGRADE_REJECTED);
 		}
 		ctx.fireChannelInboundEvent(evt);
+		if (removeThisHandler) {
+			ctx.pipeline().remove(this);
+		}
 	}
 
 	void sendNewState(Connection connection, ConnectionObserver.State state) {

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

@@ -19,6 +19,7 @@
 import io.netty5.buffer.api.Buffer;
 import io.netty5.channel.ChannelHandlerAdapter;
 import io.netty5.channel.ChannelHandlerContext;
+import io.netty5.handler.codec.http.HttpContent;
 import io.netty5.handler.codec.http.HttpRequest;
 import io.netty5.handler.codec.http.HttpResponse;
 import io.netty5.handler.codec.http.HttpResponseStatus;
@@ -114,9 +115,12 @@ public Future<Void> write(ChannelHandlerContext ctx, Object msg) {
 			dataSent += extractProcessedDataFromBuffer(msg);
 
 			if (msg instanceof LastHttpContent) {
+				// The listeners are now invoked asynchronously (see https://github.com/netty/netty/pull/9489),
+				// and it seems we need to first obtain the channelOps, which may not be present anymore
+				// when the listener will be invoked.
+				ChannelOperations<?, ?> channelOps = ChannelOperations.get(ctx.channel());
 				return ctx.write(msg)
 						.addListener(future -> {
-								ChannelOperations<?, ?> channelOps = ChannelOperations.get(ctx.channel());
 								if (channelOps instanceof HttpServerOperations ops) {
 									try {
 										recordWrite(ops, uriTagValue == null ? ops.path : uriTagValue.apply(ops.path),
@@ -208,6 +212,9 @@ private long extractProcessedDataFromBuffer(Object msg) {
 		else if (msg instanceof Buffer buffer) {
 			return buffer.readableBytes();
 		}
+		else if (msg instanceof HttpContent<?> httpContent) {
+			return httpContent.payload().readableBytes();
+		}
 		return 0;
 	}