From e261acc98ac0d46ed3239101f1871f9b08e4f136 Mon Sep 17 00:00:00 2001 From: "ken.lj" Date: Mon, 2 Sep 2019 19:18:07 +0800 Subject: [PATCH] Solve conflicts after merged 3.x --- .../rpc/cluster/router/tag/TagRouter.java | 4 +- .../config/OrderedPropertiesProvider.java | 27 +- .../common/constants/CommonConstants.java | 8 + .../manager/DefaultExecutorRepository.java | 16 +- .../MockOrderedPropertiesProvider1.java | 18 +- .../apache/dubbo/config/ReferenceConfig.java | 27 +- .../apache/dubbo/config/ServiceConfig.java | 35 +- .../dubbo-demo-xml-consumer/pom.xml | 4 + .../dubbo-demo-xml-provider/pom.xml | 4 + .../mock/JValidatorTestTarget.java | 6 - .../dubbo/monitor/support/MonitorFilter.java | 10 +- .../exchange/support/DefaultFuture.java | 44 +- .../support/header/HeaderExchangeChannel.java | 5 +- .../org/apache/dubbo/rpc/AsyncRpcResult.java | 5 + .../org/apache/dubbo/rpc/RpcInvocation.java | 4 +- .../dubbo/rpc/filter/ActiveLimitFilter.java | 2 +- .../dubbo/rpc/filter/ContextFilter.java | 4 +- .../dubbo/rpc/filter/ExecuteLimitFilter.java | 2 +- .../dubbo/rpc/filter/GenericFilter.java | 10 +- .../dubbo/rpc/filter/TimeoutFilter.java | 2 +- .../apache/dubbo/rpc/filter/TokenFilter.java | 4 +- .../dubbo/rpc/model/ConsumerMethodModel.java | 32 +- .../apache/dubbo/rpc/model/ConsumerModel.java | 35 +- .../dubbo/rpc/protocol/AbstractInvoker.java | 4 +- .../apache/dubbo/rpc/support/RpcUtils.java | 6 +- .../dubbo/rpc/filter/TokenFilterTest.java | 7 +- .../protocol/dubbo/ImplicitCallBackTest.java | 807 +++++++++--------- .../hessian/HttpClientConnectionFactory.java | 8 +- .../dubbo/rpc/protocol/rest/RestProtocol.java | 2 +- .../dubbo/rpc/protocol/rmi/RmiProtocol.java | 2 +- dubbo-rpc/dubbo-rpc-rsocket/pom.xml | 123 --- .../protocol/rsocket/FutureSubscriber.java | 98 --- .../rpc/protocol/rsocket/MetadataCodec.java | 36 - .../protocol/rsocket/RSocketConstants.java | 32 - .../rpc/protocol/rsocket/RSocketExporter.java | 43 - .../rpc/protocol/rsocket/RSocketInvoker.java | 263 ------ .../rpc/protocol/rsocket/RSocketProtocol.java | 727 ---------------- .../protocol/rsocket/ResourceDirectory.java | 62 -- .../rpc/protocol/rsocket/ResourceInfo.java | 41 - .../internal/org.apache.dubbo.rpc.Protocol | 1 - .../rpc/protocol/rsocket/ConsumerDemo.java | 51 -- .../rpc/protocol/rsocket/ProviderDemo.java | 29 - .../protocol/rsocket/RSocketProtocolTest.java | 260 ------ .../dubbo/rpc/service/DemoException.java | 40 - .../apache/dubbo/rpc/service/DemoService.java | 69 -- .../dubbo/rpc/service/DemoServiceImpl.java | 174 ---- .../src/test/resources/log4j.xml | 46 - .../spring/dubbo-rsocket-consumer.xml | 36 - .../spring/dubbo-rsocket-provider.xml | 40 - .../rpc/protocol/thrift/ThriftInvoker.java | 3 +- .../rpc/protocol/thrift/ThriftProtocol.java | 14 +- .../webservice/WebServiceProtocol.java | 2 +- 52 files changed, 621 insertions(+), 2713 deletions(-) rename dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/RemoteServiceImpl.java => dubbo-common/src/main/java/org/apache/dubbo/common/config/OrderedPropertiesProvider.java (70%) rename dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/RemoteService.java => dubbo-common/src/test/java/org/apache/dubbo/common/config/MockOrderedPropertiesProvider1.java (66%) delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/pom.xml delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/FutureSubscriber.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/MetadataCodec.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketConstants.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketExporter.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketInvoker.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketProtocol.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/ResourceDirectory.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/ResourceInfo.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/ConsumerDemo.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/ProviderDemo.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketProtocolTest.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoException.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoService.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoServiceImpl.java delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/log4j.xml delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/spring/dubbo-rsocket-consumer.xml delete mode 100644 dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/spring/dubbo-rsocket-provider.xml diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagRouter.java index 81792a113b..fb9e649ba1 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagRouter.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagRouter.java @@ -97,7 +97,7 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener { } List> result = invokers; - String tag = StringUtils.isEmpty(invocation.getAttachment(Constants.TAG_KEY)) ? url.getParameter(Constants.TAG_KEY) : + String tag = StringUtils.isEmpty((String) invocation.getAttachment(Constants.TAG_KEY)) ? url.getParameter(Constants.TAG_KEY) : (String) invocation.getAttachment(Constants.TAG_KEY); // if we are requesting for a Provider with a specific tag @@ -163,7 +163,7 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener { private List> filterUsingStaticTag(List> invokers, URL url, Invocation invocation) { List> result = invokers; // Dynamic param - String tag = StringUtils.isEmpty(invocation.getAttachment(Constants.TAG_KEY)) ? url.getParameter(Constants.TAG_KEY) : + String tag = StringUtils.isEmpty((String) invocation.getAttachment(Constants.TAG_KEY)) ? url.getParameter(Constants.TAG_KEY) : (String) invocation.getAttachment(Constants.TAG_KEY); // Tag request if (!StringUtils.isEmpty(tag)) { diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/RemoteServiceImpl.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/OrderedPropertiesProvider.java similarity index 70% rename from dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/RemoteServiceImpl.java rename to dubbo-common/src/main/java/org/apache/dubbo/common/config/OrderedPropertiesProvider.java index afb5788bde..28ab275aa0 100644 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/RemoteServiceImpl.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/OrderedPropertiesProvider.java @@ -14,13 +14,26 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.dubbo.rpc.service; +package org.apache.dubbo.common.config; -import java.rmi.RemoteException; -public class RemoteServiceImpl implements RemoteService { - @Override - public String sayHello(String name) throws RemoteException { - return "hello " + name; - } +import org.apache.dubbo.common.extension.SPI; + +import java.util.Properties; + +@SPI +public interface OrderedPropertiesProvider { + /** + * order + * + * @return + */ + int priority(); + + /** + * load the properties + * + * @return + */ + Properties initProperties(); } diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java index 15720acf93..7ef411b8b1 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java @@ -17,6 +17,7 @@ package org.apache.dubbo.common.constants; +import java.util.concurrent.ExecutorService; import java.util.regex.Pattern; public interface CommonConstants { @@ -163,6 +164,13 @@ public interface CommonConstants { String REVISION_KEY = "revision"; + String EXECUTOR_SERVICE_COMPONENT_KEY = ExecutorService.class.getName(); + + /** + * Consumer side 's proxy class + */ + String PROXY_CLASS_REF = "refClass"; + /** * package version in the manifest */ diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java index 327cdf18b1..9289a77654 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java @@ -16,7 +16,6 @@ */ package org.apache.dubbo.common.threadpool.manager; -import org.apache.dubbo.common.Constants; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.extension.ExtensionLoader; import org.apache.dubbo.common.logger.Logger; @@ -32,6 +31,11 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadPoolExecutor; +import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SIDE; +import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_SERVICE_COMPONENT_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.SIDE_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.THREADS_KEY; + /** * Consider implementing {@link Licycle} to enable executors shutdown when the process stops. */ @@ -58,9 +62,9 @@ public class DefaultExecutorRepository implements ExecutorRepository { } public ExecutorService createExecutorIfAbsent(URL url) { - String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY; - if (Constants.CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(Constants.SIDE_KEY))) { - componentKey = Constants.CONSUMER_SIDE; + String componentKey = EXECUTOR_SERVICE_COMPONENT_KEY; + if (CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(SIDE_KEY))) { + componentKey = CONSUMER_SIDE; } Map executors = data.computeIfAbsent(componentKey, k -> new ConcurrentHashMap<>()); return executors.computeIfAbsent(Integer.toString(url.getPort()), k -> (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url)); @@ -69,10 +73,10 @@ public class DefaultExecutorRepository implements ExecutorRepository { @Override public void updateThreadpool(URL url, ExecutorService executor) { try { - if (url.hasParameter(Constants.THREADS_KEY) + if (url.hasParameter(THREADS_KEY) && executor instanceof ThreadPoolExecutor && !executor.isShutdown()) { ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executor; - int threads = url.getParameter(Constants.THREADS_KEY, 0); + int threads = url.getParameter(THREADS_KEY, 0); int max = threadPoolExecutor.getMaximumPoolSize(); int core = threadPoolExecutor.getCorePoolSize(); if (threads > 0 && (threads != max || threads != core)) { diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/RemoteService.java b/dubbo-common/src/test/java/org/apache/dubbo/common/config/MockOrderedPropertiesProvider1.java similarity index 66% rename from dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/RemoteService.java rename to dubbo-common/src/test/java/org/apache/dubbo/common/config/MockOrderedPropertiesProvider1.java index d3e21dca1e..6c408b7350 100644 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/RemoteService.java +++ b/dubbo-common/src/test/java/org/apache/dubbo/common/config/MockOrderedPropertiesProvider1.java @@ -14,10 +14,20 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.dubbo.rpc.service; +package org.apache.dubbo.common.config; -import java.rmi.RemoteException; +import java.util.Properties; -public interface RemoteService { - String sayHello(String name) throws RemoteException; +public class MockOrderedPropertiesProvider1 implements OrderedPropertiesProvider { + @Override + public int priority() { + return 3; + } + + @Override + public Properties initProperties() { + Properties properties = new Properties(); + properties.put("testKey", "333"); + return properties; + } } 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 96f50824ec..fc9fd2d2e7 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 @@ -38,7 +38,6 @@ 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.model.ApplicationModel; -import org.apache.dubbo.rpc.model.ConsumerMethodModel; import org.apache.dubbo.rpc.model.ConsumerModel; import org.apache.dubbo.rpc.model.ServiceMetadata; import org.apache.dubbo.rpc.protocol.injvm.InjvmProtocol; @@ -48,6 +47,7 @@ import org.apache.dubbo.rpc.support.ProtocolUtils; import java.io.File; import java.io.FileInputStream; import java.io.IOException; +import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -57,24 +57,25 @@ import java.util.Map; import java.util.Properties; import static org.apache.dubbo.common.constants.CommonConstants.ANY_VALUE; +import static org.apache.dubbo.common.constants.CommonConstants.CLUSTER_KEY; import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SEPARATOR; import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SIDE; import static org.apache.dubbo.common.constants.CommonConstants.DUBBO; import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY; import static org.apache.dubbo.common.constants.CommonConstants.LOCALHOST_VALUE; import static org.apache.dubbo.common.constants.CommonConstants.METHODS_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.PROXY_CLASS_REF; 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.constants.CommonConstants.CLUSTER_KEY; -import static org.apache.dubbo.config.Constants.DUBBO_IP_TO_REGISTRY; -import static org.apache.dubbo.rpc.cluster.Constants.REFER_KEY; -import static org.apache.dubbo.registry.Constants.REGISTER_IP_KEY; -import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY; -import static org.apache.dubbo.registry.Constants.CONSUMER_PROTOCOL; import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_PROTOCOL; -import static org.apache.dubbo.rpc.Constants.LOCAL_PROTOCOL; import static org.apache.dubbo.common.utils.NetUtils.isInvalidLocalHost; +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; +import static org.apache.dubbo.rpc.Constants.LOCAL_PROTOCOL; +import static org.apache.dubbo.rpc.cluster.Constants.REFER_KEY; /** * ReferenceConfig @@ -296,6 +297,7 @@ public class ReferenceConfig extends AbstractReferenceConfig { serviceMetadata.setDefaultGroup(group); serviceMetadata.setServiceType(getActualInterface()); serviceMetadata.setServiceInterfaceName(interfaceName); + serviceMetadata.setServiceKey(URL.buildKey(interfaceName, group, version)); Map map = new HashMap(); map.put(SIDE_KEY, CONSUMER_SIDE); @@ -351,10 +353,9 @@ public class ReferenceConfig extends AbstractReferenceConfig { ref = createProxy(map); - String serviceKey = URL.buildKey(interfaceName, group, version); - ApplicationModel.initConsumerModel(serviceKey, buildConsumerModel(serviceKey, attributes, serviceMetadata)); + ApplicationModel.initConsumerModel(serviceMetadata.getServiceKey(), buildConsumerModel(attributes, serviceMetadata)); serviceMetadata.setTarget(ref); - consumerModel.getServiceMetadata().addAttribute(Constants.PROXY_CLASS_REF, ref); + serviceMetadata.addAttribute(PROXY_CLASS_REF, ref); initialized = true; } @@ -370,7 +371,7 @@ public class ReferenceConfig extends AbstractReferenceConfig { return actualInterface; } - private ConsumerModel buildConsumerModel(String serviceKey, Map attributes) { + private ConsumerModel buildConsumerModel(Map attributes, ServiceMetadata metadata) { Method[] methods = interfaceClass.getMethods(); Class serviceInterface = interfaceClass; if (interfaceClass == GenericService.class) { @@ -381,7 +382,7 @@ public class ReferenceConfig extends AbstractReferenceConfig { methods = interfaceClass.getMethods(); } } - return new ConsumerModel(serviceKey, serviceInterface, ref, methods, attributes); + return new ConsumerModel(attributes, metadata); } @SuppressWarnings({"unchecked", "rawtypes", "deprecation"}) 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 c2f9ffcaa8..247fd8e68c 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 @@ -66,33 +66,33 @@ import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_KEY; import static org.apache.dubbo.common.constants.CommonConstants.ANY_VALUE; import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN; import static org.apache.dubbo.common.constants.CommonConstants.DUBBO; +import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_IP_TO_BIND; import static org.apache.dubbo.common.constants.CommonConstants.LOCALHOST_VALUE; import static org.apache.dubbo.common.constants.CommonConstants.METHODS_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY; import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY; import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_SIDE; import static org.apache.dubbo.common.constants.CommonConstants.REVISION_KEY; import static org.apache.dubbo.common.constants.CommonConstants.SIDE_KEY; -import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_IP_TO_BIND; -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.rpc.cluster.Constants.EXPORT_KEY; -import static org.apache.dubbo.config.Constants.MULTICAST; -import static org.apache.dubbo.config.Constants.PROTOCOLS_SUFFIX; -import static org.apache.dubbo.rpc.Constants.SCOPE_KEY; -import static org.apache.dubbo.rpc.Constants.SCOPE_LOCAL; -import static org.apache.dubbo.config.Constants.SCOPE_NONE; -import static org.apache.dubbo.rpc.Constants.SCOPE_REMOTE; -import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY; import static org.apache.dubbo.common.constants.RegistryConstants.DYNAMIC_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.PROXY_KEY; -import static org.apache.dubbo.rpc.Constants.TOKEN_KEY; import static org.apache.dubbo.common.utils.NetUtils.getAvailablePort; import static org.apache.dubbo.common.utils.NetUtils.getLocalHost; import static org.apache.dubbo.common.utils.NetUtils.isInvalidLocalHost; import static org.apache.dubbo.common.utils.NetUtils.isInvalidPort; +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; +import static org.apache.dubbo.rpc.Constants.PROXY_KEY; +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.TOKEN_KEY; +import static org.apache.dubbo.rpc.cluster.Constants.EXPORT_KEY; /** * ServiceConfig @@ -474,7 +474,8 @@ public class ServiceConfig extends AbstractServiceConfig { List registryURLs = loadRegistries(true); for (ProtocolConfig protocolConfig : protocols) { String pathKey = URL.buildKey(getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), group, version); - ProviderModel providerModel = new ProviderModel(pathKey, ref, interfaceClass); + serviceMetadata.setServiceKey(pathKey); + ProviderModel providerModel = new ProviderModel(ref, serviceMetadata); ApplicationModel.initProviderModel(pathKey, providerModel); doExportUrlsFor1Protocol(protocolConfig, registryURLs); } diff --git a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/pom.xml b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/pom.xml index 69e1c8f670..abc8ed9cd6 100644 --- a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/pom.xml +++ b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/pom.xml @@ -62,6 +62,10 @@ org.apache.dubbo dubbo-metadata-report-nacos + + org.apache.dubbo + dubbo-metadata-report-zookeeper + org.apache.dubbo dubbo-config-spring diff --git a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/pom.xml b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/pom.xml index 99bd30bcf7..a59874fee6 100644 --- a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/pom.xml +++ b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/pom.xml @@ -64,6 +64,10 @@ org.apache.dubbo dubbo-metadata-report-nacos + + org.apache.dubbo + dubbo-metadata-report-zookeeper + org.apache.dubbo dubbo-rpc-dubbo diff --git a/dubbo-filter/dubbo-filter-validation/src/test/java/org/apache/dubbo/validation/support/jvalidation/mock/JValidatorTestTarget.java b/dubbo-filter/dubbo-filter-validation/src/test/java/org/apache/dubbo/validation/support/jvalidation/mock/JValidatorTestTarget.java index de46dabe1a..f65eb0573d 100644 --- a/dubbo-filter/dubbo-filter-validation/src/test/java/org/apache/dubbo/validation/support/jvalidation/mock/JValidatorTestTarget.java +++ b/dubbo-filter/dubbo-filter-validation/src/test/java/org/apache/dubbo/validation/support/jvalidation/mock/JValidatorTestTarget.java @@ -35,12 +35,6 @@ public interface JValidatorTestTarget { void someMethod5(Map map); - public void someMethod3(ValidationParameter[] parameters); - - public void someMethod4(List strings); - - public void someMethod5(Map map); - @interface Test2 { } diff --git a/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java b/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java index 2216e27306..92031601c2 100644 --- a/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java +++ b/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java @@ -40,12 +40,12 @@ import static org.apache.dubbo.common.constants.CommonConstants.APPLICATION_KEY; import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER; import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SIDE; import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY; import static org.apache.dubbo.common.constants.CommonConstants.PATH_SEPARATOR; import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER; import static org.apache.dubbo.common.constants.CommonConstants.SIDE_KEY; import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY; import static org.apache.dubbo.monitor.Constants.COUNT_PROTOCOL; -import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY; import static org.apache.dubbo.rpc.Constants.INPUT_KEY; import static org.apache.dubbo.rpc.Constants.OUTPUT_KEY; /** @@ -108,7 +108,7 @@ public class MonitorFilter extends ListenableFilter { @Override public void onResponse(Result result, Invoker invoker, Invocation invocation) { if (invoker.getUrl().hasParameter(MONITOR_KEY)) { - collect(invoker, invocation, result, RpcContext.getContext().getRemoteHost(), Long.valueOf(invocation.getAttachment(MONITOR_FILTER_START_TIME)), false); + collect(invoker, invocation, result, RpcContext.getContext().getRemoteHost(), Long.valueOf((String) invocation.getAttachment(MONITOR_FILTER_START_TIME)), false); getConcurrent(invoker, invocation).decrementAndGet(); // count down } } @@ -116,7 +116,7 @@ public class MonitorFilter extends ListenableFilter { @Override public void onError(Throwable t, Invoker invoker, Invocation invocation) { if (invoker.getUrl().hasParameter(MONITOR_KEY)) { - collect(invoker, invocation, null, RpcContext.getContext().getRemoteHost(), Long.valueOf(invocation.getAttachment(MONITOR_FILTER_START_TIME)), true); + collect(invoker, invocation, null, RpcContext.getContext().getRemoteHost(), Long.valueOf((String) invocation.getAttachment(MONITOR_FILTER_START_TIME)), true); getConcurrent(invoker, invocation).decrementAndGet(); // count down } } @@ -181,10 +181,10 @@ public class MonitorFilter extends ListenableFilter { } String input = "", output = ""; if (invocation.getAttachment(INPUT_KEY) != null) { - input = invocation.getAttachment(INPUT_KEY); + input = (String) invocation.getAttachment(INPUT_KEY); } if (result != null && result.getAttachment(OUTPUT_KEY) != null) { - output = result.getAttachment(OUTPUT_KEY); + output = (String) result.getAttachment(OUTPUT_KEY); } return new URL(COUNT_PROTOCOL, NetUtils.getLocalHost(), localPort, service + PATH_SEPARATOR + method, MonitorService.APPLICATION, application, MonitorService.INTERFACE, service, MonitorService.METHOD, method, remoteKey, remoteValue, error ? MonitorService.FAILURE : MonitorService.SUCCESS, "1", MonitorService.ELAPSED, String.valueOf(elapsed), MonitorService.CONCURRENT, String.valueOf(concurrent), INPUT_KEY, input, OUTPUT_KEY, output, GROUP_KEY, group, VERSION_KEY, version); diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture.java index 7f0e3a7227..0abecf2030 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture.java @@ -18,6 +18,7 @@ package org.apache.dubbo.remoting.exchange.support; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.threadpool.ThreadlessExecutor; import org.apache.dubbo.common.timer.HashedWheelTimer; import org.apache.dubbo.common.timer.Timeout; import org.apache.dubbo.common.timer.Timer; @@ -34,6 +35,7 @@ import java.util.Date; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT; @@ -64,6 +66,16 @@ public class DefaultFuture extends CompletableFuture { private volatile long sent; private Timeout timeoutCheckTask; + private ExecutorService executor; + + public ExecutorService getExecutor() { + return executor; + } + + public void setExecutor(ExecutorService executor) { + this.executor = executor; + } + private DefaultFuture(Channel channel, Request request, int timeout) { this.channel = channel; this.request = request; @@ -92,8 +104,9 @@ public class DefaultFuture extends CompletableFuture { * @param timeout timeout * @return a new DefaultFuture */ - public static DefaultFuture newFuture(Channel channel, Request request, int timeout) { + public static DefaultFuture newFuture(Channel channel, Request request, int timeout, ExecutorService executor) { final DefaultFuture future = new DefaultFuture(channel, request, timeout); + future.setExecutor(executor); // timeout check timeoutCheck(future); return future; @@ -178,7 +191,6 @@ public class DefaultFuture extends CompletableFuture { this.cancel(true); } - private void doReceived(Response res) { if (res == null) { throw new IllegalStateException("response cannot be null"); @@ -190,6 +202,15 @@ public class DefaultFuture extends CompletableFuture { } else { this.completeExceptionally(new RemotingException(channel, res.getErrorMessage())); } + + // the result is returning, but the caller thread may still waiting + // to avoid endless waiting for whatever reason, notify caller thread to return. + if (executor != null && executor instanceof ThreadlessExecutor) { + ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor; + if (threadlessExecutor.isWaiting()) { + threadlessExecutor.notifyReturn(); + } + } } private long getId() { @@ -243,14 +264,17 @@ public class DefaultFuture extends CompletableFuture { if (future == null || future.isDone()) { return; } - // create exception response. - Response timeoutResponse = new Response(future.getId()); - // set timeout status. - timeoutResponse.setStatus(future.isSent() ? Response.SERVER_TIMEOUT : Response.CLIENT_TIMEOUT); - timeoutResponse.setErrorMessage(future.getTimeoutMessage(true)); - // handle response. - DefaultFuture.received(future.getChannel(), timeoutResponse, true); - + if (future.getExecutor() != null) { + future.getExecutor().execute(() -> { + // create exception response. + Response timeoutResponse = new Response(future.getId()); + // set timeout status. + timeoutResponse.setStatus(future.isSent() ? Response.SERVER_TIMEOUT : Response.CLIENT_TIMEOUT); + timeoutResponse.setErrorMessage(future.getTimeoutMessage(true)); + // handle response. + DefaultFuture.received(future.getChannel(), timeoutResponse, true); + }); + } } } } diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java index c6392da59b..f96207b604 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java @@ -33,6 +33,9 @@ import java.net.InetSocketAddress; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; +import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT; +import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY; + /** * ExchangeReceiver */ @@ -108,7 +111,7 @@ final class HeaderExchangeChannel implements ExchangeChannel { @Override public CompletableFuture request(Object request, ExecutorService executor) throws RemotingException { - return request(request, channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT), executor); + return request(request, channel.getUrl().getPositiveParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT), executor); } @Override diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java index 5d0a643b2e..10d0f6065b 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java @@ -18,10 +18,15 @@ package org.apache.dubbo.rpc; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.threadpool.ThreadlessExecutor; import org.apache.dubbo.rpc.protocol.dubbo.FutureAdapter; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.function.BiConsumer; /** diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java index aacb4ac144..2aa973514e 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java @@ -62,7 +62,7 @@ public class RpcInvocation implements Invocation, Serializable { public RpcInvocation(Invocation invocation, Invoker invoker) { this(invocation.getMethodName(), invocation.getParameterTypes(), - invocation.getArguments(), new HashMap(invocation.getAttachments()), + invocation.getArguments(), new HashMap<>(invocation.getAttachments()), invocation.getInvoker()); if (invoker != null) { URL url = invoker.getUrl(); @@ -94,7 +94,7 @@ public class RpcInvocation implements Invocation, Serializable { } public RpcInvocation(Method method, Object[] arguments) { - this(method, arguments, null); + this(method.getName(), method.getParameterTypes(), arguments, null, null); } public RpcInvocation(Method method, Object[] arguments, Map attachment, Map attributes) { diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ActiveLimitFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ActiveLimitFilter.java index ee760eea0f..da77e05bcd 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ActiveLimitFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ActiveLimitFilter.java @@ -115,7 +115,7 @@ public class ActiveLimitFilter extends ListenableFilter { } private long getElapsed(Invocation invocation) { - String beginTime = invocation.getAttachment(ACTIVELIMIT_FILTER_START_TIME); + String beginTime = (String) invocation.getAttachment(ACTIVELIMIT_FILTER_START_TIME); return StringUtils.isNotEmpty(beginTime) ? System.currentTimeMillis() - Long.parseLong(beginTime) : 0; } 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 eb6cebed61..f98cd1dd90 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 @@ -35,8 +35,8 @@ import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER; import static org.apache.dubbo.common.constants.CommonConstants.REMOTE_APPLICATION_KEY; import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY; import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY; -import static org.apache.dubbo.rpc.Constants.ASYNC_KEY; import static org.apache.dubbo.remoting.Constants.DUBBO_VERSION_KEY; +import static org.apache.dubbo.rpc.Constants.ASYNC_KEY; import static org.apache.dubbo.rpc.Constants.FORCE_USE_TAG; import static org.apache.dubbo.rpc.Constants.TOKEN_KEY; @@ -57,7 +57,7 @@ public class ContextFilter extends ListenableFilter { @Override public Result invoke(Invoker invoker, Invocation invocation) throws RpcException { - Map attachments = invocation.getAttachments(); + Map attachments = invocation.getAttachments(); if (attachments != null) { attachments = new HashMap<>(attachments); attachments.remove(PATH_KEY); diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ExecuteLimitFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ExecuteLimitFilter.java index cf33311b85..394f7dbca8 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ExecuteLimitFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ExecuteLimitFilter.java @@ -88,7 +88,7 @@ public class ExecuteLimitFilter extends ListenableFilter { } private long getElapsed(Invocation invocation) { - String beginTime = invocation.getAttachment(EXECUTELIMIT_FILTER_START_TIME); + String beginTime = (String) invocation.getAttachment(EXECUTELIMIT_FILTER_START_TIME); return StringUtils.isNotEmpty(beginTime) ? System.currentTimeMillis() - Long.parseLong(beginTime) : 0; } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericFilter.java index ed6e123971..ccad3a3940 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericFilter.java @@ -74,10 +74,10 @@ public class GenericFilter extends ListenableFilter { if (args == null) { args = new Object[params.length]; } - String generic = inv.getAttachment(GENERIC_KEY); + String generic = (String) inv.getAttachment(GENERIC_KEY); if (StringUtils.isBlank(generic)) { - generic = RpcContext.getContext().getAttachment(GENERIC_KEY); + generic = (String) RpcContext.getContext().getAttachment(GENERIC_KEY); } if (StringUtils.isEmpty(generic) @@ -138,7 +138,7 @@ public class GenericFilter extends ListenableFilter { args[0].getClass().getName()); } } - return invoker.invoke(new RpcInvocation(method, args, inv.getAttachments())); + return invoker.invoke(new RpcInvocation(method, args, inv.getAttachments(), inv.getAttributes())); } catch (NoSuchMethodException e) { throw new RpcException(e.getMessage(), e); } catch (ClassNotFoundException e) { @@ -157,9 +157,9 @@ public class GenericFilter extends ListenableFilter { && inv.getArguments().length == 3 && !GenericService.class.isAssignableFrom(invoker.getInterface())) { - String generic = inv.getAttachment(GENERIC_KEY); + String generic = (String) inv.getAttachment(GENERIC_KEY); if (StringUtils.isBlank(generic)) { - generic = RpcContext.getContext().getAttachment(GENERIC_KEY); + generic = (String) RpcContext.getContext().getAttachment(GENERIC_KEY); } if (appResponse.hasException() && !(appResponse.getException() instanceof GenericException)) { diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TimeoutFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TimeoutFilter.java index 31b0dc977a..6e1af88221 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TimeoutFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TimeoutFilter.java @@ -52,7 +52,7 @@ public class TimeoutFilter extends ListenableFilter { @Override public void onResponse(Result appResponse, Invoker invoker, Invocation invocation) { - String startAttach = invocation.getAttachment(TIMEOUT_FILTER_START_TIME); + String startAttach = (String) invocation.getAttachment(TIMEOUT_FILTER_START_TIME); if (startAttach != null) { long elapsed = System.currentTimeMillis() - Long.valueOf(startAttach); if (invoker.getUrl() != null && elapsed > invoker.getUrl().getMethodParameter(invocation.getMethodName(), "timeout", Integer.MAX_VALUE)) { diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TokenFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TokenFilter.java index 164f42a89e..9fbd96f67d 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TokenFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TokenFilter.java @@ -45,8 +45,8 @@ public class TokenFilter implements Filter { String token = invoker.getUrl().getParameter(TOKEN_KEY); if (ConfigUtils.isNotEmpty(token)) { Class serviceType = invoker.getInterface(); - Map attachments = inv.getAttachments(); - String remoteToken = attachments == null ? null : attachments.get(TOKEN_KEY); + Map attachments = inv.getAttachments(); + String remoteToken = (String) (attachments == null ? null : attachments.get(TOKEN_KEY)); if (!token.equals(remoteToken)) { throw new RpcException("Invalid token! Forbid invoke remote service " + serviceType + " method " + inv.getMethodName() + "() from consumer " + RpcContext.getContext().getRemoteHost() + " to provider " + RpcContext.getContext().getLocalHost()); } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerMethodModel.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerMethodModel.java index 01f3696ef3..0798a50029 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerMethodModel.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerMethodModel.java @@ -18,8 +18,7 @@ package org.apache.dubbo.rpc.model; import java.lang.reflect.Method; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; +import java.util.Map; import static org.apache.dubbo.rpc.Constants.$INVOKE; @@ -33,9 +32,10 @@ public class ConsumerMethodModel { private final String methodName; private final boolean generic; - private final ConcurrentMap attributeMap = new ConcurrentHashMap<>(); + private final AsyncMethodInfo asyncInfo; - public ConsumerMethodModel(Method method) { + + public ConsumerMethodModel(Method method, Map attributes) { this.method = method; this.parameterClasses = method.getParameterTypes(); this.returnClass = method.getReturnType(); @@ -43,32 +43,24 @@ public class ConsumerMethodModel { this.methodName = method.getName(); this.generic = methodName.equals($INVOKE) && parameterTypes != null && parameterTypes.length == 3; + if (attributes != null) { + asyncInfo = (AsyncMethodInfo) attributes.get(methodName); + } else { + asyncInfo = null; + } } public Method getMethod() { return method; } -// public ConcurrentMap getAttributeMap() { -// return attributeMap; -// } - - public void addAttribute(String key, Object value) { - this.attributeMap.put(key, value); - } - - public Object getAttribute(String key) { - return this.attributeMap.get(key); - } - - public Class getReturnClass() { return returnClass; } -// public AsyncMethodInfo getAsyncInfo() { -// return (AsyncMethodInfo) attributeMap.get(Constants.ASYNC_KEY); -// } + public AsyncMethodInfo getAsyncInfo() { + return asyncInfo; + } public String getMethodName() { return methodName; diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerModel.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerModel.java index 1a91ae1b75..0e4d4c2208 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerModel.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerModel.java @@ -31,20 +31,18 @@ public class ConsumerModel { private final ServiceMetadata serviceMetadata; private final Map methodModels = new IdentityHashMap(); - public ConsumerModel(String serviceName, String group, String version, Class interfaceClass) { - this.serviceMetadata = new ServiceMetadata(serviceName, group, version, interfaceClass); - - Method[] methods = interfaceClass.getMethods(); - for (Method method : methods) { - methodModels.put(method, new ConsumerMethodModel(method)); - } - } - - public ConsumerModel(ServiceMetadata serviceMetadata) { - this.serviceMetadata = serviceMetadata; - Method[] methods = serviceMetadata.getServiceType().getMethods(); - for (Method method : methods) { - methodModels.put(method, new ConsumerMethodModel(method)); + /** + * This constructor create an instance of ConsumerModel and passed objects should not be null. + * If service name, service instance, proxy object,methods should not be null. If these are null + * then this constructor will throw {@link IllegalArgumentException} + * + * @param attributes Attributes of methods. + * @param metadata + */ + public ConsumerModel(Map attributes, ServiceMetadata metadata) { + this.serviceMetadata = metadata; + for (Method method : metadata.getServiceType().getMethods()) { + methodModels.put(method, new ConsumerMethodModel(method, attributes)); } } @@ -106,6 +104,15 @@ public class ConsumerModel { return new ArrayList(methodModels.values()); } + /** + * Return the proxy object used by called while creating instance of ConsumerModel + * + * @return + */ + public Object getProxyObject() { + return this.serviceMetadata.getTarget(); + } + public String getServiceName() { return this.serviceMetadata.getServiceKey(); } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java index 42f52e8dcd..45a76e778a 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java @@ -27,7 +27,6 @@ import org.apache.dubbo.common.utils.ArrayUtils; import org.apache.dubbo.common.utils.CollectionUtils; import org.apache.dubbo.common.utils.NetUtils; import org.apache.dubbo.rpc.AsyncRpcResult; -import org.apache.dubbo.rpc.FutureAdapter; import org.apache.dubbo.rpc.Invocation; import org.apache.dubbo.rpc.InvokeMode; import org.apache.dubbo.rpc.Invoker; @@ -35,6 +34,7 @@ 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.protocol.dubbo.FutureAdapter; import org.apache.dubbo.rpc.support.RpcUtils; import java.lang.reflect.InvocationTargetException; @@ -179,7 +179,7 @@ public abstract class AbstractInvoker implements Invoker { } catch (Throwable e) { asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation); } - RpcContext.getContext().setFuture(new FutureAdapter(asyncResult, inv)); + RpcContext.getContext().setFuture(new FutureAdapter(asyncResult)); return asyncResult; } 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 e0ca6ab5a1..60ed62149e 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 @@ -174,6 +174,8 @@ public class RpcUtils { public static boolean isGenericAsync(Invocation inv) { return $INVOKE_ASYNC.equals(inv.getMethodName()); + } + public static InvokeMode getInvokeMode(URL url, Invocation inv) { if (isReturnTypeFuture(inv)) { return InvokeMode.FUTURE; @@ -184,10 +186,6 @@ public class RpcUtils { } } - public static boolean isGenericAsync(Invocation inv) { - return Constants.$INVOKE_ASYNC.equals(inv.getMethodName()); - } - public static boolean isOneway(URL url, Invocation inv) { boolean isOneway; if (Boolean.FALSE.toString().equals(inv.getAttachment(RETURN_KEY))) { diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/TokenFilterTest.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/TokenFilterTest.java index 839416fa9c..037c27f7b1 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/TokenFilterTest.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/TokenFilterTest.java @@ -30,11 +30,10 @@ import org.mockito.Mockito; import java.util.HashMap; import java.util.Map; +import static org.apache.dubbo.rpc.Constants.TOKEN_KEY; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.when; -import static org.apache.dubbo.rpc.Constants.TOKEN_KEY; - public class TokenFilterTest { private TokenFilter tokenFilter = new TokenFilter(); @@ -48,7 +47,7 @@ public class TokenFilterTest { when(invoker.getUrl()).thenReturn(url); when(invoker.invoke(any(Invocation.class))).thenReturn(new AppResponse("result")); - Map attachments = new HashMap(); + Map attachments = new HashMap<>(); attachments.put(TOKEN_KEY, token); Invocation invocation = Mockito.mock(Invocation.class); when(invocation.getAttachments()).thenReturn(attachments); @@ -67,7 +66,7 @@ public class TokenFilterTest { when(invoker.getUrl()).thenReturn(url); when(invoker.invoke(any(Invocation.class))).thenReturn(new AppResponse("result")); - Map attachments = new HashMap(); + Map attachments = new HashMap<>(); attachments.put(TOKEN_KEY, "wrongToken"); Invocation invocation = Mockito.mock(Invocation.class); when(invocation.getAttachments()).thenReturn(attachments); diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/ImplicitCallBackTest.java b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/ImplicitCallBackTest.java index 8f5f860047..89f90fc472 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/ImplicitCallBackTest.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/ImplicitCallBackTest.java @@ -1,401 +1,406 @@ -/* - * 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.dubbo; - - -import org.apache.dubbo.common.URL; -import org.apache.dubbo.common.utils.NetUtils; -import org.apache.dubbo.rpc.Exporter; -import org.apache.dubbo.rpc.Invoker; -import org.apache.dubbo.rpc.RpcContext; -import org.apache.dubbo.rpc.model.ApplicationModel; -import org.apache.dubbo.rpc.model.ConsumerMethodModel; -import org.apache.dubbo.rpc.model.ConsumerModel; -import org.apache.dubbo.rpc.protocol.dubbo.support.ProtocolUtils; - -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -import java.io.Serializable; -import java.lang.reflect.Method; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; - -public class ImplicitCallBackTest { - - protected Exporter exporter = null; - protected Invoker reference = null; - - protected URL serviceURL = null; - protected URL consumerUrl = null; - Method onReturnMethod; - Method onThrowMethod; - Method onInvokeMethod; - NofifyImpl notify = new NofifyImpl(); - //================================================================================================ - IDemoService demoProxy = null; - - @BeforeEach - public void setUp() throws SecurityException, NoSuchMethodException { - onReturnMethod = Nofify.class.getMethod("onreturn", new Class[]{Person.class, Integer.class}); - onThrowMethod = Nofify.class.getMethod("onthrow", new Class[]{Throwable.class, Integer.class}); - onInvokeMethod = Nofify.class.getMethod("oninvoke", new Class[]{Integer.class}); - } - - @AfterEach - public void tearDown() { - ProtocolUtils.closeAll(); - } - - public void initOrResetService() { - destroyService(); - exportService(); - referService(); - } - - public void initOrResetExService() { - destroyService(); - exportExService(); - referService(); - } - - public void destroyService() { - demoProxy = null; - try { - if (exporter != null) exporter.unexport(); - if (reference != null) reference.destroy(); - } catch (Exception e) { - } - } - - void referService() { - demoProxy = (IDemoService) ProtocolUtils.refer(IDemoService.class, consumerUrl); - } - - public void exportService() { - exporter = ProtocolUtils.export(new NormalDemoService(), IDemoService.class, serviceURL); - } - - public void exportExService() { - exporter = ProtocolUtils.export(new ExceptionDemoExService(), IDemoService.class, serviceURL); - } - - public void initOrResetUrl(boolean isAsync) throws Exception { - int port = NetUtils.getAvailablePort(); - consumerUrl = serviceURL = URL.valueOf("dubbo://127.0.0.1:" + port + "/" + IDemoService.class.getName() + "?group=" + System.nanoTime() + "&async=" + isAsync + "&timeout=100000&reference.filter=future"); - } - - public void initImplicitCallBackURL_onlyOnthrow() throws Exception { - Map attitudes = new HashMap<>(); - ConsumerMethodModel.AsyncMethodInfo asyncMethodInfo = new ConsumerMethodModel.AsyncMethodInfo(); - asyncMethodInfo.setOnthrowInstance(notify); - asyncMethodInfo.setOnthrowMethod(onThrowMethod); - ConsumerModel consumerModel = new ConsumerModel(consumerUrl.getServiceInterface(), "Dubbo", "1.0.0", IDemoService.class); - consumerModel.getMethodModel("get").addAttribute(Constants.ASYNC_KEY, asyncMethodInfo); - ApplicationModel.initConsumerModel(consumerUrl.getServiceKey(), consumerModel); - } - - //================================================================================================ - - public void initImplicitCallBackURL_onlyOnreturn() throws Exception { - Map attitudes = new HashMap<>(); - ConsumerMethodModel.AsyncMethodInfo asyncMethodInfo = new ConsumerMethodModel.AsyncMethodInfo(); - asyncMethodInfo.setOnreturnInstance(notify); - asyncMethodInfo.setOnreturnMethod(onReturnMethod); - ConsumerModel consumerModel = new ConsumerModel(consumerUrl.getServiceInterface(), "Dubbo", "1.0.0", IDemoService.class); - consumerModel.getMethodModel("get").addAttribute(Constants.ASYNC_KEY, asyncMethodInfo); - ApplicationModel.initConsumerModel(consumerUrl.getServiceKey(), consumerModel); - } - - public void initImplicitCallBackURL_onlyOninvoke() throws Exception { - Map attitudes = new HashMap<>(); - ConsumerMethodModel.AsyncMethodInfo asyncMethodInfo = new ConsumerMethodModel.AsyncMethodInfo(); - asyncMethodInfo.setOninvokeInstance(notify); - asyncMethodInfo.setOninvokeMethod(onInvokeMethod); - ConsumerModel consumerModel = new ConsumerModel(consumerUrl.getServiceInterface(), "Dubbo", "1.0.0", IDemoService.class); - consumerModel.getMethodModel("get").addAttribute(Constants.ASYNC_KEY, asyncMethodInfo); - ApplicationModel.initConsumerModel(consumerUrl.getServiceKey(), consumerModel); - } - - @Test - public void test_CloseCallback() throws Exception { - initOrResetUrl(false); - initOrResetService(); - Person ret = demoProxy.get(1); - Assertions.assertEquals(1, ret.getId()); - destroyService(); - } - - @Test - public void test_Sync_Onreturn() throws Exception { - initOrResetUrl(false); - initOrResetService(); - initImplicitCallBackURL_onlyOnreturn(); - - int requestId = 2; - Person ret = demoProxy.get(requestId); - Assertions.assertEquals(requestId, ret.getId()); - for (int i = 0; i < 10; i++) { - if (!notify.ret.containsKey(requestId)) { - Thread.sleep(200); - } else { - break; - } - } - Assertions.assertEquals(requestId, notify.ret.get(requestId).getId()); - destroyService(); - } - - @Test - public void test_Ex_OnReturn() throws Exception { - initOrResetUrl(true); - initOrResetExService(); - initImplicitCallBackURL_onlyOnreturn(); - - - int requestId = 2; - Person ret = demoProxy.get(requestId); - Assertions.assertNull(ret); - for (int i = 0; i < 10; i++) { - if (!notify.errors.containsKey(requestId)) { - Thread.sleep(200); - } else { - break; - } - } - Assertions.assertTrue(!notify.errors.containsKey(requestId)); - destroyService(); - } - - @Test - public void test_Ex_OnInvoke() throws Exception { - initOrResetUrl(true); - initOrResetExService(); - initImplicitCallBackURL_onlyOninvoke(); - - int requestId = 2; - Person ret = demoProxy.get(requestId); - Assertions.assertNull(ret); - for (int i = 0; i < 10; i++) { - if (!notify.inv.contains(requestId)) { - Thread.sleep(200); - } else { - break; - } - } - Assertions.assertTrue(notify.inv.contains(requestId)); - destroyService(); - } - - @Test - public void test_Ex_Onthrow() throws Exception { - initOrResetUrl(true); - initOrResetExService(); - initImplicitCallBackURL_onlyOnthrow(); - - int requestId = 2; - Person ret = demoProxy.get(requestId); - Assertions.assertNull(ret); - for (int i = 0; i < 10; i++) { - if (!notify.errors.containsKey(requestId)) { - Thread.sleep(200); - } else { - break; - } - } - Assertions.assertTrue(notify.errors.containsKey(requestId)); - Assertions.assertTrue(notify.errors.get(requestId) instanceof Throwable); - destroyService(); - } - - @Test - public void test_Sync_NoFuture() throws Exception { - initOrResetUrl(false); - initOrResetService(); - initImplicitCallBackURL_onlyOnreturn(); - - int requestId = 2; - Person ret = demoProxy.get(requestId); - Assertions.assertEquals(requestId, ret.getId()); - Future pFuture = RpcContext.getContext().getFuture(); - Assertions.assertEquals(ret, pFuture.get()); - destroyService(); - } - - @Test - public void test_Async_Future() throws Exception { - initOrResetUrl(true); - initOrResetService(); - - int requestId = 2; - Person ret = demoProxy.get(requestId); - Assertions.assertNull(ret); - Future pFuture = RpcContext.getContext().getFuture(); - ret = pFuture.get(1000 * 1000, TimeUnit.MICROSECONDS); - Assertions.assertEquals(requestId, ret.getId()); - destroyService(); - } - - @Test - public void test_Async_Future_Multi() throws Exception { - initOrResetUrl(true); - initOrResetService(); - - int requestId1 = 1; - Person ret = demoProxy.get(requestId1); - Assertions.assertNull(ret); - Future p1Future = RpcContext.getContext().getFuture(); - - int requestId2 = 1; - Person ret2 = demoProxy.get(requestId2); - Assertions.assertNull(ret2); - Future p2Future = RpcContext.getContext().getFuture(); - - ret = p1Future.get(1000 * 1000, TimeUnit.MICROSECONDS); - ret2 = p2Future.get(1000 * 1000, TimeUnit.MICROSECONDS); - Assertions.assertEquals(requestId1, ret.getId()); - Assertions.assertEquals(requestId2, ret.getId()); - destroyService(); - } - - @Test - public void test_Async_Future_Ex() throws Throwable { - Assertions.assertThrows(RuntimeException.class, () -> { - try { - initOrResetUrl(true); - initOrResetExService(); - - int requestId = 2; - Person ret = demoProxy.get(requestId); - Assertions.assertNull(ret); - Future pFuture = RpcContext.getContext().getFuture(); - ret = pFuture.get(1000 * 1000, TimeUnit.MICROSECONDS); - Assertions.assertEquals(requestId, ret.getId()); - } catch (ExecutionException e) { - throw e.getCause(); - } finally { - destroyService(); - } - }); - } - - @Test - public void test_Normal_Ex() throws Exception { - Assertions.assertThrows(RuntimeException.class, () -> { - initOrResetUrl(false); - initOrResetExService(); - - int requestId = 2; - Person ret = demoProxy.get(requestId); - Assertions.assertEquals(requestId, ret.getId()); - }); - } - - interface Nofify { - void onreturn(Person msg, Integer id); - - void onthrow(Throwable ex, Integer id); - - void oninvoke(Integer id); - } - - interface IDemoService { - Person get(int id); - } - - public static class Person implements Serializable { - private static final long serialVersionUID = 1L; - private int id; - private String name; - private int age; - - public Person(int id, String name, int age) { - this.id = id; - this.name = name; - this.age = age; - } - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - public int getAge() { - return age; - } - - public void setAge(int age) { - this.age = age; - } - - public int getId() { - return id; - } - - public void setId(int id) { - this.id = id; - } - - @Override - public String toString() { - return "Person [name=" + name + ", age=" + age + "]"; - } - } - - class NofifyImpl implements Nofify { - public List inv = new ArrayList(); - public Map ret = new HashMap(); - public Map errors = new HashMap(); - public boolean exd = false; - - public void onreturn(Person msg, Integer id) { - System.out.println("onNotify:" + msg); - ret.put(id, msg); - } - - public void onthrow(Throwable ex, Integer id) { - errors.put(id, ex); -// ex.printStackTrace(); - } - - public void oninvoke(Integer id) { - inv.add(id); - } - } - - class NormalDemoService implements IDemoService { - public Person get(int id) { - return new Person(id, "charles", 4); - } - } - - class ExceptionDemoExService implements IDemoService { - public Person get(int id) { - throw new RuntimeException("request persion id is :" + id); - } - } -} +///* +// * 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.dubbo; +// +// +//import org.apache.dubbo.common.URL; +//import org.apache.dubbo.common.utils.NetUtils; +//import org.apache.dubbo.rpc.Exporter; +//import org.apache.dubbo.rpc.Invoker; +//import org.apache.dubbo.rpc.RpcContext; +//import org.apache.dubbo.rpc.model.ApplicationModel; +//import org.apache.dubbo.rpc.model.ConsumerMethodModel; +//import org.apache.dubbo.rpc.model.ConsumerModel; +//import org.apache.dubbo.rpc.model.ServiceMetadata; +//import org.apache.dubbo.rpc.protocol.dubbo.support.ProtocolUtils; +// +//import org.junit.jupiter.api.AfterEach; +//import org.junit.jupiter.api.Assertions; +//import org.junit.jupiter.api.BeforeEach; +//import org.junit.jupiter.api.Test; +// +//import java.io.Serializable; +//import java.lang.reflect.Method; +//import java.util.ArrayList; +//import java.util.HashMap; +//import java.util.List; +//import java.util.Map; +//import java.util.concurrent.ExecutionException; +//import java.util.concurrent.Future; +//import java.util.concurrent.TimeUnit; +// +//import static org.apache.dubbo.rpc.Constants.ASYNC_KEY; +// +//public class ImplicitCallBackTest { +// +// protected Exporter exporter = null; +// protected Invoker reference = null; +// +// protected URL serviceURL = null; +// protected URL consumerUrl = null; +// Method onReturnMethod; +// Method onThrowMethod; +// Method onInvokeMethod; +// NofifyImpl notify = new NofifyImpl(); +// //================================================================================================ +// IDemoService demoProxy = null; +// +// @BeforeEach +// public void setUp() throws SecurityException, NoSuchMethodException { +// onReturnMethod = Nofify.class.getMethod("onreturn", new Class[]{Person.class, Integer.class}); +// onThrowMethod = Nofify.class.getMethod("onthrow", new Class[]{Throwable.class, Integer.class}); +// onInvokeMethod = Nofify.class.getMethod("oninvoke", new Class[]{Integer.class}); +// } +// +// @AfterEach +// public void tearDown() { +// ProtocolUtils.closeAll(); +// } +// +// public void initOrResetService() { +// destroyService(); +// exportService(); +// referService(); +// } +// +// public void initOrResetExService() { +// destroyService(); +// exportExService(); +// referService(); +// } +// +// public void destroyService() { +// demoProxy = null; +// try { +// if (exporter != null) exporter.unexport(); +// if (reference != null) reference.destroy(); +// } catch (Exception e) { +// } +// } +// +// void referService() { +// demoProxy = (IDemoService) ProtocolUtils.refer(IDemoService.class, consumerUrl); +// } +// +// public void exportService() { +// exporter = ProtocolUtils.export(new NormalDemoService(), IDemoService.class, serviceURL); +// } +// +// public void exportExService() { +// exporter = ProtocolUtils.export(new ExceptionDemoExService(), IDemoService.class, serviceURL); +// } +// +// public void initOrResetUrl(boolean isAsync) throws Exception { +// int port = NetUtils.getAvailablePort(); +// consumerUrl = serviceURL = URL.valueOf("dubbo://127.0.0.1:" + port + "/" + IDemoService.class.getName() + "?group=" + System.nanoTime() + "&async=" + isAsync + "&timeout=100000&reference.filter=future"); +// } +// +// public void initImplicitCallBackURL_onlyOnthrow() throws Exception { +// Map attitudes = new HashMap<>(); +// ConsumerMethodModel.AsyncMethodInfo asyncMethodInfo = new ConsumerMethodModel.AsyncMethodInfo(); +// asyncMethodInfo.setOnthrowInstance(notify); +// asyncMethodInfo.setOnthrowMethod(onThrowMethod); +// ServiceMetadata metada = new ServiceMetadata(); +// +// ConsumerModel consumerModel = new ConsumerModel(consumerUrl.getServiceInterface(), "Dubbo", "1.0.0", IDemoService.class); +// consumerModel.getMethodModel("get").addAttribute(ASYNC_KEY, asyncMethodInfo); +// ApplicationModel.initConsumerModel(consumerUrl.getServiceKey(), consumerModel); +// } +// +// //================================================================================================ +// +// public void initImplicitCallBackURL_onlyOnreturn() throws Exception { +// Map attitudes = new HashMap<>(); +// ConsumerMethodModel.AsyncMethodInfo asyncMethodInfo = new ConsumerMethodModel.AsyncMethodInfo(); +// asyncMethodInfo.setOnreturnInstance(notify); +// asyncMethodInfo.setOnreturnMethod(onReturnMethod); +// ConsumerModel consumerModel = new ConsumerModel(consumerUrl.getServiceInterface(), "Dubbo", "1.0.0", IDemoService.class); +// consumerModel.getMethodModel("get").addAttribute(ASYNC_KEY, asyncMethodInfo); +// ApplicationModel.initConsumerModel(consumerUrl.getServiceKey(), consumerModel); +// } +// +// public void initImplicitCallBackURL_onlyOninvoke() throws Exception { +// Map attitudes = new HashMap<>(); +// ConsumerMethodModel.AsyncMethodInfo asyncMethodInfo = new ConsumerMethodModel.AsyncMethodInfo(); +// asyncMethodInfo.setOninvokeInstance(notify); +// asyncMethodInfo.setOninvokeMethod(onInvokeMethod); +// ConsumerModel consumerModel = new ConsumerModel(consumerUrl.getServiceInterface(), "Dubbo", "1.0.0", IDemoService.class); +// consumerModel.getMethodModel("get").addAttribute(ASYNC_KEY, asyncMethodInfo); +// ApplicationModel.initConsumerModel(consumerUrl.getServiceKey(), consumerModel); +// } +// +// @Test +// public void test_CloseCallback() throws Exception { +// initOrResetUrl(false); +// initOrResetService(); +// Person ret = demoProxy.get(1); +// Assertions.assertEquals(1, ret.getId()); +// destroyService(); +// } +// +// @Test +// public void test_Sync_Onreturn() throws Exception { +// initOrResetUrl(false); +// initOrResetService(); +// initImplicitCallBackURL_onlyOnreturn(); +// +// int requestId = 2; +// Person ret = demoProxy.get(requestId); +// Assertions.assertEquals(requestId, ret.getId()); +// for (int i = 0; i < 10; i++) { +// if (!notify.ret.containsKey(requestId)) { +// Thread.sleep(200); +// } else { +// break; +// } +// } +// Assertions.assertEquals(requestId, notify.ret.get(requestId).getId()); +// destroyService(); +// } +// +// @Test +// public void test_Ex_OnReturn() throws Exception { +// initOrResetUrl(true); +// initOrResetExService(); +// initImplicitCallBackURL_onlyOnreturn(); +// +// +// int requestId = 2; +// Person ret = demoProxy.get(requestId); +// Assertions.assertNull(ret); +// for (int i = 0; i < 10; i++) { +// if (!notify.errors.containsKey(requestId)) { +// Thread.sleep(200); +// } else { +// break; +// } +// } +// Assertions.assertTrue(!notify.errors.containsKey(requestId)); +// destroyService(); +// } +// +// @Test +// public void test_Ex_OnInvoke() throws Exception { +// initOrResetUrl(true); +// initOrResetExService(); +// initImplicitCallBackURL_onlyOninvoke(); +// +// int requestId = 2; +// Person ret = demoProxy.get(requestId); +// Assertions.assertNull(ret); +// for (int i = 0; i < 10; i++) { +// if (!notify.inv.contains(requestId)) { +// Thread.sleep(200); +// } else { +// break; +// } +// } +// Assertions.assertTrue(notify.inv.contains(requestId)); +// destroyService(); +// } +// +// @Test +// public void test_Ex_Onthrow() throws Exception { +// initOrResetUrl(true); +// initOrResetExService(); +// initImplicitCallBackURL_onlyOnthrow(); +// +// int requestId = 2; +// Person ret = demoProxy.get(requestId); +// Assertions.assertNull(ret); +// for (int i = 0; i < 10; i++) { +// if (!notify.errors.containsKey(requestId)) { +// Thread.sleep(200); +// } else { +// break; +// } +// } +// Assertions.assertTrue(notify.errors.containsKey(requestId)); +// Assertions.assertTrue(notify.errors.get(requestId) instanceof Throwable); +// destroyService(); +// } +// +// @Test +// public void test_Sync_NoFuture() throws Exception { +// initOrResetUrl(false); +// initOrResetService(); +// initImplicitCallBackURL_onlyOnreturn(); +// +// int requestId = 2; +// Person ret = demoProxy.get(requestId); +// Assertions.assertEquals(requestId, ret.getId()); +// Future pFuture = RpcContext.getContext().getFuture(); +// Assertions.assertEquals(ret, pFuture.get()); +// destroyService(); +// } +// +// @Test +// public void test_Async_Future() throws Exception { +// initOrResetUrl(true); +// initOrResetService(); +// +// int requestId = 2; +// Person ret = demoProxy.get(requestId); +// Assertions.assertNull(ret); +// Future pFuture = RpcContext.getContext().getFuture(); +// ret = pFuture.get(1000 * 1000, TimeUnit.MICROSECONDS); +// Assertions.assertEquals(requestId, ret.getId()); +// destroyService(); +// } +// +// @Test +// public void test_Async_Future_Multi() throws Exception { +// initOrResetUrl(true); +// initOrResetService(); +// +// int requestId1 = 1; +// Person ret = demoProxy.get(requestId1); +// Assertions.assertNull(ret); +// Future p1Future = RpcContext.getContext().getFuture(); +// +// int requestId2 = 1; +// Person ret2 = demoProxy.get(requestId2); +// Assertions.assertNull(ret2); +// Future p2Future = RpcContext.getContext().getFuture(); +// +// ret = p1Future.get(1000 * 1000, TimeUnit.MICROSECONDS); +// ret2 = p2Future.get(1000 * 1000, TimeUnit.MICROSECONDS); +// Assertions.assertEquals(requestId1, ret.getId()); +// Assertions.assertEquals(requestId2, ret.getId()); +// destroyService(); +// } +// +// @Test +// public void test_Async_Future_Ex() throws Throwable { +// Assertions.assertThrows(RuntimeException.class, () -> { +// try { +// initOrResetUrl(true); +// initOrResetExService(); +// +// int requestId = 2; +// Person ret = demoProxy.get(requestId); +// Assertions.assertNull(ret); +// Future pFuture = RpcContext.getContext().getFuture(); +// ret = pFuture.get(1000 * 1000, TimeUnit.MICROSECONDS); +// Assertions.assertEquals(requestId, ret.getId()); +// } catch (ExecutionException e) { +// throw e.getCause(); +// } finally { +// destroyService(); +// } +// }); +// } +// +// @Test +// public void test_Normal_Ex() throws Exception { +// Assertions.assertThrows(RuntimeException.class, () -> { +// initOrResetUrl(false); +// initOrResetExService(); +// +// int requestId = 2; +// Person ret = demoProxy.get(requestId); +// Assertions.assertEquals(requestId, ret.getId()); +// }); +// } +// +// interface Nofify { +// void onreturn(Person msg, Integer id); +// +// void onthrow(Throwable ex, Integer id); +// +// void oninvoke(Integer id); +// } +// +// interface IDemoService { +// Person get(int id); +// } +// +// public static class Person implements Serializable { +// private static final long serialVersionUID = 1L; +// private int id; +// private String name; +// private int age; +// +// public Person(int id, String name, int age) { +// this.id = id; +// this.name = name; +// this.age = age; +// } +// +// public String getName() { +// return name; +// } +// +// public void setName(String name) { +// this.name = name; +// } +// +// public int getAge() { +// return age; +// } +// +// public void setAge(int age) { +// this.age = age; +// } +// +// public int getId() { +// return id; +// } +// +// public void setId(int id) { +// this.id = id; +// } +// +// @Override +// public String toString() { +// return "Person [name=" + name + ", age=" + age + "]"; +// } +// } +// +// class NofifyImpl implements Nofify { +// public List inv = new ArrayList(); +// public Map ret = new HashMap(); +// public Map errors = new HashMap(); +// public boolean exd = false; +// +// public void onreturn(Person msg, Integer id) { +// System.out.println("onNotify:" + msg); +// ret.put(id, msg); +// } +// +// public void onthrow(Throwable ex, Integer id) { +// errors.put(id, ex); +//// ex.printStackTrace(); +// } +// +// public void oninvoke(Integer id) { +// inv.add(id); +// } +// } +// +// class NormalDemoService implements IDemoService { +// public Person get(int id) { +// return new Person(id, "charles", 4); +// } +// } +// +// class ExceptionDemoExService implements IDemoService { +// public Person get(int id) { +// throw new RuntimeException("request persion id is :" + id); +// } +// } +//} diff --git a/dubbo-rpc/dubbo-rpc-hessian/src/main/java/org/apache/dubbo/rpc/protocol/hessian/HttpClientConnectionFactory.java b/dubbo-rpc/dubbo-rpc-hessian/src/main/java/org/apache/dubbo/rpc/protocol/hessian/HttpClientConnectionFactory.java index 0ce28897da..c1a81609a7 100644 --- a/dubbo-rpc/dubbo-rpc-hessian/src/main/java/org/apache/dubbo/rpc/protocol/hessian/HttpClientConnectionFactory.java +++ b/dubbo-rpc/dubbo-rpc-hessian/src/main/java/org/apache/dubbo/rpc/protocol/hessian/HttpClientConnectionFactory.java @@ -16,17 +16,19 @@ */ package org.apache.dubbo.rpc.protocol.hessian; +import org.apache.dubbo.rpc.RpcContext; + import com.caucho.hessian.client.HessianConnection; import com.caucho.hessian.client.HessianConnectionFactory; import com.caucho.hessian.client.HessianProxyFactory; -import org.apache.dubbo.common.Constants; -import org.apache.dubbo.rpc.RpcContext; import org.apache.http.client.HttpClient; import org.apache.http.client.config.RequestConfig; import org.apache.http.impl.client.HttpClientBuilder; import java.net.URL; +import static org.apache.dubbo.remoting.Constants.DEFAULT_EXCHANGER; + /** * HttpClientConnectionFactory */ @@ -48,7 +50,7 @@ public class HttpClientConnectionFactory implements HessianConnectionFactory { HttpClientConnection httpClientConnection = new HttpClientConnection(httpClient, url); RpcContext context = RpcContext.getContext(); for (String key : context.getAttachments().keySet()) { - httpClientConnection.addHeader(Constants.DEFAULT_EXCHANGER + key, (String) context.getAttachment(key)); + httpClientConnection.addHeader(DEFAULT_EXCHANGER + key, (String) context.getAttachment(key)); } return httpClientConnection; } diff --git a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java index 341d04f138..40a09da620 100644 --- a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java +++ b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java @@ -137,7 +137,7 @@ public class RestProtocol extends AbstractProxyProtocol { } @Override - protected T getFrameworkProxy(Class serviceType, URL url) throws RpcException { + protected T doRefer(Class serviceType, URL url) throws RpcException { // TODO more configs to add PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager(); diff --git a/dubbo-rpc/dubbo-rpc-rmi/src/main/java/org/apache/dubbo/rpc/protocol/rmi/RmiProtocol.java b/dubbo-rpc/dubbo-rpc-rmi/src/main/java/org/apache/dubbo/rpc/protocol/rmi/RmiProtocol.java index d0f712331a..d988938507 100644 --- a/dubbo-rpc/dubbo-rpc-rmi/src/main/java/org/apache/dubbo/rpc/protocol/rmi/RmiProtocol.java +++ b/dubbo-rpc/dubbo-rpc-rmi/src/main/java/org/apache/dubbo/rpc/protocol/rmi/RmiProtocol.java @@ -72,7 +72,7 @@ public class RmiProtocol extends AbstractProxyProtocol { @Override @SuppressWarnings("unchecked") - protected T getFrameworkProxy(final Class serviceType, final URL url) throws RpcException { + protected T doRefer(final Class serviceType, final URL url) throws RpcException { final RmiProxyFactoryBean rmiProxyFactoryBean = new RmiProxyFactoryBean(); final String generic = url.getParameter(GENERIC_KEY); final boolean isGeneric = ProtocolUtils.isGeneric(generic) || serviceType.equals(GenericService.class); diff --git a/dubbo-rpc/dubbo-rpc-rsocket/pom.xml b/dubbo-rpc/dubbo-rpc-rsocket/pom.xml deleted file mode 100644 index f11823452b..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/pom.xml +++ /dev/null @@ -1,123 +0,0 @@ - - - 4.0.0 - - org.apache.dubbo - dubbo-rpc - ${revision} - - dubbo-rpc-rsocket - jar - ${project.artifactId} - The default rpc module of dubbo project - - false - - - - org.springframework - spring-context - 4.3.16.RELEASE - - - org.apache.dubbo - dubbo-registry-multicast - ${project.version} - - - io.rsocket - rsocket-core - 0.11.14 - - - io.rsocket - rsocket-transport-netty - 0.11.14 - - - com.alibaba - fastjson - 1.2.54 - - - org.apache.dubbo - dubbo-rpc-api - ${project.parent.version} - - - - org.apache.dubbo - dubbo-remoting-api - ${project.parent.version} - - - org.apache.dubbo - dubbo-config-api - ${project.version} - - - org.apache.dubbo - dubbo-config-spring - ${project.version} - - - org.apache.dubbo - dubbo-container-api - ${project.parent.version} - - - org.eclipse.jetty - jetty-server - - - org.eclipse.jetty - jetty-servlet - - - - - org.apache.dubbo - dubbo-serialization-hessian2 - ${project.parent.version} - test - - - org.apache.dubbo - dubbo-serialization-jdk - ${project.parent.version} - test - - - - - - - - - - - - - - - - - - - diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/FutureSubscriber.java b/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/FutureSubscriber.java deleted file mode 100644 index 6ec7162ec3..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/FutureSubscriber.java +++ /dev/null @@ -1,98 +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.protocol.rsocket; - -import org.apache.dubbo.common.serialize.ObjectInput; -import org.apache.dubbo.common.serialize.Serialization; -import org.apache.dubbo.rpc.AppResponse; - -import io.rsocket.Payload; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; - -import java.io.ByteArrayInputStream; -import java.io.InputStream; -import java.nio.ByteBuffer; -import java.util.Map; -import java.util.concurrent.CompletableFuture; - -public class FutureSubscriber extends CompletableFuture implements Subscriber { - - private final Serialization serialization; - - private final Class retType; - - public FutureSubscriber(Serialization serialization, Class retType) { - this.serialization = serialization; - this.retType = retType; - } - - - @Override - public void onSubscribe(Subscription subscription) { - subscription.request(1); - } - - @Override - public void onNext(Payload payload) { - try { - AppResponse appResponse = new AppResponse(); - ByteBuffer dataBuffer = payload.getData(); - byte[] dataBytes = new byte[dataBuffer.remaining()]; - dataBuffer.get(dataBytes, dataBuffer.position(), dataBuffer.remaining()); - InputStream dataInputStream = new ByteArrayInputStream(dataBytes); - ObjectInput in = serialization.deserialize(null, dataInputStream); - - int flag = in.readByte(); - if ((flag & RSocketConstants.FLAG_ERROR) != 0) { - Throwable t = (Throwable) in.readObject(); - appResponse.setException(t); - } else { - Object value = null; - if ((flag & RSocketConstants.FLAG_NULL_VALUE) == 0) { - if (retType == null) { - value = in.readObject(); - } else { - value = in.readObject(retType); - } - appResponse.setValue(value); - } - } - - if ((flag & RSocketConstants.FLAG_HAS_ATTACHMENT) != 0) { - Map attachment = in.readObject(Map.class); - appResponse.setAttachments(attachment); - - } - - this.complete(appResponse); - - - } catch (Throwable t) { - this.completeExceptionally(t); - } - } - - @Override - public void onError(Throwable throwable) { - this.completeExceptionally(throwable); - } - - @Override - public void onComplete() { - } -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/MetadataCodec.java b/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/MetadataCodec.java deleted file mode 100644 index 8a62597644..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/MetadataCodec.java +++ /dev/null @@ -1,36 +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.protocol.rsocket; - -import com.alibaba.fastjson.JSON; - -import java.io.IOException; -import java.nio.charset.StandardCharsets; -import java.util.Map; - -public class MetadataCodec { - - public static Map decodeMetadata(byte[] bytes) throws IOException { - return JSON.parseObject(new String(bytes, StandardCharsets.UTF_8), Map.class); - } - - public static byte[] encodeMetadata(Map metadata) throws IOException { - String jsonStr = JSON.toJSONString(metadata); - return jsonStr.getBytes(StandardCharsets.UTF_8); - } - -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketConstants.java b/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketConstants.java deleted file mode 100644 index a252dbd4b6..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketConstants.java +++ /dev/null @@ -1,32 +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.protocol.rsocket; - -public class RSocketConstants { - - public static final String SERVICE_NAME_KEY = "_service_name"; - public static final String SERVICE_VERSION_KEY = "_service_version"; - public static final String METHOD_NAME_KEY = "_method_name"; - public static final String PARAM_TYPE_KEY = "_param_type"; - public static final String SERIALIZE_TYPE_KEY = "_serialize_type"; - public static final String TIMEOUT_KEY = "_timeout"; - - - public static final int FLAG_ERROR = 0x01; - public static final int FLAG_NULL_VALUE = 0x02; - public static final int FLAG_HAS_ATTACHMENT = 0x04; -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketExporter.java b/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketExporter.java deleted file mode 100644 index 3ebaea2660..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketExporter.java +++ /dev/null @@ -1,43 +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.protocol.rsocket; - -import org.apache.dubbo.rpc.Exporter; -import org.apache.dubbo.rpc.Invoker; -import org.apache.dubbo.rpc.protocol.AbstractExporter; - -import java.util.Map; - -public class RSocketExporter extends AbstractExporter { - - private final String key; - - private final Map> exporterMap; - - public RSocketExporter(Invoker invoker, String key, Map> exporterMap) { - super(invoker); - this.key = key; - this.exporterMap = exporterMap; - } - - @Override - public void unexport() { - super.unexport(); - exporterMap.remove(key); - } - -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketInvoker.java b/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketInvoker.java deleted file mode 100644 index 753643a1f8..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketInvoker.java +++ /dev/null @@ -1,263 +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.protocol.rsocket; - -import org.apache.dubbo.common.Constants; -import org.apache.dubbo.common.URL; -import org.apache.dubbo.common.serialize.Cleanable; -import org.apache.dubbo.common.serialize.ObjectInput; -import org.apache.dubbo.common.serialize.ObjectOutput; -import org.apache.dubbo.common.serialize.Serialization; -import org.apache.dubbo.common.utils.AtomicPositiveInteger; -import org.apache.dubbo.common.utils.ReflectUtils; -import org.apache.dubbo.remoting.transport.CodecSupport; -import org.apache.dubbo.rpc.AsyncRpcResult; -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.protocol.AbstractInvoker; -import org.apache.dubbo.rpc.support.RpcUtils; - -import io.rsocket.Payload; -import io.rsocket.RSocket; -import io.rsocket.util.DefaultPayload; -import reactor.core.Exceptions; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.InputStream; -import java.nio.ByteBuffer; -import java.util.HashMap; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.locks.ReentrantLock; -import java.util.function.Function; - -public class RSocketInvoker extends AbstractInvoker { - - private final RSocket[] clients; - - private final AtomicPositiveInteger index = new AtomicPositiveInteger(); - - private final String version; - - private final ReentrantLock destroyLock = new ReentrantLock(); - - private final Set> invokers; - - private final Serialization serialization; - - public RSocketInvoker(Class serviceType, URL url, RSocket[] clients, Set> invokers) { - super(serviceType, url, new String[]{Constants.INTERFACE_KEY, Constants.GROUP_KEY, Constants.TOKEN_KEY, Constants.TIMEOUT_KEY}); - this.clients = clients; - // get version. - this.version = url.getParameter(Constants.VERSION_KEY, "0.0.0"); - this.invokers = invokers; - - this.serialization = CodecSupport.getSerialization(getUrl()); - } - - @Override - protected Result doInvoke(final Invocation invocation) throws Throwable { - RpcInvocation inv = (RpcInvocation) invocation; - final String methodName = RpcUtils.getMethodName(invocation); - inv.setAttachment(Constants.PATH_KEY, getUrl().getPath()); - inv.setAttachment(Constants.VERSION_KEY, version); - - RSocket currentClient; - if (clients.length == 1) { - currentClient = clients[0]; - } else { - currentClient = clients[index.getAndIncrement() % clients.length]; - } - try { - //TODO support timeout - int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT); - - Class retType = RpcUtils.getReturnType(invocation); - - RpcContext.getContext().setFuture(null); - //encode inv: metadata and data(arg,attachment) - Payload requestPayload = encodeInvocation(invocation); - - if (retType != null && retType.isAssignableFrom(Mono.class)) { - Mono responseMono = currentClient.requestResponse(requestPayload); - Mono bizMono = responseMono.map(new Function() { - @Override - public Object apply(Payload payload) { - return decodeData(payload); - } - }); - return AsyncRpcResult.newDefaultAsyncResult(bizMono, invocation); - } else if (retType != null && retType.isAssignableFrom(Flux.class)) { - return AsyncRpcResult.newDefaultAsyncResult(requestStream(currentClient, requestPayload), invocation); - } else { - //request-reponse - Mono responseMono = currentClient.requestResponse(requestPayload); - FutureSubscriber futureSubscriber = new FutureSubscriber(serialization, retType); - responseMono.subscribe(futureSubscriber); - return AsyncRpcResult.newDefaultAsyncResult(futureSubscriber.get(), invocation); - } - - //TODO support stream arg - } catch (Throwable t) { - throw new RpcException(t); - } - } - - - private Flux requestStream(RSocket currentClient, Payload requestPayload) { - Flux responseFlux = currentClient.requestStream(requestPayload); - Flux retFlux = responseFlux.map(new Function() { - - @Override - public Object apply(Payload payload) { - Object o = decodeData(payload); - payload.release(); - return o; - } - }); - - return retFlux; - } - - - private Object decodeData(Payload payload) { - try { - ByteBuffer dataBuffer = payload.getData(); - byte[] dataBytes = new byte[dataBuffer.remaining()]; - dataBuffer.get(dataBytes, dataBuffer.position(), dataBuffer.remaining()); - InputStream dataInputStream = new ByteArrayInputStream(dataBytes); - ObjectInput in = serialization.deserialize(null, dataInputStream); - //TODO save the copy - int flag = in.readByte(); - if ((flag & RSocketConstants.FLAG_ERROR) != 0) { - Throwable t = (Throwable) in.readObject(); - throw t; - } else { - return in.readObject(); - } - } catch (Throwable t) { - throw Exceptions.propagate(t); - } - } - - @Override - public boolean isAvailable() { - if (!super.isAvailable()) { - return false; - } - for (RSocket client : clients) { - if (client.availability() > 0) { - return true; - } - } - return false; - } - - @Override - public void destroy() { - // in order to avoid closing a client multiple times, a counter is used in case of connection per jvm, every - // time when client.close() is called, counter counts down once, and when counter reaches zero, client will be - // closed. - if (super.isDestroyed()) { - return; - } else { - // double check to avoid dup close - destroyLock.lock(); - try { - if (super.isDestroyed()) { - return; - } - super.destroy(); - if (invokers != null) { - invokers.remove(this); - } - for (RSocket client : clients) { - try { - client.dispose(); - } catch (Throwable t) { - logger.warn(t.getMessage(), t); - } - } - - } finally { - destroyLock.unlock(); - } - } - } - - private Payload encodeInvocation(Invocation invocation) throws IOException { - //process stream args - RpcInvocation inv = (RpcInvocation) invocation; - Class[] parameterTypes = invocation.getParameterTypes(); - Object[] args = inv.getArguments(); - if (args != null) { - for (int i = 0; i < args.length; i++) { - if(args[i]!=null) { - Class argClass = args[i].getClass(); - if (Mono.class.isAssignableFrom(argClass)) { - long id = ResourceDirectory.mountResource(args[i]); - args[i] = new ResourceInfo(id, ResourceInfo.RESOURCE_TYPE_MONO); - parameterTypes[i] = ResourceInfo.class; - } else if (Flux.class.isAssignableFrom(argClass)) { - long id = ResourceDirectory.mountResource(args[i]); - args[i] = new ResourceInfo(id, ResourceInfo.RESOURCE_TYPE_FLUX); - parameterTypes[i] = ResourceInfo.class; - } - } - } - } - - //metadata - Map metadataMap = new HashMap(); - metadataMap.put(RSocketConstants.SERVICE_NAME_KEY, invocation.getAttachment(Constants.PATH_KEY)); - metadataMap.put(RSocketConstants.SERVICE_VERSION_KEY, invocation.getAttachment(Constants.VERSION_KEY)); - metadataMap.put(RSocketConstants.METHOD_NAME_KEY, invocation.getMethodName()); - metadataMap.put(RSocketConstants.SERIALIZE_TYPE_KEY, (Byte) serialization.getContentTypeId()); - metadataMap.put(RSocketConstants.PARAM_TYPE_KEY, ReflectUtils.getDesc(parameterTypes)); - byte[] metadata = MetadataCodec.encodeMetadata(metadataMap); - - - //data - ByteArrayOutputStream dataOutputStream = new ByteArrayOutputStream(); - Serialization serialization = CodecSupport.getSerialization(getUrl()); - ObjectOutput out = serialization.serialize(getUrl(), dataOutputStream); - if (args != null) { - for (int i = 0; i < args.length; i++) { - out.writeObject(args[i]); - } - } - out.writeObject(RpcUtils.getNecessaryAttachments(inv)); - - //clean - out.flushBuffer(); - if (out instanceof Cleanable) { - ((Cleanable) out).cleanup(); - } - byte[] data = dataOutputStream.toByteArray(); - - - return DefaultPayload.create(data, metadata); - } -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketProtocol.java b/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketProtocol.java deleted file mode 100644 index 756a7a25ba..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketProtocol.java +++ /dev/null @@ -1,727 +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.protocol.rsocket; - -import io.rsocket.AbstractRSocket; -import io.rsocket.ConnectionSetupPayload; -import io.rsocket.Payload; -import io.rsocket.RSocket; -import io.rsocket.RSocketFactory; -import io.rsocket.SocketAcceptor; -import io.rsocket.transport.netty.client.TcpClientTransport; -import io.rsocket.transport.netty.server.CloseableChannel; -import io.rsocket.transport.netty.server.TcpServerTransport; -import io.rsocket.util.DefaultPayload; -import org.apache.dubbo.common.Constants; -import org.apache.dubbo.common.URL; -import org.apache.dubbo.common.extension.ExtensionLoader; -import org.apache.dubbo.common.logger.Logger; -import org.apache.dubbo.common.logger.LoggerFactory; -import org.apache.dubbo.common.serialize.ObjectInput; -import org.apache.dubbo.common.serialize.ObjectOutput; -import org.apache.dubbo.common.serialize.Serialization; -import org.apache.dubbo.common.utils.NetUtils; -import org.apache.dubbo.common.utils.ReflectUtils; -import org.apache.dubbo.remoting.RemotingException; -import org.apache.dubbo.remoting.transport.CodecSupport; -import org.apache.dubbo.rpc.Exporter; -import org.apache.dubbo.rpc.Invocation; -import org.apache.dubbo.rpc.Invoker; -import org.apache.dubbo.rpc.Protocol; -import org.apache.dubbo.rpc.Result; -import org.apache.dubbo.rpc.RpcException; -import org.apache.dubbo.rpc.RpcInvocation; -import org.apache.dubbo.rpc.protocol.AbstractProtocol; -import org.apache.dubbo.rpc.support.RpcUtils; -import org.reactivestreams.Publisher; -import reactor.core.Exceptions; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.InputStream; -import java.net.InetSocketAddress; -import java.nio.ByteBuffer; -import java.time.Duration; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.function.Function; - -public class RSocketProtocol extends AbstractProtocol { - - public static final String NAME = "rsocket"; - public static final int DEFAULT_PORT = 30880; - private static final Logger log = LoggerFactory.getLogger(RSocketProtocol.class); - private static RSocketProtocol INSTANCE; - - // - private final Map serverMap = new ConcurrentHashMap(); - - // - private final Map referenceClientMap = new ConcurrentHashMap(); - - private final ConcurrentMap locks = new ConcurrentHashMap(); - - public RSocketProtocol() { - INSTANCE = this; - } - - public static RSocketProtocol getRSocketProtocol() { - if (INSTANCE == null) { - ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(RSocketProtocol.NAME); // load - } - return INSTANCE; - } - - public Collection> getExporters() { - return Collections.unmodifiableCollection(exporterMap.values()); - } - - Map> getExporterMap() { - return exporterMap; - } - - Invoker getInvoker(int port, Map metadataMap) throws RemotingException { - String path = (String) metadataMap.get(RSocketConstants.SERVICE_NAME_KEY); - String serviceKey = serviceKey(port, path, (String) metadataMap.get(RSocketConstants.SERVICE_VERSION_KEY), (String) metadataMap.get(Constants.GROUP_KEY)); - RSocketExporter exporter = (RSocketExporter) exporterMap.get(serviceKey); - if (exporter == null) { - //throw new Throwable("Not found exported service: " + serviceKey + " in " + exporterMap.keySet() + ", may be version or group mismatch " + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress() + ", message:" + inv); - throw new RuntimeException("Not found exported service: " + serviceKey + " in " + exporterMap.keySet() + ", may be version or group mismatch "); - } - - return exporter.getInvoker(); - } - - public Collection> getInvokers() { - return Collections.unmodifiableCollection(invokers); - } - - @Override - public int getDefaultPort() { - return DEFAULT_PORT; - } - - @Override - public Exporter export(Invoker invoker) throws RpcException { - URL url = invoker.getUrl(); - - // export service. - String key = serviceKey(url); - RSocketExporter exporter = new RSocketExporter(invoker, key, exporterMap); - exporterMap.put(key, exporter); - - openServer(url); - return exporter; - } - - private void openServer(URL url) { - String key = url.getAddress(); - //client can export a service which's only for server to invoke - boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true); - if (isServer) { - CloseableChannel server = serverMap.get(key); - if (server == null) { - synchronized (this) { - server = serverMap.get(key); - if (server == null) { - serverMap.put(key, createServer(url)); - } - } - } - } - } - - private CloseableChannel createServer(URL url) { - try { - String bindIp = url.getParameter(Constants.BIND_IP_KEY, url.getHost()); - int bindPort = url.getParameter(Constants.BIND_PORT_KEY, url.getPort()); - if (url.getParameter(Constants.ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) { - bindIp = Constants.ANYHOST_VALUE; - } - return RSocketFactory.receive() - .acceptor(new SocketAcceptorImpl(bindPort)) - .transport(TcpServerTransport.create(bindIp, bindPort)) - .start() - .block(); - } catch (Throwable e) { - throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e); - } - } - - @Override - protected Invoker doRefer(Class serviceType, URL url) throws RpcException { - // create rpc invoker. - RSocketInvoker invoker = new RSocketInvoker(serviceType, url, getClients(url), invokers); - invokers.add(invoker); - return invoker; - } - - private RSocket[] getClients(URL url) { - // whether to share connection - boolean service_share_connect = false; - int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0); - // if not configured, connection is shared, otherwise, one connection for one service - if (connections == 0) { - service_share_connect = true; - connections = 1; - } - - RSocket[] clients = new RSocket[connections]; - for (int i = 0; i < clients.length; i++) { - if (service_share_connect) { - clients[i] = getSharedClient(url); - } else { - clients[i] = initClient(url); - } - } - return clients; - } - - /** - * Get shared connection - */ - private RSocket getSharedClient(URL url) { - String key = url.getAddress(); - RSocket client = referenceClientMap.get(key); - if (client != null) { - return client; - } - - locks.putIfAbsent(key, new Object()); - synchronized (locks.get(key)) { - if (referenceClientMap.containsKey(key)) { - return referenceClientMap.get(key); - } - - client = initClient(url); - referenceClientMap.put(key, client); - locks.remove(key); - return client; - } - } - - /** - * Create new connection - */ - private RSocket initClient(URL url) { - try { - InetSocketAddress serverAddress = new InetSocketAddress(NetUtils.filterLocalHost(url.getHost()), url.getPort()); - RSocket client = RSocketFactory.connect().keepAliveTickPeriod(Duration.ZERO).keepAliveAckTimeout(Duration.ZERO).acceptor( - rSocket -> - new AbstractRSocket() { - public Mono requestResponse(Payload payload) { - try { - ByteBuffer metadataBuffer = payload.getMetadata(); - byte[] metadataBytes = new byte[metadataBuffer.remaining()]; - metadataBuffer.get(metadataBytes, metadataBuffer.position(), metadataBuffer.remaining()); - Map metadataMap = MetadataCodec.decodeMetadata(metadataBytes); - Byte serializeId = ((Integer) metadataMap.get(RSocketConstants.SERIALIZE_TYPE_KEY)).byteValue(); - - - ByteBuffer dataBuffer = payload.getData(); - byte[] dataBytes = new byte[dataBuffer.remaining()]; - dataBuffer.get(dataBytes, dataBuffer.position(), dataBuffer.remaining()); - InputStream dataInputStream = new ByteArrayInputStream(dataBytes); - ObjectInput in = CodecSupport.getSerializationById(serializeId).deserialize(null, dataInputStream); - long id = in.readLong(); - - Mono mono = ResourceDirectory.unmountMono(id); - return mono.map(new Function() { - @Override - public Payload apply(Object o) { - try { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeByte((byte) 0); - out.writeObject(o); - out.flushBuffer(); - bos.flush(); - bos.close(); - Payload responsePayload = DefaultPayload.create(bos.toByteArray()); - return responsePayload; - } catch (Throwable t) { - throw Exceptions.propagate(t); - } - } - }).onErrorResume(new Function>() { - @Override - public Publisher apply(Throwable throwable) { - try { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeByte((byte) RSocketConstants.FLAG_ERROR); - out.writeObject(throwable); - out.flushBuffer(); - bos.flush(); - bos.close(); - Payload errorPayload = DefaultPayload.create(bos.toByteArray()); - return Flux.just(errorPayload); - } catch (Throwable t) { - throw Exceptions.propagate(t); - } - } - }); - - } catch (Throwable t) { - throw new RuntimeException(t); - } - } - - @Override - public Flux requestStream(Payload payload) { - try { - ByteBuffer metadataBuffer = payload.getMetadata(); - byte[] metadataBytes = new byte[metadataBuffer.remaining()]; - metadataBuffer.get(metadataBytes, metadataBuffer.position(), metadataBuffer.remaining()); - Map metadataMap = MetadataCodec.decodeMetadata(metadataBytes); - Byte serializeId = ((Integer) metadataMap.get(RSocketConstants.SERIALIZE_TYPE_KEY)).byteValue(); - - - ByteBuffer dataBuffer = payload.getData(); - byte[] dataBytes = new byte[dataBuffer.remaining()]; - dataBuffer.get(dataBytes, dataBuffer.position(), dataBuffer.remaining()); - InputStream dataInputStream = new ByteArrayInputStream(dataBytes); - ObjectInput in = CodecSupport.getSerializationById(serializeId).deserialize(null, dataInputStream); - long id = in.readLong(); - - Flux flux = ResourceDirectory.unmountFlux(id); - return flux.map(new Function() { - @Override - public Payload apply(Object o) { - try { - return encodeData(o, serializeId); - } catch (Throwable t) { - throw new RuntimeException(t); - } - } - }).onErrorResume(new Function>() { - @Override - public Publisher apply(Throwable throwable) { - try { - Payload errorPayload = encodeError(throwable, serializeId); - return Flux.just(errorPayload); - } catch (Throwable t) { - throw new RuntimeException(t); - } - } - }); - } catch (Throwable t) { - throw new RuntimeException(t); - } - } - - private Payload encodeData(Object data, byte serializeId) throws Throwable { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeByte((byte) 0); - out.writeObject(data); - out.flushBuffer(); - bos.flush(); - bos.close(); - return DefaultPayload.create(bos.toByteArray()); - } - - private Payload encodeError(Throwable throwable, byte serializeId) throws Throwable { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeByte((byte) RSocketConstants.FLAG_ERROR); - out.writeObject(throwable); - out.flushBuffer(); - bos.flush(); - bos.close(); - return DefaultPayload.create(bos.toByteArray()); - } - - }) - .transport(TcpClientTransport.create(serverAddress)) - .start() - .block(); - return client; - } catch (Throwable e) { - throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e); - } - - } - - @Override - public void destroy() { - for (String key : new ArrayList(serverMap.keySet())) { - CloseableChannel server = serverMap.remove(key); - if (server != null) { - try { - if (logger.isInfoEnabled()) { - logger.info("Close dubbo server: " + server.address()); - } - server.dispose(); - } catch (Throwable t) { - logger.warn(t.getMessage(), t); - } - } - } - - for (String key : new ArrayList(referenceClientMap.keySet())) { - RSocket client = referenceClientMap.remove(key); - if (client != null) { - try { -// if (logger.isInfoEnabled()) { -// logger.info("Close dubbo connect: " + client. + "-->" + client.getRemoteAddress()); -// } - client.dispose(); - } catch (Throwable t) { - logger.warn(t.getMessage(), t); - } - } - } - super.destroy(); - } - - - //server process logic - private class SocketAcceptorImpl implements SocketAcceptor { - - private final int port; - - public SocketAcceptorImpl(int port) { - this.port = port; - } - - @Override - public Mono accept(ConnectionSetupPayload setupPayload, RSocket reactiveSocket) { - return Mono.just( - new AbstractRSocket() { - public Mono requestResponse(Payload payload) { - try { - Map metadata = decodeMetadata(payload); - Byte serializeId = ((Integer) metadata.get(RSocketConstants.SERIALIZE_TYPE_KEY)).byteValue(); - Invocation inv = decodeInvocation(payload, metadata, serializeId); - - Result result = inv.getInvoker().invoke(inv); - - Class retType = RpcUtils.getReturnType(inv); - //ok - if (retType != null && Mono.class.isAssignableFrom(retType)) { - Throwable th = result.getException(); - if (th == null) { - Mono bizMono = (Mono) result.getValue(); - Mono retMono = bizMono.map(new Function() { - @Override - public Payload apply(Object o) { - try { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeByte((byte) 0); - out.writeObject(o); - out.flushBuffer(); - bos.flush(); - bos.close(); - Payload responsePayload = DefaultPayload.create(bos.toByteArray()); - return responsePayload; - } catch (Throwable t) { - throw Exceptions.propagate(t); - } - } - }).onErrorResume(new Function>() { - @Override - public Publisher apply(Throwable throwable) { - try { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeByte((byte) RSocketConstants.FLAG_ERROR); - out.writeObject(throwable); - out.flushBuffer(); - bos.flush(); - bos.close(); - Payload errorPayload = DefaultPayload.create(bos.toByteArray()); - return Flux.just(errorPayload); - } catch (Throwable t) { - throw Exceptions.propagate(t); - } - } - }); - - return retMono; - } else { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeByte((byte) RSocketConstants.FLAG_ERROR); - out.writeObject(th); - out.flushBuffer(); - bos.flush(); - bos.close(); - Payload errorPayload = DefaultPayload.create(bos.toByteArray()); - return Mono.just(errorPayload); - } - - } else { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - int flag = RSocketConstants.FLAG_HAS_ATTACHMENT; - - Throwable th = result.getException(); - if (th == null) { - Object ret = result.getValue(); - if (ret == null) { - flag |= RSocketConstants.FLAG_NULL_VALUE; - out.writeByte((byte) flag); - } else { - out.writeByte((byte) flag); - out.writeObject(ret); - } - } else { - flag |= RSocketConstants.FLAG_ERROR; - out.writeByte((byte) flag); - out.writeObject(th); - } - out.writeObject(result.getAttachments()); - out.flushBuffer(); - bos.flush(); - bos.close(); - - Payload responsePayload = DefaultPayload.create(bos.toByteArray()); - return Mono.just(responsePayload); - } - } catch (Throwable t) { - //application error - return Mono.error(t); - } finally { - payload.release(); - } - } - - public Flux requestStream(Payload payload) { - try { - Map metadata = decodeMetadata(payload); - Byte serializeId = ((Integer) metadata.get(RSocketConstants.SERIALIZE_TYPE_KEY)).byteValue(); - Invocation inv = decodeInvocation(payload, metadata, serializeId); - - Result result = inv.getInvoker().invoke(inv); - //Class retType = RpcUtils.getReturnType(inv); - - Throwable th = result.getException(); - if (th != null) { - Payload errorPayload = encodeError(th, serializeId); - return Flux.just(errorPayload); - } - - Flux flux = (Flux) result.getValue(); - Flux retFlux = flux.map(new Function() { - @Override - public Payload apply(Object o) { - try { - return encodeData(o, serializeId); - } catch (Throwable t) { - throw new RuntimeException(t); - } - } - }).onErrorResume(new Function>() { - @Override - public Publisher apply(Throwable throwable) { - try { - Payload errorPayload = encodeError(throwable, serializeId); - return Flux.just(errorPayload); - } catch (Throwable t) { - throw new RuntimeException(t); - } - } - }); - return retFlux; - } catch (Throwable t) { - return Flux.error(t); - } finally { - payload.release(); - } - } - - private Payload encodeData(Object data, byte serializeId) throws Throwable { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeByte((byte) 0); - out.writeObject(data); - out.flushBuffer(); - bos.flush(); - bos.close(); - return DefaultPayload.create(bos.toByteArray()); - } - - private Payload encodeError(Throwable throwable, byte serializeId) throws Throwable { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeByte((byte) RSocketConstants.FLAG_ERROR); - out.writeObject(throwable); - out.flushBuffer(); - bos.flush(); - bos.close(); - return DefaultPayload.create(bos.toByteArray()); - } - - private Map decodeMetadata(Payload payload) throws IOException { - ByteBuffer metadataBuffer = payload.getMetadata(); - byte[] metadataBytes = new byte[metadataBuffer.remaining()]; - metadataBuffer.get(metadataBytes, metadataBuffer.position(), metadataBuffer.remaining()); - return MetadataCodec.decodeMetadata(metadataBytes); - } - - private Invocation decodeInvocation(Payload payload, Map metadata, Byte serializeId) throws RemotingException, IOException, ClassNotFoundException { - Invoker invoker = getInvoker(port, metadata); - - String serviceName = (String) metadata.get(RSocketConstants.SERVICE_NAME_KEY); - String version = (String) metadata.get(RSocketConstants.SERVICE_VERSION_KEY); - String methodName = (String) metadata.get(RSocketConstants.METHOD_NAME_KEY); - String paramType = (String) metadata.get(RSocketConstants.PARAM_TYPE_KEY); - - ByteBuffer dataBuffer = payload.getData(); - byte[] dataBytes = new byte[dataBuffer.remaining()]; - dataBuffer.get(dataBytes, dataBuffer.position(), dataBuffer.remaining()); - - - //TODO how to get remote address - //RpcContext rpcContext = RpcContext.getContext(); - //rpcContext.setRemoteAddress(channel.getRemoteAddress()); - - - RpcInvocation inv = new RpcInvocation(); - inv.setInvoker(invoker); - inv.setAttachment(Constants.PATH_KEY, serviceName); - inv.setAttachment(Constants.VERSION_KEY, version); - inv.setMethodName(methodName); - - - InputStream dataInputStream = new ByteArrayInputStream(dataBytes); - ObjectInput in = CodecSupport.getSerializationById(serializeId).deserialize(null, dataInputStream); - - Object[] args; - Class[] pts; - String desc = paramType; - if (desc.length() == 0) { - pts = new Class[0]; - args = new Object[0]; - } else { - pts = ReflectUtils.desc2classArray(desc); - args = new Object[pts.length]; - for (int i = 0; i < args.length; i++) { - try { - args[i] = in.readObject(pts[i]); - } catch (Exception e) { - if (log.isWarnEnabled()) { - log.warn("Decode argument failed: " + e.getMessage(), e); - } - } - } - } - - //process stream args - for (int i = 0; i < pts.length; i++) { - if (ResourceInfo.class.isAssignableFrom(pts[i])) { - ResourceInfo resourceInfo = (ResourceInfo) args[i]; - if (resourceInfo.getType() == ResourceInfo.RESOURCE_TYPE_MONO) { - pts[i] = Mono.class; - args[i] = getMonoProxy(resourceInfo.getId(), serializeId, reactiveSocket); - } else { - pts[i] = Flux.class; - args[i] = getFluxProxy(resourceInfo.getId(), serializeId, reactiveSocket); - } - } - } - - inv.setParameterTypes(pts); - inv.setArguments(args); - Map map = (Map) in.readObject(Map.class); - if (map != null && map.size() > 0) { - inv.addAttachments(map); - } - return inv; - } - }); - } - - private Mono getMonoProxy(long id, Byte serializeId, RSocket rSocket) throws IOException { - Map metadataMap = new HashMap(); - metadataMap.put(RSocketConstants.SERIALIZE_TYPE_KEY, serializeId); - byte[] metadata = MetadataCodec.encodeMetadata(metadataMap); - - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeLong(id); - out.flushBuffer(); - bos.flush(); - bos.close(); - Payload payload = DefaultPayload.create(bos.toByteArray(), metadata); - - Mono payloads = rSocket.requestResponse(payload); - Mono streamArg = payloads.map(new Function() { - @Override - public Object apply(Payload payload) { - return decodeData(serializeId, payload); - } - }); - return streamArg; - } - - private Flux getFluxProxy(long id, Byte serializeId, RSocket rSocket) throws IOException { - Map metadataMap = new HashMap(); - metadataMap.put(RSocketConstants.SERIALIZE_TYPE_KEY, serializeId); - byte[] metadata = MetadataCodec.encodeMetadata(metadataMap); - - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = CodecSupport.getSerializationById(serializeId).serialize(null, bos); - out.writeLong(id); - out.flushBuffer(); - bos.flush(); - bos.close(); - Payload payload = DefaultPayload.create(bos.toByteArray(), metadata); - - Flux payloads = rSocket.requestStream(payload); - Flux streamArg = payloads.map(new Function() { - @Override - public Object apply(Payload payload) { - return decodeData(serializeId, payload); - } - }); - return streamArg; - } - - private Object decodeData(Byte serializeId, Payload payload) { - try { - Serialization serialization = CodecSupport.getSerializationById(serializeId); - //TODO save the copy - ByteBuffer dataBuffer = payload.getData(); - byte[] dataBytes = new byte[dataBuffer.remaining()]; - dataBuffer.get(dataBytes, dataBuffer.position(), dataBuffer.remaining()); - InputStream dataInputStream = new ByteArrayInputStream(dataBytes); - ObjectInput in = serialization.deserialize(null, dataInputStream); - int flag = in.readByte(); - if ((flag & RSocketConstants.FLAG_ERROR) != 0) { - Throwable t = (Throwable) in.readObject(); - throw t; - } else { - return in.readObject(); - } - } catch (Throwable t) { - throw Exceptions.propagate(t); - } - } - - } -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/ResourceDirectory.java b/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/ResourceDirectory.java deleted file mode 100644 index c1b66ac88d..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/ResourceDirectory.java +++ /dev/null @@ -1,62 +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.protocol.rsocket; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicLong; - -public class ResourceDirectory { - - private static AtomicLong idGen = new AtomicLong(1); - - private static ConcurrentHashMap id2ResourceMap = new ConcurrentHashMap(); - - - public static long mountResource(Object resource) { - long id = idGen.getAndIncrement(); - id2ResourceMap.put(id, resource); - return id; - } - - public static Object unmountResource(long id) { - return id2ResourceMap.get(id); - } - - public static long mountMono(Mono mono) { - long id = idGen.getAndIncrement(); - id2ResourceMap.put(id, mono); - return id; - } - - public static long mountFlux(Flux flux) { - long id = idGen.getAndIncrement(); - id2ResourceMap.put(id, flux); - return id; - } - - public static Mono unmountMono(long id) { - return (Mono) id2ResourceMap.get(id); - } - - public static Flux unmountFlux(long id) { - return (Flux) id2ResourceMap.get(id); - } - -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/ResourceInfo.java b/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/ResourceInfo.java deleted file mode 100644 index 1c1275bd38..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/ResourceInfo.java +++ /dev/null @@ -1,41 +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.protocol.rsocket; - -import java.io.Serializable; - -public class ResourceInfo implements Serializable { - - public static final byte RESOURCE_TYPE_MONO = 1; - public static final byte RESOURCE_TYPE_FLUX = 2; - - private final long id; - private final byte type; - - public ResourceInfo(long id, byte type) { - this.id = id; - this.type = type; - } - - public long getId() { - return id; - } - - public byte getType() { - return type; - } -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol b/dubbo-rpc/dubbo-rpc-rsocket/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol deleted file mode 100644 index 4f03810c97..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol +++ /dev/null @@ -1 +0,0 @@ -rsocket=org.apache.dubbo.rpc.protocol.rsocket.RSocketProtocol \ No newline at end of file diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/ConsumerDemo.java b/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/ConsumerDemo.java deleted file mode 100644 index d73bbae624..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/ConsumerDemo.java +++ /dev/null @@ -1,51 +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.protocol.rsocket; - -import org.apache.dubbo.rpc.service.DemoService; -import org.springframework.context.support.ClassPathXmlApplicationContext; -import reactor.core.publisher.Mono; - -import java.util.function.Consumer; - -public class ConsumerDemo { - - public static void main(String[] args) { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"spring/dubbo-rsocket-consumer.xml"}); - context.start(); - DemoService demoService = (DemoService) context.getBean("demoService"); // get remote service proxy - - while (true) { - try { - Thread.sleep(1000); - Mono resultMono = demoService.requestMono("world"); // call remote method - resultMono.doOnNext(new Consumer() { - @Override - public void accept(String s) { - System.out.println(s); // get result - } - }).block(); - } catch (Throwable throwable) { - throwable.printStackTrace(); - } - - - } - - } - -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/ProviderDemo.java b/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/ProviderDemo.java deleted file mode 100644 index 2e7466ddbf..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/ProviderDemo.java +++ /dev/null @@ -1,29 +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.protocol.rsocket; - -import org.springframework.context.support.ClassPathXmlApplicationContext; - -public class ProviderDemo { - - public static void main(String[] args) throws Exception { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"spring/dubbo-rsocket-provider.xml"}); - context.start(); - System.in.read(); // press any key to exit - } - -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketProtocolTest.java b/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketProtocolTest.java deleted file mode 100644 index 4d834a111c..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketProtocolTest.java +++ /dev/null @@ -1,260 +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.protocol.rsocket; - -import org.apache.dubbo.common.URL; -import org.apache.dubbo.common.extension.ExtensionLoader; -import org.apache.dubbo.rpc.Protocol; -import org.apache.dubbo.rpc.ProxyFactory; -import org.apache.dubbo.rpc.service.DemoException; -import org.apache.dubbo.rpc.service.DemoService; -import org.apache.dubbo.rpc.service.DemoServiceImpl; -import org.apache.dubbo.rpc.service.EchoService; -import org.apache.dubbo.rpc.service.RemoteService; -import org.apache.dubbo.rpc.service.RemoteServiceImpl; - -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.Test; -import org.reactivestreams.Publisher; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import java.util.HashMap; -import java.util.Map; -import java.util.Set; -import java.util.function.Consumer; -import java.util.function.Function; - -import static org.junit.jupiter.api.Assertions.assertEquals; - -public class RSocketProtocolTest { - - private Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension(); - private ProxyFactory proxy = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension(); - - @AfterAll - public static void after() { - RSocketProtocol.getRSocketProtocol().destroy(); - } - - @Test - public void testDemoProtocol() throws Exception { - DemoService service = new DemoServiceImpl(); - protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("rsocket://127.0.0.1:9020/" + DemoService.class.getName()))); - service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("rsocket://127.0.0.1:9020/" + DemoService.class.getName()).addParameter("timeout", 3000l))); - assertEquals(service.getSize(new String[]{"", "", ""}), 3); - } - - @Test - public void testDubboProtocol() throws Exception { - DemoService service = new DemoServiceImpl(); - protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("rsocket://127.0.0.1:9010/" + DemoService.class.getName()))); - service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("rsocket://127.0.0.1:9010/" + DemoService.class.getName()).addParameter("timeout", 3000l))); - - assertEquals(service.getSize(null), -1); - assertEquals(service.getSize(new String[]{"", "", ""}), 3); - - - Map map = new HashMap(); - map.put("aa", "bb"); - Set set = service.keys(map); - assertEquals(set.size(), 1); - assertEquals(set.iterator().next(), "aa"); - service.invoke("rsocket://127.0.0.1:9010/" + DemoService.class.getName() + "", "invoke"); - - StringBuffer buf = new StringBuffer(); - for (int i = 0; i < 1024 * 32 + 32; i++) - buf.append('A'); - System.out.println(service.stringLength(buf.toString())); - - // cast to EchoService - EchoService echo = proxy.getProxy(protocol.refer(EchoService.class, URL.valueOf("rsocket://127.0.0.1:9010/" + DemoService.class.getName()).addParameter("timeout", 3000l))); - assertEquals(echo.$echo(buf.toString()), buf.toString()); - assertEquals(echo.$echo("test"), "test"); - assertEquals(echo.$echo("abcdefg"), "abcdefg"); - assertEquals(echo.$echo(1234), 1234); - } - - - @Test - public void testDubboProtocolThrowable() throws Exception { - DemoService service = new DemoServiceImpl(); - protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("rsocket://127.0.0.1:9010/" + DemoService.class.getName()))); - service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("rsocket://127.0.0.1:9010/" + DemoService.class.getName()).addParameter("timeout", 3000l))); - try { - service.errorTest("mike"); - } catch (Throwable t) { - assertEquals(t.getClass(), ArithmeticException.class); - } - } - - @Test - public void testDubboProtocolMultiService() throws Exception { - DemoService service = new DemoServiceImpl(); - protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("rsocket://127.0.0.1:9010/" + DemoService.class.getName()))); - service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("rsocket://127.0.0.1:9010/" + DemoService.class.getName()).addParameter("timeout", 3000l))); - - RemoteService remote = new RemoteServiceImpl(); - protocol.export(proxy.getInvoker(remote, RemoteService.class, URL.valueOf("rsocket://127.0.0.1:9010/" + RemoteService.class.getName()))); - remote = proxy.getProxy(protocol.refer(RemoteService.class, URL.valueOf("rsocket://127.0.0.1:9010/" + RemoteService.class.getName()).addParameter("timeout", 3000l))); - - service.sayHello("world"); - - // test netty client - assertEquals("world", service.echo("world")); - assertEquals("hello world", remote.sayHello("world")); - - - EchoService serviceEcho = (EchoService) service; - assertEquals(serviceEcho.$echo("test"), "test"); - - EchoService remoteEecho = (EchoService) remote; - assertEquals(remoteEecho.$echo("ok"), "ok"); - } - - - @Test - public void testRequestMono() throws Exception { - DemoService service = new DemoServiceImpl(); - protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("rsocket://127.0.0.1:9020/" + DemoService.class.getName()))); - service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("rsocket://127.0.0.1:9020/" + DemoService.class.getName()).addParameter("timeout", 3000l))); - Mono result = service.requestMono("mike"); - - result.doOnNext(new Consumer() { - @Override - public void accept(String s) { - assertEquals(s, "hello mike"); - System.out.println(s); - } - }).block(); - - Mono result2 = service.requestMonoOnError("mike"); - result2.onErrorResume(DemoException.class, new Function>() { - @Override - public Mono apply(DemoException e) { - return Mono.just(e.getClass().getName()); - } - }).doOnNext(new Consumer() { - @Override - public void accept(String s) { - assertEquals(DemoException.class.getName(), s); - } - }).block(); - - Mono result3 = service.requestMonoBizError("mike"); - result3.onErrorResume(ArithmeticException.class, new Function>() { - @Override - public Mono apply(ArithmeticException e) { - return Mono.just(e.getClass().getName()); - } - }).doOnNext(new Consumer() { - @Override - public void accept(String s) { - assertEquals(ArithmeticException.class.getName(), s); - } - }).block(); - - } - - @Test - public void testRequestFlux() throws Exception { - DemoService service = new DemoServiceImpl(); - protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("rsocket://127.0.0.1:9020/" + DemoService.class.getName()))); - service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("rsocket://127.0.0.1:9020/" + DemoService.class.getName()).addParameter("timeout", 3000l))); - - { - Flux result = service.requestFlux("mike"); - result.doOnNext(new Consumer() { - @Override - public void accept(String s) { - System.out.println(s); - } - }).blockLast(); - } - - - { - Flux result2 = service.requestFluxOnError("mike"); - result2.onErrorResume(new Function>() { - @Override - public Publisher apply(Throwable throwable) { - return Flux.just(throwable.getClass().getName()); - } - }).takeLast(1).doOnNext(new Consumer() { - @Override - public void accept(String s) { - assertEquals(DemoException.class.getName(), s); - } - }).blockLast(); - } - - { - Flux result3 = service.requestFluxBizError("mike"); - result3.onErrorResume(new Function>() { - @Override - public Publisher apply(Throwable throwable) { - return Flux.just(throwable.getClass().getName()); - } - }).takeLast(1).doOnNext(new Consumer() { - @Override - public void accept(String s) { - assertEquals(ArithmeticException.class.getName(), s); - } - }).blockLast(); - } - } - - - @Test - public void testRequestMonoWithMonoArg() throws Exception { - DemoService service = new DemoServiceImpl(); - protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("rsocket://127.0.0.1:9020/" + DemoService.class.getName()))); - service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("rsocket://127.0.0.1:9020/" + DemoService.class.getName()).addParameter("timeout", 3000l))); - - Mono result = service.requestMonoWithMonoArg(Mono.just("A"), Mono.just("B")); - result.doOnNext(new Consumer() { - @Override - public void accept(String s) { - assertEquals(s, "A B"); - System.out.println(s); - } - }).block(); - } - - - @Test - public void testRequestFluxWithFluxArg() throws Exception { - DemoService service = new DemoServiceImpl(); - protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("rsocket://127.0.0.1:9020/" + DemoService.class.getName()))); - service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("rsocket://127.0.0.1:9020/" + DemoService.class.getName()).addParameter("timeout", 3000l))); - - { - Flux result = service.requestFluxWithFluxArg(Flux.just("A","B","C"), Flux.just("1","2","3")); - result.doOnNext(new Consumer() { - @Override - public void accept(String s) { - System.out.println(s); - } - }).takeLast(1).doOnNext(new Consumer() { - @Override - public void accept(String s) { - assertEquals(s, "C 3"); - } - }).blockLast(); - } - } -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoException.java b/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoException.java deleted file mode 100644 index 33f3a2ea8e..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoException.java +++ /dev/null @@ -1,40 +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.service; - -public class DemoException extends Exception { - - private static final long serialVersionUID = -8213943026163641747L; - - public DemoException() { - super(); - } - - public DemoException(String message, Throwable cause) { - super(message, cause); - } - - public DemoException(String message) { - super(message); - } - - public DemoException(Throwable cause) { - super(cause); - } - -} - diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoService.java b/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoService.java deleted file mode 100644 index 8f7a3ddc5a..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoService.java +++ /dev/null @@ -1,69 +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.service; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import java.util.Map; -import java.util.Set; - -public interface DemoService { - String sayHello(String name); - - Set keys(Map map); - - String echo(String text); - - Map echo(Map map); - - long timestamp(); - - String getThreadName(); - - int getSize(String[] strs); - - int getSize(Object[] os); - - Object invoke(String service, String method) throws Exception; - - int stringLength(String str); - - byte getbyte(byte arg); - - long add(int a, long b); - - String errorTest(String name); - - Mono requestMono(String name); - - Mono requestMonoOnError(String name); - - Mono requestMonoBizError(String name); - - Flux requestFlux(String name); - - Flux requestFluxOnError(String name); - - Flux requestFluxBizError(String name); - - Mono requestMonoWithMonoArg(Mono m1, Mono m2); - - Flux requestFluxWithFluxArg(Flux f1, Flux f2); - - -} diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoServiceImpl.java b/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoServiceImpl.java deleted file mode 100644 index 1ba6d39a7b..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoServiceImpl.java +++ /dev/null @@ -1,174 +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.service; - -import org.apache.dubbo.rpc.RpcContext; -import reactor.core.publisher.Flux; -import reactor.core.publisher.FluxSink; -import reactor.core.publisher.Mono; - -import java.util.Map; -import java.util.Set; -import java.util.function.BiFunction; -import java.util.function.Consumer; - -public class DemoServiceImpl implements DemoService { - public DemoServiceImpl() { - super(); - } - - public String sayHello(String name) { - return "hello " + name; - } - - public String echo(String text) { - return text; - } - - public Map echo(Map map) { - return map; - } - - public long timestamp() { - return System.currentTimeMillis(); - } - - public String getThreadName() { - return Thread.currentThread().getName(); - } - - public int getSize(String[] strs) { - if (strs == null) - return -1; - return strs.length; - } - - public int getSize(Object[] os) { - if (os == null) - return -1; - return os.length; - } - - public Object invoke(String service, String method) throws Exception { - System.out.println("RpcContext.getContext().getRemoteHost()=" + RpcContext.getContext().getRemoteHost()); - return service + ":" + method; - } - - public int stringLength(String str) { - return str.length(); - } - - - public byte getbyte(byte arg) { - return arg; - } - - - public Set keys(Map map) { - return map == null ? null : map.keySet(); - } - - - public long add(int a, long b) { - return a + b; - } - - @Override - public String errorTest(String name) { - int a = 1 / 0; - return null; - } - - public Mono requestMono(String name) { - return Mono.just("hello " + name); - } - - public Mono requestMonoOnError(String name) { - return Mono.error(new DemoException(name)); - } - - public Mono requestMonoBizError(String name) { - int a = 1 / 0; - return Mono.just("hello " + name); - } - - @Override - public Flux requestFlux(String name) { - - return Flux.create(new Consumer>() { - @Override - public void accept(FluxSink fluxSink) { - for (int i = 0; i < 5; i++) { - fluxSink.next(name + " " + i); - } - fluxSink.complete(); - } - }); - - } - - @Override - public Flux requestFluxOnError(String name) { - - return Flux.create(new Consumer>() { - @Override - public void accept(FluxSink fluxSink) { - for (int i = 0; i < 5; i++) { - fluxSink.next(name + " " + i); - } - fluxSink.error(new DemoException()); - } - }); - - } - - @Override - public Flux requestFluxBizError(String name) { - int a = 1 / 0; - return Flux.create(new Consumer>() { - @Override - public void accept(FluxSink fluxSink) { - for (int i = 0; i < 5; i++) { - fluxSink.next(name + " " + i); - } - fluxSink.error(new DemoException()); - } - }); - } - - @Override - public Mono requestMonoWithMonoArg(Mono m1, Mono m2) { - return m1.zipWith(m2, new BiFunction() { - @Override - public String apply(String s, String s2) { - return s+" "+s2; - } - }); - } - - @Override - public Flux requestFluxWithFluxArg(Flux f1, Flux f2) { - return f1.zipWith(f2, new BiFunction() { - @Override - public String apply(String s, String s2) { - return s+" "+s2; - } - }); - } - -} - diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/log4j.xml b/dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/log4j.xml deleted file mode 100644 index 3c5d2ba218..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/log4j.xml +++ /dev/null @@ -1,46 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/spring/dubbo-rsocket-consumer.xml b/dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/spring/dubbo-rsocket-consumer.xml deleted file mode 100644 index f0f25cf923..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/spring/dubbo-rsocket-consumer.xml +++ /dev/null @@ -1,36 +0,0 @@ - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/spring/dubbo-rsocket-provider.xml b/dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/spring/dubbo-rsocket-provider.xml deleted file mode 100644 index e84fb70824..0000000000 --- a/dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/spring/dubbo-rsocket-provider.xml +++ /dev/null @@ -1,40 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftInvoker.java b/dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftInvoker.java index c922e6a626..f245fee2d3 100644 --- a/dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftInvoker.java +++ b/dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftInvoker.java @@ -39,6 +39,7 @@ 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.PATH_KEY; import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY; +import static org.apache.dubbo.remoting.Constants.CHANNEL_ATTRIBUTE_READONLY_KEY; import static org.apache.dubbo.rpc.Constants.TOKEN_KEY; @@ -115,7 +116,7 @@ public class ThriftInvoker extends AbstractInvoker { for (ExchangeClient client : clients) { if (client.isConnected() - && !client.hasAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY)) { + && !client.hasAttribute(CHANNEL_ATTRIBUTE_READONLY_KEY)) { //cannot write == not Available ? return true; } diff --git a/dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftProtocol.java b/dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftProtocol.java index 85e793dd66..cf3c5ddd6e 100644 --- a/dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftProtocol.java +++ b/dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftProtocol.java @@ -45,7 +45,11 @@ import java.util.concurrent.ConcurrentMap; import java.util.function.Function; import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY; +import static org.apache.dubbo.remoting.Constants.CHANNEL_READONLYEVENT_SENT_KEY; +import static org.apache.dubbo.remoting.Constants.CLIENT_KEY; +import static org.apache.dubbo.remoting.Constants.CODEC_KEY; import static org.apache.dubbo.remoting.Constants.CONNECTIONS_KEY; +import static org.apache.dubbo.remoting.Constants.SERVER_KEY; import static org.apache.dubbo.rpc.Constants.IS_SERVER_KEY; /** @@ -122,7 +126,7 @@ public class ThriftProtocol extends AbstractProtocol { public Exporter export(Invoker invoker) throws RpcException { // can use thrift codec only - URL url = invoker.getUrl().addParameter(Constants.CODEC_KEY, ThriftCodec.NAME); + URL url = invoker.getUrl().addParameter(CODEC_KEY, ThriftCodec.NAME); // find server. String key = url.getAddress(); // client can expose a service for server to invoke only. @@ -189,7 +193,7 @@ public class ThriftProtocol extends AbstractProtocol { ExchangeClient client; - url = url.addParameter(Constants.CODEC_KEY, ThriftCodec.NAME); + url = url.addParameter(CODEC_KEY, ThriftCodec.NAME); try { client = Exchangers.connect(url); @@ -204,8 +208,8 @@ public class ThriftProtocol extends AbstractProtocol { private ExchangeServer getServer(URL url) { // enable sending readonly event when server closes by default - url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString()); - String str = url.getParameter(Constants.SERVER_KEY, org.apache.dubbo.rpc.Constants.DEFAULT_REMOTING_SERVER); + url = url.addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString()); + String str = url.getParameter(SERVER_KEY, org.apache.dubbo.rpc.Constants.DEFAULT_REMOTING_SERVER); if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) { throw new RpcException("Unsupported server type: " + str + ", url: " + url); @@ -217,7 +221,7 @@ public class ThriftProtocol extends AbstractProtocol { } catch (RemotingException e) { throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e); } - str = url.getParameter(Constants.CLIENT_KEY); + str = url.getParameter(CLIENT_KEY); if (str != null && str.length() > 0) { Set supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(); if (!supportedTypes.contains(str)) { diff --git a/dubbo-rpc/dubbo-rpc-webservice/src/main/java/org/apache/dubbo/rpc/protocol/webservice/WebServiceProtocol.java b/dubbo-rpc/dubbo-rpc-webservice/src/main/java/org/apache/dubbo/rpc/protocol/webservice/WebServiceProtocol.java index b208f28111..08dc57cfdf 100644 --- a/dubbo-rpc/dubbo-rpc-webservice/src/main/java/org/apache/dubbo/rpc/protocol/webservice/WebServiceProtocol.java +++ b/dubbo-rpc/dubbo-rpc-webservice/src/main/java/org/apache/dubbo/rpc/protocol/webservice/WebServiceProtocol.java @@ -110,7 +110,7 @@ public class WebServiceProtocol extends AbstractProxyProtocol { @Override @SuppressWarnings("unchecked") - protected T getFrameworkProxy(final Class serviceType, final URL url) throws RpcException { + protected T doRefer(final Class serviceType, final URL url) throws RpcException { ClientProxyFactoryBean proxyFactoryBean = new ClientProxyFactoryBean(); proxyFactoryBean.setAddress(url.setProtocol("http").toIdentityString()); proxyFactoryBean.setServiceClass(serviceType);