You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@dubbo.apache.org by GitBox <gi...@apache.org> on 2022/11/06 14:38:25 UTC
[GitHub] [dubbo] aamingaa opened a new pull request, #10885: Feature/modify xds subscribe
aamingaa opened a new pull request, #10885:
URL: https://github.com/apache/dubbo/pull/10885
modify xds factor to push
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] codecov-commenter commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1304824796
# [Codecov](https://codecov.io/gh/apache/dubbo/pull/10885?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#10885](https://codecov.io/gh/apache/dubbo/pull/10885?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (6909d74) into [3.2](https://codecov.io/gh/apache/dubbo/commit/0b5767fc92dcf569b651dec0152e8f8d14229428?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (0b5767f) will **increase** coverage by `0.14%`.
> The diff coverage is `72.77%`.
```diff
@@ Coverage Diff @@
## 3.2 #10885 +/- ##
============================================
+ Coverage 64.76% 64.91% +0.14%
+ Complexity 390 389 -1
============================================
Files 1342 1353 +11
Lines 57323 57496 +173
Branches 8457 8481 +24
============================================
+ Hits 37124 37322 +198
+ Misses 16217 16182 -35
- Partials 3982 3992 +10
```
| [Impacted Files](https://codecov.io/gh/apache/dubbo/pull/10885?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/dubbo/common/constants/CommonConstants.java](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kdWJiby9jb21tb24vY29uc3RhbnRzL0NvbW1vbkNvbnN0YW50cy5qYXZh) | `100.00% <ø> (ø)` | |
| [...a/org/apache/dubbo/rpc/model/ApplicationModel.java](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kdWJiby9ycGMvbW9kZWwvQXBwbGljYXRpb25Nb2RlbC5qYXZh) | `82.35% <0.00%> (ø)` | |
| [...fig/bootstrap/builders/AbstractServiceBuilder.java](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tY29uZmlnL2R1YmJvLWNvbmZpZy1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL2NvbmZpZy9ib290c3RyYXAvYnVpbGRlcnMvQWJzdHJhY3RTZXJ2aWNlQnVpbGRlci5qYXZh) | `88.76% <0.00%> (-4.18%)` | :arrow_down: |
| [...o/registry/xds/util/protocol/AbstractProtocol.java](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8teGRzL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kdWJiby9yZWdpc3RyeS94ZHMvdXRpbC9wcm90b2NvbC9BYnN0cmFjdFByb3RvY29sLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [.../protocol/dubbo/DubboIsolationExecutorSupport.java](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy1kdWJiby9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZHViYm8vcnBjL3Byb3RvY29sL2R1YmJvL0R1YmJvSXNvbGF0aW9uRXhlY3V0b3JTdXBwb3J0LmphdmE=) | `58.82% <58.82%> (ø)` | |
| [...in/java/org/apache/dubbo/config/ServiceConfig.java](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tY29uZmlnL2R1YmJvLWNvbmZpZy1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL2NvbmZpZy9TZXJ2aWNlQ29uZmlnLmphdmE=) | `70.09% <66.66%> (-0.16%)` | :arrow_down: |
| [.../tri/transport/TripleIsolationExecutorSupport.java](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvdHJhbnNwb3J0L1RyaXBsZUlzb2xhdGlvbkV4ZWN1dG9yU3VwcG9ydC5qYXZh) | `70.58% <70.58%> (ø)` | |
| [.../common/threadpool/manager/ExecutorRepository.java](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kdWJiby9jb21tb24vdGhyZWFkcG9vbC9tYW5hZ2VyL0V4ZWN1dG9yUmVwb3NpdG9yeS5qYXZh) | `80.00% <80.00%> (ø)` | |
| [...rpc/executor/AbstractIsolationExecutorSupport.java](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kdWJiby9ycGMvZXhlY3V0b3IvQWJzdHJhY3RJc29sYXRpb25FeGVjdXRvclN1cHBvcnQuamF2YQ==) | `82.75% <82.75%> (ø)` | |
| [...ng/transport/dispatcher/WrappedChannelHandler.java](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcmVtb3RpbmcvZHViYm8tcmVtb3RpbmctYXBpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kdWJiby9yZW1vdGluZy90cmFuc3BvcnQvZGlzcGF0Y2hlci9XcmFwcGVkQ2hhbm5lbEhhbmRsZXIuamF2YQ==) | `56.86% <83.33%> (-0.29%)` | :arrow_down: |
| ... and [90 more](https://codecov.io/gh/apache/dubbo/pull/10885/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1314633108
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [18 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] aamingaa commented on a diff in pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
aamingaa commented on code in PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#discussion_r1017811034
##########
dubbo-dependencies-bom/pom.xml:
##########
@@ -142,7 +142,7 @@
<tomcat_embed_version>8.5.78</tomcat_embed_version>
<jetcd_version>0.5.3</jetcd_version>
<nacos_version>2.1.0</nacos_version>
- <grpc.version>1.47.0</grpc.version>
+ <grpc.version>1.41.0</grpc.version>
Review Comment:
because the 1.47.0 version can't send ca request to envoy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1372320133
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [14 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [49.2% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] aamingaa commented on a diff in pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
aamingaa commented on code in PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#discussion_r1018232085
##########
dubbo-dependencies-bom/pom.xml:
##########
@@ -180,7 +180,7 @@
<portlet_version>2.0</portlet_version>
<maven_flatten_version>1.1.0</maven_flatten_version>
<commons_compress_version>1.21</commons_compress_version>
- <revision>3.2.0-beta.1-SNAPSHOT</revision>
+ <revision>3.3</revision>
Review Comment:
just ignore it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] AlbumenJ commented on a diff in pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
AlbumenJ commented on code in PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#discussion_r1048221006
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java:
##########
@@ -16,182 +16,127 @@
*/
package org.apache.dubbo.registry.xds.util.protocol;
+
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+
+
import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.NamedThreadFactory;
+import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.registry.xds.util.XdsChannel;
-
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import io.grpc.stub.StreamObserver;
+import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.Collections;
-import java.util.Map;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
+import java.util.List;
+import java.util.Map;
+import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_REQUEST_XDS;
-public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements XdsProtocol<T> {
+public abstract class AbstractProtocol<T, S extends DeltaResource<T>, R> implements XdsProtocol<T> {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractProtocol.class);
- protected final XdsChannel xdsChannel;
+ protected XdsChannel xdsChannel;
protected final Node node;
- /**
- * Store Request Parameter ( resourceNames )
- * K - requestId, V - resourceNames
- */
- protected final Map<Long, Set<String>> requestParam = new ConcurrentHashMap<>();
-
- /**
- * Store ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, StreamObserver<DiscoveryRequest>> requestObserverMap = new ConcurrentHashMap<>();
-
- /**
- * Store Delta-ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, ScheduledFuture<?>> observeScheduledMap = new ConcurrentHashMap<>();
-
- /**
- * Store CompletableFuture for Request ( used to fetch async result in ResponseObserver )
- * K - requestId, V - CompletableFuture
- */
- private final Map<Long, CompletableFuture<T>> streamResult = new ConcurrentHashMap<>();
+ private final int pollingTimeout;
- private final ScheduledExecutorService pollingExecutor;
+ private final Object consumerObserveMapUpdate = new Object();
Review Comment:
Replace to `ReentrantLock`
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java:
##########
@@ -16,182 +16,127 @@
*/
package org.apache.dubbo.registry.xds.util.protocol;
+
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+
+
import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.NamedThreadFactory;
+import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.registry.xds.util.XdsChannel;
-
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import io.grpc.stub.StreamObserver;
+import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.Collections;
-import java.util.Map;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
+import java.util.List;
+import java.util.Map;
+import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_REQUEST_XDS;
-public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements XdsProtocol<T> {
+public abstract class AbstractProtocol<T, S extends DeltaResource<T>, R> implements XdsProtocol<T> {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractProtocol.class);
- protected final XdsChannel xdsChannel;
+ protected XdsChannel xdsChannel;
protected final Node node;
- /**
- * Store Request Parameter ( resourceNames )
- * K - requestId, V - resourceNames
- */
- protected final Map<Long, Set<String>> requestParam = new ConcurrentHashMap<>();
-
- /**
- * Store ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, StreamObserver<DiscoveryRequest>> requestObserverMap = new ConcurrentHashMap<>();
-
- /**
- * Store Delta-ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, ScheduledFuture<?>> observeScheduledMap = new ConcurrentHashMap<>();
-
- /**
- * Store CompletableFuture for Request ( used to fetch async result in ResponseObserver )
- * K - requestId, V - CompletableFuture
- */
- private final Map<Long, CompletableFuture<T>> streamResult = new ConcurrentHashMap<>();
+ private final int pollingTimeout;
- private final ScheduledExecutorService pollingExecutor;
+ private final Object consumerObserveMapUpdate = new Object();
- private final int pollingTimeout;
+ private Set<String> observeResourcesName;
- protected final static AtomicLong requestId = new AtomicLong(0);
+ private final Map<Set<String>, List<Consumer<T>>> consumerObserveMap = new ConcurrentHashMap<>();
- public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) {
+ public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout) {
this.xdsChannel = xdsChannel;
this.node = node;
- this.pollingExecutor = new ScheduledThreadPoolExecutor(pollingPoolSize, new NamedThreadFactory("Dubbo-registry-xds"));
this.pollingTimeout = pollingTimeout;
}
+ protected Map<String, R> resourcesMap = new ConcurrentHashMap<>();
+
+ private CompletableFuture<T> future;
+
+ private StreamObserver<DiscoveryRequest> requestObserver;
+
/**
* Abstract method to obtain Type-URL from sub-class
*
* @return Type-URL of xDS
*/
public abstract String getTypeUrl();
- @Override
- public T getResource(Set<String> resourceNames) {
- long request = requestId.getAndIncrement();
- resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
-
- // Store Request Parameter, which will be used for ACK
- requestParam.put(request, resourceNames);
-
- // create observer
- StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
-
- // use future to get async result
- CompletableFuture<T> future = new CompletableFuture<>();
- requestObserverMap.put(request, requestObserver);
- streamResult.put(request, future);
-
- // send request to control panel
- requestObserver.onNext(buildDiscoveryRequest(resourceNames));
-
- try {
- // get result
- return future.get();
- } catch (InterruptedException | ExecutionException e) {
- logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Error occur when request control panel.");
- return null;
- } finally {
- // close observer
- //requestObserver.onCompleted();
-
- // remove temp
- streamResult.remove(request);
- requestObserverMap.remove(request);
- requestParam.remove(request);
- }
- }
+ public abstract boolean isExistResource(Set<String> resourceNames);
Review Comment:
Remove this abstract method. Should be handled in `AbstractProtocol`.
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java:
##########
@@ -16,182 +16,127 @@
*/
package org.apache.dubbo.registry.xds.util.protocol;
+
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+
+
import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.NamedThreadFactory;
+import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.registry.xds.util.XdsChannel;
-
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import io.grpc.stub.StreamObserver;
+import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.Collections;
-import java.util.Map;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
+import java.util.List;
+import java.util.Map;
+import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_REQUEST_XDS;
-public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements XdsProtocol<T> {
+public abstract class AbstractProtocol<T, S extends DeltaResource<T>, R> implements XdsProtocol<T> {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractProtocol.class);
- protected final XdsChannel xdsChannel;
+ protected XdsChannel xdsChannel;
protected final Node node;
- /**
- * Store Request Parameter ( resourceNames )
- * K - requestId, V - resourceNames
- */
- protected final Map<Long, Set<String>> requestParam = new ConcurrentHashMap<>();
-
- /**
- * Store ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, StreamObserver<DiscoveryRequest>> requestObserverMap = new ConcurrentHashMap<>();
-
- /**
- * Store Delta-ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, ScheduledFuture<?>> observeScheduledMap = new ConcurrentHashMap<>();
-
- /**
- * Store CompletableFuture for Request ( used to fetch async result in ResponseObserver )
- * K - requestId, V - CompletableFuture
- */
- private final Map<Long, CompletableFuture<T>> streamResult = new ConcurrentHashMap<>();
+ private final int pollingTimeout;
Review Comment:
rename to `checkInterval`
##########
dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/istio/IstioEnvMock.java:
##########
@@ -0,0 +1,47 @@
+package org.apache.dubbo.registry.xds.istio;
Review Comment:
Add ASF license header for all of the newly created files
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java:
##########
@@ -248,13 +196,30 @@ private void returnResult(T result) {
@Override
public void onCompleted() {
- logger.info("xDS Client completed, requestId: " + requestId);
- clear();
- }
-
- private void clear() {
- requestObserverMap.remove(requestId);
+ logger.info("xDS Client completed");
}
}
+ private void triggerReConnectTask() {
+ AtomicBoolean isConnectFail = new AtomicBoolean(false);
+ ScheduledExecutorService scheduledFuture = ApplicationModel.defaultModel().getFrameworkModel().getBeanFactory()
+ .getBean(FrameworkExecutorRepository.class).getSharedScheduledExecutor();
Review Comment:
Should fetch `ApplicationModel` in contructor. `ApplicationModel.defaultModel()` is forbidden here.
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java:
##########
@@ -16,182 +16,127 @@
*/
package org.apache.dubbo.registry.xds.util.protocol;
+
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+
+
import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.NamedThreadFactory;
+import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.registry.xds.util.XdsChannel;
-
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import io.grpc.stub.StreamObserver;
+import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.Collections;
-import java.util.Map;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
+import java.util.List;
+import java.util.Map;
+import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_REQUEST_XDS;
-public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements XdsProtocol<T> {
+public abstract class AbstractProtocol<T, S extends DeltaResource<T>, R> implements XdsProtocol<T> {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractProtocol.class);
- protected final XdsChannel xdsChannel;
+ protected XdsChannel xdsChannel;
protected final Node node;
- /**
- * Store Request Parameter ( resourceNames )
- * K - requestId, V - resourceNames
- */
- protected final Map<Long, Set<String>> requestParam = new ConcurrentHashMap<>();
-
- /**
- * Store ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, StreamObserver<DiscoveryRequest>> requestObserverMap = new ConcurrentHashMap<>();
-
- /**
- * Store Delta-ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, ScheduledFuture<?>> observeScheduledMap = new ConcurrentHashMap<>();
-
- /**
- * Store CompletableFuture for Request ( used to fetch async result in ResponseObserver )
- * K - requestId, V - CompletableFuture
- */
- private final Map<Long, CompletableFuture<T>> streamResult = new ConcurrentHashMap<>();
+ private final int pollingTimeout;
- private final ScheduledExecutorService pollingExecutor;
+ private final Object consumerObserveMapUpdate = new Object();
- private final int pollingTimeout;
+ private Set<String> observeResourcesName;
- protected final static AtomicLong requestId = new AtomicLong(0);
+ private final Map<Set<String>, List<Consumer<T>>> consumerObserveMap = new ConcurrentHashMap<>();
- public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) {
+ public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout) {
this.xdsChannel = xdsChannel;
this.node = node;
- this.pollingExecutor = new ScheduledThreadPoolExecutor(pollingPoolSize, new NamedThreadFactory("Dubbo-registry-xds"));
this.pollingTimeout = pollingTimeout;
}
+ protected Map<String, R> resourcesMap = new ConcurrentHashMap<>();
+
+ private CompletableFuture<T> future;
+
+ private StreamObserver<DiscoveryRequest> requestObserver;
+
/**
* Abstract method to obtain Type-URL from sub-class
*
* @return Type-URL of xDS
*/
public abstract String getTypeUrl();
- @Override
- public T getResource(Set<String> resourceNames) {
- long request = requestId.getAndIncrement();
- resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
-
- // Store Request Parameter, which will be used for ACK
- requestParam.put(request, resourceNames);
-
- // create observer
- StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
-
- // use future to get async result
- CompletableFuture<T> future = new CompletableFuture<>();
- requestObserverMap.put(request, requestObserver);
- streamResult.put(request, future);
-
- // send request to control panel
- requestObserver.onNext(buildDiscoveryRequest(resourceNames));
-
- try {
- // get result
- return future.get();
- } catch (InterruptedException | ExecutionException e) {
- logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Error occur when request control panel.");
- return null;
- } finally {
- // close observer
- //requestObserver.onCompleted();
-
- // remove temp
- streamResult.remove(request);
- requestObserverMap.remove(request);
- requestParam.remove(request);
- }
- }
+ public abstract boolean isExistResource(Set<String> resourceNames);
- @Override
- public long observeResource(Set<String> resourceNames, Consumer<T> consumer) {
- long request = requestId.getAndIncrement();
- resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
+ public abstract void updateResourceCollection(R resourceCollection, Set<String> resourceNames);
- // Store Request Parameter, which will be used for ACK
- requestParam.put(request, resourceNames);
-
- // call once for full data
- consumer.accept(getResource(resourceNames));
+ public abstract R getResourceCollection();
- // channel reused
- StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
- requestObserverMap.put(request, requestObserver);
+ public abstract T getDsResult(R resourceCollection);
- ScheduledFuture<?> scheduledFuture = pollingExecutor.scheduleAtFixedRate(() -> {
+ public T getCacheResource(Set<String> resourceNames) {
+ R resourceCollection = getResourceCollection();
+ if (!resourceNames.isEmpty() && isExistResource(resourceNames)) {
+ updateResourceCollection(resourceCollection, resourceNames);
+ } else {
+ if (requestObserver == null) {
+ future = new CompletableFuture<>();
+ requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(future));
Review Comment:
Observe resource in `consumerObserveMap`
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java:
##########
@@ -16,182 +16,127 @@
*/
package org.apache.dubbo.registry.xds.util.protocol;
+
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+
+
import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.NamedThreadFactory;
+import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.registry.xds.util.XdsChannel;
-
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import io.grpc.stub.StreamObserver;
+import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.Collections;
-import java.util.Map;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
+import java.util.List;
+import java.util.Map;
+import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_REQUEST_XDS;
-public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements XdsProtocol<T> {
+public abstract class AbstractProtocol<T, S extends DeltaResource<T>, R> implements XdsProtocol<T> {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractProtocol.class);
- protected final XdsChannel xdsChannel;
+ protected XdsChannel xdsChannel;
protected final Node node;
- /**
- * Store Request Parameter ( resourceNames )
- * K - requestId, V - resourceNames
- */
- protected final Map<Long, Set<String>> requestParam = new ConcurrentHashMap<>();
-
- /**
- * Store ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, StreamObserver<DiscoveryRequest>> requestObserverMap = new ConcurrentHashMap<>();
-
- /**
- * Store Delta-ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, ScheduledFuture<?>> observeScheduledMap = new ConcurrentHashMap<>();
-
- /**
- * Store CompletableFuture for Request ( used to fetch async result in ResponseObserver )
- * K - requestId, V - CompletableFuture
- */
- private final Map<Long, CompletableFuture<T>> streamResult = new ConcurrentHashMap<>();
+ private final int pollingTimeout;
- private final ScheduledExecutorService pollingExecutor;
+ private final Object consumerObserveMapUpdate = new Object();
- private final int pollingTimeout;
+ private Set<String> observeResourcesName;
- protected final static AtomicLong requestId = new AtomicLong(0);
+ private final Map<Set<String>, List<Consumer<T>>> consumerObserveMap = new ConcurrentHashMap<>();
- public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) {
+ public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout) {
this.xdsChannel = xdsChannel;
this.node = node;
- this.pollingExecutor = new ScheduledThreadPoolExecutor(pollingPoolSize, new NamedThreadFactory("Dubbo-registry-xds"));
this.pollingTimeout = pollingTimeout;
}
+ protected Map<String, R> resourcesMap = new ConcurrentHashMap<>();
+
+ private CompletableFuture<T> future;
+
+ private StreamObserver<DiscoveryRequest> requestObserver;
+
/**
* Abstract method to obtain Type-URL from sub-class
*
* @return Type-URL of xDS
*/
public abstract String getTypeUrl();
- @Override
- public T getResource(Set<String> resourceNames) {
- long request = requestId.getAndIncrement();
- resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
-
- // Store Request Parameter, which will be used for ACK
- requestParam.put(request, resourceNames);
-
- // create observer
- StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
-
- // use future to get async result
- CompletableFuture<T> future = new CompletableFuture<>();
- requestObserverMap.put(request, requestObserver);
- streamResult.put(request, future);
-
- // send request to control panel
- requestObserver.onNext(buildDiscoveryRequest(resourceNames));
-
- try {
- // get result
- return future.get();
- } catch (InterruptedException | ExecutionException e) {
- logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Error occur when request control panel.");
- return null;
- } finally {
- // close observer
- //requestObserver.onCompleted();
-
- // remove temp
- streamResult.remove(request);
- requestObserverMap.remove(request);
- requestParam.remove(request);
- }
- }
+ public abstract boolean isExistResource(Set<String> resourceNames);
- @Override
- public long observeResource(Set<String> resourceNames, Consumer<T> consumer) {
- long request = requestId.getAndIncrement();
- resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
+ public abstract void updateResourceCollection(R resourceCollection, Set<String> resourceNames);
- // Store Request Parameter, which will be used for ACK
- requestParam.put(request, resourceNames);
-
- // call once for full data
- consumer.accept(getResource(resourceNames));
+ public abstract R getResourceCollection();
- // channel reused
- StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
- requestObserverMap.put(request, requestObserver);
+ public abstract T getDsResult(R resourceCollection);
- ScheduledFuture<?> scheduledFuture = pollingExecutor.scheduleAtFixedRate(() -> {
+ public T getCacheResource(Set<String> resourceNames) {
+ R resourceCollection = getResourceCollection();
+ if (!resourceNames.isEmpty() && isExistResource(resourceNames)) {
+ updateResourceCollection(resourceCollection, resourceNames);
+ } else {
+ if (requestObserver == null) {
+ future = new CompletableFuture<>();
+ requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(future));
Review Comment:
remove this `future`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] CrazyHZM commented on a diff in pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
CrazyHZM commented on code in PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#discussion_r1017365394
##########
dubbo-dependencies-bom/pom.xml:
##########
@@ -142,7 +142,7 @@
<tomcat_embed_version>8.5.78</tomcat_embed_version>
<jetcd_version>0.5.3</jetcd_version>
<nacos_version>2.1.0</nacos_version>
- <grpc.version>1.47.0</grpc.version>
+ <grpc.version>1.41.0</grpc.version>
Review Comment:
Why lower the grpc version?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1314647222
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [18 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1316322733
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [18 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] AlbumenJ commented on a diff in pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
AlbumenJ commented on code in PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#discussion_r1015125754
##########
dubbo-dependencies-bom/pom.xml:
##########
@@ -180,7 +180,7 @@
<portlet_version>2.0</portlet_version>
<maven_flatten_version>1.1.0</maven_flatten_version>
<commons_compress_version>1.21</commons_compress_version>
- <revision>3.2.0-beta.1-SNAPSHOT</revision>
+ <revision>3.3</revision>
Review Comment:
Why change this line
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java:
##########
@@ -139,48 +132,43 @@ public long observeResource(Set<String> resourceNames, Consumer<T> consumer) {
consumer.accept(getResource(resourceNames));
// channel reused
- StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
+ StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request, consumer));
requestObserverMap.put(request, requestObserver);
- ScheduledFuture<?> scheduledFuture = pollingExecutor.scheduleAtFixedRate(() -> {
- try {
- // origin request, may changed by updateObserve
- Set<String> names = requestParam.get(request);
-
- // use future to get async result, future complete on StreamObserver onNext
- CompletableFuture<T> future = new CompletableFuture<>();
- streamResult.put(request, future);
-
- // observer reused
- StreamObserver<DiscoveryRequest> observer = requestObserverMap.get(request);
-
- if (observer == null) {
- observer = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
- requestObserverMap.put(request, observer);
- }
-
- // send request to control panel
- observer.onNext(buildDiscoveryRequest(names));
-
- try {
- // get result
- consumer.accept(future.get());
- } catch (InterruptedException | ExecutionException e) {
- logger.error("Error occur when request control panel.");
- } finally {
- // close observer
- //requestObserver.onCompleted();
-
- // remove temp
- streamResult.remove(request);
- }
- } catch (Throwable t) {
- logger.error("Error when requesting observe data. Type: " + getTypeUrl(), t);
- }
- }, pollingTimeout, pollingTimeout, TimeUnit.SECONDS);
+ try {
+ // origin request, may changed by updateObserve
+ Set<String> names = requestParam.get(request);
+
+ // use future to get async result, future complete on StreamObserver onNext
+ CompletableFuture<T> future = new CompletableFuture<>();
+ streamResult.put(request, future);
+
+ // observer reused
+ StreamObserver<DiscoveryRequest> observer = requestObserverMap.get(request);
- observeScheduledMap.put(request, scheduledFuture);
+ if (observer == null) {
+ observer = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request, consumer));
+ requestObserverMap.put(request, observer);
+ }
+ // send request to control panel
+ observer.onNext(buildDiscoveryRequest(names));
+
+// try {
+// // get result
+// consumer.accept(future.get());
+// } catch (InterruptedException | ExecutionException e) {
+// logger.error("Error occur when request control panel.");
+// } finally {
+// // close observer
+// //requestObserver.onCompleted();
+//
+// // remove temp
+// streamResult.remove(request);
+// }
Review Comment:
remove unused code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1341254497
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [3 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] AlbumenJ merged pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
AlbumenJ merged PR #10885:
URL: https://github.com/apache/dubbo/pull/10885
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] AlbumenJ commented on a diff in pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
AlbumenJ commented on code in PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#discussion_r1051331022
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java:
##########
@@ -45,153 +42,120 @@
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractProtocol.class);
- protected final XdsChannel xdsChannel;
+ protected XdsChannel xdsChannel;
protected final Node node;
- /**
- * Store Request Parameter ( resourceNames )
- * K - requestId, V - resourceNames
- */
- protected final Map<Long, Set<String>> requestParam = new ConcurrentHashMap<>();
+ private final int checkInterval;
- /**
- * Store ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, StreamObserver<DiscoveryRequest>> requestObserverMap = new ConcurrentHashMap<>();
+ private final ReentrantLock lock = new ReentrantLock();
- /**
- * Store Delta-ADS Request Observer ( StreamObserver in Streaming Request )
- * K - requestId, V - StreamObserver
- */
- private final Map<Long, ScheduledFuture<?>> observeScheduledMap = new ConcurrentHashMap<>();
-
- /**
- * Store CompletableFuture for Request ( used to fetch async result in ResponseObserver )
- * K - requestId, V - CompletableFuture
- */
- private final Map<Long, CompletableFuture<T>> streamResult = new ConcurrentHashMap<>();
+ private Set<String> observeResourcesName;
- private final ScheduledExecutorService pollingExecutor;
+ private final Map<Set<String>, List<Consumer<T>>> consumerObserveMap = new ConcurrentHashMap<>();
- private final int pollingTimeout;
+ private ApplicationModel applicationModel;
- protected final static AtomicLong requestId = new AtomicLong(0);
-
- public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) {
+ public AbstractProtocol(XdsChannel xdsChannel, Node node, int checkInterval, ApplicationModel applicationModel) {
this.xdsChannel = xdsChannel;
this.node = node;
- this.pollingExecutor = new ScheduledThreadPoolExecutor(pollingPoolSize, new NamedThreadFactory("Dubbo-registry-xds"));
- this.pollingTimeout = pollingTimeout;
+ this.checkInterval = checkInterval;
+ this.applicationModel = applicationModel;
}
+ protected Map<String, Object> resourcesMap = new ConcurrentHashMap<>();
+
+ private StreamObserver<DiscoveryRequest> requestObserver;
+
/**
* Abstract method to obtain Type-URL from sub-class
*
* @return Type-URL of xDS
*/
public abstract String getTypeUrl();
- @Override
- public T getResource(Set<String> resourceNames) {
- long request = requestId.getAndIncrement();
- resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
-
- // Store Request Parameter, which will be used for ACK
- requestParam.put(request, resourceNames);
-
- // create observer
- StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
-
- // use future to get async result
- CompletableFuture<T> future = new CompletableFuture<>();
- requestObserverMap.put(request, requestObserver);
- streamResult.put(request, future);
+ public abstract T getDsResult(Set<String> resourceNames);
- // send request to control panel
- requestObserver.onNext(buildDiscoveryRequest(resourceNames));
+// public abstract T getDsResult();
- try {
- // get result
- return future.get();
- } catch (InterruptedException | ExecutionException e) {
- logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Error occur when request control panel.");
- return null;
- } finally {
- // close observer
- //requestObserver.onCompleted();
-
- // remove temp
- streamResult.remove(request);
- requestObserverMap.remove(request);
- requestParam.remove(request);
+ public boolean isExistResource(Set<String> resourceNames) {
+ for (String resourceName : resourceNames) {
+ if ("".equals(resourceName)) {
Review Comment:
```suggestion
if (StringUtils.isEmpty(resourceName)) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] chickenlj commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
chickenlj commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1324666992
Thanks. I will have a look with this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1372188388
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [14 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [49.2% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1341901182
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [3 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [4 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1342206021
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [4 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [13.9% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1341298945
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [3 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [4 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1316516567
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [18 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] sonarcloud[bot] commented on pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#issuecomment-1341878847
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_dubbo&pullRequest=10885)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG) [3 Bugs](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo&pullRequest=10885&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL) [4 Code Smells](https://sonarcloud.io/project/issues?id=apache_dubbo&pullRequest=10885&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_dubbo&pullRequest=10885&metric=new_duplicated_lines_density&view=list)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] AlbumenJ commented on a diff in pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
AlbumenJ commented on code in PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#discussion_r1043000896
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java:
##########
@@ -32,27 +34,75 @@
import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.Rds;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
-import java.util.Collections;
-import java.util.Objects;
+import java.util.HashMap;
import java.util.Set;
-import java.util.function.Consumer;
+import java.util.Objects;
+import java.util.HashSet;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
+import java.util.function.Consumer;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_RESPONSE_XDS;
public class LdsProtocol extends AbstractProtocol<ListenerResult, DeltaListener> {
-
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(LdsProtocol.class);
- public LdsProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) {
- super(xdsChannel, node, pollingPoolSize, pollingTimeout);
+ private StreamObserver<DiscoveryRequest> requestObserver;
+
+ private CompletableFuture<ListenerResult> future;
+ public LdsProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout) {
+ super(xdsChannel, node, pollingTimeout);
}
@Override
public String getTypeUrl() {
return "type.googleapis.com/envoy.config.listener.v3.Listener";
}
+ private HashMap<String, Object> resourcesMap = new HashMap<>();
+
+ @Override
+ public boolean isExistResource(Set<String> resourceNames) {
+ for (String resourceName : resourceNames) {
+ if (!resourcesMap.containsKey(resourceName)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public ListenerResult getCacheResource(Set<String> resourceNames) {
+ Set<String> resourceSet = new HashSet<>();
+ if (!resourceNames.isEmpty() && isExistResource(resourceNames)) {
+ for (String resourceName : resourceNames) {
+ resourceSet.add((String) resourcesMap.get(resourceName));
+ }
+ } else {
+ if (requestObserver == null) {
+ future = new CompletableFuture<>();
+ requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(future));
+ }
+ resourceNames.addAll(resourcesMap.keySet());
+ requestObserver.onNext(buildDiscoveryRequest(resourceNames));
+ try {
+ return future.get();
+ } catch (InterruptedException e) {
+ logger.error("InterruptedException occur when request control panel. error={}", e);
+ Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ logger.error("Error occur when request control panel. error=. ",e);
+ }
+ }
+ return new ListenerResult(resourceSet);
+ }
Review Comment:
Move this method impl to `AbstractProtocol`.
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java:
##########
@@ -43,15 +48,61 @@ public class EdsProtocol extends AbstractProtocol<EndpointResult, DeltaEndpoint>
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(EdsProtocol.class);
- public EdsProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) {
- super(xdsChannel, node, pollingPoolSize, pollingTimeout);
+ private StreamObserver<DiscoveryRequest> requestObserver;
+
+ private HashMap<String, Object> resourcesMap = new HashMap<>();
+
+
+ public EdsProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout) {
+ super(xdsChannel, node, pollingTimeout);
}
@Override
public String getTypeUrl() {
return "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment";
}
+ @Override
+ public boolean isExistResource(Set<String> resourceNames) {
+ for (String resourceName : resourceNames) {
+ if (!resourcesMap.containsKey(resourceName)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public EndpointResult getCacheResource(Set<String> resourceNames) {
+ Set<Endpoint> resourceSet = new HashSet<>();
+ if (!resourceNames.isEmpty() && isExistResource(resourceNames)) {
+ for (String resourceName : resourceNames) {
+ resourceSet.addAll((Set<Endpoint>) resourcesMap.get(resourceName));
+ }
+ } else {
+ CompletableFuture<EndpointResult> future = new CompletableFuture<>();
+ if (requestObserver == null) {
+ requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(future));
+ }
Review Comment:
`future` will never get response if `requestObserver` is not null when `resourceNames` update.
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java:
##########
@@ -43,22 +46,73 @@ public class RdsProtocol extends AbstractProtocol<RouteResult, DeltaRoute> {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(RdsProtocol.class);
- public RdsProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) {
- super(xdsChannel, node, pollingPoolSize, pollingTimeout);
+ private StreamObserver<DiscoveryRequest> requestObserver;
+
+ private HashMap<String, Object> resourcesMap = new HashMap<>();
Review Comment:
Use concurrent map
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java:
##########
@@ -93,103 +65,23 @@ public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, i
*/
public abstract String getTypeUrl();
+ public abstract boolean isExistResource(Set<String> resourceNames);
+
+ public abstract T getCacheResource(Set<String> resourceNames);
+
+ public abstract StreamObserver<DiscoveryRequest> getStreamObserver();
@Override
public T getResource(Set<String> resourceNames) {
- long request = requestId.getAndIncrement();
resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
-
- // Store Request Parameter, which will be used for ACK
- requestParam.put(request, resourceNames);
-
- // create observer
- StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
-
- // use future to get async result
- CompletableFuture<T> future = new CompletableFuture<>();
- requestObserverMap.put(request, requestObserver);
- streamResult.put(request, future);
-
- // send request to control panel
- requestObserver.onNext(buildDiscoveryRequest(resourceNames));
-
- try {
- // get result
- return future.get();
- } catch (InterruptedException | ExecutionException e) {
- logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Error occur when request control panel.");
- return null;
- } finally {
- // close observer
- //requestObserver.onCompleted();
-
- // remove temp
- streamResult.remove(request);
- requestObserverMap.remove(request);
- requestParam.remove(request);
- }
+ return getCacheResource(resourceNames);
}
-
@Override
- public long observeResource(Set<String> resourceNames, Consumer<T> consumer) {
- long request = requestId.getAndIncrement();
+ public void observeResource(Set<String> resourceNames, Consumer<T> consumer) {
resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
-
- // Store Request Parameter, which will be used for ACK
- requestParam.put(request, resourceNames);
-
// call once for full data
consumer.accept(getResource(resourceNames));
-
- // channel reused
- StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
- requestObserverMap.put(request, requestObserver);
-
- ScheduledFuture<?> scheduledFuture = pollingExecutor.scheduleAtFixedRate(() -> {
- try {
- // origin request, may changed by updateObserve
- Set<String> names = requestParam.get(request);
-
- // use future to get async result, future complete on StreamObserver onNext
- CompletableFuture<T> future = new CompletableFuture<>();
- streamResult.put(request, future);
-
- // observer reused
- StreamObserver<DiscoveryRequest> observer = requestObserverMap.get(request);
-
- if (observer == null) {
- observer = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
- requestObserverMap.put(request, observer);
- }
-
- // send request to control panel
- observer.onNext(buildDiscoveryRequest(names));
-
- try {
- // get result
- consumer.accept(future.get());
- } catch (InterruptedException | ExecutionException e) {
- logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Error occur when request control panel.");
- } finally {
- // close observer
- //requestObserver.onCompleted();
-
- // remove temp
- streamResult.remove(request);
- }
- } catch (Throwable t) {
- logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Error when requesting observe data. Type: " + getTypeUrl(), t);
- }
- }, pollingTimeout, pollingTimeout, TimeUnit.SECONDS);
-
- observeScheduledMap.put(request, scheduledFuture);
-
- return request;
- }
-
- @Override
- public void updateObserve(long request, Set<String> resourceNames) {
- // send difference in resourceNames
- requestParam.put(request, resourceNames);
+ this.observeResourcesName = resourceNames;
+ this.observeConsumer = consumer;
Review Comment:
Should support multi observe.
E.g. Consumer 1 consumer resource names 1. Consumer 2 consumer resource names 2.
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java:
##########
@@ -45,61 +46,51 @@ public class PilotExchanger {
private ListenerResult listenerResult;
private RouteResult routeResult;
-
- private final AtomicLong observeRouteRequest = new AtomicLong(-1);
-
- private final Map<String, Long> domainObserveRequest = new ConcurrentHashMap<>();
+ private final AtomicBoolean isRdsObserve = new AtomicBoolean(false);
+ private final HashSet<String> domainObserveRequest = new HashSet<>();
private final Map<String, Set<Consumer<Set<Endpoint>>>> domainObserveConsumer = new ConcurrentHashMap<>();
private PilotExchanger(URL url) {
xdsChannel = new XdsChannel(url);
- int pollingPoolSize = url.getParameter("pollingPoolSize", 10);
int pollingTimeout = url.getParameter("pollingTimeout", 10);
- LdsProtocol ldsProtocol = new LdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout);
- this.rdsProtocol = new RdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout);
- this.edsProtocol = new EdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout);
+ LdsProtocol ldsProtocol = new LdsProtocol(xdsChannel, NodeBuilder.build(), pollingTimeout);
+ this.rdsProtocol = new RdsProtocol(xdsChannel, NodeBuilder.build(), pollingTimeout);
+ this.edsProtocol = new EdsProtocol(xdsChannel, NodeBuilder.build(), pollingTimeout);
this.listenerResult = ldsProtocol.getListeners();
this.routeResult = rdsProtocol.getResource(listenerResult.getRouteConfigNames());
// Observer RDS update
if (CollectionUtils.isNotEmpty(listenerResult.getRouteConfigNames())) {
- this.observeRouteRequest.set(createRouteObserve());
+ createRouteObserve();
+ isRdsObserve.set(true);
}
+
// Observe LDS updated
ldsProtocol.observeListeners((newListener) -> {
// update local cache
if (!newListener.equals(listenerResult)) {
this.listenerResult = newListener;
// update RDS observation
- synchronized (observeRouteRequest) {
- if (observeRouteRequest.get() == -1) {
- this.observeRouteRequest.set(createRouteObserve());
- } else {
- rdsProtocol.updateObserve(observeRouteRequest.get(), newListener.getRouteConfigNames());
- }
+ synchronized (isRdsObserve) {
Review Comment:
Why sync on a atomic boolean object?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo] AlbumenJ commented on a diff in pull request #10885: Feature/modify xds subscribe
Posted by GitBox <gi...@apache.org>.
AlbumenJ commented on code in PR #10885:
URL: https://github.com/apache/dubbo/pull/10885#discussion_r1051359333
##########
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java:
##########
@@ -26,85 +26,81 @@
import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult;
import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult;
import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult;
+import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.Collections;
+import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
public class PilotExchanger {
- private final XdsChannel xdsChannel;
+ protected final XdsChannel xdsChannel;
- private final RdsProtocol rdsProtocol;
+ protected final LdsProtocol ldsProtocol;
- private final EdsProtocol edsProtocol;
+ protected final RdsProtocol rdsProtocol;
- private ListenerResult listenerResult;
+ protected final EdsProtocol edsProtocol;
- private RouteResult routeResult;
+ protected ListenerResult listenerResult;
- private final AtomicLong observeRouteRequest = new AtomicLong(-1);
+ protected RouteResult routeResult;
- private final Map<String, Long> domainObserveRequest = new ConcurrentHashMap<>();
+ private final AtomicBoolean isRdsObserve = new AtomicBoolean(false);
+ private final HashSet<String> domainObserveRequest = new HashSet<>();
private final Map<String, Set<Consumer<Set<Endpoint>>>> domainObserveConsumer = new ConcurrentHashMap<>();
- private PilotExchanger(URL url) {
+ protected PilotExchanger(URL url) {
xdsChannel = new XdsChannel(url);
- int pollingPoolSize = url.getParameter("pollingPoolSize", 10);
int pollingTimeout = url.getParameter("pollingTimeout", 10);
- LdsProtocol ldsProtocol = new LdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout);
- this.rdsProtocol = new RdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout);
- this.edsProtocol = new EdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout);
+ ApplicationModel applicationModel = ApplicationModel.defaultModel();
Review Comment:
```suggestion
ApplicationModel applicationModel = url.getorDefaultApplicationModel();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org