Merge reactor#2357 from netty5 branch. Use socks-proxy from netty-con…

…trib main branch.

.github/workflows/check_transport.yml

@@ -68,10 +68,8 @@ jobs:
           path: codec-extras
       - uses: actions/checkout@v3
         with:
-          # TODO temporary https://github.com/netty-contrib/socks-proxy/pull/11
-          repository: pderop/socks-proxy
+          repository: netty-contrib/socks-proxy
           path: socks-proxy
-          ref: blocking-methods-have-moved-from-future-to-futurecompletablestage
       - uses: actions/checkout@v3
         with:
           # TODO temporary https://github.com/netty-contrib/codec-multipart/pull/1

docs/asciidoc/getting-started.adoc

@@ -22,7 +22,7 @@ backpressure-ready network engines for `HTTP` (including Websockets), `TCP`, and
 
 It has transitive dependencies on:
 
-* Reactive Streams v1.0.3
+* Reactive Streams v1.0.4
 * Reactor Core v3.x
 * Netty v4.1.x
 

docs/asciidoc/http-client-conn-provider.adoc

@@ -12,7 +12,7 @@ and further attempts are declined with an error.
 [source,java,indent=0]
 ../../../reactor-netty-core/src/main/java/reactor/netty/ReactorNetty.java
 ----
-include::./../../reactor-netty-core/src/main/java/reactor/netty/ReactorNetty.java[lines=117..159]
+include::./../../reactor-netty-core/src/main/java/reactor/netty/ReactorNetty.java[lines=117..165]
 ----
 ====
 

docs/asciidoc/http-client.adoc

@@ -195,7 +195,7 @@ By default, the `HTTP` client is configured with the following settings:
 [source,java,indent=0]
 .{sourcedir}/reactor/netty/http/HttpDecoderSpec.java
 ----
-include::{sourcedir}/reactor/netty/http/HttpDecoderSpec.java[lines=29..34]
+include::{sourcedir}/reactor/netty/http/HttpDecoderSpec.java[lines=29..38]
 ----
 ====
 ====
@@ -265,7 +265,7 @@ to extend the default `TCP` client configuration (add an option, bind address et
 [source,java,indent=0]
 .{examplesdir}/channeloptions/Application.java
 ----
-include::{examplesdir}/channeloptions/Application.java[lines=18..46]
+include::{examplesdir}/channeloptions/Application.java[lines=18..54]
 ----
 <1> Configures the connection establishment timeout to 10 seconds.
 <2> Enables TCP `keepalive`. This means that TCP starts sending `keepalive` probes when a connection is idle for some time.
@@ -626,7 +626,7 @@ To customize the default settings, you can configure `HttpClient` as follows:
 [source,java,indent=0]
 .{examplesdir}/channeloptions/Application.java
 ----
-include::{examplesdir}/channeloptions/Application.java[lines=18..52]
+include::{examplesdir}/channeloptions/Application.java[lines=18..54]
 ----
 <1> Configures the connection establishment timeout to 10 seconds.
 <2> Enables TCP `keepalive`. This means that TCP starts sending `keepalive` probes when a connection is idle for some time.
@@ -674,7 +674,7 @@ To customize the default settings, you can configure `HttpClient` as follows:
 [source,java,indent=0]
 .{examplesdir}/proxy/Application.java
 ----
-include::{examplesdir}/proxy/Application.java[lines=18..41]
+include::{examplesdir}/proxy/Application.java[lines=18..42]
 ----
 <1> Configures the connection establishment timeout to 20 seconds.
 ====

docs/asciidoc/http-server.adoc

@@ -314,7 +314,7 @@ By default, the `HTTP` server is configured with the following settings:
 [source,java,indent=0]
 .{sourcedir}/reactor/netty/http/HttpDecoderSpec.java
 ----
-include::{sourcedir}/reactor/netty/http/HttpDecoderSpec.java[lines=29..34]
+include::{sourcedir}/reactor/netty/http/HttpDecoderSpec.java[lines=29..38]
 ----
 ====
 ====
@@ -560,6 +560,8 @@ The following table provides information for the HTTP server metrics:
 [width="100%",options="header"]
 |=======
 | metric name | type | description
+| reactor.netty.http.server.streams.active | Gauge | The number of active HTTP/2 streams.
+See <<observability-metrics-streams-active>>
 | reactor.netty.http.server.connections.active | Gauge | The number of http connections currently processing requests.
 See <<observability-metrics-connections-active>>
 | reactor.netty.http.server.connections.total | Gauge | The number of all opened connections.

docs/asciidoc/proxy.adoc

@@ -17,7 +17,7 @@ The following example uses `ProxyProvider`:
 [source,java,indent=0]
 .{examplesdir}/proxy/Application.java
 ----
-include::{examplesdir}/proxy/Application.java[lines=18..38]
+include::{examplesdir}/proxy/Application.java[lines=18..42]
 ----
 <1> Configures the connection establishment timeout to 20 seconds.
 ====

docs/asciidoc/wire-logger.adoc

@@ -11,7 +11,7 @@ and apply the following configuration:
 [source,java,indent=0]
 .{examplesdir}/wiretap/Application.java
 ----
-include::{examplesdir}/wiretap/Application.java[lines=18..32]
+include::{examplesdir}/wiretap/Application.java[lines=18..35]
 ----
 <1> Enables the wire logging
 ====
@@ -56,7 +56,7 @@ When you need to change the default formatter you can configure it as follows:
 [source,java,indent=0]
 .{examplesdir}/wiretap/custom/Application.java
 ----
-include::{examplesdir}/wiretap/custom/Application.java[lines=18..34]
+include::{examplesdir}/wiretap/custom/Application.java[lines=18..38]
 ----
 <1> Enables the wire logging,
 {javadoc}/reactor/netty/transport/logging/AdvancedByteBufFormat.html#TEXTUAL[AdvancedByteBufFormat#TEXTUAL] is used for

reactor-netty-core/src/main/java/reactor/netty/Metrics.java

@@ -256,6 +256,11 @@ public class Metrics {
 	 */
 	public static final String PENDING_TASKS = ".pending.tasks";
 
+	// HttpServer Metrics
+	/**
+	 * The number of active HTTP/2 streams
+	 */
+	public static final String STREAMS_ACTIVE = ".streams.active";
 
 	// Tags
 	public static final String LOCAL_ADDRESS = "local.address";

reactor-netty-http/build.gradle

@@ -214,6 +214,9 @@ task japicmp(type: JapicmpTask) {
 			'reactor.netty.http.HttpDecoderSpec#DEFAULT_MAX_CHUNK_SIZE'
 	]
 	methodExcludes = [
+			"reactor.netty.http.server.HttpServerMetricsRecorder#recordStreamClosed(java.net.SocketAddress)",
+			"reactor.netty.http.server.HttpServerMetricsRecorder#recordStreamOpened(java.net.SocketAddress)",
+
 			'reactor.netty.http.HttpOperations#send(org.reactivestreams.Publisher)',
 
 			'reactor.netty.http.client.HttpClient$ResponseReceiver#responseContent()',

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

@@ -46,12 +46,10 @@ abstract class AbstractHttpServerMetricsHandler extends ChannelHandlerAdapter {
 
 	long dataSent;
 
-
 	long dataReceivedTime;
 
 	long dataSentTime;
 
-
 	final Function<String, String> uriTagValue;
 
 	protected AbstractHttpServerMetricsHandler(@Nullable Function<String, String> uriTagValue) {
@@ -130,12 +128,16 @@ public Future<Void> write(ChannelHandlerContext ctx, Object msg) {
 										log.warn("Exception caught while recording metrics.", e);
 										// Allow request-response exchange to continue, unaffected by metrics problem
 									}
-									if (!ops.isHttp2() && ops.hostAddress() != null) {
-										// This metric is not applicable for HTTP/2
-										// ops.hostAddress() == null when request decoding failed, in this case
-										// we do not report active connection, so we do not report inactive connection
+									// ops.hostAddress() == null when request decoding failed, in this case
+									// we do not report active connection, so we do not report inactive connection
+									if (ops.hostAddress() != null) {
 										try {
-											recordInactiveConnection(ops);
+											if (ops.isHttp2()) {
+												recordClosedStream(ops);
+											}
+											else {
+												recordInactiveConnection(ops);
+											}
 										}
 										catch (RuntimeException e) {
 											log.warn("Exception caught while recording metrics.", e);
@@ -161,8 +163,10 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
 			if (msg instanceof HttpRequest) {
 				ChannelOperations<?, ?> channelOps = ChannelOperations.get(ctx.channel());
 				if (channelOps instanceof HttpServerOperations ops) {
-					if (!ops.isHttp2()) {
-						// This metric is not applicable for HTTP/2
+					if (ops.isHttp2()) {
+						recordOpenStream(ops);
+					}
+					else {
 						recordActiveConnection(ops);
 					}
 					startRead(ops, uriTagValue == null ? ops.path : uriTagValue.apply(ops.path), ops.method().name());
@@ -255,6 +259,14 @@ protected void recordInactiveConnection(HttpServerOperations ops) {
 		recorder().recordServerConnectionInactive(ops.hostAddress());
 	}
 
+	protected void recordOpenStream(HttpServerOperations ops) {
+		recorder().recordStreamOpened(ops.hostAddress());
+	}
+
+	protected void recordClosedStream(HttpServerOperations ops) {
+		recorder().recordStreamClosed(ops.hostAddress());
+	}
+
 	protected void startRead(HttpServerOperations ops, String path, String method) {
 		dataReceivedTime = System.nanoTime();
 	}

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

@@ -47,6 +47,26 @@ public Meter.Type getType() {
 		}
 	},
 
+	/**
+	 * The number of HTTP/2 streams currently active on the server
+	 */
+	STREAMS_ACTIVE {
+		@Override
+		public String getName() {
+			return "reactor.netty.http.server.streams.active";
+		}
+
+		@Override
+		public KeyName[] getKeyNames() {
+			return StreamsActiveTags.values();
+		}
+
+		@Override
+		public Meter.Type getType() {
+			return Meter.Type.GAUGE;
+		}
+	},
+
 	/**
 	 * Amount of the data received, in bytes.
 	 */
@@ -157,6 +177,29 @@ public Meter.Type getType() {
 		}
 	};
 
+	enum StreamsActiveTags implements KeyName {
+
+		/**
+		 * Local address.
+		 */
+		LOCAL_ADDRESS {
+			@Override
+			public String getKeyName() {
+				return "local.address";
+			}
+		},
+
+		/**
+		 * URI.
+		 */
+		URI {
+			@Override
+			public String getKeyName() {
+				return "uri";
+			}
+		}
+	}
+
 	enum ConnectionsActiveTags implements KeyName {
 
 		/**

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

@@ -71,4 +71,21 @@ default void recordServerConnectionActive(SocketAddress localAddress) { }
 	 * @since 1.0.15
 	 */
 	default void recordServerConnectionInactive(SocketAddress localAddress) { }
+
+	/**
+	 * Record an opened HTTP/2 stream
+	 *
+	 * @param localAddress the local server address
+	 * @since 1.0.21
+	 */
+	default void recordStreamOpened(SocketAddress localAddress) { }
+
+	/**
+	 * Record a closed HTTP/2 stream
+	 *
+	 * @param localAddress the local server address
+	 * @since 1.0.21
+	 */
+	default void recordStreamClosed(SocketAddress localAddress) { }
+
 }

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

@@ -42,6 +42,7 @@
 import static reactor.netty.Metrics.STATUS;
 import static reactor.netty.Metrics.URI;
 import static reactor.netty.http.server.HttpServerMeters.CONNECTIONS_ACTIVE;
+import static reactor.netty.http.server.HttpServerMeters.STREAMS_ACTIVE;
 
 /**
  * @author Violeta Georgieva
@@ -52,7 +53,9 @@ final class MicrometerHttpServerMetricsRecorder extends MicrometerHttpMetricsRec
 	final static MicrometerHttpServerMetricsRecorder INSTANCE = new MicrometerHttpServerMetricsRecorder();
 	private final static String PROTOCOL_VALUE_HTTP = "http";
 	private final LongAdder activeConnectionsAdder = new LongAdder();
+	private final LongAdder activeStreamsAdder = new LongAdder();
 	private final ConcurrentMap<String, LongAdder> activeConnectionsCache = new ConcurrentHashMap<>();
+	private final ConcurrentMap<String, LongAdder> activeStreamsCache = new ConcurrentHashMap<>();
 	private final ConcurrentMap<String, DistributionSummary> dataReceivedCache = new ConcurrentHashMap<>();
 	private final ConcurrentMap<String, DistributionSummary> dataSentCache = new ConcurrentHashMap<>();
 	private final ConcurrentMap<String, Counter> errorsCache = new ConcurrentHashMap<>();
@@ -156,6 +159,22 @@ public void recordServerConnectionInactive(SocketAddress localAddress) {
 		}
 	}
 
+	@Override
+	public void recordStreamOpened(SocketAddress localAddress) {
+		LongAdder adder = getActiveStreamsAdder(localAddress);
+		if (adder != null) {
+			adder.increment();
+		}
+	}
+
+	@Override
+	public void recordStreamClosed(SocketAddress localAddress) {
+		LongAdder adder = getActiveStreamsAdder(localAddress);
+		if (adder != null) {
+			adder.decrement();
+		}
+	}
+
 	@Override
 	public void recordDataReceived(SocketAddress remoteAddress, long bytes) {
 		// noop
@@ -186,6 +205,21 @@ public void recordResolveAddressTime(SocketAddress remoteAddress, Duration time,
 		throw new UnsupportedOperationException();
 	}
 
+	@Nullable
+	private LongAdder getActiveStreamsAdder(SocketAddress localAddress) {
+		String address = reactor.netty.Metrics.formatSocketAddress(localAddress);
+		return MapUtils.computeIfAbsent(activeStreamsCache, address,
+				key -> {
+					Gauge gauge = filter(
+							Gauge.builder(STREAMS_ACTIVE.getName(), activeStreamsAdder, LongAdder::longValue)
+									.tags(HttpServerMeters.StreamsActiveTags.URI.getKeyName(), PROTOCOL_VALUE_HTTP,
+											HttpServerMeters.StreamsActiveTags.LOCAL_ADDRESS.getKeyName(), address)
+									.register(REGISTRY));
+					return gauge != null ? activeStreamsAdder : null;
+				});
+	}
+
+	@Nullable
 	private LongAdder getServerConnectionAdder(SocketAddress localAddress) {
 		String address = reactor.netty.Metrics.formatSocketAddress(localAddress);
 		return MapUtils.computeIfAbsent(activeConnectionsCache, address,

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

@@ -89,6 +89,7 @@
 import static reactor.netty.Metrics.REMOTE_ADDRESS;
 import static reactor.netty.Metrics.RESPONSE_TIME;
 import static reactor.netty.Metrics.STATUS;
+import static reactor.netty.Metrics.STREAMS_ACTIVE;
 import static reactor.netty.Metrics.TLS_HANDSHAKE_TIME;
 import static reactor.netty.Metrics.URI;
 import static reactor.netty.Metrics.formatSocketAddress;
@@ -610,6 +611,7 @@ void testServerConnectionsMicrometer(HttpProtocol[] serverProtocols, HttpProtoco
 		}
 		else {
 			checkGauge(SERVER_CONNECTIONS_TOTAL, true, 1, URI, HTTP, LOCAL_ADDRESS, address);
+			checkGauge(SERVER_STREAMS_ACTIVE, true, 0, URI, HTTP, LOCAL_ADDRESS, address);
 		}
 
 		// These metrics are meant only for the servers,
@@ -747,7 +749,10 @@ private void checkServerConnectionsMicrometer(HttpServerRequest request) {
 		String address = formatSocketAddress(request.hostAddress());
 		boolean isHttp2 = request.requestHeaders().contains(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text());
 		checkGauge(SERVER_CONNECTIONS_TOTAL, true, 1, URI, HTTP, LOCAL_ADDRESS, address);
-		if (!isHttp2) {
+		if (isHttp2) {
+			checkGauge(SERVER_STREAMS_ACTIVE, true, 1, URI, HTTP, LOCAL_ADDRESS, address);
+		}
+		else {
 			checkGauge(SERVER_CONNECTIONS_ACTIVE, true, 1, URI, HTTP, LOCAL_ADDRESS, address);
 		}
 	}
@@ -977,6 +982,7 @@ static Stream<Arguments> httpCompatibleProtocols() {
 
 	private static final String SERVER_CONNECTIONS_ACTIVE = HTTP_SERVER_PREFIX + CONNECTIONS_ACTIVE;
 	private static final String SERVER_CONNECTIONS_TOTAL = HTTP_SERVER_PREFIX + CONNECTIONS_TOTAL;
+	private static final String SERVER_STREAMS_ACTIVE = HTTP_SERVER_PREFIX + STREAMS_ACTIVE;
 	private static final String SERVER_RESPONSE_TIME = HTTP_SERVER_PREFIX + RESPONSE_TIME;
 	private static final String SERVER_DATA_SENT_TIME = HTTP_SERVER_PREFIX + DATA_SENT_TIME;
 	private static final String SERVER_DATA_RECEIVED_TIME = HTTP_SERVER_PREFIX + DATA_RECEIVED_TIME;