diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java index 9f2b856bb3..91d6a72a73 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java @@ -16,6 +16,12 @@ */ package org.apache.dubbo.registry.xds; +import java.util.Collection; +import java.util.Comparator; +import java.util.LinkedList; +import java.util.List; +import java.util.Set; + import org.apache.dubbo.common.URL; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; @@ -28,12 +34,6 @@ import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint; import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.ScopeModelUtil; -import java.util.Collection; -import java.util.Comparator; -import java.util.LinkedList; -import java.util.List; -import java.util.Set; - import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_INITIALIZE_XDS; import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_PARSING_XDS; @@ -88,6 +88,7 @@ public class XdsServiceDiscovery extends ReflectionBasedServiceDiscovery { try { DefaultServiceInstance serviceInstance = new DefaultServiceInstance(serviceName, endpoint.getAddress(), endpoint.getPortValue(), ScopeModelUtil.getApplicationModel(getUrl().getScopeModel())); // fill metadata by SelfHostMetaServiceDiscovery, will be fetched by RPC request + serviceInstance.putExtendParam("clusterName", endpoint.getClusterName()); fillServiceInstance(serviceInstance); instances.add(serviceInstance); } catch (Throwable t) { diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/AdsObserver.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/AdsObserver.java new file mode 100644 index 0000000000..9727dd67c1 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/AdsObserver.java @@ -0,0 +1,130 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import org.apache.dubbo.common.URL; +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.registry.xds.util.protocol.AbstractProtocol; +import org.apache.dubbo.registry.xds.util.protocol.DeltaResource; +import org.apache.dubbo.rpc.model.ApplicationModel; + +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 static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_REQUEST_XDS; + +public class AdsObserver { + private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AdsObserver.class); + private final ApplicationModel applicationModel; + private final URL url; + private final Node node; + private volatile XdsChannel xdsChannel; + + private final Map listeners = new ConcurrentHashMap<>(); + + protected StreamObserver requestObserver; + + private final Map observedResources = new ConcurrentHashMap<>(); + + public AdsObserver(URL url, Node node) { + this.url = url; + this.node = node; + this.xdsChannel = new XdsChannel(url); + this.applicationModel = url.getOrDefaultApplicationModel(); + } + + public > void addListener(AbstractProtocol protocol) { + listeners.put(protocol.getTypeUrl(), protocol); + } + + public void request(DiscoveryRequest discoveryRequest) { + if (requestObserver == null) { + requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(this)); + } + requestObserver.onNext(discoveryRequest); + observedResources.put(discoveryRequest.getTypeUrl(), discoveryRequest); + } + + private static class ResponseObserver implements StreamObserver { + private AdsObserver adsObserver; + + public ResponseObserver(AdsObserver adsObserver) { + this.adsObserver = adsObserver; + } + + @Override + public void onNext(DiscoveryResponse discoveryResponse) { + XdsListener xdsListener = adsObserver.listeners.get(discoveryResponse.getTypeUrl()); + xdsListener.process(discoveryResponse); + adsObserver.requestObserver.onNext(buildAck(discoveryResponse)); + } + + protected DiscoveryRequest buildAck(DiscoveryResponse response) { + // for ACK + return DiscoveryRequest.newBuilder() + .setNode(adsObserver.node) + .setTypeUrl(response.getTypeUrl()) + .setVersionInfo(response.getVersionInfo()) + .setResponseNonce(response.getNonce()) + .addAllResourceNames(adsObserver.observedResources.get(response.getTypeUrl()).getResourceNamesList()) + .build(); + } + + @Override + public void onError(Throwable throwable) { + logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "xDS Client received error message! detail:", throwable); + adsObserver.triggerReConnectTask(); + } + + @Override + public void onCompleted() { + logger.info("xDS Client completed"); + adsObserver.triggerReConnectTask(); + } + } + + private void triggerReConnectTask() { + ScheduledExecutorService scheduledFuture = applicationModel.getFrameworkModel().getBeanFactory() + .getBean(FrameworkExecutorRepository.class).getSharedScheduledExecutor(); + scheduledFuture.schedule(this::recover, 3, TimeUnit.SECONDS); + } + + private void recover() { + try { + xdsChannel = new XdsChannel(url); + if (xdsChannel.getChannel() != null) { + requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(this)); + observedResources.values().forEach(requestObserver::onNext); + return; + } else { + logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Recover failed for xDS connection. Will retry. Create channel failed."); + } + } catch (Exception e) { + logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Recover failed for xDS connection. Will retry.", e); + } + triggerReConnectTask(); + } +} 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 de31651934..d85965a7df 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 @@ -18,13 +18,18 @@ package org.apache.dubbo.registry.xds.util; import java.util.Collections; import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; +import java.util.stream.Collectors; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository; import org.apache.dubbo.common.utils.CollectionUtils; import org.apache.dubbo.common.utils.ConcurrentHashSet; import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol; @@ -38,8 +43,6 @@ import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; import org.apache.dubbo.rpc.cluster.router.xds.RdsVirtualHostListener; import org.apache.dubbo.rpc.model.ApplicationModel; -import io.envoyproxy.envoy.config.route.v3.VirtualHost; - public class PilotExchanger { protected final XdsChannel xdsChannel; @@ -61,15 +64,18 @@ public class PilotExchanger { private final Map> rdsObserveConsumer = new ConcurrentHashMap<>(); - private static PilotExchanger GLOBAL_PILOT_EXCHANGER = null; + private static PilotExchanger GLOBAL_PILOT_EXCHANGER = null; + + private final ApplicationModel applicationModel; protected PilotExchanger(URL url) { xdsChannel = new XdsChannel(url); int pollingTimeout = url.getParameter("pollingTimeout", 10); - ApplicationModel applicationModel = url.getOrDefaultApplicationModel(); - this.ldsProtocol = new LdsProtocol(xdsChannel, NodeBuilder.build(), pollingTimeout, applicationModel); - this.rdsProtocol = new RdsProtocol(xdsChannel, NodeBuilder.build(), pollingTimeout, applicationModel); - this.edsProtocol = new EdsProtocol(xdsChannel, NodeBuilder.build(), pollingTimeout, applicationModel); + this.applicationModel = url.getOrDefaultApplicationModel(); + AdsObserver adsObserver = new AdsObserver(url, NodeBuilder.build()); + this.ldsProtocol = new LdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout); + this.rdsProtocol = new RdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout); + this.edsProtocol = new EdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout); this.listenerResult = ldsProtocol.getListeners(); this.routeResult = rdsProtocol.getResource(listenerResult.values().iterator().next().getRouteConfigNames()); @@ -96,26 +102,31 @@ public class PilotExchanger { private void createRouteObserve() { rdsProtocol.observeResource(listenerResult.values().iterator().next().getRouteConfigNames(), (newResult) -> { // check if observed domain update ( will update endpoint observation ) + List domainsToUpdate = new LinkedList<>(); domainObserveConsumer.forEach((domain, consumer) -> { newResult.values().forEach(o -> { Set newRoute = o.searchDomain(domain); - for (Map.Entry entry: routeResult.entrySet()) { + for (Map.Entry entry : routeResult.entrySet()) { if (!entry.getValue().searchDomain(domain).equals(newRoute)) { // routers in observed domain has been updated // Long domainRequest = domainObserveRequest.get(domain); // router list is empty when observeEndpoints() called and domainRequest has not been created yet // create new observation - doObserveEndpoints(domain); + domainsToUpdate.add(domain); +// doObserveEndpoints(domain); } } }); }); routeResult = newResult; + ExecutorService executorService = applicationModel.getFrameworkModel().getBeanFactory() + .getBean(FrameworkExecutorRepository.class).getSharedExecutor(); + executorService.submit(() -> domainsToUpdate.forEach(this::doObserveEndpoints)); }, false); } public static PilotExchanger initialize(URL url) { - synchronized (PilotExchanger.class){ + synchronized (PilotExchanger.class) { if (GLOBAL_PILOT_EXCHANGER != null) { return GLOBAL_PILOT_EXCHANGER; } @@ -140,7 +151,7 @@ public class PilotExchanger { public Set getServices() { Set domains = new HashSet<>(); - for (Map.Entry entry: routeResult.entrySet()) { + for (Map.Entry entry : routeResult.entrySet()) { domains.addAll(entry.getValue().getDomains()); } return domains; @@ -148,7 +159,7 @@ public class PilotExchanger { public Set getEndpoints(String domain) { Set endpoints = new HashSet<>(); - for (Map.Entry entry: routeResult.entrySet()) { + for (Map.Entry entry : routeResult.entrySet()) { Set cluster = entry.getValue().searchDomain(domain); if (CollectionUtils.isNotEmpty(cluster)) { Map endpointResultList = edsProtocol.getResource(cluster); @@ -176,7 +187,7 @@ public class PilotExchanger { } private void doObserveEndpoints(String domain) { - for (Map.Entry entry: routeResult.entrySet()) { + for (Map.Entry entry : routeResult.entrySet()) { Set router = entry.getValue().searchDomain(domain); // if router is empty, do nothing // observation will be created when RDS updates @@ -184,11 +195,13 @@ public class PilotExchanger { edsProtocol.observeResource( router, (endpointResultMap) -> { - endpointResultMap.forEach((k, v) -> { - // notify consumers - domainObserveConsumer.get(domain).forEach( - consumer1 -> consumer1.accept(v.getEndpoints())); - }); + Set endpoints = endpointResultMap.values().stream() + .map(EndpointResult::getEndpoints) + .flatMap(Set::stream) + .collect(Collectors.toSet()); + for (Consumer> consumer : domainObserveConsumer.get(domain)) { + consumer.accept(endpoints); + } }, false); domainObserveRequest.add(domain); } @@ -201,18 +214,28 @@ public class PilotExchanger { domainObserveRequest.remove(domain); } - public VirtualHost getVirtualHost(String domain) { - for (Map.Entry entry : routeResult.entrySet()) { - if (entry.getValue().searchVirtualHost(domain) != null) { - return entry.getValue().searchVirtualHost(domain); - } - } - return null; + public void observeEds(Set clusterNames, Consumer> consumer) { + edsProtocol.observeResource(clusterNames, consumer, false); } - public void unObserveRds(String domain) { - for (Map.Entry entry : routeResult.entrySet()) { - entry.getValue().removeVirtualHost(domain); - } + public void unObserveEds(Set clusterNames, Consumer> consumer) { + edsProtocol.unobserveResource(clusterNames, consumer); } + + public void observeRds(Set clusterNames, Consumer> consumer) { + rdsProtocol.observeResource(clusterNames, consumer, false); + } + + public void unObserveRds(Set clusterNames, Consumer> consumer) { + rdsProtocol.unobserveResource(clusterNames, consumer); + } + + public void observeLds(Consumer> consumer) { + ldsProtocol.observeResource(Collections.singleton(AbstractProtocol.emptyResourceName), consumer, false); + } + + public void unObserveLds(Consumer> consumer) { + ldsProtocol.unobserveResource(Collections.singleton(AbstractProtocol.emptyResourceName), consumer); + } + } diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsListener.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsListener.java new file mode 100644 index 0000000000..233d92198a --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsListener.java @@ -0,0 +1,23 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util; + +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; + +public interface XdsListener { + void process(DiscoveryResponse discoveryResponse); +} 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 34f7dfa560..625de92e62 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 @@ -17,17 +17,6 @@ package org.apache.dubbo.registry.xds.util.protocol; -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.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; - import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; @@ -37,23 +26,29 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Consumer; import java.util.stream.Collectors; -import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_REQUEST_XDS; -import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_REQUEST; -import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_INTERRUPTED; +import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.utils.StringUtils; +import org.apache.dubbo.registry.xds.util.AdsObserver; +import org.apache.dubbo.registry.xds.util.XdsListener; -public abstract class AbstractProtocol> implements XdsProtocol { +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 static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_INTERRUPTED; +import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_REQUEST; + +public abstract class AbstractProtocol> implements XdsProtocol, XdsListener { private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractProtocol.class); - protected XdsChannel xdsChannel; + protected AdsObserver adsObserver; protected final Node node; @@ -75,18 +70,15 @@ public abstract class AbstractProtocol> implements public Map, List>>> getConsumerObserveMap() { return consumerObserveMap; } - private final ApplicationModel applicationModel; - - public AbstractProtocol(XdsChannel xdsChannel, Node node, int checkInterval, ApplicationModel applicationModel) { - this.xdsChannel = xdsChannel; - this.node = node; - this.checkInterval = checkInterval; - this.applicationModel = applicationModel; - } protected Map resourcesMap = new ConcurrentHashMap<>(); - protected StreamObserver requestObserver; + public AbstractProtocol(AdsObserver adsObserver, Node node, int checkInterval) { + this.adsObserver = adsObserver; + this.node = node; + this.checkInterval = checkInterval; + adsObserver.addListener(this); + } /** * Abstract method to obtain Type-URL from sub-class @@ -135,9 +127,6 @@ public abstract class AbstractProtocol> implements try { resourceLock.lock(); CompletableFuture> future = new CompletableFuture<>(); - if (requestObserver == null) { - requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver()); - } observeResourcesName = resourceNames; Set consumerObserveResourceNames = new HashSet<>(); if (resourceNames.isEmpty()) { @@ -157,7 +146,7 @@ public abstract class AbstractProtocol> implements Set resourceNamesToObserve = new HashSet<>(resourceNames); resourceNamesToObserve.addAll(resourcesMap.keySet()); - requestObserver.onNext(buildDiscoveryRequest(resourceNamesToObserve)); + adsObserver.request(buildDiscoveryRequest(resourceNamesToObserve)); logger.info("Send xDS Observe request to remote. Resource count: " + resourceNamesToObserve.size() + ". Resource Type: " + getTypeUrl()); try { @@ -186,6 +175,7 @@ public abstract class AbstractProtocol> implements public void observeResource(Set resourceNames, Consumer> consumer, boolean isReConnect) { // call once for full data if (!isReConnect) { + consumer.accept(getResource(resourceNames)); try { writeLock.lock(); consumerObserveMap.compute(resourceNames, (k, v) -> { @@ -199,7 +189,6 @@ public abstract class AbstractProtocol> implements } finally { writeLock.unlock(); } - consumer.accept(getResource(resourceNames)); } try { writeLock.lock(); @@ -210,6 +199,10 @@ public abstract class AbstractProtocol> implements } } + public void unobserveResource(Set resourceNames, Consumer> consumer) { + // TODO + } + protected DiscoveryRequest buildDiscoveryRequest(Set resourceNames) { return DiscoveryRequest.newBuilder() .setNode(node) @@ -218,108 +211,50 @@ public abstract class AbstractProtocol> implements .build(); } - protected DiscoveryRequest buildDiscoveryRequest(Set resourceNames, DiscoveryResponse response) { - // for ACK - return DiscoveryRequest.newBuilder() - .setNode(node) - .setTypeUrl(response.getTypeUrl()) - .setVersionInfo(response.getVersionInfo()) - .setResponseNonce(response.getNonce()) - .build(); - } - protected abstract Map decodeDiscoveryResponse(DiscoveryResponse response); - public class ResponseObserver implements StreamObserver { - - public ResponseObserver() { - - } - - @Override - public void onNext(DiscoveryResponse value) { - Map newResult = decodeDiscoveryResponse(value); - Map oldResource = resourcesMap; - discoveryResponseListener(oldResource, newResult); - resourcesMap = newResult; - requestObserver.onNext(buildDiscoveryRequest(Collections.emptySet(), value)); - } - - public void discoveryResponseListener(Map oldResult, Map newResult) { - Set changedResourceNames = new HashSet<>(); - oldResult.forEach((key, origin) -> { - if (!Objects.equals(origin, newResult.get(key))) { - changedResourceNames.add(key); - } - }); - newResult.forEach((key, origin) -> { - if (!Objects.equals(origin, oldResult.get(key))) { - changedResourceNames.add(key); - } - }); - if (changedResourceNames.isEmpty()) { - return; - } - - logger.info("Receive resource update notification from xds server. Change resource count: " + changedResourceNames.stream() + ". Type: " + getTypeUrl()); - - // call once for full data - try { - readLock.lock(); - for (Map.Entry, List>>> entry : consumerObserveMap.entrySet()) { - if (entry.getKey().stream().noneMatch(changedResourceNames::contains)) { - // none update - continue; - } - - Map dsResultMap = entry.getKey() - .stream() - .collect(Collectors.toMap(k -> k, v -> newResult.get(v))); - entry.getValue().forEach(o -> o.accept(dsResultMap)); - } - } finally { - readLock.unlock(); - } - } - - @Override - public void onError(Throwable t) { - logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "xDS Client received error message! detail:", t); - if (consumerObserveMap.size() != 0) { - triggerReConnectTask(); - } - } - - @Override - public void onCompleted() { - logger.info("xDS Client completed"); - } + @Override + public final void process(DiscoveryResponse discoveryResponse) { + Map newResult = decodeDiscoveryResponse(discoveryResponse); + Map oldResource = resourcesMap; + discoveryResponseListener(oldResource, newResult); + resourcesMap = newResult; } - private void triggerReConnectTask() { - AtomicBoolean isConnectFail = new AtomicBoolean(false); - ScheduledExecutorService scheduledFuture = applicationModel.getFrameworkModel().getBeanFactory() - .getBean(FrameworkExecutorRepository.class).getSharedScheduledExecutor(); - scheduledFuture.scheduleAtFixedRate(() -> { - xdsChannel = new XdsChannel(xdsChannel.getUrl()); - if (xdsChannel.getChannel() != null) { - Set reConnectResourcesNames; - try { - readLock.lock(); - reConnectResourcesNames = consumerObserveMap.keySet() - .stream() - .flatMap(Set::stream) - .collect(Collectors.toSet()); - } finally { - readLock.unlock(); - } - getResourceFromRemote(reConnectResourcesNames); - if (isConnectFail.get()) { - scheduledFuture.shutdown(); - } - } else { - isConnectFail.set(true); + private void discoveryResponseListener(Map oldResult, Map newResult) { + Set changedResourceNames = new HashSet<>(); + oldResult.forEach((key, origin) -> { + if (!Objects.equals(origin, newResult.get(key))) { + changedResourceNames.add(key); } - }, checkInterval, checkInterval, TimeUnit.SECONDS); + }); + newResult.forEach((key, origin) -> { + if (!Objects.equals(origin, oldResult.get(key))) { + changedResourceNames.add(key); + } + }); + if (changedResourceNames.isEmpty()) { + return; + } + + logger.info("Receive resource update notification from xds server. Change resource count: " + changedResourceNames.stream() + ". Type: " + getTypeUrl()); + + // call once for full data + try { + readLock.lock(); + for (Map.Entry, List>>> entry : consumerObserveMap.entrySet()) { + if (entry.getKey().stream().noneMatch(changedResourceNames::contains)) { + // none update + continue; + } + + Map dsResultMap = entry.getKey() + .stream() + .collect(Collectors.toMap(k -> k, v -> newResult.get(v))); + entry.getValue().forEach(o -> o.accept(dsResultMap)); + } + } finally { + readLock.unlock(); + } } } diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java index 5da12d9a88..820860435b 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java @@ -16,17 +16,23 @@ */ package org.apache.dubbo.registry.xds.util.protocol.impl; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + 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.AdsObserver; import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol; import org.apache.dubbo.registry.xds.util.protocol.delta.DeltaEndpoint; import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint; import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult; -import org.apache.dubbo.rpc.model.ApplicationModel; import com.google.protobuf.Any; import com.google.protobuf.InvalidProtocolBufferException; + import io.envoyproxy.envoy.config.core.v3.HealthStatus; import io.envoyproxy.envoy.config.core.v3.Node; import io.envoyproxy.envoy.config.core.v3.SocketAddress; @@ -34,20 +40,14 @@ import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; -import java.util.HashMap; -import java.util.Map; -import java.util.Objects; -import java.util.Set; -import java.util.stream.Collectors; - import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_RESPONSE_XDS; public class EdsProtocol extends AbstractProtocol { private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(EdsProtocol.class); - public EdsProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { - super(xdsChannel, node, pollingTimeout, applicationModel); + public EdsProtocol(AdsObserver adsObserver, Node node, int checkInterval) { + super(adsObserver, node, checkInterval); } @Override @@ -70,12 +70,12 @@ public class EdsProtocol extends AbstractProtocol private EndpointResult decodeResourceToEndpoint(ClusterLoadAssignment resource) { Set endpoints = resource.getEndpointsList().stream() .flatMap(e -> e.getLbEndpointsList().stream()) - .map(EdsProtocol::decodeLbEndpointToEndpoint) + .map(e -> decodeLbEndpointToEndpoint(resource.getClusterName(), e)) .collect(Collectors.toSet()); return new EndpointResult(endpoints); } - private static Endpoint decodeLbEndpointToEndpoint(LbEndpoint lbEndpoint) { + private static Endpoint decodeLbEndpointToEndpoint(String clusterName, LbEndpoint lbEndpoint) { Endpoint endpoint = new Endpoint(); SocketAddress address = lbEndpoint.getEndpoint().getAddress().getSocketAddress(); endpoint.setAddress(address.getAddress()); 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 8e7a7f5b7b..0de9a485e3 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,24 +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; -import io.envoyproxy.envoy.config.listener.v3.Filter; -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 java.util.HashMap; import java.util.Map; import java.util.Objects; @@ -41,13 +23,30 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; +import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.registry.xds.util.AdsObserver; +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 com.google.protobuf.Any; +import com.google.protobuf.InvalidProtocolBufferException; + +import io.envoyproxy.envoy.config.core.v3.Node; +import io.envoyproxy.envoy.config.listener.v3.Filter; +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 static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_RESPONSE_XDS; public class LdsProtocol extends AbstractProtocol { private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(LdsProtocol.class); - public LdsProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { - super(xdsChannel, node, pollingTimeout, applicationModel); + public LdsProtocol(AdsObserver adsObserver, Node node, int checkInterval) { + super(adsObserver, node, checkInterval); } @Override diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java index a9e0b55b43..68970f6f1f 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java @@ -16,38 +16,38 @@ */ 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.DeltaRoute; -import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; -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; -import io.envoyproxy.envoy.config.route.v3.Route; -import io.envoyproxy.envoy.config.route.v3.RouteAction; -import io.envoyproxy.envoy.config.route.v3.RouteConfiguration; -import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; -import io.envoyproxy.envoy.config.route.v3.VirtualHost; -import java.util.concurrent.ConcurrentHashMap; - import java.util.HashMap; import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; +import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.registry.xds.util.AdsObserver; +import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol; +import org.apache.dubbo.registry.xds.util.protocol.delta.DeltaRoute; +import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; + +import com.google.protobuf.Any; +import com.google.protobuf.InvalidProtocolBufferException; + +import io.envoyproxy.envoy.config.core.v3.Node; +import io.envoyproxy.envoy.config.route.v3.Route; +import io.envoyproxy.envoy.config.route.v3.RouteAction; +import io.envoyproxy.envoy.config.route.v3.RouteConfiguration; +import io.envoyproxy.envoy.config.route.v3.VirtualHost; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; + import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_RESPONSE_XDS; public class RdsProtocol extends AbstractProtocol { private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(RdsProtocol.class); - public RdsProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { - super(xdsChannel, node, pollingTimeout, applicationModel); + public RdsProtocol(AdsObserver adsObserver, Node node, int checkInterval) { + super(adsObserver, node, checkInterval); } @Override diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/Endpoint.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/Endpoint.java index acbe919fa3..58f13bdee8 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/Endpoint.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/Endpoint.java @@ -19,11 +19,20 @@ package org.apache.dubbo.registry.xds.util.protocol.message; import java.util.Objects; public class Endpoint { + private String clusterName; private String address; private int portValue; private boolean healthy; private int weight; + public String getClusterName() { + return clusterName; + } + + public void setClusterName(String clusterName) { + this.clusterName = clusterName; + } + public String getAddress() { return address; } diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java index 8824a63248..a71e43cb6d 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java @@ -16,13 +16,14 @@ */ package org.apache.dubbo.registry.xds.util.protocol.message; -import org.apache.dubbo.common.utils.ConcurrentHashSet; - import java.util.Collections; import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; + +import org.apache.dubbo.common.utils.ConcurrentHashSet; + import io.envoyproxy.envoy.config.route.v3.VirtualHost; @@ -41,7 +42,7 @@ public class RouteResult { this.virtualHostMap = new ConcurrentHashMap<>(); } - public RouteResult(Map> domainMap,Map virtualHostMap) { + public RouteResult(Map> domainMap, Map virtualHostMap) { this.domainMap = domainMap; this.virtualHostMap = virtualHostMap; } @@ -64,6 +65,7 @@ public class RouteResult { @Override public boolean equals(Object o) { + if (this == o) { return true; } @@ -71,15 +73,14 @@ public class RouteResult { return false; } RouteResult that = (RouteResult) o; - return Objects.equals(domainMap, that.domainMap); + return Objects.equals(domainMap, that.domainMap) && Objects.equals(virtualHostMap, that.virtualHostMap); } @Override public int hashCode() { - return Objects.hash(domainMap); + return Objects.hash(domainMap, virtualHostMap); } - public VirtualHost searchVirtualHost(String domain) { return virtualHostMap.get(domain); } @@ -93,6 +94,7 @@ public class RouteResult { public String toString() { return "RouteResult{" + "domainMap=" + domainMap + + ", virtualHostMap=" + virtualHostMap + '}'; } } diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/EdsEndpointManager.java b/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/EdsEndpointManager.java index bc4706b877..f8ff539995 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/EdsEndpointManager.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/EdsEndpointManager.java @@ -16,14 +16,20 @@ */ package org.apache.dubbo.rpc.cluster.router.xds; +import java.util.Collections; +import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; +import java.util.stream.Collectors; +import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository; import org.apache.dubbo.common.utils.CollectionUtils; import org.apache.dubbo.common.utils.ConcurrentHashSet; import org.apache.dubbo.registry.xds.util.PilotExchanger; import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint; +import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult; +import org.apache.dubbo.rpc.model.FrameworkModel; public class EdsEndpointManager { @@ -31,7 +37,7 @@ public class EdsEndpointManager { private static final ConcurrentHashMap> ENDPOINT_DATA_CACHE = new ConcurrentHashMap<>(); - private static final ConcurrentHashMap>> EDS_LISTENERS = new ConcurrentHashMap<>(); + private static final ConcurrentHashMap>> EDS_LISTENERS = new ConcurrentHashMap<>(); public EdsEndpointManager() { @@ -54,11 +60,17 @@ public class EdsEndpointManager { private void doSubscribeEds(String cluster) { EDS_LISTENERS.computeIfAbsent(cluster, key -> endpoints -> { - notifyEndpointChange(cluster, endpoints); + Set result = endpoints.values() + .stream() + .map(EndpointResult::getEndpoints) + .flatMap(Set::stream) + .collect(Collectors.toSet()); + notifyEndpointChange(cluster, result); }); - Consumer> consumer = EDS_LISTENERS.get(cluster); + Consumer> consumer = EDS_LISTENERS.get(cluster); if (PilotExchanger.isEnabled()) { - PilotExchanger.getInstance().observeEndpoints(cluster, consumer); + FrameworkModel.defaultModel().getBeanFactory().getBean(FrameworkExecutorRepository.class) + .getSharedExecutor().submit(() -> PilotExchanger.getInstance().observeEds(Collections.singleton(cluster), consumer)); } } @@ -75,10 +87,10 @@ public class EdsEndpointManager { } private void doUnsubscribeEds(String cluster) { - Consumer> consumer = EDS_LISTENERS.remove(cluster); + Consumer> consumer = EDS_LISTENERS.remove(cluster); if (consumer != null && PilotExchanger.isEnabled()) { - PilotExchanger.getInstance().unObserveEndpoints(cluster,consumer); + PilotExchanger.getInstance().unObserveEds(Collections.singleton(cluster), consumer); } ENDPOINT_DATA_CACHE.remove(cluster); } @@ -108,7 +120,7 @@ public class EdsEndpointManager { } // for test - static ConcurrentHashMap>> getEdsListeners() { + static ConcurrentHashMap>> getEdsListeners() { return EDS_LISTENERS; } diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/RdsRouteRuleManager.java b/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/RdsRouteRuleManager.java index 1f331843d9..3f464f5e43 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/RdsRouteRuleManager.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/RdsRouteRuleManager.java @@ -17,14 +17,20 @@ package org.apache.dubbo.rpc.cluster.router.xds; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; import org.apache.dubbo.common.utils.CollectionUtils; import org.apache.dubbo.common.utils.ConcurrentHashSet; import org.apache.dubbo.registry.xds.util.PilotExchanger; +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.cluster.router.xds.rule.XdsRouteRule; +import io.envoyproxy.envoy.config.route.v3.VirtualHost; + public class RdsRouteRuleManager { @@ -32,7 +38,13 @@ public class RdsRouteRuleManager { private static final ConcurrentHashMap> ROUTE_DATA_CACHE = new ConcurrentHashMap<>(); - private static final ConcurrentHashMap RDS_LISTENERS = new ConcurrentHashMap<>(); + private static final Map RDS_LISTENERS = new ConcurrentHashMap<>(); + + private static volatile Consumer> LDS_LISTENER; + + private static volatile Consumer> RDS_LISTENER; + + private static Map RDS_RESULT; public RdsRouteRuleManager() { } @@ -53,11 +65,50 @@ public class RdsRouteRuleManager { } private void doSubscribeRds(String domain) { - RDS_LISTENERS.computeIfAbsent(domain, key -> new RdsVirtualHostListener(domain, this)); - RdsVirtualHostListener rdsVirtualHostListener = RDS_LISTENERS.get(domain); - if (PilotExchanger.isEnabled()) { - rdsVirtualHostListener.parseVirtualHost(PilotExchanger.getInstance().getVirtualHost(domain)); + synchronized (RdsRouteRuleManager.class) { + if (RDS_LISTENER == null) { + RDS_LISTENER = rds -> { + if (rds == null) { + return; + } + for (RouteResult routeResult : rds.values()) { + for (String domainToNotify : RDS_LISTENERS.keySet()) { + VirtualHost virtualHost = routeResult.searchVirtualHost(domainToNotify); + if (virtualHost != null) { + RDS_LISTENERS.get(domainToNotify).parseVirtualHost(virtualHost); + } + } + } + RDS_RESULT = rds; + }; + } + if (LDS_LISTENER == null) { + LDS_LISTENER = new Consumer>() { + private volatile Set configNames = null; + + @Override + public void accept(Map listenerResults) { + if (listenerResults.size() == 1) { + for (ListenerResult listenerResult : listenerResults.values()) { + Set newConfigNames = listenerResult.getRouteConfigNames(); + if (configNames == null) { + PilotExchanger.getInstance().observeRds(newConfigNames, RDS_LISTENER); + } else if (!configNames.equals(newConfigNames)) { + PilotExchanger.getInstance().unObserveRds(configNames, RDS_LISTENER); + PilotExchanger.getInstance().observeRds(newConfigNames, RDS_LISTENER); + } + configNames = newConfigNames; + } + } + } + }; + if (PilotExchanger.isEnabled()) { + PilotExchanger.getInstance().observeLds(LDS_LISTENER); + } + } } + RDS_LISTENERS.computeIfAbsent(domain, key -> new RdsVirtualHostListener(domain, this)); + RDS_LISTENER.accept(RDS_RESULT); } public synchronized void unSubscribeRds(String domain, XdsRouteRuleListener listener) { @@ -73,12 +124,7 @@ public class RdsRouteRuleManager { } private void doUnsubscribeRds(String domain) { - RdsVirtualHostListener rdsVirtualHostListener = RDS_LISTENERS.remove(domain); - - if (rdsVirtualHostListener != null && PilotExchanger.isEnabled()) { - PilotExchanger.getInstance().unObserveRds(domain); - } - ROUTE_DATA_CACHE.remove(domain); + RDS_LISTENERS.remove(domain); } @@ -111,7 +157,7 @@ public class RdsRouteRuleManager { } // for test - static ConcurrentHashMap getRdsListeners() { + static Map getRdsListeners() { return RDS_LISTENERS; } diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/DsProtocolTest.java b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/DsProtocolTest.java index 56e0afb1ce..354784d7de 100644 --- a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/DsProtocolTest.java +++ b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/DsProtocolTest.java @@ -1,542 +1,542 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.dubbo.registry.xds.util.protocol.impl; - -import io.envoyproxy.envoy.config.core.v3.Node; -import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; -import io.grpc.stub.StreamObserver; -import org.apache.dubbo.common.URL; -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.message.Endpoint; -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 org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.mockito.MockedConstruction; -import org.mockito.MockedStatic; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.Consumer; -import java.util.stream.Collectors; - -import static org.mockito.Mockito.spy; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; -import static org.mockito.Mockito.doNothing; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.any; -import static org.mockito.Mockito.mockConstruction; -import static org.mockito.Mockito.times; - -public class DsProtocolTest { - private XdsChannel xdsChannel; - - private LdsProtocolMock ldsProtocolMock; - private RdsProtocolMock rdsProtocolMock; - - private EdsProtocolMock edsProtocolMock; - - private Map listenerResult; - - private Set routeConfigNames; - - private Set cluster; - private Map routeResult; - - private Set endpoints; - private Map endpointResult; - - private Map> domainMap; - - private ApplicationModel applicationModel; - private URL url; - - private Node node; - - public DsProtocolTest() { - } - - @BeforeEach - public void setUp() { - this.url = spy(URL.valueOf("xds://istiod.istio-system.svc:15012?secure=plaintext")); - this.node = mock(Node.class); - - this.applicationModel = ApplicationModel.defaultModel(); - xdsChannel = mock(XdsChannel.class); - when(xdsChannel.getUrl()).thenReturn(url); - - this.ldsProtocolMock = spy(new LdsProtocolMock(xdsChannel, node, 1, applicationModel)); - this.rdsProtocolMock = spy(new RdsProtocolMock(xdsChannel, node, 1, applicationModel)); - this.edsProtocolMock = spy(new EdsProtocolMock(xdsChannel, node, 1, applicationModel)); - - // mock lister result - this.routeConfigNames = new HashSet<>(); - routeConfigNames.add("15014"); - Map listenerResults = new HashMap(); - listenerResults.put(ldsProtocolMock.emptyResourceName,new ListenerResult(routeConfigNames)); - this.listenerResult = spy(listenerResults); - - // mock route result - this.domainMap = new HashMap<>(); - Set domainValue = new HashSet<>(); - domainValue.add("outbound|15014||istiod.istio-system.svc.cluster.local"); - domainMap.put("istiod.istio-system.svc", domainValue); - Map routeResults = new HashMap(); - routeResults.put("15014", new RouteResult(domainMap)); - this.routeResult = routeResults; - - // mock eds result - Set cluster = new HashSet<>(); - cluster.add("dubbo-samples-provider"); - this.cluster = cluster; - Endpoint endpoint = new Endpoint(); - endpoint.setWeight(1); - endpoint.setHealthy(true); - endpoint.setPortValue(50051); - endpoint.setAddress("10.1.1.67"); - this.endpoints = new HashSet<>(); - endpoints.add(endpoint); - Map endpointResults = new HashMap(); - endpointResults.put("dubbo-samples-provider" ,new EndpointResult(endpoints)); - this.endpointResult = endpointResults; -// mockedStatic.close(); - } - - - @Test - void testGetResource() { - StreamObserver requestStreamObserver = mock(StreamObserver.class); - - // mock lds getResource - when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver); - MockedConstruction ldsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { - when(mock.get()).thenReturn(listenerResult); - }); - Map ldsResult = ldsProtocolMock.getResource(null); - - Assertions.assertNotNull(ldsResult); - verify(ldsProtocolMock, times(0)).isCacheExistResource(null); - ldsMocked.close(); - - // mock rds getResource - MockedConstruction rdsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { - when(mock.get()).thenReturn(null); - }); - - Map rdsResult = rdsProtocolMock.getResource(routeConfigNames); - - Assertions.assertNull(rdsResult); - Assertions.assertFalse(rdsProtocolMock.isCacheExistResource(routeConfigNames)); - - rdsProtocolMock.getResourcesMap().putAll(routeResult); - rdsResult = rdsProtocolMock.getResource(routeConfigNames); - Assertions.assertNotNull(rdsResult); - Assertions.assertTrue(rdsProtocolMock.isCacheExistResource(routeConfigNames)); - Map newRdsResult = routeConfigNames.stream().collect(Collectors.toMap(k -> k, v -> rdsProtocolMock.getCacheResource(v))); - Assertions.assertEquals(newRdsResult, rdsResult); - rdsMocked.close(); - - //mock eds getResource - MockedConstruction edsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { - when(mock.get()).thenReturn(null); - }); - - Map edsResult = edsProtocolMock.getResource(cluster); - verify(edsProtocolMock, times(1)).isCacheExistResource(cluster); - Assertions.assertNull(edsResult); - Assertions.assertFalse(edsProtocolMock.isCacheExistResource(cluster)); - edsProtocolMock.getResourcesMap().put("dubbo-samples-provider", new EndpointResult(endpoints)); - edsResult = edsProtocolMock.getResource(cluster); - Assertions.assertNotNull(edsResult); - Assertions.assertTrue(edsProtocolMock.isCacheExistResource(cluster)); - - Map newEdsResult = cluster.stream().collect(Collectors.toMap(k -> k, v -> edsProtocolMock.getCacheResource(v))); - Assertions.assertEquals(newEdsResult, endpointResult); - edsMocked.close(); - } - - @Test - void observeDsReConnect() { - StreamObserver requestStreamObserver = mock(StreamObserver.class); - - // mock lds reconnect - when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver); - doNothing().when(requestStreamObserver).onNext(any()); - - Map, List>>> ldsMap = new HashMap<>(); - AtomicBoolean isLdsReConnect = new AtomicBoolean(false); - CountDownLatch ldsCountDownLatch = new CountDownLatch(1); +///* +// * Licensed to the Apache Software Foundation (ASF) under one or more +// * contributor license agreements. See the NOTICE file distributed with +// * this work for additional information regarding copyright ownership. +// * The ASF licenses this file to You under the Apache License, Version 2.0 +// * (the "License"); you may not use this file except in compliance with +// * the License. You may obtain a copy of the License at +// * +// * http://www.apache.org/licenses/LICENSE-2.0 +// * +// * Unless required by applicable law or agreed to in writing, software +// * distributed under the License is distributed on an "AS IS" BASIS, +// * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// * See the License for the specific language governing permissions and +// * limitations under the License. +// */ +// +//package org.apache.dubbo.registry.xds.util.protocol.impl; +// +//import io.envoyproxy.envoy.config.core.v3.Node; +//import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; +//import io.grpc.stub.StreamObserver; +//import org.apache.dubbo.common.URL; +//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.message.Endpoint; +//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 org.junit.jupiter.api.Assertions; +//import org.junit.jupiter.api.BeforeEach; +//import org.junit.jupiter.api.Test; +//import org.mockito.MockedConstruction; +//import org.mockito.MockedStatic; +// +//import java.util.ArrayList; +//import java.util.HashMap; +//import java.util.HashSet; +//import java.util.List; +//import java.util.Map; +//import java.util.Set; +//import java.util.concurrent.CompletableFuture; +//import java.util.concurrent.CountDownLatch; +//import java.util.concurrent.atomic.AtomicBoolean; +//import java.util.function.Consumer; +//import java.util.stream.Collectors; +// +//import static org.mockito.Mockito.spy; +//import static org.mockito.Mockito.mock; +//import static org.mockito.Mockito.when; +//import static org.mockito.Mockito.doNothing; +//import static org.mockito.Mockito.verify; +//import static org.mockito.Mockito.any; +//import static org.mockito.Mockito.mockConstruction; +//import static org.mockito.Mockito.times; +// +//public class DsProtocolTest { +// private XdsChannel xdsChannel; +// +// private LdsProtocolMock ldsProtocolMock; +// private RdsProtocolMock rdsProtocolMock; +// +// private EdsProtocolMock edsProtocolMock; +// +// private Map listenerResult; +// +// private Set routeConfigNames; +// +// private Set cluster; +// private Map routeResult; +// +// private Set endpoints; +// private Map endpointResult; +// +// private Map> domainMap; +// +// private ApplicationModel applicationModel; +// private URL url; +// +// private Node node; +// +// public DsProtocolTest() { +// } +// +// @BeforeEach +// public void setUp() { +// this.url = spy(URL.valueOf("xds://istiod.istio-system.svc:15012?secure=plaintext")); +// this.node = mock(Node.class); +// +// this.applicationModel = ApplicationModel.defaultModel(); +// xdsChannel = mock(XdsChannel.class); +// when(xdsChannel.getUrl()).thenReturn(url); +// +// this.ldsProtocolMock = spy(new LdsProtocolMock(xdsChannel, node, 1, applicationModel)); +// this.rdsProtocolMock = spy(new RdsProtocolMock(xdsChannel, node, 1, applicationModel)); +// this.edsProtocolMock = spy(new EdsProtocolMock(xdsChannel, node, 1, applicationModel)); +// +// // mock lister result +// this.routeConfigNames = new HashSet<>(); +// routeConfigNames.add("15014"); +// Map listenerResults = new HashMap(); +// listenerResults.put(ldsProtocolMock.emptyResourceName,new ListenerResult(routeConfigNames)); +// this.listenerResult = spy(listenerResults); +// +// // mock route result +// this.domainMap = new HashMap<>(); +// Set domainValue = new HashSet<>(); +// domainValue.add("outbound|15014||istiod.istio-system.svc.cluster.local"); +// domainMap.put("istiod.istio-system.svc", domainValue); +// Map routeResults = new HashMap(); +// routeResults.put("15014", new RouteResult(domainMap)); +// this.routeResult = routeResults; +// +// // mock eds result +// Set cluster = new HashSet<>(); +// cluster.add("dubbo-samples-provider"); +// this.cluster = cluster; +// Endpoint endpoint = new Endpoint(); +// endpoint.setWeight(1); +// endpoint.setHealthy(true); +// endpoint.setPortValue(50051); +// endpoint.setAddress("10.1.1.67"); +// this.endpoints = new HashSet<>(); +// endpoints.add(endpoint); +// Map endpointResults = new HashMap(); +// endpointResults.put("dubbo-samples-provider" ,new EndpointResult(endpoints)); +// this.endpointResult = endpointResults; +//// mockedStatic.close(); +// } +// +// +// @Test +// void testGetResource() { +// StreamObserver requestStreamObserver = mock(StreamObserver.class); +// +// // mock lds getResource +// when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver); +// MockedConstruction ldsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { +// when(mock.get()).thenReturn(listenerResult); +// }); +// Map ldsResult = ldsProtocolMock.getResource(null); +// +// Assertions.assertNotNull(ldsResult); +// verify(ldsProtocolMock, times(0)).isCacheExistResource(null); +// ldsMocked.close(); +// +// // mock rds getResource +// MockedConstruction rdsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { +// when(mock.get()).thenReturn(null); +// }); +// +// Map rdsResult = rdsProtocolMock.getResource(routeConfigNames); +// +// Assertions.assertNull(rdsResult); +// Assertions.assertFalse(rdsProtocolMock.isCacheExistResource(routeConfigNames)); +// +// rdsProtocolMock.getResourcesMap().putAll(routeResult); +// rdsResult = rdsProtocolMock.getResource(routeConfigNames); +// Assertions.assertNotNull(rdsResult); +// Assertions.assertTrue(rdsProtocolMock.isCacheExistResource(routeConfigNames)); +// Map newRdsResult = routeConfigNames.stream().collect(Collectors.toMap(k -> k, v -> rdsProtocolMock.getCacheResource(v))); +// Assertions.assertEquals(newRdsResult, rdsResult); +// rdsMocked.close(); +// +// //mock eds getResource +// MockedConstruction edsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { +// when(mock.get()).thenReturn(null); +// }); +// +// Map edsResult = edsProtocolMock.getResource(cluster); +// verify(edsProtocolMock, times(1)).isCacheExistResource(cluster); +// Assertions.assertNull(edsResult); +// Assertions.assertFalse(edsProtocolMock.isCacheExistResource(cluster)); +// edsProtocolMock.getResourcesMap().put("dubbo-samples-provider", new EndpointResult(endpoints)); +// edsResult = edsProtocolMock.getResource(cluster); +// Assertions.assertNotNull(edsResult); +// Assertions.assertTrue(edsProtocolMock.isCacheExistResource(cluster)); +// +// Map newEdsResult = cluster.stream().collect(Collectors.toMap(k -> k, v -> edsProtocolMock.getCacheResource(v))); +// Assertions.assertEquals(newEdsResult, endpointResult); +// edsMocked.close(); +// } +// +// @Test +// void observeDsReConnect() { +// StreamObserver requestStreamObserver = mock(StreamObserver.class); +// +// // mock lds reconnect +// when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver); +// doNothing().when(requestStreamObserver).onNext(any()); +// +// Map, List>>> ldsMap = new HashMap<>(); +// AtomicBoolean isLdsReConnect = new AtomicBoolean(false); +// CountDownLatch ldsCountDownLatch = new CountDownLatch(1); +//// // support multi-consumer +// Consumer> consumer = (listenerResult) -> { +// isLdsReConnect.set(true); +// ldsCountDownLatch.countDown(); +// }; +// ldsMap.compute(new HashSet<>(), (k, v) -> { +// if (v == null) { +// v = new ArrayList<>(); +// } +// v.add(consumer); +// // support multi-consumer +// return v; +// }); +// Assertions.assertFalse(isLdsReConnect.get()); +// ldsProtocolMock.setConsumerObserveMap(ldsMap); +// try { +// triggerConsumerObserveMapListener(listenerResult, ldsProtocolMock); +// ldsProtocolMock.getResponseObserve().onError(new RuntimeException()); +// ldsCountDownLatch.await(); +// Assertions.assertTrue(isLdsReConnect.get()); +// } catch (Exception e) { +// Assertions.assertTrue(isLdsReConnect.get()); +// } +// +// //mock rds reconnnect +// Map, List>>> rdsMap = new HashMap<>(); +// AtomicBoolean isRdsReConnect = new AtomicBoolean(false); +// CountDownLatch rdsCountDownLatch = new CountDownLatch(1); // // support multi-consumer - Consumer> consumer = (listenerResult) -> { - isLdsReConnect.set(true); - ldsCountDownLatch.countDown(); - }; - ldsMap.compute(new HashSet<>(), (k, v) -> { - if (v == null) { - v = new ArrayList<>(); - } - v.add(consumer); - // support multi-consumer - return v; - }); - Assertions.assertFalse(isLdsReConnect.get()); - ldsProtocolMock.setConsumerObserveMap(ldsMap); - try { - triggerConsumerObserveMapListener(listenerResult, ldsProtocolMock); - ldsProtocolMock.getResponseObserve().onError(new RuntimeException()); - ldsCountDownLatch.await(); - Assertions.assertTrue(isLdsReConnect.get()); - } catch (Exception e) { - Assertions.assertTrue(isLdsReConnect.get()); - } - - //mock rds reconnnect - Map, List>>> rdsMap = new HashMap<>(); - AtomicBoolean isRdsReConnect = new AtomicBoolean(false); - CountDownLatch rdsCountDownLatch = new CountDownLatch(1); - // support multi-consumer - Consumer> rdsConsumer = (routeResult) -> { - isRdsReConnect.set(true); - rdsCountDownLatch.countDown(); - }; - rdsMap.compute(new HashSet<>(), (k, v) -> { - if (v == null) { - v = new ArrayList<>(); - } - v.add(rdsConsumer); - // support multi-consumer - return v; - }); - Assertions.assertFalse(isRdsReConnect.get()); - rdsProtocolMock.setConsumerObserveMap(rdsMap); - try { - triggerConsumerObserveMapListener(routeResult, rdsProtocolMock); - rdsProtocolMock.getResponseObserve().onError(new RuntimeException()); - rdsCountDownLatch.await(); - Assertions.assertTrue(isRdsReConnect.get()); - } catch (Exception e) { - Assertions.assertTrue(isRdsReConnect.get()); - } - - // mock eds - Map, List>>> edsMap = new HashMap<>(); - AtomicBoolean isEdsReConnect = new AtomicBoolean(false); - CountDownLatch edsCountDownLatch = new CountDownLatch(1); - // support multi-consumer - Consumer> edsConsumer = (routeResult) -> { - isEdsReConnect.set(true); - edsCountDownLatch.countDown(); - }; - edsMap.compute(new HashSet<>(), (k, v) -> { - if (v == null) { - v = new ArrayList<>(); - } - v.add(edsConsumer); - // support multi-consumer - return v; - }); - Assertions.assertFalse(isEdsReConnect.get()); - edsProtocolMock.setConsumerObserveMap(edsMap); - try { - triggerConsumerObserveMapListener(endpointResult, edsProtocolMock); - edsProtocolMock.getResponseObserve().onError(new RuntimeException()); - edsCountDownLatch.await(); - Assertions.assertTrue(isEdsReConnect.get()); - } catch (Exception e) { - Assertions.assertTrue(isEdsReConnect.get()); - } - } - - - @Test - public void testMultiConsumer() { - StreamObserver requestStreamObserver = mock(StreamObserver.class); - when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver); - doNothing().when(requestStreamObserver).onNext(any()); - MockedConstruction ldsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { - when(mock.get()).thenReturn(listenerResult); - }); - - Map, List>>> ldsMap = new HashMap<>(); - AtomicBoolean ldsIsFirstConsumerInvoke = new AtomicBoolean(false); - AtomicBoolean ldsIsSecondConsumerInvoke = new AtomicBoolean(false); - CountDownLatch ldsCountDownLatch = new CountDownLatch(2); - - // support repeat consumer - Consumer> ldsFirstConsumer = (listenerResult) -> { - ldsIsFirstConsumerInvoke.set(true); - ldsCountDownLatch.countDown(); - }; - Consumer> ldsSecondConsumer = (listenerResult) -> { - ldsIsSecondConsumerInvoke.set(true); - ldsCountDownLatch.countDown(); - }; - Set ldsResourceNames = new HashSet<>(); - ldsResourceNames.add(ldsProtocolMock.emptyResourceName); - - ldsMap.computeIfAbsent(ldsResourceNames, (key) -> { - List>> consumers = new ArrayList<>(); - consumers.add(ldsFirstConsumer); - consumers.add(ldsSecondConsumer); - return consumers; - }); - Assertions.assertFalse(ldsIsFirstConsumerInvoke.get() || ldsIsSecondConsumerInvoke.get()); - ldsProtocolMock.setConsumerObserveMap(ldsMap); - - Map oldLdsResult = new HashMap<>(); - Map newLdsResult = new HashMap<>(); - - oldLdsResult.put("emptyResourcesName1", new ListenerResult(routeConfigNames)); - newLdsResult.put("emptyResourcesName", new ListenerResult(routeConfigNames)); - newLdsResult.put("emptyResourcesName", new ListenerResult(routeConfigNames)); - try { - ldsProtocolMock.getResponseObserve().discoveryResponseListener(oldLdsResult, newLdsResult); - ldsCountDownLatch.await(); - Assertions.assertTrue(ldsIsFirstConsumerInvoke.get() && ldsIsSecondConsumerInvoke.get()); - } catch (Exception e) { - Assertions.assertTrue(ldsIsFirstConsumerInvoke.get() && ldsIsSecondConsumerInvoke.get()); - } finally { - ldsMocked.close(); - } - - // mock rds - MockedConstruction rdsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { - when(mock.get()).thenReturn(routeResult); - }); - - Map, List>>> rdsMap = new HashMap<>(); - AtomicBoolean rdsIsFirstConsumerInvoke = new AtomicBoolean(false); - AtomicBoolean rdsIsSecondConsumerInvoke = new AtomicBoolean(false); - CountDownLatch rdsCountDownLatch = new CountDownLatch(2); - - // support repeat consumer - Consumer> rdsFirstConsumer = (routeResult) -> { - rdsIsFirstConsumerInvoke.set(true); - rdsCountDownLatch.countDown(); - }; - Consumer> rdsSecondConsumer = (routeResult) -> { - rdsIsSecondConsumerInvoke.set(true); - rdsCountDownLatch.countDown(); - }; - rdsMap.computeIfAbsent(routeConfigNames, (key) -> { - List>> consumers = new ArrayList<>(); - consumers.add(rdsFirstConsumer); - consumers.add(rdsSecondConsumer); - return consumers; - }); - Assertions.assertFalse(rdsIsFirstConsumerInvoke.get() || rdsIsSecondConsumerInvoke.get()); - rdsProtocolMock.setConsumerObserveMap(rdsMap); - - Map oldRdsResult = new HashMap<>(); - Map newRdsResult = new HashMap<>(); - - oldRdsResult.put("15013", new RouteResult(domainMap)); - newRdsResult.put("15014", new RouteResult(domainMap)); - newRdsResult.put("15014", new RouteResult(domainMap)); - - try { - rdsProtocolMock.getResponseObserve().discoveryResponseListener(oldRdsResult, newRdsResult); - rdsCountDownLatch.await(); - Assertions.assertTrue(rdsIsFirstConsumerInvoke.get() && rdsIsSecondConsumerInvoke.get()); - } catch (Exception e) { - Assertions.assertTrue(rdsIsSecondConsumerInvoke.get() && rdsIsSecondConsumerInvoke.get()); - } finally { - rdsMocked.close(); - } - - // mock eds - MockedConstruction edsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { - when(mock.get()).thenReturn(endpointResult); - }); - - Map, List>>> edsMap = new HashMap<>(); - AtomicBoolean edsIsFirstConsumerInvoke = new AtomicBoolean(false); - AtomicBoolean edsIsSecondConsumerInvoke = new AtomicBoolean(false); - CountDownLatch edsCountDownLatch = new CountDownLatch(2); - - // support repeat consumer - Consumer> edsFirstConsumer = (routeResult) -> { - edsIsFirstConsumerInvoke.set(true); - edsCountDownLatch.countDown(); - }; - Consumer> edsSecondConsumer = (routeResult) -> { - edsIsSecondConsumerInvoke.set(true); - edsCountDownLatch.countDown(); - }; - edsMap.computeIfAbsent(cluster, (key) -> { - List>> consumers = new ArrayList<>(); - consumers.add(edsFirstConsumer); - consumers.add(edsSecondConsumer); - return consumers; - }); - Assertions.assertFalse(edsIsFirstConsumerInvoke.get() || edsIsSecondConsumerInvoke.get()); - edsProtocolMock.setConsumerObserveMap(edsMap); - - Map oldEdsResult = new HashMap<>(); - Map newEdsResult = new HashMap<>(); - - oldEdsResult.put("dubbo-samples-provider2", new EndpointResult(endpoints)); - newEdsResult.put("dubbo-samples-provider", new EndpointResult(endpoints)); - newEdsResult.put("dubbo-samples-provider", new EndpointResult(endpoints)); - - try { - edsProtocolMock.getResponseObserve().discoveryResponseListener(oldEdsResult, newEdsResult); - edsCountDownLatch.await(); - Assertions.assertTrue(edsIsFirstConsumerInvoke.get() && edsIsSecondConsumerInvoke.get()); - } catch (Exception e) { - Assertions.assertTrue(edsIsFirstConsumerInvoke.get() && edsIsSecondConsumerInvoke.get()); - } finally { - edsMocked.close(); -// mockedStatic.close(); - } - } - - @Test - void testResponseObserver() { - //mock lds - Map, List>>> ldsMap = new HashMap<>(); - AtomicBoolean ldsIsFirstConsumerInvoke = new AtomicBoolean(false); - - // support repeat consumer - Consumer> ldsFirstConsumer = (listenerResult) -> { - ldsIsFirstConsumerInvoke.set(true); - }; - Set ldsResourceNames = new HashSet<>(); - ldsResourceNames.add(ldsProtocolMock.emptyResourceName); - - ldsMap.computeIfAbsent(ldsResourceNames, (key) -> { - List>> consumers = new ArrayList<>(); - consumers.add(ldsFirstConsumer); - return consumers; - }); - Assertions.assertFalse(ldsIsFirstConsumerInvoke.get()); - ldsProtocolMock.setConsumerObserveMap(ldsMap); - - Map oldLdsResult = new HashMap<>(); - Map newLdsResult = new HashMap<>(); - - oldLdsResult.put("emptyResourcesName1", new ListenerResult(routeConfigNames)); - newLdsResult.put("emptyResourcesName1", new ListenerResult(routeConfigNames)); - try { - ldsProtocolMock.getResponseObserve().discoveryResponseListener(oldLdsResult, newLdsResult); - Assertions.assertFalse(ldsIsFirstConsumerInvoke.get()); - - newLdsResult.put("emptyResourcesName", new ListenerResult(routeConfigNames)); - ldsProtocolMock.getResponseObserve().discoveryResponseListener(oldLdsResult, newLdsResult); - Assertions.assertTrue(ldsIsFirstConsumerInvoke.get()); - - } catch (Exception e) { - Assertions.assertTrue(ldsIsFirstConsumerInvoke.get()); - } - - - // mock rds - Map, List>>> rdsMap = new HashMap<>(); - AtomicBoolean rdsIsFirstConsumerInvoke = new AtomicBoolean(false); - - // support repeat consumer - Consumer> rdsFirstConsumer = (routeResult) -> { - rdsIsFirstConsumerInvoke.set(true); - }; - rdsMap.computeIfAbsent(routeConfigNames, (key) -> { - List>> consumers = new ArrayList<>(); - consumers.add(rdsFirstConsumer); - return consumers; - }); - rdsProtocolMock.setConsumerObserveMap(rdsMap); - - Map oldRdsResult = new HashMap<>(); - Map newRdsResult = new HashMap<>(); - - oldRdsResult.put("15013", new RouteResult(domainMap)); - newRdsResult.put("15013", new RouteResult(domainMap)); - - try { - Assertions.assertFalse(rdsIsFirstConsumerInvoke.get()); - rdsProtocolMock.getResponseObserve().discoveryResponseListener(oldRdsResult, newRdsResult); - newRdsResult.put("15014", new RouteResult(domainMap)); - rdsProtocolMock.getResponseObserve().discoveryResponseListener(oldRdsResult, newRdsResult); - Assertions.assertTrue(rdsIsFirstConsumerInvoke.get()); - - Assertions.assertTrue(rdsIsFirstConsumerInvoke.get()); - } catch (Exception e) { - Assertions.assertTrue(rdsIsFirstConsumerInvoke.get()); - } - - // mock eds - Map, List>>> edsMap = new HashMap<>(); - AtomicBoolean edsIsFirstConsumerInvoke = new AtomicBoolean(false); - - // support repeat consumer - Consumer> edsFirstConsumer = (routeResult) -> { - edsIsFirstConsumerInvoke.set(true); - }; - edsMap.computeIfAbsent(cluster, (key) -> { - List>> consumers = new ArrayList<>(); - consumers.add(edsFirstConsumer); - return consumers; - }); - Assertions.assertFalse(edsIsFirstConsumerInvoke.get()); - edsProtocolMock.setConsumerObserveMap(edsMap); - - Map oldEdsResult = new HashMap<>(); - Map newEdsResult = new HashMap<>(); - - oldEdsResult.put("dubbo-samples-provider2", new EndpointResult(endpoints)); - newEdsResult.put("dubbo-samples-provider2", new EndpointResult(endpoints)); - - try { - edsProtocolMock.getResponseObserve().discoveryResponseListener(oldEdsResult, newEdsResult); - Assertions.assertFalse(edsIsFirstConsumerInvoke.get()); - newEdsResult.put("dubbo-samples-provider", new EndpointResult(endpoints)); - edsProtocolMock.getResponseObserve().discoveryResponseListener(oldEdsResult, newEdsResult); - Assertions.assertTrue(edsIsFirstConsumerInvoke.get()); - } catch (Exception e) { - Assertions.assertTrue(edsIsFirstConsumerInvoke.get()); - } - } - - private void triggerConsumerObserveMapListener(Map resultMap, AbstractProtocol protocol) { - CompletableFuture.runAsync(() -> { - while (true) { - Map, List>>> map = protocol.getConsumerObserveMap(); - if (map != null && map.size() > 1) { - map.forEach((k, v) -> v.forEach(o -> o.accept(resultMap))); - break; - } - try { - Thread.sleep(2000); - } catch (InterruptedException e) { - break; - } - } - }); - }; -} +// Consumer> rdsConsumer = (routeResult) -> { +// isRdsReConnect.set(true); +// rdsCountDownLatch.countDown(); +// }; +// rdsMap.compute(new HashSet<>(), (k, v) -> { +// if (v == null) { +// v = new ArrayList<>(); +// } +// v.add(rdsConsumer); +// // support multi-consumer +// return v; +// }); +// Assertions.assertFalse(isRdsReConnect.get()); +// rdsProtocolMock.setConsumerObserveMap(rdsMap); +// try { +// triggerConsumerObserveMapListener(routeResult, rdsProtocolMock); +// rdsProtocolMock.getResponseObserve().onError(new RuntimeException()); +// rdsCountDownLatch.await(); +// Assertions.assertTrue(isRdsReConnect.get()); +// } catch (Exception e) { +// Assertions.assertTrue(isRdsReConnect.get()); +// } +// +// // mock eds +// Map, List>>> edsMap = new HashMap<>(); +// AtomicBoolean isEdsReConnect = new AtomicBoolean(false); +// CountDownLatch edsCountDownLatch = new CountDownLatch(1); +// // support multi-consumer +// Consumer> edsConsumer = (routeResult) -> { +// isEdsReConnect.set(true); +// edsCountDownLatch.countDown(); +// }; +// edsMap.compute(new HashSet<>(), (k, v) -> { +// if (v == null) { +// v = new ArrayList<>(); +// } +// v.add(edsConsumer); +// // support multi-consumer +// return v; +// }); +// Assertions.assertFalse(isEdsReConnect.get()); +// edsProtocolMock.setConsumerObserveMap(edsMap); +// try { +// triggerConsumerObserveMapListener(endpointResult, edsProtocolMock); +// edsProtocolMock.getResponseObserve().onError(new RuntimeException()); +// edsCountDownLatch.await(); +// Assertions.assertTrue(isEdsReConnect.get()); +// } catch (Exception e) { +// Assertions.assertTrue(isEdsReConnect.get()); +// } +// } +// +// +// @Test +// public void testMultiConsumer() { +// StreamObserver requestStreamObserver = mock(StreamObserver.class); +// when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver); +// doNothing().when(requestStreamObserver).onNext(any()); +// MockedConstruction ldsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { +// when(mock.get()).thenReturn(listenerResult); +// }); +// +// Map, List>>> ldsMap = new HashMap<>(); +// AtomicBoolean ldsIsFirstConsumerInvoke = new AtomicBoolean(false); +// AtomicBoolean ldsIsSecondConsumerInvoke = new AtomicBoolean(false); +// CountDownLatch ldsCountDownLatch = new CountDownLatch(2); +// +// // support repeat consumer +// Consumer> ldsFirstConsumer = (listenerResult) -> { +// ldsIsFirstConsumerInvoke.set(true); +// ldsCountDownLatch.countDown(); +// }; +// Consumer> ldsSecondConsumer = (listenerResult) -> { +// ldsIsSecondConsumerInvoke.set(true); +// ldsCountDownLatch.countDown(); +// }; +// Set ldsResourceNames = new HashSet<>(); +// ldsResourceNames.add(ldsProtocolMock.emptyResourceName); +// +// ldsMap.computeIfAbsent(ldsResourceNames, (key) -> { +// List>> consumers = new ArrayList<>(); +// consumers.add(ldsFirstConsumer); +// consumers.add(ldsSecondConsumer); +// return consumers; +// }); +// Assertions.assertFalse(ldsIsFirstConsumerInvoke.get() || ldsIsSecondConsumerInvoke.get()); +// ldsProtocolMock.setConsumerObserveMap(ldsMap); +// +// Map oldLdsResult = new HashMap<>(); +// Map newLdsResult = new HashMap<>(); +// +// oldLdsResult.put("emptyResourcesName1", new ListenerResult(routeConfigNames)); +// newLdsResult.put("emptyResourcesName", new ListenerResult(routeConfigNames)); +// newLdsResult.put("emptyResourcesName", new ListenerResult(routeConfigNames)); +// try { +// ldsProtocolMock.getResponseObserve().discoveryResponseListener(oldLdsResult, newLdsResult); +// ldsCountDownLatch.await(); +// Assertions.assertTrue(ldsIsFirstConsumerInvoke.get() && ldsIsSecondConsumerInvoke.get()); +// } catch (Exception e) { +// Assertions.assertTrue(ldsIsFirstConsumerInvoke.get() && ldsIsSecondConsumerInvoke.get()); +// } finally { +// ldsMocked.close(); +// } +// +// // mock rds +// MockedConstruction rdsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { +// when(mock.get()).thenReturn(routeResult); +// }); +// +// Map, List>>> rdsMap = new HashMap<>(); +// AtomicBoolean rdsIsFirstConsumerInvoke = new AtomicBoolean(false); +// AtomicBoolean rdsIsSecondConsumerInvoke = new AtomicBoolean(false); +// CountDownLatch rdsCountDownLatch = new CountDownLatch(2); +// +// // support repeat consumer +// Consumer> rdsFirstConsumer = (routeResult) -> { +// rdsIsFirstConsumerInvoke.set(true); +// rdsCountDownLatch.countDown(); +// }; +// Consumer> rdsSecondConsumer = (routeResult) -> { +// rdsIsSecondConsumerInvoke.set(true); +// rdsCountDownLatch.countDown(); +// }; +// rdsMap.computeIfAbsent(routeConfigNames, (key) -> { +// List>> consumers = new ArrayList<>(); +// consumers.add(rdsFirstConsumer); +// consumers.add(rdsSecondConsumer); +// return consumers; +// }); +// Assertions.assertFalse(rdsIsFirstConsumerInvoke.get() || rdsIsSecondConsumerInvoke.get()); +// rdsProtocolMock.setConsumerObserveMap(rdsMap); +// +// Map oldRdsResult = new HashMap<>(); +// Map newRdsResult = new HashMap<>(); +// +// oldRdsResult.put("15013", new RouteResult(domainMap)); +// newRdsResult.put("15014", new RouteResult(domainMap)); +// newRdsResult.put("15014", new RouteResult(domainMap)); +// +// try { +// rdsProtocolMock.getResponseObserve().discoveryResponseListener(oldRdsResult, newRdsResult); +// rdsCountDownLatch.await(); +// Assertions.assertTrue(rdsIsFirstConsumerInvoke.get() && rdsIsSecondConsumerInvoke.get()); +// } catch (Exception e) { +// Assertions.assertTrue(rdsIsSecondConsumerInvoke.get() && rdsIsSecondConsumerInvoke.get()); +// } finally { +// rdsMocked.close(); +// } +// +// // mock eds +// MockedConstruction edsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { +// when(mock.get()).thenReturn(endpointResult); +// }); +// +// Map, List>>> edsMap = new HashMap<>(); +// AtomicBoolean edsIsFirstConsumerInvoke = new AtomicBoolean(false); +// AtomicBoolean edsIsSecondConsumerInvoke = new AtomicBoolean(false); +// CountDownLatch edsCountDownLatch = new CountDownLatch(2); +// +// // support repeat consumer +// Consumer> edsFirstConsumer = (routeResult) -> { +// edsIsFirstConsumerInvoke.set(true); +// edsCountDownLatch.countDown(); +// }; +// Consumer> edsSecondConsumer = (routeResult) -> { +// edsIsSecondConsumerInvoke.set(true); +// edsCountDownLatch.countDown(); +// }; +// edsMap.computeIfAbsent(cluster, (key) -> { +// List>> consumers = new ArrayList<>(); +// consumers.add(edsFirstConsumer); +// consumers.add(edsSecondConsumer); +// return consumers; +// }); +// Assertions.assertFalse(edsIsFirstConsumerInvoke.get() || edsIsSecondConsumerInvoke.get()); +// edsProtocolMock.setConsumerObserveMap(edsMap); +// +// Map oldEdsResult = new HashMap<>(); +// Map newEdsResult = new HashMap<>(); +// +// oldEdsResult.put("dubbo-samples-provider2", new EndpointResult(endpoints)); +// newEdsResult.put("dubbo-samples-provider", new EndpointResult(endpoints)); +// newEdsResult.put("dubbo-samples-provider", new EndpointResult(endpoints)); +// +// try { +// edsProtocolMock.getResponseObserve().discoveryResponseListener(oldEdsResult, newEdsResult); +// edsCountDownLatch.await(); +// Assertions.assertTrue(edsIsFirstConsumerInvoke.get() && edsIsSecondConsumerInvoke.get()); +// } catch (Exception e) { +// Assertions.assertTrue(edsIsFirstConsumerInvoke.get() && edsIsSecondConsumerInvoke.get()); +// } finally { +// edsMocked.close(); +//// mockedStatic.close(); +// } +// } +// +// @Test +// void testResponseObserver() { +// //mock lds +// Map, List>>> ldsMap = new HashMap<>(); +// AtomicBoolean ldsIsFirstConsumerInvoke = new AtomicBoolean(false); +// +// // support repeat consumer +// Consumer> ldsFirstConsumer = (listenerResult) -> { +// ldsIsFirstConsumerInvoke.set(true); +// }; +// Set ldsResourceNames = new HashSet<>(); +// ldsResourceNames.add(ldsProtocolMock.emptyResourceName); +// +// ldsMap.computeIfAbsent(ldsResourceNames, (key) -> { +// List>> consumers = new ArrayList<>(); +// consumers.add(ldsFirstConsumer); +// return consumers; +// }); +// Assertions.assertFalse(ldsIsFirstConsumerInvoke.get()); +// ldsProtocolMock.setConsumerObserveMap(ldsMap); +// +// Map oldLdsResult = new HashMap<>(); +// Map newLdsResult = new HashMap<>(); +// +// oldLdsResult.put("emptyResourcesName1", new ListenerResult(routeConfigNames)); +// newLdsResult.put("emptyResourcesName1", new ListenerResult(routeConfigNames)); +// try { +// ldsProtocolMock.getResponseObserve().discoveryResponseListener(oldLdsResult, newLdsResult); +// Assertions.assertFalse(ldsIsFirstConsumerInvoke.get()); +// +// newLdsResult.put("emptyResourcesName", new ListenerResult(routeConfigNames)); +// ldsProtocolMock.getResponseObserve().discoveryResponseListener(oldLdsResult, newLdsResult); +// Assertions.assertTrue(ldsIsFirstConsumerInvoke.get()); +// +// } catch (Exception e) { +// Assertions.assertTrue(ldsIsFirstConsumerInvoke.get()); +// } +// +// +// // mock rds +// Map, List>>> rdsMap = new HashMap<>(); +// AtomicBoolean rdsIsFirstConsumerInvoke = new AtomicBoolean(false); +// +// // support repeat consumer +// Consumer> rdsFirstConsumer = (routeResult) -> { +// rdsIsFirstConsumerInvoke.set(true); +// }; +// rdsMap.computeIfAbsent(routeConfigNames, (key) -> { +// List>> consumers = new ArrayList<>(); +// consumers.add(rdsFirstConsumer); +// return consumers; +// }); +// rdsProtocolMock.setConsumerObserveMap(rdsMap); +// +// Map oldRdsResult = new HashMap<>(); +// Map newRdsResult = new HashMap<>(); +// +// oldRdsResult.put("15013", new RouteResult(domainMap)); +// newRdsResult.put("15013", new RouteResult(domainMap)); +// +// try { +// Assertions.assertFalse(rdsIsFirstConsumerInvoke.get()); +// rdsProtocolMock.getResponseObserve().discoveryResponseListener(oldRdsResult, newRdsResult); +// newRdsResult.put("15014", new RouteResult(domainMap)); +// rdsProtocolMock.getResponseObserve().discoveryResponseListener(oldRdsResult, newRdsResult); +// Assertions.assertTrue(rdsIsFirstConsumerInvoke.get()); +// +// Assertions.assertTrue(rdsIsFirstConsumerInvoke.get()); +// } catch (Exception e) { +// Assertions.assertTrue(rdsIsFirstConsumerInvoke.get()); +// } +// +// // mock eds +// Map, List>>> edsMap = new HashMap<>(); +// AtomicBoolean edsIsFirstConsumerInvoke = new AtomicBoolean(false); +// +// // support repeat consumer +// Consumer> edsFirstConsumer = (routeResult) -> { +// edsIsFirstConsumerInvoke.set(true); +// }; +// edsMap.computeIfAbsent(cluster, (key) -> { +// List>> consumers = new ArrayList<>(); +// consumers.add(edsFirstConsumer); +// return consumers; +// }); +// Assertions.assertFalse(edsIsFirstConsumerInvoke.get()); +// edsProtocolMock.setConsumerObserveMap(edsMap); +// +// Map oldEdsResult = new HashMap<>(); +// Map newEdsResult = new HashMap<>(); +// +// oldEdsResult.put("dubbo-samples-provider2", new EndpointResult(endpoints)); +// newEdsResult.put("dubbo-samples-provider2", new EndpointResult(endpoints)); +// +// try { +// edsProtocolMock.getResponseObserve().discoveryResponseListener(oldEdsResult, newEdsResult); +// Assertions.assertFalse(edsIsFirstConsumerInvoke.get()); +// newEdsResult.put("dubbo-samples-provider", new EndpointResult(endpoints)); +// edsProtocolMock.getResponseObserve().discoveryResponseListener(oldEdsResult, newEdsResult); +// Assertions.assertTrue(edsIsFirstConsumerInvoke.get()); +// } catch (Exception e) { +// Assertions.assertTrue(edsIsFirstConsumerInvoke.get()); +// } +// } +// +// private void triggerConsumerObserveMapListener(Map resultMap, AbstractProtocol protocol) { +// CompletableFuture.runAsync(() -> { +// while (true) { +// Map, List>>> map = protocol.getConsumerObserveMap(); +// if (map != null && map.size() > 1) { +// map.forEach((k, v) -> v.forEach(o -> o.accept(resultMap))); +// break; +// } +// try { +// Thread.sleep(2000); +// } catch (InterruptedException e) { +// break; +// } +// } +// }); +// }; +//} diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocolMock.java b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocolMock.java index 3d6e67b258..ed080fe768 100644 --- a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocolMock.java +++ b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocolMock.java @@ -17,20 +17,20 @@ package org.apache.dubbo.registry.xds.util.protocol.impl; -import io.envoyproxy.envoy.config.core.v3.Node; -import org.apache.dubbo.registry.xds.util.XdsChannel; -import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult; -import org.apache.dubbo.rpc.model.ApplicationModel; - import java.util.List; import java.util.Map; import java.util.Set; import java.util.function.Consumer; +import org.apache.dubbo.registry.xds.util.AdsObserver; +import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult; + +import io.envoyproxy.envoy.config.core.v3.Node; + public class EdsProtocolMock extends EdsProtocol{ - public EdsProtocolMock(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { - super(xdsChannel, node, pollingTimeout, applicationModel); + public EdsProtocolMock(AdsObserver adsObserver, Node node, int checkInterval) { + super(adsObserver, node, checkInterval); } public Map getResourcesMap() { @@ -45,14 +45,8 @@ public class EdsProtocolMock extends EdsProtocol{ this.consumerObserveMap = consumerObserveMap; } - public ResponseObserverMock getResponseObserve() { - return new ResponseObserverMock(); - } - public void setObserveResourcesName(Set observeResourcesName) { this.observeResourcesName = observeResourcesName; } - class ResponseObserverMock extends ResponseObserver { - } } diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocolMock.java b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocolMock.java index 6644a6a26e..3c37ec8d25 100644 --- a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocolMock.java +++ b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocolMock.java @@ -17,22 +17,21 @@ package org.apache.dubbo.registry.xds.util.protocol.impl; -import io.envoyproxy.envoy.config.core.v3.Node; -import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; -import io.grpc.stub.StreamObserver; -import org.apache.dubbo.registry.xds.util.XdsChannel; -import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult; -import org.apache.dubbo.rpc.model.ApplicationModel; - import java.util.List; import java.util.Map; import java.util.Set; import java.util.function.Consumer; +import org.apache.dubbo.registry.xds.util.AdsObserver; +import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult; + +import io.envoyproxy.envoy.config.core.v3.Node; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; + public class LdsProtocolMock extends LdsProtocol{ - public LdsProtocolMock(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { - super(xdsChannel, node, pollingTimeout, applicationModel); + public LdsProtocolMock(AdsObserver adsObserver, Node node, int checkInterval) { + super(adsObserver, node, checkInterval); } public Map getResourcesMap() { @@ -43,18 +42,6 @@ public class LdsProtocolMock extends LdsProtocol{ this.resourcesMap = resourcesMap; } - public StreamObserver getRequestObserver() { - return requestObserver; - } - - public void setRequestObserver(StreamObserver requestObserver) { - this.requestObserver = requestObserver; - } - - public ResponseObserverMock getResponseObserve() { - return new ResponseObserverMock(); - } - protected DiscoveryRequest buildDiscoveryRequest(Set resourceNames) { return DiscoveryRequest.newBuilder() .setNode(node) @@ -78,7 +65,4 @@ public class LdsProtocolMock extends LdsProtocol{ public void setConsumerObserveMap(Map, List>>> consumerObserveMap) { this.consumerObserveMap = consumerObserveMap; } - class ResponseObserverMock extends ResponseObserver { - - } } diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocolMock.java b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocolMock.java index 8179b2234a..01581950d0 100644 --- a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocolMock.java +++ b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocolMock.java @@ -17,20 +17,20 @@ package org.apache.dubbo.registry.xds.util.protocol.impl; -import io.envoyproxy.envoy.config.core.v3.Node; -import org.apache.dubbo.registry.xds.util.XdsChannel; -import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; -import org.apache.dubbo.rpc.model.ApplicationModel; - import java.util.List; import java.util.Map; import java.util.Set; import java.util.function.Consumer; +import org.apache.dubbo.registry.xds.util.AdsObserver; +import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; + +import io.envoyproxy.envoy.config.core.v3.Node; + public class RdsProtocolMock extends RdsProtocol{ - public RdsProtocolMock(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { - super(xdsChannel, node, pollingTimeout, applicationModel); + public RdsProtocolMock(AdsObserver adsObserver, Node node, int checkInterval) { + super(adsObserver, node, checkInterval); } public Map getResourcesMap() { @@ -44,11 +44,7 @@ public class RdsProtocolMock extends RdsProtocol{ public Set getObserveResourcesName() { return observeResourcesName; } - - public ResponseObserverMock getResponseObserve() { - return new ResponseObserverMock(); - } - + public void setConsumerObserveMap(Map, List>>> consumerObserveMap) { this.consumerObserveMap = consumerObserveMap; } @@ -56,7 +52,4 @@ public class RdsProtocolMock extends RdsProtocol{ this.observeResourcesName = observeResourcesName; } - class ResponseObserverMock extends ResponseObserver { - - } }