This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.2
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.2 by this push:
new cc6e1aea89 feat: fix lds observe (#11300)
cc6e1aea89 is described below
commit cc6e1aea8949206be571ccc8d7719702679a4c72
Author: aamingaa <[email protected]>
AuthorDate: Sun Jan 15 21:09:21 2023 +0800
feat: fix lds observe (#11300)
* feat: fix lds observe
* feat: remove unuse import
---
.../dubbo/registry/xds/util/PilotExchanger.java | 9 +++++----
.../xds/util/protocol/AbstractProtocol.java | 14 ++++----------
.../xds/util/protocol/impl/LdsProtocol.java | 21 +++++++--------------
3 files changed, 16 insertions(+), 28 deletions(-)
diff --git
a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java
b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java
index ce84aab722..567a1ca6dd 100644
---
a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java
+++
b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java
@@ -19,6 +19,7 @@ package org.apache.dubbo.registry.xds.util;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
+import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol;
import org.apache.dubbo.registry.xds.util.protocol.impl.EdsProtocol;
import org.apache.dubbo.registry.xds.util.protocol.impl.LdsProtocol;
import org.apache.dubbo.registry.xds.util.protocol.impl.RdsProtocol;
@@ -65,15 +66,15 @@ public class PilotExchanger {
this.listenerResult = ldsProtocol.getListeners();
this.routeResult =
rdsProtocol.getResource(listenerResult.values().iterator().next().getRouteConfigNames());
-
+ Set<String> ldsResourcesName = new HashSet<>();
+ ldsResourcesName.add(AbstractProtocol.emptyResourceName);
// Observer RDS update
if
(CollectionUtils.isNotEmpty(listenerResult.values().iterator().next().getRouteConfigNames()))
{
createRouteObserve();
isRdsObserve.set(true);
}
-
// Observe LDS updated
- ldsProtocol.observeListeners((newListener) -> {
+ ldsProtocol.observeResource(ldsResourcesName, (newListener) -> {
// update local cache
if (!newListener.equals(listenerResult)) {
this.listenerResult = newListener;
@@ -82,7 +83,7 @@ public class PilotExchanger {
createRouteObserve();
}
}
- });
+ }, false);
}
private void createRouteObserve() {
diff --git
a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
index 6fcd422635..34f7dfa560 100644
---
a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
+++
b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
@@ -24,6 +24,7 @@ import io.grpc.stub.StreamObserver;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
+import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.xds.util.XdsChannel;
import org.apache.dubbo.rpc.model.ApplicationModel;
@@ -126,7 +127,7 @@ public abstract class AbstractProtocol<T, S extends
DeltaResource<T>> implements
}
private Map<String, T> getResourceFromCache(Set<String> resourceNames) {
- return resourceNames.stream()
+ return resourceNames.stream().filter(o -> !StringUtils.isEmpty(o))
.collect(Collectors.toMap(k -> k, this::getCacheResource));
}
@@ -183,18 +184,11 @@ public abstract class AbstractProtocol<T, S extends
DeltaResource<T>> implements
}
public void observeResource(Set<String> resourceNames,
Consumer<Map<String, T>> consumer, boolean isReConnect) {
- resourceNames = resourceNames == null ? Collections.emptySet() :
resourceNames;
// call once for full data
if (!isReConnect) {
try {
writeLock.lock();
- Set<String> consumerObserveResourceNames = new HashSet<>();
- if (resourceNames.isEmpty()) {
- consumerObserveResourceNames.add(emptyResourceName);
- } else {
- consumerObserveResourceNames = resourceNames;
- }
- consumerObserveMap.compute(consumerObserveResourceNames, (k,
v) -> {
+ consumerObserveMap.compute(resourceNames, (k, v) -> {
if (v == null) {
v = new ArrayList<>();
}
@@ -202,10 +196,10 @@ public abstract class AbstractProtocol<T, S extends
DeltaResource<T>> implements
v.add(consumer);
return v;
});
- consumer.accept(getResource(resourceNames));
} finally {
writeLock.unlock();
}
+ consumer.accept(getResource(resourceNames));
}
try {
writeLock.lock();
diff --git
a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java
b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java
index e1b9ce6bb7..7129337959 100644
---
a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java
+++
b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java
@@ -16,14 +16,6 @@
*/
package org.apache.dubbo.registry.xds.util.protocol.impl;
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.registry.xds.util.XdsChannel;
-import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol;
-import org.apache.dubbo.registry.xds.util.protocol.delta.DeltaListener;
-import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-
import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
import io.envoyproxy.envoy.config.core.v3.Node;
@@ -32,14 +24,19 @@ import io.envoyproxy.envoy.config.listener.v3.Listener;
import
io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager;
import
io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.Rds;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
+import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.registry.xds.util.XdsChannel;
+import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol;
+import org.apache.dubbo.registry.xds.util.protocol.delta.DeltaListener;
+import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult;
+import org.apache.dubbo.rpc.model.ApplicationModel;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.function.Consumer;
import java.util.stream.Collectors;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_RESPONSE_XDS;
@@ -60,10 +57,6 @@ public class LdsProtocol extends
AbstractProtocol<ListenerResult, DeltaListener>
return getResource(null);
}
- public void observeListeners(Consumer<Map<String, ListenerResult>>
consumer) {
- observeResource(Collections.emptySet(), consumer, false);
- }
-
@Override
protected Map<String, ListenerResult>
decodeDiscoveryResponse(DiscoveryResponse response) {
if (getTypeUrl().equals(response.getTypeUrl())) {
