virtualHosts, String hostName) {
+ // Domain search order:
+ // 1. Exact domain names: ``www.foo.com``.
+ // 2. Suffix domain wildcards: ``*.foo.com`` or ``*-bar.foo.com``.
+ // 3. Prefix domain wildcards: ``foo.*`` or ``foo-*``.
+ // 4. Special wildcard ``*`` matching any domain.
+ //
+ // The longest wildcards match first.
+ // Assuming only a single virtual host in the entire route configuration can match
+ // on ``*`` and a domain must be unique across all virtual hosts.
+ int matchingLen = -1; // longest length of wildcard pattern that matches host name
+ boolean exactMatchFound = false; // true if a virtual host with exactly matched domain found
+ VirtualHost targetVirtualHost = null; // target VirtualHost with longest matched domain
+ for (VirtualHost vHost : virtualHosts) {
+ for (String domain : vHost.getDomains()) {
+ boolean selected = false;
+ if (matchHostName(hostName, domain)) { // matching
+ if (!domain.contains("*")) { // exact matching
+ exactMatchFound = true;
+ targetVirtualHost = vHost;
+ break;
+ } else if (domain.length() > matchingLen) { // longer matching pattern
+ selected = true;
+ } else if (domain.length() == matchingLen && domain.startsWith("*")) { // suffix matching
+ selected = true;
+ }
+ }
+ if (selected) {
+ matchingLen = domain.length();
+ targetVirtualHost = vHost;
+ }
+ }
+ if (exactMatchFound) {
+ break;
+ }
+ }
+ return targetVirtualHost;
+ }
+
+ /**
+ * Returns {@code true} iff {@code hostName} matches the domain name {@code pattern} with
+ * case-insensitive.
+ *
+ * Wildcard pattern rules:
+ *
+ * - A single asterisk (*) matches any domain.
+ * - Asterisk (*) is only permitted in the left-most or the right-most part of the pattern,
+ * but not both.
+ *
+ */
+ private static boolean matchHostName(String hostName, String pattern) {
+ checkArgument(
+ hostName.length() != 0 && !hostName.startsWith(".") && !hostName.endsWith("."), "Invalid host name");
+ checkArgument(
+ pattern.length() != 0 && !pattern.startsWith(".") && !pattern.endsWith("."),
+ "Invalid pattern/domain name");
+
+ hostName = hostName.toLowerCase(Locale.US);
+ pattern = pattern.toLowerCase(Locale.US);
+ // hostName and pattern are now in lower case -- domain names are case-insensitive.
+
+ if (!pattern.contains("*")) {
+ // Not a wildcard pattern -- hostName and pattern must match exactly.
+ return hostName.equals(pattern);
+ }
+ // Wildcard pattern
+
+ if (pattern.length() == 1) {
+ return true;
+ }
+
+ int index = pattern.indexOf('*');
+
+ // At most one asterisk (*) is allowed.
+ if (pattern.indexOf('*', index + 1) != -1) {
+ return false;
+ }
+
+ // Asterisk can only match prefix or suffix.
+ if (index != 0 && index != pattern.length() - 1) {
+ return false;
+ }
+
+ // HostName must be at least as long as the pattern because asterisk has to
+ // match one or more characters.
+ if (hostName.length() < pattern.length()) {
+ return false;
+ }
+
+ if (index == 0 && hostName.endsWith(pattern.substring(1))) {
+ // Prefix matching fails.
+ return true;
+ }
+
+ // Pattern matches hostname if suffix matching succeeds.
+ return index == pattern.length() - 1 && hostName.startsWith(pattern.substring(0, pattern.length() - 1));
+ }
+
+ /**
+ * Returns {@code true} iff the given {@link RouteMatch} matches the RPC's full method name and
+ * headers.
+ */
+ static boolean matchRoute(RouteMatch routeMatch, String fullMethodName, Metadata headers, ThreadSafeRandom random) {
+ if (!matchPath(routeMatch.getPathMatcher(), fullMethodName)) {
+ return false;
+ }
+ for (HeaderMatcher headerMatcher : routeMatch.getHeaderMatchers()) {
+ if (!headerMatcher.matches(getHeaderValue(headers, headerMatcher.name()))) {
+ return false;
+ }
+ }
+ FractionMatcher fraction = routeMatch.getFractionMatcher();
+ return fraction == null || random.nextInt(fraction.getDenominator()) < fraction.getNumerator();
+ }
+
+ private static boolean matchPath(PathMatcher pathMatcher, String fullMethodName) {
+ if (pathMatcher.getPath() != null) {
+ return pathMatcher.isCaseSensitive()
+ ? pathMatcher.getPath().equals(fullMethodName)
+ : pathMatcher.getPath().equalsIgnoreCase(fullMethodName);
+ } else if (pathMatcher.getPrefix() != null) {
+ return pathMatcher.isCaseSensitive()
+ ? fullMethodName.startsWith(pathMatcher.getPrefix())
+ : fullMethodName
+ .toLowerCase(Locale.US)
+ .startsWith(pathMatcher.getPrefix().toLowerCase(Locale.US));
+ }
+ return pathMatcher.getRegEx().matches(fullMethodName);
+ }
+
+ @Nullable
+ private static String getHeaderValue(Metadata headers, String headerName) {
+ if (headerName.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
+ return null;
+ }
+ if (headerName.equals("content-type")) {
+ return "application/grpc";
+ }
+ Metadata.Key key;
+ try {
+ key = Metadata.Key.of(headerName, Metadata.ASCII_STRING_MARSHALLER);
+ } catch (IllegalArgumentException e) {
+ return null;
+ }
+ Iterable values = headers.getAll(key);
+ return values == null ? null : Joiner.on(",").join(values);
+ }
+}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/directory/XdsDirectory.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/directory/XdsDirectory.java
index e4c121e479e..441b8080db5 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/directory/XdsDirectory.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/directory/XdsDirectory.java
@@ -19,6 +19,7 @@
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.url.component.URLAddress;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@@ -28,20 +29,43 @@
import org.apache.dubbo.rpc.cluster.directory.AbstractDirectory;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.xds.PilotExchanger;
+import org.apache.dubbo.xds.XdsResourceListener;
+import org.apache.dubbo.xds.directory.XdsDirectory.LdsUpdateWatcher.RdsUpdateWatcher;
+import org.apache.dubbo.xds.resource.XdsClusterResource;
+import org.apache.dubbo.xds.resource.XdsListenerResource;
+import org.apache.dubbo.xds.resource.XdsRouteConfigureResource;
+import org.apache.dubbo.xds.resource.cluster.OutlierDetection;
+import org.apache.dubbo.xds.resource.common.Locality;
+import org.apache.dubbo.xds.resource.endpoint.DropOverload;
import org.apache.dubbo.xds.resource.endpoint.LbEndpoint;
+import org.apache.dubbo.xds.resource.endpoint.LocalityLbEndpoints;
+import org.apache.dubbo.xds.resource.filter.NamedFilterConfig;
+import org.apache.dubbo.xds.resource.listener.HttpConnectionManager;
+import org.apache.dubbo.xds.resource.listener.security.UpstreamTlsContext;
import org.apache.dubbo.xds.resource.route.ClusterWeight;
import org.apache.dubbo.xds.resource.route.Route;
import org.apache.dubbo.xds.resource.route.RouteAction;
import org.apache.dubbo.xds.resource.route.VirtualHost;
+import org.apache.dubbo.xds.resource.update.CdsUpdate;
+import org.apache.dubbo.xds.resource.update.CdsUpdate.ClusterType;
import org.apache.dubbo.xds.resource.update.EdsUpdate;
+import org.apache.dubbo.xds.resource.update.LdsUpdate;
+import org.apache.dubbo.xds.resource.update.RdsUpdate;
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
+
+import com.google.common.collect.Sets;
public class XdsDirectory extends AbstractDirectory {
@@ -63,6 +87,11 @@ public class XdsDirectory extends AbstractDirectory {
private static ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(XdsDirectory.class);
+ private Map ldsWatchers = new HashMap<>();
+ private Map rdsWatchers = new HashMap<>();
+ private Map cdsWatchers = new HashMap<>();
+ private Map edsWatchers = new HashMap<>();
+
public XdsDirectory(Directory directory) {
super(directory.getUrl(), null, true, directory.getConsumerUrl());
this.serviceType = directory.getInterface();
@@ -76,7 +105,9 @@ public XdsDirectory(Directory directory) {
// subscribe resource
for (String applicationName : applicationNames) {
- pilotExchanger.subscribeRds(applicationName, this);
+ LdsUpdateWatcher ldsUpdateWatcher = new LdsUpdateWatcher(applicationName);
+ ldsWatchers.putIfAbsent(applicationName, ldsUpdateWatcher);
+ pilotExchanger.subscribeXdsResource(applicationName, XdsListenerResource.getInstance(), ldsUpdateWatcher);
}
}
@@ -112,13 +143,6 @@ public List> getAllInvokers() {
return super.getInvokers();
}
- public void onRdsChange(String applicationName, VirtualHost xdsVirtualHost) {
- Set oldCluster = getAllCluster();
- xdsVirtualHostMap.put(applicationName, xdsVirtualHost);
- Set newCluster = getAllCluster();
- changeClusterSubscribe(oldCluster, newCluster);
- }
-
private Set getAllCluster() {
if (CollectionUtils.isEmptyMap(xdsVirtualHostMap)) {
return new HashSet<>();
@@ -139,53 +163,283 @@ private Set getAllCluster() {
return clusters;
}
- private void changeClusterSubscribe(Set oldCluster, Set newCluster) {
- Set removeSubscribe = new HashSet<>(oldCluster);
- Set addSubscribe = new HashSet<>(newCluster);
+ @Override
+ public boolean isAvailable() {
+ return true;
+ }
+
+ @Override
+ public void destroy() {
+ super.destroy();
+ //
+ // pilotExchanger.unSubscribeXdsResource(resourceName, this);
+ }
+
+ public class LdsUpdateWatcher implements XdsResourceListener {
+ private final String ldsResourceName;
+
+ @Nullable
+ private Set existingClusters; // clusters to which new requests can be routed
- removeSubscribe.removeAll(newCluster);
- addSubscribe.removeAll(oldCluster);
+ @Nullable
+ private RdsUpdateWatcher rdsUpdateWatcher;
- // remove subscribe cluster
- for (String cluster : removeSubscribe) {
- pilotExchanger.unSubscribeCds(cluster, this);
- xdsEndpointMap.remove(cluster);
- // TODO: delete invokers which belong unsubscribed cluster
+ public LdsUpdateWatcher(String ldsResourceName) {
+ this.ldsResourceName = ldsResourceName;
}
- // add subscribe cluster
- for (String cluster : addSubscribe) {
- pilotExchanger.subscribeCds(cluster, this);
+
+ @Override
+ public void onResourceUpdate(LdsUpdate update) {
+ HttpConnectionManager httpConnectionManager = update.getHttpConnectionManager();
+ List virtualHosts = httpConnectionManager.getVirtualHosts();
+ String rdsName = httpConnectionManager.getRdsName();
+
+ if (virtualHosts != null) {
+ updateRoutes(
+ virtualHosts,
+ httpConnectionManager.getHttpMaxStreamDurationNano(),
+ httpConnectionManager.getHttpFilterConfigs());
+ } else {
+ rdsUpdateWatcher = new RdsUpdateWatcher(
+ rdsName,
+ httpConnectionManager.getHttpMaxStreamDurationNano(),
+ httpConnectionManager.getHttpFilterConfigs());
+ rdsWatchers.putIfAbsent(rdsName, rdsUpdateWatcher);
+ pilotExchanger.subscribeXdsResource(rdsName, XdsRouteConfigureResource.getInstance(), rdsUpdateWatcher);
+ }
+ }
+
+ private void updateRoutes(
+ List virtualHosts,
+ long httpMaxStreamDurationNano,
+ @Nullable List filterConfigs) {
+ // String authority = overrideAuthority != null ? overrideAuthority : ldsResourceName;
+ String authority = ldsResourceName;
+ VirtualHost virtualHost = RoutingUtils.findVirtualHostForHostName(virtualHosts, authority);
+ if (virtualHost == null) {
+ return;
+ }
+
+ List routes = virtualHost.getRoutes();
+
+ // Populate all clusters to which requests can be routed to through the virtual host.
+ Set clusters = new HashSet<>();
+ // uniqueName -> clusterName
+ Map clusterNameMap = new HashMap<>();
+ for (Route route : routes) {
+ RouteAction action = route.getRouteAction();
+ String clusterName;
+ if (action != null) {
+ if (action.getCluster() != null) {
+ clusterName = action.getCluster();
+ clusters.add(clusterName);
+ clusterNameMap.put(clusterName, action.getCluster());
+ } else if (action.getWeightedClusters() != null) {
+ for (ClusterWeight weighedCluster : action.getWeightedClusters()) {
+ clusterName = weighedCluster.getName();
+ clusters.add(clusterName);
+ clusterNameMap.put(clusterName, weighedCluster.getName());
+ }
+ }
+ }
+ }
+
+ boolean shouldUpdateResult = existingClusters == null;
+ Set addedClusters =
+ existingClusters == null ? clusters : Sets.difference(clusters, existingClusters);
+ Set deletedClusters =
+ existingClusters == null ? Collections.emptySet() : Sets.difference(existingClusters, clusters);
+ existingClusters = clusters;
+ for (String cluster : addedClusters) {
+ CdsUpdateNodeDirectory cdsUpdateWatcher = new CdsUpdateNodeDirectory();
+ cdsWatchers.putIfAbsent(cluster, cdsUpdateWatcher);
+ pilotExchanger.subscribeXdsResource(cluster, XdsClusterResource.getInstance(), cdsUpdateWatcher);
+ }
+ }
+
+ public class RdsUpdateWatcher implements XdsResourceListener {
+ private String rdsName;
+
+ private final long httpMaxStreamDurationNano;
+
+ @Nullable
+ private final List filterConfigs;
+
+ public RdsUpdateWatcher(
+ String rdsName, long httpMaxStreamDurationNano, @Nullable List filterConfigs) {
+ this.rdsName = rdsName;
+ this.httpMaxStreamDurationNano = httpMaxStreamDurationNano;
+ this.filterConfigs = filterConfigs;
+ }
+
+ @Override
+ public void onResourceUpdate(RdsUpdate update) {
+ if (RdsUpdateWatcher.this != rdsUpdateWatcher) {
+ return;
+ }
+ updateRoutes(update.getVirtualHosts(), httpMaxStreamDurationNano, filterConfigs);
+ }
}
}
- public void onEdsChange(String clusterName, EdsUpdate edsUpdate) {
- xdsEndpointMap.put(clusterName, edsUpdate);
- // String lbPolicy = xdsCluster.getLbPolicy();
- List xdsEndpoints = edsUpdate.getLocalityLbEndpointsMap().values().stream()
- .flatMap(e -> e.getEndpoints().stream())
- .collect(Collectors.toList());
- BitList> invokers = new BitList<>(Collections.emptyList());
- xdsEndpoints.forEach(e -> {
- String ip = e.getAddresses().getFirst().getAddress();
- int port = e.getAddresses().getFirst().getPort();
- URL url = new URL(this.protocolName, ip, port, this.serviceType.getName(), this.url.getParameters());
- // set cluster name
- url = url.addParameter("clusterID", clusterName);
- // set load balance policy
- // url = url.addParameter("loadbalance", lbPolicy);
- // cluster to invoker
- Invoker invoker = this.protocol.refer(this.serviceType, url);
- invokers.add(invoker);
- });
- // TODO: Consider cases where some clients are not available
- // super.getInvokers().addAll(invokers);
- // TODO: Need add new api which can add invokers, because a XdsDirectory need monitor multi clusters.
- super.setInvokers(invokers);
- // xdsCluster.setInvokers(invokers);
+ /**
+ * This is the internal node of the Directory tree, which is responsible for creating invokers from clusters.
+ *
+ * Each invoker instance created in this should be representing a cluster pointing to another Directory instead of a specific instance invoker.
+ */
+ public class CdsUpdateNodeDirectory implements XdsResourceListener {
+ @Override
+ public void onResourceUpdate(CdsUpdate update) {
+ // 啥都不干，就是把 aggregate logicalDns eds 三种做个分类处理，其中eds的不用做什么事情
+ if (update.getClusterType() == ClusterType.AGGREGATE) {
+ String clusterName = update.getClusterName();
+ for (String cluster : update.getPrioritizedClusterNames()) {
+ // create internal node directory.
+ }
+ } else if (update.getClusterType() == ClusterType.EDS) {
+ // create leaf directory.
+ } else {
+
+ }
+ }
}
- @Override
- public boolean isAvailable() {
- return true;
+ /**
+ * This is the leaf node of the Directory tree, which is responsible for creating invokers from endpoints.
+ *
+ * Each invoker instance created in this should be representing a specific dubbo provider instance.
+ */
+ public class EdsUpdateLeafDirectory implements XdsResourceListener {
+ private final String clusterName;
+ private final String edsResourceName;
+
+ @Nullable
+ protected final Long maxConcurrentRequests;
+
+ @Nullable
+ protected final UpstreamTlsContext tlsContext;
+
+ @Nullable
+ protected final OutlierDetection outlierDetection;
+
+ private Map localityPriorityNames = Collections.emptyMap();
+
+ int priorityNameGenId = 1;
+
+ public EdsUpdateLeafDirectory(
+ String clusterName,
+ String edsResourceName,
+ @Nullable Long maxConcurrentRequests,
+ @Nullable UpstreamTlsContext tlsContext,
+ @Nullable OutlierDetection outlierDetection) {
+ this.clusterName = clusterName;
+ this.edsResourceName = edsResourceName;
+ this.maxConcurrentRequests = maxConcurrentRequests;
+ this.tlsContext = tlsContext;
+ this.outlierDetection = outlierDetection;
+ }
+
+ @Override
+ public void onResourceUpdate(EdsUpdate update) {
+ Map localityLbEndpoints = update.getLocalityLbEndpointsMap();
+ List dropOverloads = update.getDropPolicies();
+ List addresses = new ArrayList<>();
+ Map> prioritizedLocalityWeights = new HashMap<>();
+ List sortedPriorityNames = generatePriorityNames(clusterName, localityLbEndpoints);
+ for (Locality locality : localityLbEndpoints.keySet()) {
+ LocalityLbEndpoints localityLbInfo = localityLbEndpoints.get(locality);
+ String priorityName = localityPriorityNames.get(locality);
+ boolean discard = true;
+ for (LbEndpoint endpoint : localityLbInfo.getEndpoints()) {
+ if (endpoint.isHealthy()) {
+ discard = false;
+ long weight = localityLbInfo.getLocalityWeight();
+ if (endpoint.getLoadBalancingWeight() != 0) {
+ weight *= endpoint.getLoadBalancingWeight();
+ }
+ addresses.add(endpoint.getAddresses().get(0));
+ }
+ }
+ if (discard) {
+ logger.info("Discard locality {0} with 0 healthy endpoints", locality);
+ continue;
+ }
+ if (!prioritizedLocalityWeights.containsKey(priorityName)) {
+ prioritizedLocalityWeights.put(priorityName, new HashMap());
+ }
+ prioritizedLocalityWeights.get(priorityName).put(locality, localityLbInfo.getLocalityWeight());
+ }
+
+ sortedPriorityNames.retainAll(prioritizedLocalityWeights.keySet());
+ }
+
+ private List generatePriorityNames(
+ String name, Map localityLbEndpoints) {
+ TreeMap> todo = new TreeMap<>();
+ for (Locality locality : localityLbEndpoints.keySet()) {
+ int priority = localityLbEndpoints.get(locality).getPriority();
+ if (!todo.containsKey(priority)) {
+ todo.put(priority, new ArrayList<>());
+ }
+ todo.get(priority).add(locality);
+ }
+ Map newNames = new HashMap<>();
+ Set usedNames = new HashSet<>();
+ List ret = new ArrayList<>();
+ for (Integer priority : todo.keySet()) {
+ String foundName = "";
+ for (Locality locality : todo.get(priority)) {
+ if (localityPriorityNames.containsKey(locality)
+ && usedNames.add(localityPriorityNames.get(locality))) {
+ foundName = localityPriorityNames.get(locality);
+ break;
+ }
+ }
+ if ("".equals(foundName)) {
+ foundName = String.format(Locale.US, "%s[child%d]", name, priorityNameGenId++);
+ }
+ for (Locality locality : todo.get(priority)) {
+ newNames.put(locality, foundName);
+ }
+ ret.add(foundName);
+ }
+ localityPriorityNames = newNames;
+ return ret;
+ }
}
+
+ //
+ // public void onResourceUpdate(CdsUpdate cdsUpdate) {
+ // // for eds cluster, do nothing
+ //
+ // // for aggregate clusters, do subscription
+ // String clusterName = cdsUpdate.getClusterName();
+ // this.pilotExchanger.subscribeCds(clusterName, this);
+ // }
+ //
+ // public void onResourceUpdate(String clusterName, EdsUpdate edsUpdate) {
+ // xdsEndpointMap.put(clusterName, edsUpdate);
+ // // String lbPolicy = xdsCluster.getLbPolicy();
+ // List xdsEndpoints = edsUpdate.getLocalityLbEndpointsMap().values().stream()
+ // .flatMap(e -> e.getEndpoints().stream())
+ // .collect(Collectors.toList());
+ // BitList> invokers = new BitList<>(Collections.emptyList());
+ // xdsEndpoints.forEach(e -> {
+ // String ip = e.getAddresses().get(0).getAddress();
+ // int port = e.getAddresses().get(0).getPort();
+ // URL url = new URL(this.protocolName, ip, port, this.serviceType.getName(), this.url.getParameters());
+ // // set cluster name
+ // url = url.addParameter("clusterID", clusterName);
+ // // set load balance policy
+ // // url = url.addParameter("loadbalance", lbPolicy);
+ // // cluster to invoker
+ // Invoker invoker = this.protocol.refer(this.serviceType, url);
+ // invokers.add(invoker);
+ // });
+ // // TODO: Consider cases where some clients are not available
+ // // super.getInvokers().addAll(invokers);
+ // // TODO: Need add new api which can add invokers, because a XdsDirectory need monitor multi clusters.
+ // super.setInvokers(invokers);
+ // // xdsCluster.setInvokers(invokers);
+ // }
}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/CdsListener.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/CdsListener.java
index 2e875d6fc66..bc8498cbbb1 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/CdsListener.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/CdsListener.java
@@ -18,8 +18,11 @@
import org.apache.dubbo.common.extension.ExtensionScope;
import org.apache.dubbo.common.extension.SPI;
-import org.apache.dubbo.xds.protocol.XdsResourceListener;
import org.apache.dubbo.xds.resource.update.CdsUpdate;
+import java.util.List;
+
@SPI(scope = ExtensionScope.APPLICATION)
-public interface CdsListener extends XdsResourceListener {}
+public interface CdsListener {
+ void onResourceUpdate(List resource);
+}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/LdsListener.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/LdsListener.java
index e73e652c645..f874e2e0866 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/LdsListener.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/LdsListener.java
@@ -18,8 +18,11 @@
import org.apache.dubbo.common.extension.ExtensionScope;
import org.apache.dubbo.common.extension.SPI;
-import org.apache.dubbo.xds.protocol.XdsResourceListener;
import org.apache.dubbo.xds.resource.update.LdsUpdate;
+import java.util.List;
+
@SPI(scope = ExtensionScope.APPLICATION)
-public interface LdsListener extends XdsResourceListener {}
+public interface LdsListener {
+ void onResourceUpdate(List resource);
+}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/UpstreamTlsConfigListener.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/UpstreamTlsConfigListener.java
index 6380f365708..d5a6e0f5543 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/UpstreamTlsConfigListener.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/listener/UpstreamTlsConfigListener.java
@@ -52,7 +52,6 @@ public UpstreamTlsConfigListener(ApplicationModel application) {
this.tlsConfigRepository = application.getBeanFactory().getOrRegisterBean(XdsTlsConfigRepository.class);
}
- @Override
public void onResourceUpdate(List resources) {
Map configs = new ConcurrentHashMap<>(16);
List clusters =
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/AbstractProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/AbstractProtocol.java
deleted file mode 100644
index 7f04e1d8a5a..00000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/AbstractProtocol.java
+++ /dev/null
@@ -1,228 +0,0 @@
-/*
- * 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.xds.protocol;
-
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
-import org.apache.dubbo.common.utils.StringUtils;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-import org.apache.dubbo.xds.AdsObserver;
-import org.apache.dubbo.xds.XdsListener;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.function.Consumer;
-import java.util.stream.Collectors;
-
-import io.envoyproxy.envoy.config.core.v3.Node;
-import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
-import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
-
-public abstract class AbstractProtocol implements XdsProtocol, XdsListener {
-
- private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractProtocol.class);
-
- protected AdsObserver adsObserver;
-
- protected final Node node;
-
- private final int checkInterval;
-
- protected final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
-
- protected final ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
-
- protected final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
-
- protected Set observeResourcesName;
-
- public static final String emptyResourceName = "emptyResourcesName";
- private final ReentrantLock resourceLock = new ReentrantLock();
-
- protected Map, List>>> consumerObserveMap = new ConcurrentHashMap<>();
-
- public Map, List>>> getConsumerObserveMap() {
- return consumerObserveMap;
- }
-
- protected Map resourcesMap = new ConcurrentHashMap<>();
-
- protected List> resourceListeners = new CopyOnWriteArrayList<>();
-
- protected ApplicationModel applicationModel;
-
- public AbstractProtocol(AdsObserver adsObserver, Node node, int checkInterval, ApplicationModel applicationModel) {
- this.adsObserver = adsObserver;
- this.node = node;
- this.checkInterval = checkInterval;
- this.applicationModel = applicationModel;
- adsObserver.addListener(this);
- }
-
- public void registerListen(XdsResourceListener listener) {
- this.resourceListeners.add(listener);
- }
-
- /**
- * Abstract method to obtain Type-URL from sub-class
- *
- * @return Type-URL of xDS
- */
- public abstract String getTypeUrl();
-
- public boolean isCacheExistResource(Set resourceNames) {
- for (String resourceName : resourceNames) {
- if ("".equals(resourceName)) {
- continue;
- }
- if (!resourcesMap.containsKey(resourceName)) {
- return false;
- }
- }
- return true;
- }
-
- public T getCacheResource(String resourceName) {
- if (resourceName == null || resourceName.length() == 0) {
- return null;
- }
- return resourcesMap.get(resourceName);
- }
-
- @Override
- public void subscribeResource(Set resourceNames) {
- resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
-
- if (!resourceNames.isEmpty() && isCacheExistResource(resourceNames)) {
- getResourceFromCache(resourceNames);
- } else {
- getResourceFromRemote(resourceNames);
- }
- }
-
- private Map getResourceFromCache(Set resourceNames) {
- return resourceNames.stream()
- .filter(o -> !StringUtils.isEmpty(o))
- .collect(Collectors.toMap(k -> k, this::getCacheResource));
- }
-
- public Map getResourceFromRemote(Set resourceNames) {
- try {
- resourceLock.lock();
- CompletableFuture> future = new CompletableFuture<>();
- observeResourcesName = resourceNames;
- Set consumerObserveResourceNames = new HashSet<>();
- if (resourceNames.isEmpty()) {
- consumerObserveResourceNames.add(emptyResourceName);
- } else {
- consumerObserveResourceNames = resourceNames;
- }
-
- Consumer> futureConsumer = future::complete;
- try {
- writeLock.lock();
- ConcurrentHashMapUtils.computeIfAbsent(
- (ConcurrentHashMap, List>>>) consumerObserveMap,
- consumerObserveResourceNames,
- key -> new ArrayList<>())
- .add(futureConsumer);
- } finally {
- writeLock.unlock();
- }
-
- Set resourceNamesToObserve = new HashSet<>(resourceNames);
- resourceNamesToObserve.addAll(resourcesMap.keySet());
- adsObserver.request(buildDiscoveryRequest(resourceNamesToObserve));
- logger.info("Send xDS Observe request to remote. Resource count: " + resourceNamesToObserve.size()
- + ". Resource Type: " + getTypeUrl());
- } finally {
- resourceLock.unlock();
- }
- return Collections.emptyMap();
- }
-
- protected DiscoveryRequest buildDiscoveryRequest(Set resourceNames) {
- return DiscoveryRequest.newBuilder()
- .setNode(node)
- .setTypeUrl(getTypeUrl())
- .addAllResourceNames(resourceNames)
- .build();
- }
-
- // protected abstract Map decodeDiscoveryResponse(DiscoveryResponse response);
-
- protected abstract Map decodeDiscoveryResponse(DiscoveryResponse response);
-
- @Override
- public final void process(DiscoveryResponse discoveryResponse) {
- // Map newResult = decodeDiscoveryResponse(discoveryResponse);
- Map oldResource = resourcesMap;
- // discoveryResponseListener(oldResource, newResult);
-
- Map newResult = decodeDiscoveryResponse(discoveryResponse);
- resourceListeners.forEach(l -> l.onResourceUpdate(new ArrayList<>(newResult.values())));
- resourcesMap = newResult;
- }
-
- 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);
- }
- });
- 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/xds/protocol/XdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/XdsProtocol.java
deleted file mode 100644
index 838988ad518..00000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/XdsProtocol.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.xds.protocol;
-
-import java.util.Set;
-
-public interface XdsProtocol {
- /**
- * Gets all resources by the specified resource name.
- * For LDS, the {@param resourceNames} is ignored
- *
- * @param resourceNames specified resource name
- * @return resources, null if request failed
- */
- void subscribeResource(Set resourceNames);
-
- /**
- * Add a observer resource with {@link Consumer}
- *
- * @param resourceNames specified resource name
- * @param consumer resource notifier, will be called when resource updated
- * @return requestId, used when resourceNames update with {@link XdsProtocol#updateObserve(long, Set)}
- */
- // void observeResource(Set resourceNames, Consumer> consumer, boolean isReConnect);
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/CdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/CdsProtocol.java
deleted file mode 100644
index 33525699a45..00000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/CdsProtocol.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * 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.xds.protocol.impl;
-
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-import org.apache.dubbo.xds.AdsObserver;
-import org.apache.dubbo.xds.listener.CdsListener;
-import org.apache.dubbo.xds.protocol.AbstractProtocol;
-import org.apache.dubbo.xds.resource.XdsClusterResource;
-import org.apache.dubbo.xds.resource.XdsResourceType;
-import org.apache.dubbo.xds.resource.update.CdsUpdate;
-import org.apache.dubbo.xds.resource.update.ValidatedResourceUpdate;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.function.Consumer;
-import java.util.stream.Collectors;
-
-import io.envoyproxy.envoy.config.core.v3.Node;
-import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_PARSING_XDS;
-
-public class CdsProtocol extends AbstractProtocol {
- private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(CdsProtocol.class);
-
- public void setUpdateCallback(Consumer> updateCallback) {
- this.updateCallback = updateCallback;
- }
-
- private static final XdsClusterResource xdsClusterResource = XdsClusterResource.getInstance();
-
- private Consumer> updateCallback;
-
- public CdsProtocol(AdsObserver adsObserver, Node node, int checkInterval, ApplicationModel applicationModel) {
- super(adsObserver, node, checkInterval, applicationModel);
- List ldsListeners =
- applicationModel.getExtensionLoader(CdsListener.class).getActivateExtensions();
- ldsListeners.forEach(this::registerListen);
- }
-
- @Override
- public String getTypeUrl() {
- return "type.googleapis.com/envoy.config.cluster.v3.Cluster";
- }
-
- public void subscribeClusters() {
- subscribeResource(null);
- }
-
- @Override
- protected Map decodeDiscoveryResponse(DiscoveryResponse response) {
- if (!getTypeUrl().equals(response.getTypeUrl())) {
- return Collections.emptyMap();
- }
- ValidatedResourceUpdate validatedResourceUpdate =
- xdsClusterResource.parse(XdsResourceType.xdsResourceTypeArgs, response.getResourcesList());
- if (!validatedResourceUpdate.getErrors().isEmpty()) {
- logger.error(
- REGISTRY_ERROR_PARSING_XDS,
- validatedResourceUpdate.getErrors().toArray());
- }
- return validatedResourceUpdate.getParsedResources().entrySet().stream()
- .collect(Collectors.toConcurrentMap(
- Entry::getKey, e -> e.getValue().getResourceUpdate()));
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/EdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/EdsProtocol.java
deleted file mode 100644
index 11314fc0137..00000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/EdsProtocol.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.xds.protocol.impl;
-
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-import org.apache.dubbo.xds.AdsObserver;
-import org.apache.dubbo.xds.protocol.AbstractProtocol;
-import org.apache.dubbo.xds.protocol.XdsResourceListener;
-import org.apache.dubbo.xds.resource.XdsEndpointResource;
-import org.apache.dubbo.xds.resource.XdsResourceType;
-import org.apache.dubbo.xds.resource.update.CdsUpdate;
-import org.apache.dubbo.xds.resource.update.EdsUpdate;
-import org.apache.dubbo.xds.resource.update.ValidatedResourceUpdate;
-
-import java.util.Collections;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-import io.envoyproxy.envoy.config.core.v3.Node;
-import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_PARSING_XDS;
-
-public class EdsProtocol extends AbstractProtocol {
-
- private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(EdsProtocol.class);
-
- private static final XdsEndpointResource xdsEndpointResource = XdsEndpointResource.getInstance();
-
- private XdsResourceListener clusterListener = clusters -> {
- Set clusterNames =
- clusters.stream().map(CdsUpdate::getClusterName).collect(Collectors.toSet());
- this.subscribeResource(clusterNames);
- };
-
- public EdsProtocol(AdsObserver adsObserver, Node node, int checkInterval, ApplicationModel applicationModel) {
- super(adsObserver, node, checkInterval, applicationModel);
- }
-
- @Override
- public String getTypeUrl() {
- return "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment";
- }
-
- public XdsResourceListener getCdsListener() {
- return clusterListener;
- }
-
- @Override
- protected Map decodeDiscoveryResponse(DiscoveryResponse response) {
- if (!getTypeUrl().equals(response.getTypeUrl())) {
- return Collections.emptyMap();
- }
- ValidatedResourceUpdate validatedResourceUpdate =
- xdsEndpointResource.parse(XdsResourceType.xdsResourceTypeArgs, response.getResourcesList());
- if (!validatedResourceUpdate.getErrors().isEmpty()) {
- logger.error(
- REGISTRY_ERROR_PARSING_XDS,
- validatedResourceUpdate.getErrors().toArray());
- }
- return validatedResourceUpdate.getParsedResources().entrySet().stream()
- .collect(Collectors.toConcurrentMap(
- Entry::getKey, e -> e.getValue().getResourceUpdate()));
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/LdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/LdsProtocol.java
deleted file mode 100644
index bc8cd0c5480..00000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/LdsProtocol.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.xds.protocol.impl;
-
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-import org.apache.dubbo.xds.AdsObserver;
-import org.apache.dubbo.xds.listener.LdsListener;
-import org.apache.dubbo.xds.protocol.AbstractProtocol;
-import org.apache.dubbo.xds.resource.XdsListenerResource;
-import org.apache.dubbo.xds.resource.XdsResourceType;
-import org.apache.dubbo.xds.resource.update.LdsUpdate;
-import org.apache.dubbo.xds.resource.update.ValidatedResourceUpdate;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.stream.Collectors;
-
-import io.envoyproxy.envoy.config.core.v3.Node;
-import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_PARSING_XDS;
-
-public class LdsProtocol extends AbstractProtocol {
-
- private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(LdsProtocol.class);
-
- private static final XdsListenerResource xdsListenerResource = XdsListenerResource.getInstance();
-
- public LdsProtocol(AdsObserver adsObserver, Node node, int checkInterval, ApplicationModel applicationModel) {
- super(adsObserver, node, checkInterval, applicationModel);
- List ldsListeners =
- applicationModel.getExtensionLoader(LdsListener.class).getActivateExtensions();
- ldsListeners.forEach(this::registerListen);
- }
-
- @Override
- public String getTypeUrl() {
- return "type.googleapis.com/envoy.config.listener.v3.Listener";
- }
-
- public void subscribeListeners() {
- subscribeResource(null);
- }
-
- @Override
- protected Map decodeDiscoveryResponse(DiscoveryResponse response) {
- if (!getTypeUrl().equals(response.getTypeUrl())) {
- return Collections.emptyMap();
- }
-
- if (!getTypeUrl().equals(response.getTypeUrl())) {
- return Collections.emptyMap();
- }
- ValidatedResourceUpdate validatedResourceUpdate =
- xdsListenerResource.parse(XdsResourceType.xdsResourceTypeArgs, response.getResourcesList());
- if (!validatedResourceUpdate.getErrors().isEmpty()) {
- logger.error(
- REGISTRY_ERROR_PARSING_XDS,
- validatedResourceUpdate.getErrors().toArray());
- }
- return validatedResourceUpdate.getParsedResources().entrySet().stream()
- .collect(Collectors.toConcurrentMap(
- Entry::getKey, e -> e.getValue().getResourceUpdate()));
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/RdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/RdsProtocol.java
deleted file mode 100644
index 5b84f2cb17d..00000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/RdsProtocol.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.xds.protocol.impl;
-
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-import org.apache.dubbo.xds.AdsObserver;
-import org.apache.dubbo.xds.protocol.AbstractProtocol;
-import org.apache.dubbo.xds.protocol.XdsResourceListener;
-import org.apache.dubbo.xds.resource.XdsResourceType;
-import org.apache.dubbo.xds.resource.XdsRouteConfigureResource;
-import org.apache.dubbo.xds.resource.update.LdsUpdate;
-import org.apache.dubbo.xds.resource.update.RdsUpdate;
-import org.apache.dubbo.xds.resource.update.ValidatedResourceUpdate;
-
-import java.util.Collections;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-import io.envoyproxy.envoy.config.core.v3.Node;
-import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_PARSING_XDS;
-
-public class RdsProtocol extends AbstractProtocol {
-
- private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(RdsProtocol.class);
-
- private static final XdsRouteConfigureResource xdsRouteConfigureResource = XdsRouteConfigureResource.getInstance();
-
- public RdsProtocol(AdsObserver adsObserver, Node node, int checkInterval, ApplicationModel applicationModel) {
- super(adsObserver, node, checkInterval, applicationModel);
- }
-
- @Override
- public String getTypeUrl() {
- return "type.googleapis.com/envoy.config.route.v3.RouteConfiguration";
- }
-
- @Override
- protected Map decodeDiscoveryResponse(DiscoveryResponse response) {
- if (!getTypeUrl().equals(response.getTypeUrl())) {
- return Collections.emptyMap();
- }
- ValidatedResourceUpdate updates =
- xdsRouteConfigureResource.parse(XdsResourceType.xdsResourceTypeArgs, response.getResourcesList());
- if (!updates.getInvalidResources().isEmpty()) {
- logger.error(REGISTRY_ERROR_PARSING_XDS, updates.getErrors().toArray());
- }
- return updates.getParsedResources().entrySet().stream()
- .collect(Collectors.toConcurrentMap(
- Entry::getKey, v -> v.getValue().getResourceUpdate()));
- }
-
- public XdsResourceListener getLdsListener() {
- return ldsListener;
- }
-
- private final XdsResourceListener ldsListener = resource -> {
- Set set = resource.stream()
- .flatMap(l -> l.getListener().getFilterChains().stream())
- .map(c -> c.getHttpConnectionManager().getRdsName())
- .collect(Collectors.toSet());
- this.subscribeResource(set);
- };
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsClusterResource.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsClusterResource.java
index 507244f339d..74155f67d5f 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsClusterResource.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsClusterResource.java
@@ -69,7 +69,7 @@ String typeName() {
}
@Override
- String typeUrl() {
+ public String typeUrl() {
return ADS_TYPE_URL_CDS;
}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsEndpointResource.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsEndpointResource.java
index 89915023315..b628355371a 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsEndpointResource.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsEndpointResource.java
@@ -62,7 +62,7 @@ String typeName() {
}
@Override
- String typeUrl() {
+ public String typeUrl() {
return ADS_TYPE_URL_EDS;
}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsListenerResource.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsListenerResource.java
index 74c094a7db0..6a233c436aa 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsListenerResource.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsListenerResource.java
@@ -90,7 +90,7 @@ Class unpackedClassName() {
}
@Override
- String typeUrl() {
+ public String typeUrl() {
return ADS_TYPE_URL_LDS;
}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsResourceType.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsResourceType.java
index 3690b0f73ef..2f014f636b5 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsResourceType.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsResourceType.java
@@ -72,7 +72,7 @@ public abstract class XdsResourceType {
abstract String typeName();
- abstract String typeUrl();
+ public abstract String typeUrl();
// Do not confuse with the SotW approach: it is the mechanism in which the client must specify all
// resource names it is interested in with each request. Different resource types may behave
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/matcher/PathMatcher.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/matcher/PathMatcher.java
index 30fa7228349..56d9cf64a1f 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/matcher/PathMatcher.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/matcher/PathMatcher.java
@@ -76,7 +76,7 @@ public Pattern getRegEx() {
return regEx;
}
- boolean isCaseSensitive() {
+ public boolean isCaseSensitive() {
return caseSensitive;
}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/route/RouteMatch.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/route/RouteMatch.java
index dcd11d36729..1b4512b0ba5 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/route/RouteMatch.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/route/RouteMatch.java
@@ -47,16 +47,16 @@ public RouteMatch(
this.fractionMatcher = fractionMatcher;
}
- PathMatcher pathMatcher() {
+ public PathMatcher getPathMatcher() {
return pathMatcher;
}
- List headerMatchers() {
+ public List getHeaderMatchers() {
return headerMatchers;
}
@Nullable
- FractionMatcher fractionMatcher() {
+ public FractionMatcher getFractionMatcher() {
return fractionMatcher;
}
@@ -71,11 +71,11 @@ public boolean equals(Object o) {
}
if (o instanceof RouteMatch) {
RouteMatch that = (RouteMatch) o;
- return this.pathMatcher.equals(that.pathMatcher())
- && this.headerMatchers.equals(that.headerMatchers())
+ return this.pathMatcher.equals(that.getPathMatcher())
+ && this.headerMatchers.equals(that.getHeaderMatchers())
&& (this.fractionMatcher == null
- ? that.fractionMatcher() == null
- : this.fractionMatcher.equals(that.fractionMatcher()));
+ ? that.getFractionMatcher() == null
+ : this.fractionMatcher.equals(that.getFractionMatcher()));
}
return false;
}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/update/CdsUpdate.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/update/CdsUpdate.java
index 92f2b4bceef..4f24b772d9c 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/update/CdsUpdate.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/update/CdsUpdate.java
@@ -31,13 +31,13 @@
public class CdsUpdate implements ResourceUpdate {
- enum ClusterType {
+ public enum ClusterType {
EDS,
LOGICAL_DNS,
AGGREGATE
}
- enum LbPolicy {
+ public enum LbPolicy {
ROUND_ROBIN,
RING_HASH,
LEAST_REQUEST
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouter.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouter.java
index 04917e6e0b5..64d6055bad4 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouter.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouter.java
@@ -81,7 +81,9 @@ protected BitList> doRoute(
private String matchCluster(Invocation invocation) {
String cluster = null;
String serviceName = invocation.getInvoker().getUrl().getParameter("provided-by");
- VirtualHost xdsVirtualHost = pilotExchanger.getXdsVirtualHostMap().get(serviceName);
+ // VirtualHost xdsVirtualHost = pilotExchanger.getXdsVirtualHostMap().get(serviceName);
+ // FIXME
+ VirtualHost xdsVirtualHost = xdsVirtualHostMap.get(serviceName);
// match route
for (Route xdsRoute : xdsVirtualHost.getRoutes()) {