Merge #2932 into 2.0.0-M4

reactor · Oct 12, 2023 · f3dd3e8 · f3dd3e8
2 parents aac727e + ae82154
commit f3dd3e8
Show file tree

Hide file tree

Showing 2 changed files with 73 additions and 7 deletions.
diff --git a/reactor-netty5-core/src/main/java/reactor/netty5/resources/PooledConnectionProvider.java b/reactor-netty5-core/src/main/java/reactor/netty5/resources/PooledConnectionProvider.java
@@ -364,7 +364,19 @@ final void disposeInactivePoolsInBackground() {
 					if (log.isDebugEnabled()) {
 						log.debug("ConnectionProvider[name={}]: Disposing inactive pool for [{}]", name, e.getKey().holder);
 					}
-					e.getValue().dispose();
+					SocketAddress address = e.getKey().holder;
+					String id = e.getKey().hashCode() + "";
+					PoolFactory<T> poolFactory = poolFactory(address);
+					e.getValue().disposeLater().then(
+							Mono.<Void>fromRunnable(() -> {
+								if (poolFactory.registrar != null) {
+									poolFactory.registrar.get().deRegisterMetrics(name, id, address);
+								}
+								else if (Metrics.isMicrometerAvailable()) {
+									deRegisterDefaultMetrics(id, address);
+								}
+							})
+					).subscribe();
 				}
 			});
 		}

diff --git a/...tty5-http/src/test/java/reactor/netty5/resources/DefaultPooledConnectionProviderTest.java b/...tty5-http/src/test/java/reactor/netty5/resources/DefaultPooledConnectionProviderTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2022 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2020-2023 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -74,6 +74,7 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
@@ -652,7 +653,7 @@ void testMinConnections() throws Exception {
 
 	@ParameterizedTest
 	@MethodSource("disposeInactivePoolsInBackgroundCombinations")
-	void testDisposeInactivePoolsInBackground(boolean enableEvictInBackground, boolean isHttp2) throws Exception {
+	void testDisposeInactivePoolsInBackground(boolean enableEvictInBackground, boolean isHttp2, boolean isBuiltInMetrics) throws Exception {
 		disposableServer =
 				createServer()
 				        .wiretap(false)
@@ -671,6 +672,19 @@ void testDisposeInactivePoolsInBackground(boolean enableEvictInBackground, boole
 			builder.evictInBackground(Duration.ofMillis(50));
 		}
 
+		MeterRegistrarImpl meterRegistrar;
+		String metricsName = "";
+		if (isBuiltInMetrics) {
+			meterRegistrar = null;
+			builder.metrics(true);
+
+			metricsName = isHttp2 ? "http2.testDisposeInactivePoolsInBackground" : "testDisposeInactivePoolsInBackground";
+		}
+		else {
+			meterRegistrar = new MeterRegistrarImpl();
+			builder.metrics(true, () -> meterRegistrar);
+		}
+
 		CountDownLatch latch = new CountDownLatch(10);
 		DefaultPooledConnectionProvider provider = (DefaultPooledConnectionProvider) builder.build();
 		HttpClient client =
@@ -697,6 +711,12 @@ void testDisposeInactivePoolsInBackground(boolean enableEvictInBackground, boole
 			    .verify(Duration.ofSeconds(5));
 
 			assertThat(provider.channelPools.size()).isEqualTo(1);
+			if (meterRegistrar != null) {
+				assertThat(meterRegistrar.registered.get()).isTrue();
+			}
+			else {
+				assertThat(getGaugeValue(CONNECTION_PROVIDER_PREFIX + ACTIVE_CONNECTIONS, NAME, metricsName)).isNotEqualTo(-1);
+			}
 
 			if (enableEvictInBackground) {
 				assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
@@ -709,6 +729,17 @@ void testDisposeInactivePoolsInBackground(boolean enableEvictInBackground, boole
 			               .isEqualTo(enableEvictInBackground ? 0 : 1));
 
 			assertThat(provider.isDisposed()).isEqualTo(enableEvictInBackground);
+			if (meterRegistrar != null) {
+				assertThat(meterRegistrar.deRegistered.get()).isEqualTo(enableEvictInBackground);
+			}
+			else {
+				if (enableEvictInBackground) {
+					assertThat(getGaugeValue(CONNECTION_PROVIDER_PREFIX + ACTIVE_CONNECTIONS, NAME, metricsName)).isEqualTo(-1);
+				}
+				else {
+					assertThat(getGaugeValue(CONNECTION_PROVIDER_PREFIX + ACTIVE_CONNECTIONS, NAME, metricsName)).isNotEqualTo(-1);
+				}
+			}
 		}
 		finally {
 			if (!enableEvictInBackground) {
@@ -720,10 +751,15 @@ void testDisposeInactivePoolsInBackground(boolean enableEvictInBackground, boole
 
 	static Stream<Arguments> disposeInactivePoolsInBackgroundCombinations() {
 		return Stream.of(
-				Arguments.of(false, false),
-				Arguments.of(false, true),
-				Arguments.of(true, false),
-				Arguments.of(true, true)
+				// enableEvictInBackground, isHttp2, isBuiltInMetrics
+				Arguments.of(false, false, false),
+				Arguments.of(false, false, true),
+				Arguments.of(false, true, false),
+				Arguments.of(false, true, true),
+				Arguments.of(true, false, false),
+				Arguments.of(true, false, true),
+				Arguments.of(true, true, false),
+				Arguments.of(true, true, true)
 		);
 	}
 
@@ -825,4 +861,22 @@ void testHttp2PoolAndGoAway() {
 			        .block(Duration.ofSeconds(5));
 		}
 	}
+
+	static final class MeterRegistrarImpl implements ConnectionProvider.MeterRegistrar {
+		AtomicBoolean registered = new AtomicBoolean();
+		AtomicBoolean deRegistered = new AtomicBoolean();
+
+		MeterRegistrarImpl() {
+		}
+
+		@Override
+		public void registerMetrics(String poolName, String id, SocketAddress remoteAddress, ConnectionPoolMetrics metrics) {
+			registered.compareAndSet(false, true);
+		}
+
+		@Override
+		public void deRegisterMetrics(String poolName, String id, SocketAddress remoteAddress) {
+			deRegistered.compareAndSet(false, true);
+		}
+	}
 }