Merge branch '2020.0' into dev/sean/refactor_cb_2020

# Conflicts:
#	spring-cloud-starter-tencent-polaris-discovery/src/main/java/com/tencent/cloud/polaris/loadbalancer/PolarisLoadBalancerClientConfiguration.java

CHANGELOG.md

@@ -5,3 +5,5 @@
 - [feat:support webclient and gateway report call metrics](https://github.com/Tencent/spring-cloud-tencent/pull/942)
 - [docs:prevent the release of the final version of the sdk.](https://github.com/Tencent/spring-cloud-tencent/pull/945)
 - [fix: fix nacos CircuitBreaker disable bug.](https://github.com/Tencent/spring-cloud-tencent/pull/947)
+- [feature: add config for customized local port.](https://github.com/Tencent/spring-cloud-tencent/pull/956)
+- [refactor:refactor reactor code in router module.](https://github.com/Tencent/spring-cloud-tencent/pull/958)

spring-cloud-starter-tencent-polaris-discovery/src/main/java/com/tencent/cloud/polaris/registry/PolarisRegistration.java

@@ -25,6 +25,7 @@ import java.util.Objects;
 
 import com.tencent.cloud.common.metadata.StaticMetadataManager;
 import com.tencent.cloud.polaris.PolarisDiscoveryProperties;
+import com.tencent.cloud.polaris.context.config.PolarisContextProperties;
 import com.tencent.cloud.polaris.extend.consul.ConsulContextProperties;
 import com.tencent.cloud.polaris.extend.nacos.NacosContextProperties;
 import com.tencent.polaris.client.api.SDKContext;
@@ -76,6 +77,7 @@ public class PolarisRegistration implements Registration {
 
 	public PolarisRegistration(
 			PolarisDiscoveryProperties polarisDiscoveryProperties,
+			@Nullable PolarisContextProperties polarisContextProperties,
 			@Nullable ConsulContextProperties consulContextProperties,
 			SDKContext context, StaticMetadataManager staticMetadataManager,
 			@Nullable NacosContextProperties nacosContextProperties,
@@ -89,6 +91,9 @@ public class PolarisRegistration implements Registration {
 		this.servletWebServerApplicationContext = servletWebServerApplicationContext;
 		this.reactiveWebServerApplicationContext = reactiveWebServerApplicationContext;
 		host = polarisContext.getConfig().getGlobal().getAPI().getBindIP();
+		if (polarisContextProperties != null) {
+			port = polarisContextProperties.getLocalPort();
+		}
 	}
 
 	@Override

spring-cloud-starter-tencent-polaris-discovery/src/main/java/com/tencent/cloud/polaris/registry/PolarisServiceRegistryAutoConfiguration.java

@@ -20,6 +20,7 @@ package com.tencent.cloud.polaris.registry;
 
 import com.tencent.cloud.common.metadata.StaticMetadataManager;
 import com.tencent.cloud.polaris.PolarisDiscoveryProperties;
+import com.tencent.cloud.polaris.context.config.PolarisContextProperties;
 import com.tencent.cloud.polaris.discovery.PolarisDiscoveryAutoConfiguration;
 import com.tencent.cloud.polaris.discovery.PolarisDiscoveryHandler;
 import com.tencent.cloud.polaris.extend.consul.ConsulContextProperties;
@@ -65,11 +66,12 @@ public class PolarisServiceRegistryAutoConfiguration {
 	@ConditionalOnBean(AutoServiceRegistrationProperties.class)
 	public PolarisRegistration polarisRegistration(
 			PolarisDiscoveryProperties polarisDiscoveryProperties,
+			PolarisContextProperties polarisContextProperties,
 			@Autowired(required = false) ConsulContextProperties consulContextProperties,
 			SDKContext context, StaticMetadataManager staticMetadataManager, NacosContextProperties nacosContextProperties,
 			@Autowired(required = false) ServletWebServerApplicationContext servletWebServerApplicationContext,
 			@Autowired(required = false) ReactiveWebServerApplicationContext reactiveWebServerApplicationContext) {
-		return new PolarisRegistration(polarisDiscoveryProperties, consulContextProperties, context,
+		return new PolarisRegistration(polarisDiscoveryProperties, polarisContextProperties, consulContextProperties, context,
 				staticMetadataManager, nacosContextProperties,
 				servletWebServerApplicationContext, reactiveWebServerApplicationContext);
 	}

spring-cloud-starter-tencent-polaris-discovery/src/test/java/com/tencent/cloud/polaris/registry/PolarisRegistrationTest.java

@@ -22,6 +22,7 @@ import java.util.Map;
 
 import com.tencent.cloud.common.metadata.StaticMetadataManager;
 import com.tencent.cloud.polaris.PolarisDiscoveryProperties;
+import com.tencent.cloud.polaris.context.config.PolarisContextProperties;
 import com.tencent.cloud.polaris.extend.consul.ConsulContextProperties;
 import com.tencent.cloud.polaris.extend.nacos.NacosContextProperties;
 import com.tencent.polaris.api.config.Configuration;
@@ -62,6 +63,10 @@ public class PolarisRegistrationTest {
 
 	private PolarisRegistration polarisRegistration3;
 
+	private PolarisRegistration polarisRegistration4;
+
+	private static int testLocalPort = 10086;
+
 	@BeforeEach
 	void setUp() {
 		// mock PolarisDiscoveryProperties
@@ -70,23 +75,27 @@ public class PolarisRegistrationTest {
 		doReturn("http").when(polarisDiscoveryProperties).getProtocol();
 		doReturn(true).when(polarisDiscoveryProperties).isRegisterEnabled();
 
+		// mock PolarisContextProperties
+		PolarisContextProperties polarisContextProperties = mock(PolarisContextProperties.class);
+		doReturn(testLocalPort).when(polarisContextProperties).getLocalPort();
+
 		// mock ConsulContextProperties
 		ConsulContextProperties consulContextProperties = mock(ConsulContextProperties.class);
 		doReturn(true).when(consulContextProperties).isEnabled();
 		doReturn(true).when(consulContextProperties).isRegister();
 
 		// mock NacosContextProperties
-		nacosContextProperties = new NacosContextProperties();
+		// mock NacosContextProperties
-		nacosContextProperties.setEnabled(true);
+		nacosContextProperties = mock(NacosContextProperties.class);
-		nacosContextProperties.setRegisterEnabled(true);
+		doReturn(true).when(nacosContextProperties).isEnabled();
-		nacosContextProperties.setContextPath("/");
+		doReturn(true).when(nacosContextProperties).isRegisterEnabled();
-		nacosContextProperties.setClusterName("cluster");
+		doReturn("/").when(nacosContextProperties).getContextPath();
-		nacosContextProperties.setGroup("");
+		doReturn("cluster").when(nacosContextProperties).getClusterName();
-		nacosContextProperties.setDiscoveryEnabled(true);
+		doReturn("").when(nacosContextProperties).getGroup();
-		nacosContextProperties.setPassword("");
+		doReturn(true).when(nacosContextProperties).isDiscoveryEnabled();
-		nacosContextProperties.setUsername("");
+		doReturn("").when(nacosContextProperties).getPassword();
-		nacosContextProperties.setServerAddr("");
+		doReturn("").when(nacosContextProperties).getUsername();
-		nacosContextProperties.toString();
+		doReturn("").when(nacosContextProperties).getServerAddr();
 
 		// mock SDKContext
 		APIConfig apiConfig = mock(APIConfig.class);
@@ -114,15 +123,19 @@ public class PolarisRegistrationTest {
 		ReactiveWebServerApplicationContext reactiveWebServerApplicationContext = mock(ReactiveWebServerApplicationContext.class);
 		doReturn(reactiveWebServer).when(reactiveWebServerApplicationContext).getWebServer();
 
-		polarisRegistration1 = new PolarisRegistration(polarisDiscoveryProperties, consulContextProperties,
+		polarisRegistration1 = new PolarisRegistration(polarisDiscoveryProperties, null, consulContextProperties,
 				polarisContext, staticMetadataManager, nacosContextProperties,
 				servletWebServerApplicationContext, null);
 
-		polarisRegistration2 = new PolarisRegistration(polarisDiscoveryProperties, consulContextProperties,
+		polarisRegistration2 = new PolarisRegistration(polarisDiscoveryProperties, null, consulContextProperties,
 				polarisContext, staticMetadataManager, nacosContextProperties,
 				null, reactiveWebServerApplicationContext);
 
-		polarisRegistration3 = new PolarisRegistration(polarisDiscoveryProperties, consulContextProperties,
+		polarisRegistration3 = new PolarisRegistration(polarisDiscoveryProperties, null, consulContextProperties,
+				polarisContext, staticMetadataManager, nacosContextProperties,
+				null, null);
+
+		polarisRegistration4 = new PolarisRegistration(polarisDiscoveryProperties, polarisContextProperties, consulContextProperties,
 				polarisContext, staticMetadataManager, nacosContextProperties,
 				null, null);
 	}
@@ -147,6 +160,7 @@ public class PolarisRegistrationTest {
 		catch (RuntimeException e) {
 			assertThat(e.getMessage()).isEqualTo("Unsupported web type.");
 		}
+		assertThat(polarisRegistration4.getPort()).isEqualTo(testLocalPort);
 	}
 
 	@Test

spring-cloud-starter-tencent-polaris-router/src/main/java/com/tencent/cloud/polaris/router/RouterUtils.java

@@ -17,11 +17,10 @@
 
 package com.tencent.cloud.polaris.router;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
+import java.util.function.Function;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
 import com.tencent.cloud.common.metadata.MetadataContext;
@@ -30,6 +29,7 @@ import com.tencent.polaris.api.pojo.DefaultServiceInstances;
 import com.tencent.polaris.api.pojo.Instance;
 import com.tencent.polaris.api.pojo.ServiceInstances;
 import com.tencent.polaris.api.pojo.ServiceKey;
+import org.reactivestreams.Publisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
@@ -57,33 +57,19 @@ public final class RouterUtils {
 	 * @return ServiceInstances
 	 */
 	public static ServiceInstances transferServersToServiceInstances(Flux<List<ServiceInstance>> servers, InstanceTransformer instanceTransformer) {
-		CountDownLatch latch = new CountDownLatch(1);
+		List<Instance> instanceList = Collections.synchronizedList(new ArrayList<>());
-
+		servers.flatMap((Function<List<ServiceInstance>, Publisher<?>>) serviceInstances ->
-		AtomicReference<List<Instance>> instancesRef = new AtomicReference<>();
+				Flux.fromIterable(serviceInstances.stream()
-		servers.subscribe(serviceInstances -> {
+						.map(instanceTransformer::transform)
-			instancesRef.set(serviceInstances
+						.collect(Collectors.toList()))).subscribe(instance -> instanceList.add((Instance) instance));
-					.stream()
-					.map(instanceTransformer::transform)
-					.collect(Collectors.toList()));
-
-			latch.countDown();
-		});
-
-		try {
-			latch.await(WAIT_TIME, TimeUnit.SECONDS);
-		}
-		catch (InterruptedException e) {
-			LOGGER.error("Wait get instance result error. ", e);
-		}
 
 		String serviceName = "";
-		if (!CollectionUtils.isEmpty(instancesRef.get())) {
+		if (!CollectionUtils.isEmpty(instanceList)) {
-			serviceName = instancesRef.get().get(0).getService();
+			serviceName = instanceList.get(0).getService();
 		}
 
 		ServiceKey serviceKey = new ServiceKey(MetadataContext.LOCAL_NAMESPACE, serviceName);
-		List<Instance> instances = instancesRef.get() == null ? Collections.emptyList() : instancesRef.get();
 
-		return new DefaultServiceInstances(serviceKey, instances);
+		return new DefaultServiceInstances(serviceKey, instanceList);
 	}
 }

spring-cloud-tencent-polaris-context/src/main/java/com/tencent/cloud/polaris/context/config/PolarisContextAutoConfiguration.java

@@ -52,7 +52,8 @@ public class PolarisContextAutoConfiguration {
 	@ConditionalOnMissingBean
 	public SDKContext polarisContext(PolarisContextProperties properties, Environment environment, List<PolarisConfigModifier> modifierList) throws PolarisException {
 		return SDKContext.initContextByConfig(properties.configuration(modifierList,
-				() -> environment.getProperty("spring.cloud.client.ip-address")));
+				() -> environment.getProperty("spring.cloud.client.ip-address"),
+				() -> environment.getProperty("spring.cloud.polaris.local-port", Integer.class, 0)));
 	}
 
 	@Bean

spring-cloud-tencent-polaris-context/src/main/java/com/tencent/cloud/polaris/context/config/PolarisContextProperties.java

@@ -52,6 +52,11 @@ public class PolarisContextProperties {
 	 */
 	private String localIpAddress;
 
+	/**
+	 * current server local port.
+	 */
+	private Integer localPort;
+
 	/**
 	 * If polaris enabled.
 	 */
@@ -67,7 +72,7 @@ public class PolarisContextProperties {
 	 */
 	private String service;
 
-	public Configuration configuration(List<PolarisConfigModifier> modifierList, Supplier<String> ipAddressSupplier) {
+	public Configuration configuration(List<PolarisConfigModifier> modifierList, Supplier<String> ipAddressSupplier, Supplier<Integer> portSupplier) {
 		// 1. Read user-defined polaris.yml configuration
 		ConfigurationImpl configuration = (ConfigurationImpl) ConfigAPIFactory
 				.defaultConfig(ConfigProvider.DEFAULT_CONFIG);
@@ -78,6 +83,9 @@ public class PolarisContextProperties {
 			defaultHost = ipAddressSupplier.get();
 			this.localIpAddress = defaultHost;
 		}
+		if (this.localPort == null || this.localPort <= 0) {
+			this.localPort = portSupplier.get();
+		}
 
 		configuration.getGlobal().getAPI().setBindIP(defaultHost);
 
@@ -110,6 +118,14 @@ public class PolarisContextProperties {
 		this.localIpAddress = localIpAddress;
 	}
 
+	public Integer getLocalPort() {
+		return localPort;
+	}
+
+	public void setLocalPort(Integer localPort) {
+		this.localPort = localPort;
+	}
+
 	public Boolean getEnabled() {
 		return enabled;
 	}

spring-cloud-tencent-polaris-context/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -38,7 +38,13 @@
       "name": "spring.cloud.polaris.local-ip-address",
       "type": "java.lang.String",
       "defaultValue": "",
-      "description": "current server local ip address."
+      "description": "current server local ip address to be registered."
+    },
+    {
+      "name": "spring.cloud.polaris.local-port",
+      "type": "java.lang.Integer",
+      "defaultValue": "",
+      "description": "current server local port to be registered."
     }
   ],
   "hints": []