natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: fix lds observe (#11300) 

* feat: fix lds observe

* feat: remove unuse import
This commit is contained in:
aamingaa 2023-01-15 21:09:21 +08:00 committed by GitHub
parent af55fc87e8
commit cc6e1aea89
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 16 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java
View File

@ -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() {

14
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
View File

@ -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();

21
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java
View File

@ -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())) {