You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by li...@apache.org on 2019/08/23 07:43:29 UTC
[dubbo] 01/03: 1. Add ClusterInterceptor to intercept request
before service discovery. 2. Enhance multiple registry subscription to
support better grained traffic distribution. 3. Add InfraAdapter to support
extra attributes at startup.
This is an automated email from the ASF dual-hosted git repository.
liujun pushed a commit to branch cloud-native
in repository https://gitbox.apache.org/repos/asf/dubbo.git
commit 2dd3d1a0c25b6414316c5577928b4ba207c0b310
Author: ken.lj <ke...@gmail.com>
AuthorDate: Fri Aug 23 14:41:01 2019 +0800
1. Add ClusterInterceptor to intercept request before service discovery.
2. Enhance multiple registry subscription to support better grained traffic distribution.
3. Add InfraAdapter to support extra attributes at startup.
---
.../cluster/support/AbstractClusterInvoker.java | 9 --
.../dubbo/rpc/cluster/support/FailsafeCluster.java | 5 +-
.../rpc/cluster/support/MergeableCluster.java | 5 +-
.../support/RegistryAwareClusterInvoker.java | 61 ----------
.../ZoneAwareCluster.java} | 14 +--
.../support/registry/ZoneAwareClusterInvoker.java | 97 +++++++++++++++
.../cluster/support/wrapper/AbstractCluster.java | 110 +++++++++++++++++
.../internal/org.apache.dubbo.rpc.cluster.Cluster | 2 +-
.../dubbo/common/constants/RegistryConstants.java | 4 +
.../dubbo/common/extension/ExtensionLoader.java | 22 +++-
.../apache/dubbo/common/infra/InfraAdapter.java | 83 +++++++------
.../dubbo/common/infra/support/CmdbAdapter.java | 23 ++--
.../common/infra/support/EnvironmentAdapter.java | 72 +++++++++++
.../common/infra/support/KubernetesAdapter.java | 21 ++--
.../org.apache.dubbo.common.infra.InfraAdapter | 1 +
.../org/apache/dubbo/config/ApplicationConfig.java | 67 ++++++-----
.../org/apache/dubbo/config/ReferenceConfig.java | 10 +-
.../org/apache/dubbo/config/RegistryConfig.java | 17 +++
.../org/apache/dubbo/config/ServiceConfig.java | 2 +
.../src/main/resources/META-INF/compat/dubbo.xsd | 5 +
.../src/main/resources/META-INF/dubbo.xsd | 5 +
.../org/apache/dubbo/rpc/ClusterInterceptor.java | 78 ++++++------
.../main/java/org/apache/dubbo/rpc/Constants.java | 2 +
.../main/java/org/apache/dubbo/rpc/RpcContext.java | 14 ++-
.../java/org/apache/dubbo/rpc/ZoneDetector.java | 21 ++--
.../org/apache/dubbo/rpc/filter/ContextFilter.java | 2 +
.../ConsumerContextClusterInterceptor.java} | 134 +++++++++------------
.../interceptors/ZoneAwareClusterInterceptor.java | 61 ++++++++++
.../org.apache.dubbo.rpc.ClusterInterceptor | 2 +
.../dubbo/internal/org.apache.dubbo.rpc.Filter | 1 -
.../rpc/filter/ConsumerContextFilterTest.java | 53 --------
31 files changed, 647 insertions(+), 356 deletions(-)
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java
index 61ef955..37fac07 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java
@@ -27,16 +27,13 @@ import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
-import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.cluster.LoadBalance;
import org.apache.dubbo.rpc.support.RpcUtils;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.dubbo.rpc.cluster.Constants.CLUSTER_AVAILABLE_CHECK_KEY;
@@ -240,12 +237,6 @@ public abstract class AbstractClusterInvoker<T> implements Invoker<T> {
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
- // binding attachments into invocation.
- Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
- if (contextAttachments != null && contextAttachments.size() != 0) {
- ((RpcInvocation) invocation).addAttachments(contextAttachments);
- }
-
List<Invoker<T>> invokers = list(invocation);
LoadBalance loadbalance = initLoadBalance(invokers, invocation);
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailsafeCluster.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailsafeCluster.java
index eed0b5b..422335c 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailsafeCluster.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailsafeCluster.java
@@ -18,19 +18,18 @@ package org.apache.dubbo.rpc.cluster.support;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.cluster.Cluster;
import org.apache.dubbo.rpc.cluster.Directory;
/**
* {@link FailsafeClusterInvoker}
*
*/
-public class FailsafeCluster implements Cluster {
+public class FailsafeCluster extends AbstractCluster {
public final static String NAME = "failsafe";
@Override
- public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
+ protected <T> Invoker<T> doJoin(Directory<T> directory) throws RpcException {
return new FailsafeClusterInvoker<T>(directory);
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java
index 1e0f43d..ba5c79c 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java
@@ -18,15 +18,14 @@ package org.apache.dubbo.rpc.cluster.support;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.cluster.Cluster;
import org.apache.dubbo.rpc.cluster.Directory;
-public class MergeableCluster implements Cluster {
+public class MergeableCluster extends AbstractCluster {
public static final String NAME = "mergeable";
@Override
- public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
+ protected <T> Invoker<T> doJoin(Directory<T> directory) throws RpcException {
return new MergeableClusterInvoker<T>(directory);
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/RegistryAwareClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/RegistryAwareClusterInvoker.java
deleted file mode 100644
index 42fbc8a..0000000
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/RegistryAwareClusterInvoker.java
+++ /dev/null
@@ -1,61 +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.rpc.cluster.support;
-
-import org.apache.dubbo.common.logger.Logger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.rpc.Invocation;
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.Result;
-import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.cluster.Directory;
-import org.apache.dubbo.rpc.cluster.LoadBalance;
-
-import java.util.List;
-
-import static org.apache.dubbo.common.constants.CommonConstants.PREFERRED_KEY;
-import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_KEY;
-
-/**
- *
- */
-public class RegistryAwareClusterInvoker<T> extends AbstractClusterInvoker<T> {
-
- private static final Logger logger = LoggerFactory.getLogger(RegistryAwareClusterInvoker.class);
-
- public RegistryAwareClusterInvoker(Directory<T> directory) {
- super(directory);
- }
-
- @Override
- @SuppressWarnings({"unchecked", "rawtypes"})
- public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
- // First, pick the invoker (XXXClusterInvoker) that comes from the local registry, distinguished by a 'preferred' key.
- for (Invoker<T> invoker : invokers) {
- if (invoker.isAvailable() && invoker.getUrl().getParameter(REGISTRY_KEY + "." + PREFERRED_KEY, false)) {
- return invoker.invoke(invocation);
- }
- }
- // If none of the invokers has a local signal, pick the first one available.
- for (Invoker<T> invoker : invokers) {
- if (invoker.isAvailable()) {
- return invoker.invoke(invocation);
- }
- }
- throw new RpcException("No provider available in " + invokers);
- }
-}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/RegistryAwareCluster.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/registry/ZoneAwareCluster.java
similarity index 70%
rename from dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/RegistryAwareCluster.java
rename to dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/registry/ZoneAwareCluster.java
index 81b21a3..e709e03 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/RegistryAwareCluster.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/registry/ZoneAwareCluster.java
@@ -14,23 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dubbo.rpc.cluster.support;
+package org.apache.dubbo.rpc.cluster.support.registry;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.cluster.Cluster;
import org.apache.dubbo.rpc.cluster.Directory;
+import org.apache.dubbo.rpc.cluster.support.AbstractCluster;
/**
- *
+ * See {@link ZoneAwareClusterInvoker}
*/
-public class RegistryAwareCluster implements Cluster {
+public class ZoneAwareCluster extends AbstractCluster {
- public final static String NAME = "registryaware";
+ public final static String NAME = "zone-aware";
@Override
- public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
- return new RegistryAwareClusterInvoker<T>(directory);
+ protected <T> Invoker<T> doJoin(Directory<T> directory) throws RpcException {
+ return new ZoneAwareClusterInvoker<T>(directory);
}
}
\ No newline at end of file
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/registry/ZoneAwareClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/registry/ZoneAwareClusterInvoker.java
new file mode 100644
index 0000000..49ca4f7
--- /dev/null
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/registry/ZoneAwareClusterInvoker.java
@@ -0,0 +1,97 @@
+/*
+ * 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.rpc.cluster.support.registry;
+
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.cluster.Directory;
+import org.apache.dubbo.rpc.cluster.LoadBalance;
+import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
+import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_KEY;
+import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_ZONE;
+import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_ZONE_FORCE;
+import static org.apache.dubbo.common.constants.RegistryConstants.ZONE_KEY;
+
+/**
+ * When there're more than one registry for subscription.
+ * This extension provides a strategy to decide how to distribute traffics among them:
+ * 1. registry marked as 'preferred=true' has the highest priority.
+ * 2. check the zone the current request belongs, pick the registry that has the same zone first.
+ * 3. Evenly balance traffic between all registries based on each registry's weight.
+ * 4. Pick anyone that's available.
+ */
+public class ZoneAwareClusterInvoker<T> extends AbstractClusterInvoker<T> {
+
+ private static final Logger logger = LoggerFactory.getLogger(ZoneAwareClusterInvoker.class);
+
+ public ZoneAwareClusterInvoker(Directory<T> directory) {
+ super(directory);
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
+ // First, pick the invoker (XXXClusterInvoker) that comes from the local registry, distinguish by a 'default' key.
+ for (Invoker<T> invoker : invokers) {
+ if (invoker.isAvailable() && invoker.getUrl().getParameter(REGISTRY_KEY + "." + DEFAULT_KEY, false)) {
+ return invoker.invoke(invocation);
+ }
+ }
+
+ // providers in the registry with the same
+ String zone = invocation.getAttachment(REGISTRY_ZONE);
+ if (StringUtils.isNotEmpty(zone)) {
+ for (Invoker<T> invoker : invokers) {
+ if (invoker.isAvailable() && zone.equals(invoker.getUrl().getParameter(REGISTRY_KEY + "." + ZONE_KEY))) {
+ return invoker.invoke(invocation);
+ }
+ }
+ String force = invocation.getAttachment(REGISTRY_ZONE_FORCE);
+ if (StringUtils.isNotEmpty(force) && "true".equalsIgnoreCase(force)) {
+ throw new IllegalStateException("No registry instance in zone or no available providers in the registry, zone: "
+ + zone
+ + ", registries: " + invokers.stream().map(i -> i.getUrl().toString()).collect(Collectors.joining(",")));
+ }
+ }
+
+
+ // load balance among all registries, with registry weight count in.
+ Invoker<T> balancedInvoker = select(loadbalance, invocation, invokers, null);
+ if (balancedInvoker.isAvailable()) {
+ return balancedInvoker.invoke(invocation);
+ }
+
+ // If none of the invokers has a preferred signal or is picked by the loadbalancer, pick the first one available.
+ for (Invoker<T> invoker : invokers) {
+ if (invoker.isAvailable()) {
+ return invoker.invoke(invocation);
+ }
+ }
+ throw new RpcException("No provider available in " + invokers);
+ }
+
+}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/AbstractCluster.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/AbstractCluster.java
new file mode 100644
index 0000000..2ead330
--- /dev/null
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/AbstractCluster.java
@@ -0,0 +1,110 @@
+/*
+ * 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.rpc.cluster.support;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.rpc.ClusterInterceptor;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.cluster.Cluster;
+import org.apache.dubbo.rpc.cluster.Directory;
+
+import java.util.List;
+
+import static org.apache.dubbo.rpc.Constants.REFERENCE_INTERCEPTOR_KEY;
+
+public abstract class AbstractCluster implements Cluster {
+
+ private <T> Invoker<T> buildClusterInterceptors(Invoker<T> invoker, String key) {
+ Invoker<T> last = invoker;
+ List<ClusterInterceptor> interceptors = ExtensionLoader.getExtensionLoader(ClusterInterceptor.class).getActivateExtension(invoker.getUrl(), key);
+
+ if (!interceptors.isEmpty()) {
+ for (int i = interceptors.size() - 1; i >= 0; i--) {
+ final ClusterInterceptor interceptor = interceptors.get(i);
+ final Invoker<T> next = last;
+ last = new Invoker<T>() {
+
+ @Override
+ public Class<T> getInterface() {
+ return invoker.getInterface();
+ }
+
+ @Override
+ public URL getUrl() {
+ return invoker.getUrl();
+ }
+
+ @Override
+ public boolean isAvailable() {
+ return invoker.isAvailable();
+ }
+
+ @Override
+ public Result invoke(Invocation invocation) throws RpcException {
+ Result asyncResult;
+ try {
+ interceptor.before(next, invocation);
+ asyncResult = interceptor.invoke(next, invocation);
+ } catch (Exception e) {
+ // onError callback
+ if (interceptor instanceof ClusterInterceptor.Listener) {
+ ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor;
+ listener.onError(e, invoker, invocation);
+ }
+ throw e;
+ } finally {
+ interceptor.after(next, invocation);
+ }
+ return asyncResult.whenCompleteWithContext((r, t) -> {
+ // onResponse callback
+ if (interceptor instanceof ClusterInterceptor.Listener) {
+ ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor;
+ if (t == null) {
+ listener.onResponse(r, invoker, invocation);
+ } else {
+ listener.onError(t, invoker, invocation);
+ }
+ }
+ });
+ }
+
+ @Override
+ public void destroy() {
+ invoker.destroy();
+ }
+
+ @Override
+ public String toString() {
+ return invoker.toString();
+ }
+ };
+ }
+ }
+ return last;
+ }
+
+ @Override
+ public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
+ return buildClusterInterceptors(doJoin(directory), directory.getUrl().getParameter(REFERENCE_INTERCEPTOR_KEY));
+ }
+
+ protected abstract <T> Invoker<T> doJoin(Directory<T> directory) throws RpcException;
+}
diff --git a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster
index ba30247..5808d13 100644
--- a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster
+++ b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster
@@ -7,4 +7,4 @@ forking=org.apache.dubbo.rpc.cluster.support.ForkingCluster
available=org.apache.dubbo.rpc.cluster.support.AvailableCluster
mergeable=org.apache.dubbo.rpc.cluster.support.MergeableCluster
broadcast=org.apache.dubbo.rpc.cluster.support.BroadcastCluster
-registryaware=org.apache.dubbo.rpc.cluster.support.RegistryAwareCluster
\ No newline at end of file
+zone-aware=org.apache.dubbo.rpc.cluster.support.registry.ZoneAwareCluster
\ No newline at end of file
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/RegistryConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/RegistryConstants.java
index cd10a11..ce9dc12 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/RegistryConstants.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/RegistryConstants.java
@@ -90,4 +90,8 @@ public interface RegistryConstants {
int DEFAULT_INSTANCES_REQUEST_SIZE = 100;
String ACCEPTS_KEY = "accepts";
+
+ String REGISTRY_ZONE = "registry_zone";
+ String REGISTRY_ZONE_FORCE = "registry_zone_force";
+ String ZONE_KEY = "zone";
}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/extension/ExtensionLoader.java b/dubbo-common/src/main/java/org/apache/dubbo/common/extension/ExtensionLoader.java
index 4575dbb..7b6d482 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/extension/ExtensionLoader.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/extension/ExtensionLoader.java
@@ -39,6 +39,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -274,11 +275,19 @@ public class ExtensionLoader<T> {
return true;
}
for (String key : keys) {
+ // @Active(value="key1:value1, key2:value2")
+ String keyValue = null;
+ if (key.contains(":")) {
+ String[] arr = key.split(":");
+ key = arr[0];
+ keyValue = arr[1];
+ }
+
for (Map.Entry<String, String> entry : url.getParameters().entrySet()) {
String k = entry.getKey();
String v = entry.getValue();
if ((k.equals(key) || k.endsWith("." + key))
- && ConfigUtils.isNotEmpty(v)) {
+ && ((keyValue != null && keyValue.equals(v)) || (keyValue == null && ConfigUtils.isNotEmpty(v)))) {
return true;
}
}
@@ -387,6 +396,17 @@ public class ExtensionLoader<T> {
return Collections.unmodifiableSet(new TreeSet<>(clazzes.keySet()));
}
+ public Set<T> getSupportedExtensionInstances() {
+ Set<T> instances = new HashSet<>();
+ Set<String> supportedExtensions = getSupportedExtensions();
+ if (CollectionUtils.isNotEmpty(supportedExtensions)) {
+ for (String name : supportedExtensions) {
+ instances.add(getExtension(name));
+ }
+ }
+ return instances;
+ }
+
/**
* Return default extension name, return <code>null</code> if not configured.
*/
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailsafeCluster.java b/dubbo-common/src/main/java/org/apache/dubbo/common/infra/InfraAdapter.java
similarity index 52%
copy from dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailsafeCluster.java
copy to dubbo-common/src/main/java/org/apache/dubbo/common/infra/InfraAdapter.java
index eed0b5b..b1dc16d 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailsafeCluster.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/infra/InfraAdapter.java
@@ -1,37 +1,46 @@
-/*
- * 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.rpc.cluster.support;
-
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.cluster.Cluster;
-import org.apache.dubbo.rpc.cluster.Directory;
-
-/**
- * {@link FailsafeClusterInvoker}
- *
- */
-public class FailsafeCluster implements Cluster {
-
- public final static String NAME = "failsafe";
-
- @Override
- public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
- return new FailsafeClusterInvoker<T>(directory);
- }
-
-}
+/*
+ * 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.common.infra;
+
+import org.apache.dubbo.common.extension.SPI;
+
+import java.util.Map;
+
+/**
+ * Used to interact with other systems. Typical use cases:
+ * 1. get extra attributes related to the instance on which Dubbo is deploying from underlying third-party systems or infrastructures.
+ * 2. get configurations from third-party systems which maybe useful for a specific component.
+ */
+
+@SPI
+public interface InfraAdapter {
+
+ /**
+ * get extra attributes
+ *
+ * @param params application name or hostname are most likely to be used as input params.
+ * @return
+ */
+ Map<String, String> getExtraAttributes(Map<String, String> params);
+
+ /**
+ * @param key
+ * @return
+ */
+ String getAttribute(String key);
+
+}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java b/dubbo-common/src/main/java/org/apache/dubbo/common/infra/support/CmdbAdapter.java
similarity index 64%
copy from dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java
copy to dubbo-common/src/main/java/org/apache/dubbo/common/infra/support/CmdbAdapter.java
index 1e0f43d..8b20895 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/infra/support/CmdbAdapter.java
@@ -14,20 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dubbo.rpc.cluster.support;
+package org.apache.dubbo.common.infra.support;
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.cluster.Cluster;
-import org.apache.dubbo.rpc.cluster.Directory;
+import org.apache.dubbo.common.infra.InfraAdapter;
-public class MergeableCluster implements Cluster {
+import java.util.Map;
- public static final String NAME = "mergeable";
+public class CmdbAdapter implements InfraAdapter {
+
+ public CmdbAdapter() {
+ // init;
+ }
@Override
- public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
- return new MergeableClusterInvoker<T>(directory);
+ public Map<String, String> getExtraAttributes(Map<String, String> params) {
+ return null;
}
+ @Override
+ public String getAttribute(String key) {
+ return null;
+ }
}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/infra/support/EnvironmentAdapter.java b/dubbo-common/src/main/java/org/apache/dubbo/common/infra/support/EnvironmentAdapter.java
new file mode 100644
index 0000000..4a38446
--- /dev/null
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/infra/support/EnvironmentAdapter.java
@@ -0,0 +1,72 @@
+/*
+ * 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.common.infra.support;
+
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.common.infra.InfraAdapter;
+import org.apache.dubbo.common.utils.StringUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
+import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_ENV_KEYS;
+import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_LABELS;
+import static org.apache.dubbo.common.constants.CommonConstants.EQUAL_SPLIT_PATTERN;
+import static org.apache.dubbo.common.constants.CommonConstants.SEMICOLON_SPLIT_PATTERN;
+
+@Activate
+public class EnvironmentAdapter implements InfraAdapter {
+
+ /**
+ * 1. OS Environment: DUBBO_LABELS=tag=pre;key=value
+ * 2. JVM Options: -Denv_keys = DUBBO_KEY1, DUBBO_KEY2
+ */
+ @Override
+ public Map<String, String> getExtraAttributes(Map<String, String> params) {
+ Map<String, String> parameters = new HashMap<>();
+
+ String rawLabels = ConfigurationUtils.getProperty(DUBBO_LABELS);
+ if (StringUtils.isNotEmpty(rawLabels)) {
+ String[] labelPairs = SEMICOLON_SPLIT_PATTERN.split(rawLabels);
+ for (String pair : labelPairs) {
+ String[] label = EQUAL_SPLIT_PATTERN.split(pair);
+ if (label.length == 2) {
+ parameters.put(label[0], label[1]);
+ }
+ }
+ }
+
+ String rawKeys = ConfigurationUtils.getProperty(DUBBO_ENV_KEYS);
+ if (StringUtils.isNotEmpty(rawKeys)) {
+ String[] keys = COMMA_SPLIT_PATTERN.split(rawKeys);
+ for (String key : keys) {
+ String value = ConfigurationUtils.getProperty(key);
+ if (value != null) {
+ parameters.put(key, value);
+ }
+ }
+ }
+ return parameters;
+ }
+
+ @Override
+ public String getAttribute(String key) {
+ return ConfigurationUtils.getProperty(key);
+ }
+}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java b/dubbo-common/src/main/java/org/apache/dubbo/common/infra/support/KubernetesAdapter.java
similarity index 64%
copy from dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java
copy to dubbo-common/src/main/java/org/apache/dubbo/common/infra/support/KubernetesAdapter.java
index 1e0f43d..0d5afc3 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/infra/support/KubernetesAdapter.java
@@ -14,20 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dubbo.rpc.cluster.support;
+package org.apache.dubbo.common.infra.support;
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.cluster.Cluster;
-import org.apache.dubbo.rpc.cluster.Directory;
+import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.common.infra.InfraAdapter;
-public class MergeableCluster implements Cluster {
+import java.util.Map;
- public static final String NAME = "mergeable";
+@Activate
+public class KubernetesAdapter implements InfraAdapter {
@Override
- public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
- return new MergeableClusterInvoker<T>(directory);
+ public Map<String, String> getExtraAttributes(Map<String, String> params) {
+ return null;
}
+ @Override
+ public String getAttribute(String key) {
+ return null;
+ }
}
diff --git a/dubbo-common/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.infra.InfraAdapter b/dubbo-common/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.infra.InfraAdapter
new file mode 100644
index 0000000..743027a
--- /dev/null
+++ b/dubbo-common/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.infra.InfraAdapter
@@ -0,0 +1 @@
+environment=org.apache.dubbo.common.infra.support.EnvironmentAdapter
\ No newline at end of file
diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ApplicationConfig.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ApplicationConfig.java
index 57b006f..bc5b137 100644
--- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ApplicationConfig.java
+++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ApplicationConfig.java
@@ -17,24 +17,25 @@
package org.apache.dubbo.config;
import org.apache.dubbo.common.compiler.support.AdaptiveCompiler;
-import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.infra.InfraAdapter;
+import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.support.Parameter;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import static org.apache.dubbo.common.constants.CommonConstants.APPLICATION_KEY;
-import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
-import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_ENV_KEYS;
-import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_LABELS;
import static org.apache.dubbo.common.constants.CommonConstants.DUMP_DIRECTORY;
-import static org.apache.dubbo.common.constants.CommonConstants.EQUAL_SPLIT_PATTERN;
-import static org.apache.dubbo.common.constants.CommonConstants.SEMICOLON_SPLIT_PATTERN;
+import static org.apache.dubbo.common.constants.CommonConstants.HOST_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.SHUTDOWN_WAIT_KEY;
import static org.apache.dubbo.common.constants.QosConstants.ACCEPT_FOREIGN_IP;
import static org.apache.dubbo.common.constants.QosConstants.ACCEPT_FOREIGN_IP_COMPATIBLE;
@@ -58,6 +59,7 @@ import static org.apache.dubbo.config.Constants.TEST_ENVIRONMENT;
* @export
*/
public class ApplicationConfig extends AbstractConfig {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ApplicationConfig.class);
private static final long serialVersionUID = 5508512956753757169L;
@@ -147,6 +149,8 @@ public class ApplicationConfig extends AbstractConfig {
*/
private String shutwait;
+ private String hostname;
+
/**
* Metadata type, local or remote, if choose remote, you need to further specify metadata center.
*/
@@ -384,6 +388,19 @@ public class ApplicationConfig extends AbstractConfig {
this.shutwait = shutwait;
}
+ @Parameter(excluded = true)
+ public String getHostname() {
+ if (hostname == null) {
+ try {
+ hostname = InetAddress.getLocalHost().getHostName();
+ } catch (UnknownHostException e) {
+ LOGGER.warn("Failed to get the hostname of current instance.", e);
+ hostname = "UNKNOWN";
+ }
+ }
+ return hostname;
+ }
+
@Override
@Parameter(excluded = true)
public boolean isValid() {
@@ -412,38 +429,22 @@ public class ApplicationConfig extends AbstractConfig {
appendEnvironmentProperties();
}
- /**
- * 1. OS Environment: DUBBO_LABELS=tag=pre;key=value
- * 2. JVM Options: -Denv_keys = DUBBO_KEY1, DUBBO_KEY2
- */
private void appendEnvironmentProperties() {
- String rawLabels = ConfigurationUtils.getProperty(DUBBO_LABELS);
- if (StringUtils.isNotEmpty(rawLabels)) {
- String[] labelPairs = SEMICOLON_SPLIT_PATTERN.split(rawLabels);
- for (String pair : labelPairs) {
- String[] label = EQUAL_SPLIT_PATTERN.split(pair);
- if (label.length == 2) {
- updateParameters(parameters, label[0], label[1]);
- }
- }
+ if (parameters == null) {
+ parameters = new HashMap<>();
}
- String rawKeys = ConfigurationUtils.getProperty(DUBBO_ENV_KEYS);
- if (StringUtils.isNotEmpty(rawKeys)) {
- String[] keys = COMMA_SPLIT_PATTERN.split(rawKeys);
- for (String key : keys) {
- String value = ConfigurationUtils.getProperty(key);
- if (value != null) {
- updateParameters(parameters, key, value);
+ Set<InfraAdapter> adapters = ExtensionLoader.getExtensionLoader(InfraAdapter.class).getSupportedExtensionInstances();
+ if (CollectionUtils.isNotEmpty(adapters)) {
+ Map<String, String> inputParameters = new HashMap<>();
+ inputParameters.put(APPLICATION_KEY, getName());
+ inputParameters.put(HOST_KEY, getHostname());
+ for (InfraAdapter adapter : adapters) {
+ Map<String, String> extraParameters = adapter.getExtraAttributes(inputParameters);
+ if (CollectionUtils.isNotEmptyMap(extraParameters)) {
+ parameters.putAll(extraParameters);
}
}
}
}
-
- private void updateParameters(Map<String, String> map, String key, String value) {
- if (parameters == null) {
- parameters = new HashMap<>();
- }
- parameters.put(key, value);
- }
}
diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java
index 3169aa8..5a72c7a 100644
--- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java
+++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java
@@ -41,7 +41,7 @@ import org.apache.dubbo.rpc.ProxyFactory;
import org.apache.dubbo.rpc.cluster.Cluster;
import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
import org.apache.dubbo.rpc.cluster.support.ClusterUtils;
-import org.apache.dubbo.rpc.cluster.support.RegistryAwareCluster;
+import org.apache.dubbo.rpc.cluster.support.registry.ZoneAwareCluster;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ConsumerModel;
import org.apache.dubbo.rpc.protocol.injvm.InjvmProtocol;
@@ -74,7 +74,7 @@ import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.REVISION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.SEMICOLON_SPLIT_PATTERN;
import static org.apache.dubbo.common.constants.CommonConstants.SIDE_KEY;
-import static org.apache.dubbo.common.utils.NetUtils.isInvalidLocalHost;
+import static org.apache.dubbo.common.constants.CommonConstants.CLUSTER_KEY;
import static org.apache.dubbo.config.Constants.DUBBO_IP_TO_REGISTRY;
import static org.apache.dubbo.registry.Constants.CONSUMER_PROTOCOL;
import static org.apache.dubbo.registry.Constants.REGISTER_IP_KEY;
@@ -423,9 +423,9 @@ public class ReferenceConfig<T> extends AbstractReferenceConfig {
}
}
if (registryURL != null) { // registry url is available
- // use RegistryAwareCluster only when register's CLUSTER is available
- URL u = registryURL.addParameter(CLUSTER_KEY, RegistryAwareCluster.NAME);
- // The invoker wrap relation would be: RegistryAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, will execute route) -> Invoker
+ // for multi-subscription scenario, use 'zone-aware' policy by default
+ URL u = registryURL.addParameterIfAbsent(CLUSTER_KEY, ZoneAwareCluster.NAME);
+ // The invoker wrap relation would be like: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
invoker = CLUSTER.join(new StaticDirectory(u, invokers));
} else { // not a registry url, must be direct invoke.
invoker = CLUSTER.join(new StaticDirectory(invokers));
diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/RegistryConfig.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/RegistryConfig.java
index cd81def..41e1857 100644
--- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/RegistryConfig.java
+++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/RegistryConfig.java
@@ -76,9 +76,18 @@ public class RegistryConfig extends AbstractConfig {
private String client;
+ /**
+ * Affects how traffic distributes among registries, useful when subscribing multiple registries, available options:
+ * 1. zone-aware, a certain type of traffic always goes to one Registry according to where the traffic is originated.
+ */
private String cluster;
/**
+ * The region where the registry belongs, usually used to isolate traffics
+ */
+ private String zone;
+
+ /**
* The group the services registry in
*/
private String group;
@@ -381,6 +390,14 @@ public class RegistryConfig extends AbstractConfig {
this.cluster = cluster;
}
+ public String getZone() {
+ return zone;
+ }
+
+ public void setZone(String zone) {
+ this.zone = zone;
+ }
+
public String getGroup() {
return group;
}
diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
index c7b8f75..c2560e2 100644
--- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
+++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
@@ -87,6 +87,7 @@ import static org.apache.dubbo.config.Constants.DUBBO_IP_TO_REGISTRY;
import static org.apache.dubbo.config.Constants.DUBBO_PORT_TO_BIND;
import static org.apache.dubbo.config.Constants.DUBBO_PORT_TO_REGISTRY;
import static org.apache.dubbo.config.Constants.MULTICAST;
+import static org.apache.dubbo.config.Constants.PROTOCOLS_SUFFIX;
import static org.apache.dubbo.config.Constants.SCOPE_NONE;
import static org.apache.dubbo.rpc.Constants.GENERIC_KEY;
import static org.apache.dubbo.rpc.Constants.LOCAL_PROTOCOL;
@@ -575,6 +576,7 @@ public class ServiceConfig<T> extends AbstractServiceConfig {
Integer port = this.findConfigedPorts(protocolConfig, name, map);
URL url = new URL(name, host, port, getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), map);
+ // You can customize Configurator to append extra parameters
if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
.hasExtension(url.getProtocol())) {
url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
diff --git a/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/compat/dubbo.xsd b/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/compat/dubbo.xsd
index 893cc5e..0966179 100644
--- a/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/compat/dubbo.xsd
+++ b/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/compat/dubbo.xsd
@@ -512,6 +512,11 @@
<xsd:documentation><![CDATA[ The registry cluster type. ]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
+ <xsd:attribute name="zone" type="xsd:string">
+ <xsd:annotation>
+ <xsd:documentation><![CDATA[ The registry zone type. ]]></xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
<xsd:attribute name="forks" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[ ForkingCluster forks. ]]></xsd:documentation>
diff --git a/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd b/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd
index e538751..29ea4f2 100644
--- a/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd
+++ b/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd
@@ -506,6 +506,11 @@
<xsd:documentation><![CDATA[ The registry cluster type. ]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
+ <xsd:attribute name="zone" type="xsd:string">
+ <xsd:annotation>
+ <xsd:documentation><![CDATA[ The registry zone type. ]]></xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
<xsd:attribute name="forks" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[ ForkingCluster forks. ]]></xsd:documentation>
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailsafeCluster.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/ClusterInterceptor.java
similarity index 54%
copy from dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailsafeCluster.java
copy to dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/ClusterInterceptor.java
index eed0b5b..a90b500 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailsafeCluster.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/ClusterInterceptor.java
@@ -1,37 +1,41 @@
-/*
- * 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.rpc.cluster.support;
-
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.cluster.Cluster;
-import org.apache.dubbo.rpc.cluster.Directory;
-
-/**
- * {@link FailsafeClusterInvoker}
- *
- */
-public class FailsafeCluster implements Cluster {
-
- public final static String NAME = "failsafe";
-
- @Override
- public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
- return new FailsafeClusterInvoker<T>(directory);
- }
-
-}
+/*
+ * 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.rpc;
+
+import org.apache.dubbo.common.extension.SPI;
+
+/**
+ * Different from {@link Filter}, ClusterInterceptor works on the outmost layer, before one specific address/invoker is picked.
+ */
+@SPI
+public interface ClusterInterceptor {
+
+ void before(Invoker<?> invoker, Invocation invocation);
+
+ void after(Invoker<?> invoker, Invocation invocation);
+
+ default Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
+ return invoker.invoke(invocation);
+ }
+
+ interface Listener {
+
+ void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation);
+
+ void onError(Throwable t, Invoker<?> invoker, Invocation invocation);
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java
index f515960..180d69d 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java
@@ -70,6 +70,8 @@ public interface Constants {
String REFERENCE_FILTER_KEY = "reference.filter";
+ String REFERENCE_INTERCEPTOR_KEY = "reference.interceptor";
+
String INVOKER_LISTENER_KEY = "invoker.listener";
String SERVICE_FILTER_KEY = "service.filter";
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java
index 1066c7d..954ce92 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java
@@ -105,6 +105,8 @@ public class RpcContext {
private Object response;
private AsyncContext asyncContext;
+ private boolean remove = true;
+
protected RpcContext() {
}
@@ -140,6 +142,14 @@ public class RpcContext {
return LOCAL.get();
}
+ public boolean canRemove() {
+ return remove;
+ }
+
+ public void clearAfterEachInvoke(boolean remove) {
+ this.remove = remove;
+ }
+
public static void restoreContext(RpcContext oldContext) {
LOCAL.set(oldContext);
}
@@ -150,7 +160,9 @@ public class RpcContext {
* @see org.apache.dubbo.rpc.filter.ContextFilter
*/
public static void removeContext() {
- LOCAL.remove();
+ if (LOCAL.get().canRemove()) {
+ LOCAL.remove();
+ }
}
/**
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/ZoneDetector.java
similarity index 63%
copy from dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java
copy to dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/ZoneDetector.java
index 1e0f43d..934d9d5 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/MergeableCluster.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/ZoneDetector.java
@@ -14,20 +14,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dubbo.rpc.cluster.support;
+package org.apache.dubbo.rpc;
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.cluster.Cluster;
-import org.apache.dubbo.rpc.cluster.Directory;
+import org.apache.dubbo.common.extension.SPI;
-public class MergeableCluster implements Cluster {
+/**
+ * Extend and provide your owen implementation if you want to distribute traffic around registries.
+ * Please, name it as 'default'
+ */
+@SPI
+public interface ZoneDetector {
- public static final String NAME = "mergeable";
+ String getZoneOfCurrentRequest(Invocation invocation);
- @Override
- public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
- return new MergeableClusterInvoker<T>(directory);
- }
+ String isZoneForcingEnabled(Invocation invocation, String zone);
}
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java
index 51e22b9..cca3979 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java
@@ -93,8 +93,10 @@ public class ContextFilter extends ListenableFilter {
((RpcInvocation) invocation).setInvoker(invoker);
}
try {
+ RpcContext.getContext().clearAfterEachInvoke(false);
return invoker.invoke(invocation);
} finally {
+ RpcContext.getContext().clearAfterEachInvoke(true);
// IMPORTANT! For async scenario, we must remove context from current thread, so we always create a new RpcContext for the next invoke for the same thread.
RpcContext.removeContext();
RpcContext.removeServerContext();
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ConsumerContextFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/interceptors/ConsumerContextClusterInterceptor.java
similarity index 52%
rename from dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ConsumerContextFilter.java
rename to dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/interceptors/ConsumerContextClusterInterceptor.java
index 43353f7..3b81b1a 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ConsumerContextFilter.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/interceptors/ConsumerContextClusterInterceptor.java
@@ -1,75 +1,59 @@
-/*
- * 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.rpc.filter;
-
-import org.apache.dubbo.common.extension.Activate;
-import org.apache.dubbo.common.utils.NetUtils;
-import org.apache.dubbo.rpc.Invocation;
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.ListenableFilter;
-import org.apache.dubbo.rpc.Result;
-import org.apache.dubbo.rpc.RpcContext;
-import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.RpcInvocation;
-
-import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER;
-
-/**
- * ConsumerContextFilter set current RpcContext with invoker,invocation, local host, remote host and port
- * for consumer invoker.It does it to make the requires info available to execution thread's RpcContext.
- *
- * @see org.apache.dubbo.rpc.Filter
- * @see RpcContext
- */
-@Activate(group = CONSUMER, order = -10000)
-public class ConsumerContextFilter extends ListenableFilter {
-
- public ConsumerContextFilter() {
- super.listener = new ConsumerContextListener();
- }
-
- @Override
- public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
- RpcContext.getContext()
- .setInvoker(invoker)
- .setInvocation(invocation)
- .setLocalAddress(NetUtils.getLocalHost(), 0)
- .setRemoteAddress(invoker.getUrl().getHost(),
- invoker.getUrl().getPort());
- if (invocation instanceof RpcInvocation) {
- ((RpcInvocation) invocation).setInvoker(invoker);
- }
- try {
- RpcContext.removeServerContext();
- return invoker.invoke(invocation);
- } finally {
- RpcContext.removeContext();
- }
- }
-
- static class ConsumerContextListener implements Listener {
- @Override
- public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
- RpcContext.getServerContext().setAttachments(appResponse.getAttachments());
- }
-
- @Override
- public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
-
- }
- }
-}
+/*
+ * 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.rpc.interceptors;
+
+import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.common.utils.NetUtils;
+import org.apache.dubbo.rpc.ClusterInterceptor;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.RpcInvocation;
+
+@Activate
+public class ConsumerContextClusterInterceptor implements ClusterInterceptor, ClusterInterceptor.Listener {
+
+ @Override
+ public void before(Invoker<?> invoker, Invocation invocation) {
+ RpcContext.getContext()
+ .setInvoker(invoker)
+ .setInvocation(invocation)
+ .setLocalAddress(NetUtils.getLocalHost(), 0)
+ .setRemoteAddress(invoker.getUrl().getHost(),
+ invoker.getUrl().getPort());
+ if (invocation instanceof RpcInvocation) {
+ ((RpcInvocation) invocation).setInvoker(invoker);
+ }
+ RpcContext.removeServerContext();
+ }
+
+ @Override
+ public void after(Invoker<?> invoker, Invocation invocation) {
+ RpcContext.removeContext();
+ }
+
+ @Override
+ public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
+ RpcContext.getServerContext().setAttachments(appResponse.getAttachments());
+ }
+
+ @Override
+ public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
+
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/interceptors/ZoneAwareClusterInterceptor.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/interceptors/ZoneAwareClusterInterceptor.java
new file mode 100644
index 0000000..2fdc42e
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/interceptors/ZoneAwareClusterInterceptor.java
@@ -0,0 +1,61 @@
+/*
+ * 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.rpc.interceptors;
+
+import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.rpc.ClusterInterceptor;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.ZoneDetector;
+
+import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_ZONE;
+import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_ZONE_FORCE;
+
+/**
+ * Determines the zone information of current request.
+ */
+@Activate(value = "cluster:zone-aware")
+public class ZoneAwareClusterInterceptor implements ClusterInterceptor {
+
+ @Override
+ public void before(Invoker<?> invoker, Invocation invocation) {
+ RpcContext rpcContext = RpcContext.getContext();
+ String zone = rpcContext.getAttachment(REGISTRY_ZONE);
+ String force = rpcContext.getAttachment(REGISTRY_ZONE_FORCE);
+ ExtensionLoader<ZoneDetector> loader = ExtensionLoader.getExtensionLoader(ZoneDetector.class);
+ if (StringUtils.isEmpty(zone) && loader.hasExtension("default")) {
+ ZoneDetector detector = loader.getExtension("default");
+ zone = detector.getZoneOfCurrentRequest(invocation);
+ force = detector.isZoneForcingEnabled(invocation, zone);
+ }
+
+ if (StringUtils.isNotEmpty(zone)) {
+ invocation.setAttachment(REGISTRY_ZONE, zone);
+ }
+ if (StringUtils.isNotEmpty(force)) {
+ invocation.setAttachment(REGISTRY_ZONE_FORCE, force);
+ }
+ }
+
+ @Override
+ public void after(Invoker<?> invoker, Invocation invocation) {
+
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.ClusterInterceptor b/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.ClusterInterceptor
new file mode 100644
index 0000000..2b60b1a
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.ClusterInterceptor
@@ -0,0 +1,2 @@
+consumer-context=org.apache.dubbo.rpc.interceptors.ConsumerContextClusterInterceptor
+zone-aware=org.apache.dubbo.rpc.interceptors.ZoneAwareClusterInterceptor
\ No newline at end of file
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter b/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter
index 376f966..1cc6181 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter
@@ -6,7 +6,6 @@ accesslog=org.apache.dubbo.rpc.filter.AccessLogFilter
activelimit=org.apache.dubbo.rpc.filter.ActiveLimitFilter
classloader=org.apache.dubbo.rpc.filter.ClassLoaderFilter
context=org.apache.dubbo.rpc.filter.ContextFilter
-consumercontext=org.apache.dubbo.rpc.filter.ConsumerContextFilter
exception=org.apache.dubbo.rpc.filter.ExceptionFilter
executelimit=org.apache.dubbo.rpc.filter.ExecuteLimitFilter
deprecated=org.apache.dubbo.rpc.filter.DeprecatedFilter
diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ConsumerContextFilterTest.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ConsumerContextFilterTest.java
deleted file mode 100644
index baf5752..0000000
--- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ConsumerContextFilterTest.java
+++ /dev/null
@@ -1,53 +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.rpc.filter;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.utils.NetUtils;
-import org.apache.dubbo.rpc.Filter;
-import org.apache.dubbo.rpc.Invocation;
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.Result;
-import org.apache.dubbo.rpc.RpcContext;
-import org.apache.dubbo.rpc.support.DemoService;
-import org.apache.dubbo.rpc.support.MockInvocation;
-import org.apache.dubbo.rpc.support.MyInvoker;
-
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-/**
- * ConsumerContextFilterTest.java
- */
-public class ConsumerContextFilterTest {
- Filter consumerContextFilter = new ConsumerContextFilter();
-
- @Test
- public void testSetContext() {
- URL url = URL.valueOf("test://test:11/test?group=dubbo&version=1.1");
- Invoker<DemoService> invoker = new MyInvoker<DemoService>(url);
- Invocation invocation = new MockInvocation();
- Result asyncResult = consumerContextFilter.invoke(invoker, invocation);
- asyncResult.whenCompleteWithContext((result, t) -> {
- assertEquals(invoker, RpcContext.getContext().getInvoker());
- assertEquals(invocation, RpcContext.getContext().getInvocation());
- assertEquals(NetUtils.getLocalHost() + ":0", RpcContext.getContext().getLocalAddressString());
- assertEquals("test:11", RpcContext.getContext().getRemoteAddressString());
- });
- }
-}
\ No newline at end of file