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

Make xds router run (#11319)

This commit is contained in:
Albumen Kevin 2023-01-18 10:47:14 +08:00 committed by GitHub
parent fd3fb63826
commit 0c1b7987c6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 987 additions and 836 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java
View File

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

130
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/AdsObserver.java Normal file
View File

@ -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<String, XdsListener> listeners = new ConcurrentHashMap<>();
protected StreamObserver<DiscoveryRequest> requestObserver;
private final Map<String, DiscoveryRequest> 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 <T, S extends DeltaResource<T>> void addListener(AbstractProtocol<T, S> 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<DiscoveryResponse> {
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();
}
}

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

@ -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<String, Consumer<RdsVirtualHostListener>> 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<String> domainsToUpdate = new LinkedList<>();
domainObserveConsumer.forEach((domain, consumer) -> {
newResult.values().forEach(o -> {
Set<String> newRoute = o.searchDomain(domain);
for (Map.Entry<String, RouteResult> entry: routeResult.entrySet()) {
for (Map.Entry<String, RouteResult> 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<String> getServices() {
Set<String> domains = new HashSet<>();
for (Map.Entry<String, RouteResult> entry: routeResult.entrySet()) {
for (Map.Entry<String, RouteResult> entry : routeResult.entrySet()) {
domains.addAll(entry.getValue().getDomains());
}
return domains;
@ -148,7 +159,7 @@ public class PilotExchanger {
public Set<Endpoint> getEndpoints(String domain) {
Set<Endpoint> endpoints = new HashSet<>();
for (Map.Entry<String, RouteResult> entry: routeResult.entrySet()) {
for (Map.Entry<String, RouteResult> entry : routeResult.entrySet()) {
Set<String> cluster = entry.getValue().searchDomain(domain);
if (CollectionUtils.isNotEmpty(cluster)) {
Map<String, EndpointResult> endpointResultList = edsProtocol.getResource(cluster);
@ -176,7 +187,7 @@ public class PilotExchanger {
}
private void doObserveEndpoints(String domain) {
for (Map.Entry<String, RouteResult> entry: routeResult.entrySet()) {
for (Map.Entry<String, RouteResult> entry : routeResult.entrySet()) {
Set<String> 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<Endpoint> endpoints = endpointResultMap.values().stream()
.map(EndpointResult::getEndpoints)
.flatMap(Set::stream)
.collect(Collectors.toSet());
for (Consumer<Set<Endpoint>> 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<String, RouteResult> entry : routeResult.entrySet()) {
if (entry.getValue().searchVirtualHost(domain) != null) {
return entry.getValue().searchVirtualHost(domain);
}
}
return null;
public void observeEds(Set<String> clusterNames, Consumer<Map<String, EndpointResult>> consumer) {
edsProtocol.observeResource(clusterNames, consumer, false);
}
public void unObserveRds(String domain) {
for (Map.Entry<String, RouteResult> entry : routeResult.entrySet()) {
entry.getValue().removeVirtualHost(domain);
}
public void unObserveEds(Set<String> clusterNames, Consumer<Map<String, EndpointResult>> consumer) {
edsProtocol.unobserveResource(clusterNames, consumer);
}
public void observeRds(Set<String> clusterNames, Consumer<Map<String, RouteResult>> consumer) {
rdsProtocol.observeResource(clusterNames, consumer, false);
}
public void unObserveRds(Set<String> clusterNames, Consumer<Map<String, RouteResult>> consumer) {
rdsProtocol.unobserveResource(clusterNames, consumer);
}
public void observeLds(Consumer<Map<String, ListenerResult>> consumer) {
ldsProtocol.observeResource(Collections.singleton(AbstractProtocol.emptyResourceName), consumer, false);
}
public void unObserveLds(Consumer<Map<String, ListenerResult>> consumer) {
ldsProtocol.unobserveResource(Collections.singleton(AbstractProtocol.emptyResourceName), consumer);
}
}

23
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsListener.java Normal file
View File

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

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

@ -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<T, S extends DeltaResource<T>> implements XdsProtocol<T> {
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<T, S extends DeltaResource<T>> implements XdsProtocol<T>, 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<T, S extends DeltaResource<T>> implements
public Map<Set<String>, List<Consumer<Map<String, T>>>> 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<String, T> resourcesMap = new ConcurrentHashMap<>();
protected StreamObserver<DiscoveryRequest> 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<T, S extends DeltaResource<T>> implements
try {
resourceLock.lock();
CompletableFuture<Map<String, T>> future = new CompletableFuture<>();
if (requestObserver == null) {
requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver());
}
observeResourcesName = resourceNames;
Set<String> consumerObserveResourceNames = new HashSet<>();
if (resourceNames.isEmpty()) {
@ -157,7 +146,7 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
Set<String> 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<T, S extends DeltaResource<T>> implements
public void observeResource(Set<String> resourceNames, Consumer<Map<String, T>> 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<T, S extends DeltaResource<T>> implements
} finally {
writeLock.unlock();
}
consumer.accept(getResource(resourceNames));
}
try {
writeLock.lock();
@ -210,6 +199,10 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
}
}
public void unobserveResource(Set<String> resourceNames, Consumer<Map<String, T>> consumer) {
// TODO
}
protected DiscoveryRequest buildDiscoveryRequest(Set<String> resourceNames) {
return DiscoveryRequest.newBuilder()
.setNode(node)
@ -218,108 +211,50 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
.build();
}
protected DiscoveryRequest buildDiscoveryRequest(Set<String> resourceNames, DiscoveryResponse response) {
// for ACK
return DiscoveryRequest.newBuilder()
.setNode(node)
.setTypeUrl(response.getTypeUrl())
.setVersionInfo(response.getVersionInfo())
.setResponseNonce(response.getNonce())
.build();
}
protected abstract Map<String, T> decodeDiscoveryResponse(DiscoveryResponse response);
public class ResponseObserver implements StreamObserver<DiscoveryResponse> {
public ResponseObserver() {
}
@Override
public void onNext(DiscoveryResponse value) {
Map<String, T> newResult = decodeDiscoveryResponse(value);
Map<String, T> oldResource = resourcesMap;
discoveryResponseListener(oldResource, newResult);
resourcesMap = newResult;
requestObserver.onNext(buildDiscoveryRequest(Collections.emptySet(), value));
}
public void discoveryResponseListener(Map<String, T> oldResult, Map<String, T> newResult) {
Set<String> 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<Set<String>, List<Consumer<Map<String, T>>>> entry : consumerObserveMap.entrySet()) {
if (entry.getKey().stream().noneMatch(changedResourceNames::contains)) {
// none update
continue;
}
Map<String, T> 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<String, T> newResult = decodeDiscoveryResponse(discoveryResponse);
Map<String, T> 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<String> 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<String, T> oldResult, Map<String, T> newResult) {
Set<String> 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<Set<String>, List<Consumer<Map<String, T>>>> entry : consumerObserveMap.entrySet()) {
if (entry.getKey().stream().noneMatch(changedResourceNames::contains)) {
// none update
continue;
}
Map<String, T> dsResultMap = entry.getKey()
.stream()
.collect(Collectors.toMap(k -> k, v -> newResult.get(v)));
entry.getValue().forEach(o -> o.accept(dsResultMap));
}
} finally {
readLock.unlock();
}
}
}

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

@ -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<EndpointResult, DeltaEndpoint> {
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<EndpointResult, DeltaEndpoint>
private EndpointResult decodeResourceToEndpoint(ClusterLoadAssignment resource) {
Set<Endpoint> 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());

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

@ -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<ListenerResult, DeltaListener> {
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

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

@ -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<RouteResult, DeltaRoute> {
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

9
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/Endpoint.java
View File

@ -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;
}

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

@ -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<String, Set<String>> domainMap,Map<String, VirtualHost> virtualHostMap) {
public RouteResult(Map<String, Set<String>> domainMap, Map<String, VirtualHost> 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 +
'}';
}
}

26
dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/EdsEndpointManager.java
View File

@ -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<String, Set<Endpoint>> ENDPOINT_DATA_CACHE = new ConcurrentHashMap<>();
private static final ConcurrentHashMap<String, Consumer<Set<Endpoint>>> EDS_LISTENERS = new ConcurrentHashMap<>();
private static final ConcurrentHashMap<String, Consumer<Map<String, EndpointResult>>> 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<Endpoint> result = endpoints.values()
.stream()
.map(EndpointResult::getEndpoints)
.flatMap(Set::stream)
.collect(Collectors.toSet());
notifyEndpointChange(cluster, result);
});
Consumer<Set<Endpoint>> consumer = EDS_LISTENERS.get(cluster);
Consumer<Map<String, EndpointResult>> 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<Set<Endpoint>> consumer = EDS_LISTENERS.remove(cluster);
Consumer<Map<String, EndpointResult>> 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<String, Consumer<Set<Endpoint>>> getEdsListeners() {
static ConcurrentHashMap<String, Consumer<Map<String, EndpointResult>>> getEdsListeners() {
return EDS_LISTENERS;
}

70
dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/RdsRouteRuleManager.java
View File

@ -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<String, List<XdsRouteRule>> ROUTE_DATA_CACHE = new ConcurrentHashMap<>();
private static final ConcurrentHashMap<String, RdsVirtualHostListener> RDS_LISTENERS = new ConcurrentHashMap<>();
private static final Map<String, RdsVirtualHostListener> RDS_LISTENERS = new ConcurrentHashMap<>();
private static volatile Consumer<Map<String, ListenerResult>> LDS_LISTENER;
private static volatile Consumer<Map<String, RouteResult>> RDS_LISTENER;
private static Map<String, RouteResult> 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<Map<String, ListenerResult>>() {
private volatile Set<String> configNames = null;
@Override
public void accept(Map<String, ListenerResult> listenerResults) {
if (listenerResults.size() == 1) {
for (ListenerResult listenerResult : listenerResults.values()) {
Set<String> 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<String, RdsVirtualHostListener> getRdsListeners() {
static Map<String, RdsVirtualHostListener> getRdsListeners() {
return RDS_LISTENERS;
}

1082
dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/DsProtocolTest.java
View File

File diff suppressed because it is too large Load Diff

20
dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocolMock.java
View File

@ -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<String, EndpointResult> getResourcesMap() {
@ -45,14 +45,8 @@ public class EdsProtocolMock extends EdsProtocol{
this.consumerObserveMap = consumerObserveMap;
}
public ResponseObserverMock getResponseObserve() {
return new ResponseObserverMock();
}
public void setObserveResourcesName(Set<String> observeResourcesName) {
this.observeResourcesName = observeResourcesName;
}
class ResponseObserverMock extends ResponseObserver {
}
}

32
dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocolMock.java
View File

@ -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<String, ListenerResult> getResourcesMap() {
@ -43,18 +42,6 @@ public class LdsProtocolMock extends LdsProtocol{
this.resourcesMap = resourcesMap;
}
public StreamObserver<DiscoveryRequest> getRequestObserver() {
return requestObserver;
}
public void setRequestObserver(StreamObserver<DiscoveryRequest> requestObserver) {
this.requestObserver = requestObserver;
}
public ResponseObserverMock getResponseObserve() {
return new ResponseObserverMock();
}
protected DiscoveryRequest buildDiscoveryRequest(Set<String> resourceNames) {
return DiscoveryRequest.newBuilder()
.setNode(node)
@ -78,7 +65,4 @@ public class LdsProtocolMock extends LdsProtocol{
public void setConsumerObserveMap(Map<Set<String>, List<Consumer<Map<String, ListenerResult>>>> consumerObserveMap) {
this.consumerObserveMap = consumerObserveMap;
}
class ResponseObserverMock extends ResponseObserver {
}
}

23
dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocolMock.java
View File

@ -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<String, RouteResult> getResourcesMap() {
@ -44,11 +44,7 @@ public class RdsProtocolMock extends RdsProtocol{
public Set<String> getObserveResourcesName() {
return observeResourcesName;
}
public ResponseObserverMock getResponseObserve() {
return new ResponseObserverMock();
}
public void setConsumerObserveMap(Map<Set<String>, List<Consumer<Map<String, RouteResult>>>> consumerObserveMap) {
this.consumerObserveMap = consumerObserveMap;
}
@ -56,7 +52,4 @@ public class RdsProtocolMock extends RdsProtocol{
this.observeResourcesName = observeResourcesName;
}
class ResponseObserverMock extends ResponseObserver {
}
}