refactor:refactor reactor code in router module. (#955)

CHANGELOG.md

@@ -1,9 +1,10 @@
 # Change Log
 ---
 
-- [feature: optimize polaris-ratelimit-example, add caller query params and headers, add callee custom label resolver.](https://github.com/Tencent/spring-cloud-tencent/pull/951)
 - [feature: add config for customized local port.](https://github.com/Tencent/spring-cloud-tencent/pull/923)
 - [feat:support webclient and gateway report call metrics](https://github.com/Tencent/spring-cloud-tencent/pull/924)
 - [feature: optimize polaris-discovery-example/discovery-callee-service, add client-ip return.](https://github.com/Tencent/spring-cloud-tencent/pull/939)
 - [docs:prevent the release of the final version of the sdk.](https://github.com/Tencent/spring-cloud-tencent/pull/943)
 - [fix: fix nacos CircuitBreaker disable bug.](https://github.com/Tencent/spring-cloud-tencent/pull/948)
+- [feature: optimize polaris-ratelimit-example, add caller query params and headers, add callee custom label resolver.](https://github.com/Tencent/spring-cloud-tencent/pull/951)
+- [refactor:refactor reactor code in router module.](https://github.com/Tencent/spring-cloud-tencent/pull/955)

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

@@ -25,6 +25,7 @@ import com.tencent.polaris.factory.api.DiscoveryAPIFactory;
 import com.tencent.polaris.router.api.core.RouterAPI;
 
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.cloud.client.ConditionalOnBlockingDiscoveryEnabled;
@@ -72,6 +73,7 @@ public class PolarisLoadBalancerClientConfiguration {
 
 	@Bean
 	@ConditionalOnMissingBean
+	@ConditionalOnClass(name = "org.springframework.web.reactive.function.client.ClientRequest")
 	public LoadBalancerClientRequestTransformer polarisLoadBalancerClientRequestTransformer(SDKContext sdkContext) {
 		ConsumerAPI consumerAPI = DiscoveryAPIFactory.createConsumerAPIByContext(sdkContext);
 		return new PolarisLoadBalancerClientRequestTransformer(consumerAPI);

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.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
 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);
-
-		AtomicReference<List<Instance>> instancesRef = new AtomicReference<>();
-		servers.subscribe(serviceInstances -> {
-			instancesRef.set(serviceInstances
-					.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);
-		}
+		List<Instance> instanceList = Collections.synchronizedList(new ArrayList<>());
+		servers.flatMap((Function<List<ServiceInstance>, Publisher<?>>) serviceInstances ->
+				Flux.fromIterable(serviceInstances.stream()
+						.map(instanceTransformer::transform)
+						.collect(Collectors.toList()))).subscribe(instance -> instanceList.add((Instance) instance));
 
 		String serviceName = "";
-		if (!CollectionUtils.isEmpty(instancesRef.get())) {
-			serviceName = instancesRef.get().get(0).getService();
+		if (!CollectionUtils.isEmpty(instanceList)) {
+			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/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": []