Merge branch 'apache-3.0' into apache-3.1
# Conflicts:
# dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java
# dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java
index a9d5246..02b8f6d 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java
@@ -28,10 +28,11 @@
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.cluster.LoadBalance;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_ERROR_RESPONSE;
-import static org.apache.dubbo.rpc.Constants.ASYNC_KEY;
/**
* BroadcastClusterInvoker
@@ -68,10 +69,15 @@
int failThresholdIndex = invokers.size() * broadcastFailPercent / MAX_BROADCAST_FAIL_PERCENT;
int failIndex = 0;
- for (Invoker<T> invoker : invokers) {
+ for (int i = 0, invokersSize = invokers.size(); i < invokersSize; i++) {
+ Invoker<T> invoker = invokers.get(i);
+ RpcContext.RestoreContext restoreContext = new RpcContext.RestoreContext();
try {
- RpcInvocation subInvocation = new RpcInvocation(invocation, invoker);
- subInvocation.setAttachment(ASYNC_KEY, "true");
+ RpcInvocation subInvocation = new RpcInvocation(invocation.getTargetServiceUniqueName(),
+ invocation.getServiceModel(), invocation.getMethodName(), invocation.getServiceName(), invocation.getProtocolServiceKey(),
+ invocation.getParameterTypes(), invocation.getArguments(), invocation.copyObjectAttachments(),
+ invocation.getInvoker(), Collections.synchronizedMap(new HashMap<>(invocation.getAttributes())),
+ invocation instanceof RpcInvocation ? ((RpcInvocation) invocation).getInvokeMode() : null);
result = invokeWithContext(invoker, subInvocation);
if (null != result && result.hasException()) {
Throwable resultException = result.getException();
@@ -91,6 +97,10 @@
if (failIndex == failThresholdIndex) {
break;
}
+ } finally {
+ if (i != invokersSize - 1) {
+ restoreContext.restore();
+ }
}
}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java
index bc82f26..041ba56 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java
@@ -19,6 +19,7 @@
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.threadlocal.InternalThreadLocal;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -96,9 +97,12 @@
try {
while ((r = runQueue.poll()) != null) {
try {
+ InternalThreadLocal.removeAll();
r.run();
} catch (RuntimeException e) {
LOGGER.error("Exception while executing runnable " + r, e);
+ } finally {
+ InternalThreadLocal.removeAll();
}
}
} finally {
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 f63f5f7..4f81b5d 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
@@ -536,11 +536,7 @@
URL url = new ServiceConfigURL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName(), referenceParameters);
url = url.setScopeModel(getScopeModel());
url = url.setServiceModel(consumerModel);
- Invoker<?> withFilter = protocolSPI.refer(interfaceClass, url);
- // Local Invoke ( Support Cluster Filter / Filter )
- List<Invoker<?>> invokers = new ArrayList<>();
- invokers.add(withFilter);
- invoker = Cluster.getCluster(url.getScopeModel(), Cluster.DEFAULT).join(new StaticDirectory(url, invokers), true);
+ invoker = protocolSPI.refer(interfaceClass, url);
if (logger.isInfoEnabled()) {
logger.info("Using in jvm service " + interfaceClass.getName());
diff --git a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/ReferenceConfigTest.java b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/ReferenceConfigTest.java
index 7d34883..41943fc 100644
--- a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/ReferenceConfigTest.java
+++ b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/ReferenceConfigTest.java
@@ -477,10 +477,9 @@
.initialize();
referenceConfig.init();
- Assertions.assertTrue(referenceConfig.getInvoker() instanceof MockClusterInvoker);
- Invoker<?> withFilter = ((MockClusterInvoker<?>) referenceConfig.getInvoker()).getDirectory().getAllInvokers().get(0);
- Assertions.assertTrue(withFilter instanceof ListenerInvokerWrapper);
- Assertions.assertTrue(((ListenerInvokerWrapper<?>) withFilter).getInvoker() instanceof InjvmInvoker);
+ Invoker<?> withFilter = ((ListenerInvokerWrapper<?>) referenceConfig.getInvoker()).getInvoker();
+ withFilter = ((MockClusterInvoker<?>) withFilter).getDirectory().getAllInvokers().get(0);
+ Assertions.assertTrue(withFilter instanceof InjvmInvoker);
URL url = withFilter.getUrl();
Assertions.assertEquals("application1", url.getParameter("application"));
Assertions.assertEquals("value1", url.getParameter("key1"));
diff --git a/dubbo-dependencies-bom/pom.xml b/dubbo-dependencies-bom/pom.xml
index 4f38e14..ed43d25 100644
--- a/dubbo-dependencies-bom/pom.xml
+++ b/dubbo-dependencies-bom/pom.xml
@@ -168,7 +168,7 @@
<activation_version>1.2.0</activation_version>
<test_container_version>1.15.3</test_container_version>
<etcd_launcher_version>0.5.3</etcd_launcher_version>
- <hessian_lite_version>3.2.12</hessian_lite_version>
+ <hessian_lite_version>3.2.13</hessian_lite_version>
<swagger_version>1.5.24</swagger_version>
<snappy_java_version>1.1.8.4</snappy_java_version>
diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataUtils.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataUtils.java
index 1007cec..c5129b2 100644
--- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataUtils.java
+++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataUtils.java
@@ -137,7 +137,7 @@
Protocol protocol = applicationModel.getExtensionLoader(Protocol.class).getAdaptiveExtension();
- url.setServiceModel(consumerModel);
+ url = url.setServiceModel(consumerModel);
Invoker<MetadataService> invoker = protocol.refer(MetadataService.class, url);
diff --git a/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleRegistry.java b/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleRegistry.java
index 7a24a40..a28e8ad 100644
--- a/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleRegistry.java
+++ b/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleRegistry.java
@@ -263,7 +263,7 @@
protected static class MultipleNotifyListenerWrapper implements NotifyListener {
- Map<URL, SingleNotifyListener> registryMap = new ConcurrentHashMap<URL, SingleNotifyListener>(4);
+ Map<URL, SingleNotifyListener> registryMap = new ConcurrentHashMap<>(4);
NotifyListener sourceNotifyListener;
public MultipleNotifyListenerWrapper(NotifyListener sourceNotifyListener) {
diff --git a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java
index 076e637..fa47d05 100644
--- a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java
+++ b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java
@@ -214,6 +214,7 @@
NacosInstanceManageUtil.setCorrespondingServiceNames(serviceName, serviceNames);
}
}
+
doSubscribe(url, nacosAggregateListener, serviceNames);
}
@@ -278,13 +279,17 @@
} else {
Map<NotifyListener, NacosAggregateListener> listenerMap = originToAggregateListener.get(url);
if (listenerMap == null) {
+ logger.warn(String.format("No aggregate listener found for url %s, this service might have already been unsubscribed.", url));
return;
}
NacosAggregateListener nacosAggregateListener = listenerMap.remove(listener);
if (nacosAggregateListener != null) {
- Set<String> serviceNames = getServiceNames(url, nacosAggregateListener);
+ Set<String> serviceNames = nacosAggregateListener.getServiceNames();
try {
doUnsubscribe(url, nacosAggregateListener, serviceNames);
+ for (String serviceName : serviceNames) {
+ NacosInstanceManageUtil.removeCorrespondingServiceNames(serviceName);
+ }
} catch (NacosException e) {
logger.error("Failed to unsubscribe " + url + " to nacos " + getUrl() + ", cause: " + e.getMessage(), e);
}
diff --git a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/util/NacosInstanceManageUtil.java b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/util/NacosInstanceManageUtil.java
index cb0179f..125f22a 100644
--- a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/util/NacosInstanceManageUtil.java
+++ b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/util/NacosInstanceManageUtil.java
@@ -52,6 +52,10 @@
SERVICE_INSTANCE_LIST_MAP.put(serviceName, instanceList);
}
+ public static Set<String> removeCorrespondingServiceNames(String serviceName) {
+ return CORRESPONDING_SERVICE_NAMES_MAP.remove(serviceName);
+ }
+
public static List<Instance> getAllCorrespondingServiceInstanceList(String serviceName) {
if (!CORRESPONDING_SERVICE_NAMES_MAP.containsKey(serviceName)) {
return Lists.newArrayList();
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/dispatcher/ChannelEventRunnable.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/dispatcher/ChannelEventRunnable.java
index ffd6f2a..64a4125 100644
--- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/dispatcher/ChannelEventRunnable.java
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/dispatcher/ChannelEventRunnable.java
@@ -18,6 +18,7 @@
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.threadlocal.InternalThreadLocal;
import org.apache.dubbo.remoting.Channel;
import org.apache.dubbo.remoting.ChannelHandler;
@@ -52,6 +53,7 @@
@Override
public void run() {
+ InternalThreadLocal.removeAll();
if (state == ChannelState.RECEIVED) {
try {
handler.received(channel, message);
@@ -95,7 +97,7 @@
logger.warn("unknown state: " + state + ", message is " + message);
}
}
-
+ InternalThreadLocal.removeAll();
}
/**
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 6fc453a..deb7d52 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
@@ -133,6 +133,9 @@
return invoker.invoke(invocation);
} finally {
context.clearAfterEachInvoke(true);
+ if (context.isAsyncStarted()) {
+ removeContext();
+ }
}
}
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/RpcUtils.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/RpcUtils.java
index 8232542..7473807 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/RpcUtils.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/RpcUtils.java
@@ -259,8 +259,17 @@
return timeout;
}
+ public static long getTimeoutFromInvocation(Invocation invocation, long defaultTimeout) {
+ long timeout = defaultTimeout;
+ Object genericTimeout = invocation.getObjectAttachment(TIMEOUT_KEY);
+ if (genericTimeout != null) {
+ timeout = convertToNumber(genericTimeout, defaultTimeout);
+ }
+ return timeout;
+ }
+
private static long convertToNumber(Object obj, long defaultTimeout) {
- long timeout = 0;
+ long timeout = defaultTimeout;
try {
if (obj instanceof String) {
timeout = Long.parseLong((String) obj);
diff --git a/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmInvoker.java b/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmInvoker.java
index 499fcf8..972efa5 100644
--- a/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmInvoker.java
+++ b/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmInvoker.java
@@ -42,7 +42,6 @@
import java.lang.reflect.Type;
import java.util.HashMap;
-import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -63,7 +62,7 @@
private final String key;
- private final Map<String, Exporter<?>> exporterMap;
+ private final Exporter<?> exporter;
private final ExecutorRepository executorRepository;
@@ -71,10 +70,10 @@
private final boolean shouldIgnoreSameModule;
- InjvmInvoker(Class<T> type, URL url, String key, Map<String, Exporter<?>> exporterMap) {
+ InjvmInvoker(Class<T> type, URL url, String key, Exporter<?> exporter) {
super(type, url);
this.key = key;
- this.exporterMap = exporterMap;
+ this.exporter = exporter;
this.executorRepository = url.getOrDefaultApplicationModel().getExtensionLoader(ExecutorRepository.class).getDefaultExtension();
this.paramDeepCopyUtil = url.getOrDefaultFrameworkModel().getExtensionLoader(ParamDeepCopyUtil.class)
.getExtension(url.getParameter(CommonConstants.INJVM_COPY_UTIL_KEY, DefaultParamDeepCopyUtil.NAME));
@@ -83,7 +82,6 @@
@Override
public boolean isAvailable() {
- InjvmExporter<?> exporter = (InjvmExporter<?>) exporterMap.get(key);
if (exporter == null) {
return false;
} else {
@@ -93,7 +91,6 @@
@Override
public Result doInvoke(Invocation invocation) throws Throwable {
- Exporter<?> exporter = InjvmProtocol.getExporter(exporterMap, getUrl());
if (exporter == null) {
throw new RpcException("Service [" + key + "] not found.");
}
diff --git a/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmProtocol.java b/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmProtocol.java
index 75a51f3..fdc9e10 100644
--- a/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmProtocol.java
+++ b/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmProtocol.java
@@ -18,19 +18,29 @@
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.common.utils.UrlUtils;
import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.cluster.Cluster;
+import org.apache.dubbo.rpc.cluster.ClusterInvoker;
+import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
+import org.apache.dubbo.rpc.cluster.support.MergeableCluster;
import org.apache.dubbo.rpc.model.ScopeModel;
import org.apache.dubbo.rpc.protocol.AbstractProtocol;
import org.apache.dubbo.rpc.support.ProtocolUtils;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import static org.apache.dubbo.common.constants.CommonConstants.BROADCAST_CLUSTER;
import static org.apache.dubbo.common.constants.CommonConstants.CLUSTER_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
import static org.apache.dubbo.rpc.Constants.GENERIC_KEY;
import static org.apache.dubbo.rpc.Constants.LOCAL_PROTOCOL;
import static org.apache.dubbo.rpc.Constants.SCOPE_KEY;
@@ -69,7 +79,7 @@
if (result == null) {
return null;
} else if (ProtocolUtils.isGeneric(
- result.getInvoker().getUrl().getParameter(GENERIC_KEY))) {
+ result.getInvoker().getUrl().getParameter(GENERIC_KEY))) {
return null;
} else {
return result;
@@ -88,7 +98,15 @@
@Override
public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException {
- return new InjvmInvoker<T>(serviceType, url, url.getServiceKey(), exporterMap);
+ // group="a,b" or group="*"
+ String group = url.getParameter(GROUP_KEY);
+ if (StringUtils.isNotEmpty(group)) {
+ if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
+ return doCreateInvoker(url, Cluster.getCluster(url.getScopeModel(), MergeableCluster.NAME), serviceType);
+ }
+ }
+ Cluster cluster = Cluster.getCluster(url.getScopeModel(), url.getParameter(CLUSTER_KEY));
+ return doCreateInvoker(url, cluster, serviceType);
}
public boolean isInjvmRefer(URL url) {
@@ -116,4 +134,34 @@
return false;
}
}
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ protected <T> ClusterInvoker<T> doCreateInvoker(URL url, Cluster cluster, Class<T> type) {
+ StaticDirectory directory = new StaticDirectory(url, getInvokers(exporterMap, url, type));
+ return (ClusterInvoker<T>) cluster.join(directory, true);
+ }
+
+ private <T> List<Invoker<T>> getInvokers(Map<String, Exporter<?>> map, URL url, Class<T> type) {
+ List<Invoker<T>> result = new ArrayList<>();
+
+ if (!url.getServiceKey().contains("*")) {
+ Exporter<?> exporter = map.get(url.getServiceKey());
+ InjvmInvoker<T> invoker = new InjvmInvoker<>(type, url, url.getServiceKey(), exporter);
+ result.add(invoker);
+ } else {
+ if (CollectionUtils.isNotEmptyMap(map)) {
+ for (Exporter<?> exporter : map.values()) {
+ if (UrlUtils.isServiceKeyMatch(url, exporter.getInvoker().getUrl())) {
+ URL providerUrl = exporter.getInvoker().getUrl();
+ URL consumerUrl = url.addParameter(GROUP_KEY, providerUrl.getGroup())
+ .addParameter(VERSION_KEY, providerUrl.getVersion());
+ InjvmInvoker<T> invoker = new InjvmInvoker<>(type, consumerUrl, consumerUrl.getServiceKey(), exporter);
+ result.add(invoker);
+ }
+ }
+ }
+ }
+
+ return result;
+ }
}
diff --git a/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/Hello1ServiceImpl.java b/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/Hello1ServiceImpl.java
new file mode 100644
index 0000000..ee5b773
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/Hello1ServiceImpl.java
@@ -0,0 +1,31 @@
+/*
+ * 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.protocol.injvm;
+
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class Hello1ServiceImpl implements HelloService {
+
+ @Override
+ public List<String> hellos() {
+ List<String> res = new ArrayList<>();
+ res.add("consumer-hello-1");
+ return res;
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/Hello2ServiceImpl.java b/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/Hello2ServiceImpl.java
new file mode 100644
index 0000000..8937f50
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/Hello2ServiceImpl.java
@@ -0,0 +1,33 @@
+/*
+ * 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.protocol.injvm;
+
+
+import java.util.ArrayList;
+import java.util.List;
+
+
+public class Hello2ServiceImpl implements HelloService {
+
+ @Override
+ public List<String> hellos() {
+ List<String> res = new ArrayList<>();
+ res.add("consumer-hello-2");
+ return res;
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/HelloService.java b/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/HelloService.java
new file mode 100644
index 0000000..cd6bc45
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/HelloService.java
@@ -0,0 +1,24 @@
+/*
+ * 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.protocol.injvm;
+
+import java.util.List;
+
+public interface HelloService {
+ List<String> hellos();
+}
diff --git a/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/InjvmProtocolTest.java b/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/InjvmProtocolTest.java
index 4e57d05..d7dd373 100644
--- a/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/InjvmProtocolTest.java
+++ b/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/InjvmProtocolTest.java
@@ -29,18 +29,16 @@
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
-import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
-import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.*;
import static org.apache.dubbo.rpc.Constants.ASYNC_KEY;
import static org.apache.dubbo.rpc.Constants.GENERIC_KEY;
import static org.apache.dubbo.rpc.Constants.LOCAL_PROTOCOL;
import static org.apache.dubbo.rpc.Constants.SCOPE_KEY;
import static org.apache.dubbo.rpc.Constants.SCOPE_LOCAL;
import static org.apache.dubbo.rpc.Constants.SCOPE_REMOTE;
+import static org.apache.dubbo.rpc.Constants.MERGER_KEY;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -76,7 +74,7 @@
assertEquals(service.getSize(new String[]{"", "", ""}), 3);
service.invoke("injvm://127.0.0.1/TestService", "invoke");
- InjvmInvoker<?> injvmInvoker = new InjvmInvoker<>(DemoService.class, URL.valueOf("injvm://127.0.0.1/TestService"), null, new HashMap<>());
+ InjvmInvoker<?> injvmInvoker = new InjvmInvoker<>(DemoService.class, URL.valueOf("injvm://127.0.0.1/TestService"), null, null);
assertFalse(injvmInvoker.isAvailable());
}
@@ -137,4 +135,36 @@
assertNull(service.getAsyncResult());
}
+ @Test
+ public void testLocalProtocolForMergeResult() throws Exception {
+ HelloService helloService1 = new Hello1ServiceImpl();
+ URL url = URL.valueOf("injvm://127.0.0.1/HelloService")
+ .addParameter(INTERFACE_KEY, HelloService.class.getName())
+ .addParameter(APPLICATION_KEY, "consumer")
+ .addParameter(GROUP_KEY, "g1");
+ Invoker<?> invoker1 = proxy.getInvoker(helloService1, HelloService.class, url);
+ assertTrue(invoker1.isAvailable());
+ Exporter<?> exporter1 = protocol.export(invoker1);
+ exporters.add(exporter1);
+
+ URL url2 = URL.valueOf("injvm://127.0.0.1/HelloService")
+ .addParameter(INTERFACE_KEY, HelloService.class.getName())
+ .addParameter(APPLICATION_KEY, "consumer")
+ .addParameter(GROUP_KEY, "g2");
+ HelloService helloService2 = new Hello2ServiceImpl();
+ Invoker<?> invoker2 = proxy.getInvoker(helloService2, HelloService.class, url2);
+ assertTrue(invoker2.isAvailable());
+ Exporter<?> exporter2 = protocol.export(invoker2);
+ exporters.add(exporter2);
+
+
+ URL referUrl = URL.valueOf("injvm://127.0.0.1/HelloService")
+ .addParameter(INTERFACE_KEY, HelloService.class.getName())
+ .addParameter(APPLICATION_KEY, "consumer")
+ .addParameter(GROUP_KEY, "*")
+ .addParameter(MERGER_KEY, "list");
+ List<String> list = proxy.getProxy(protocol.refer(HelloService.class, referUrl)).hellos();
+ assertEquals(2, list.size());
+ }
+
}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
index e358eff..16b4373 100644
--- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
@@ -292,7 +292,7 @@
private int calculateTimeout(Invocation invocation, String methodName) {
if (invocation.getObjectAttachment(TIMEOUT_KEY) != null) {
- return (int) invocation.getObjectAttachment(TIMEOUT_KEY);
+ return (int) RpcUtils.getTimeoutFromInvocation(invocation, 3000);
}
Object countdown = RpcContext.getClientAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY);
int timeout;
