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

Solve conflicts after merged 3.x

This commit is contained in:
ken.lj 2019-09-02 19:18:07 +08:00
parent 9f5cc83d34
commit e261acc98a
52 changed files with 621 additions and 2713 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagRouter.java
View File

@ -97,7 +97,7 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener {
}
List<Invoker<T>> 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 <T> List<Invoker<T>> filterUsingStaticTag(List<Invoker<T>> invokers, URL url, Invocation invocation) {
List<Invoker<T>> 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)) {

27
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
View File

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

8
dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
View File

@ -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
*/

16
dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java
View File

@ -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<String, ExecutorService> 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)) {

18
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
View File

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

27
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java
View File

@ -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<T> extends AbstractReferenceConfig {
serviceMetadata.setDefaultGroup(group);
serviceMetadata.setServiceType(getActualInterface());
serviceMetadata.setServiceInterfaceName(interfaceName);
serviceMetadata.setServiceKey(URL.buildKey(interfaceName, group, version));
Map<String, String> map = new HashMap<String, String>();
map.put(SIDE_KEY, CONSUMER_SIDE);
@ -351,10 +353,9 @@ public class ReferenceConfig<T> 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<T> extends AbstractReferenceConfig {
return actualInterface;
}
private ConsumerModel buildConsumerModel(String serviceKey, Map<String, Object> attributes) {
private ConsumerModel buildConsumerModel(Map<String, Object> attributes, ServiceMetadata metadata) {
Method[] methods = interfaceClass.getMethods();
Class serviceInterface = interfaceClass;
if (interfaceClass == GenericService.class) {
@ -381,7 +382,7 @@ public class ReferenceConfig<T> extends AbstractReferenceConfig {
methods = interfaceClass.getMethods();
}
}
return new ConsumerModel(serviceKey, serviceInterface, ref, methods, attributes);
return new ConsumerModel(attributes, metadata);
}
@SuppressWarnings({"unchecked", "rawtypes", "deprecation"})

35
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
View File

@ -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<T> extends AbstractServiceConfig {
List<URL> 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);
}

4
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/pom.xml
View File

@ -62,6 +62,10 @@
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-nacos</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-zookeeper</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-config-spring</artifactId>

4
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/pom.xml
View File

@ -64,6 +64,10 @@
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-nacos</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-zookeeper</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-rpc-dubbo</artifactId>

6
dubbo-filter/dubbo-filter-validation/src/test/java/org/apache/dubbo/validation/support/jvalidation/mock/JValidatorTestTarget.java
View File

@ -35,12 +35,6 @@ public interface JValidatorTestTarget {
void someMethod5(Map<String, String> map);
public void someMethod3(ValidationParameter[] parameters);
public void someMethod4(List<String> strings);
public void someMethod5(Map<String, String> map);
@interface Test2 {
}

10
dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java
View File

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

44
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture.java
View File

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

5
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java
View File

@ -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<Object> 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

5
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java
View File

@ -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;
/**

4
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
View File

@ -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<String, Object>(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<String, Object> attachment, Map<Object, Object> attributes) {

2
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ActiveLimitFilter.java
View File

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

4
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java
View File

@ -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<String, String> attachments = invocation.getAttachments();
Map<String, Object> attachments = invocation.getAttachments();
if (attachments != null) {
attachments = new HashMap<>(attachments);
attachments.remove(PATH_KEY);

2
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ExecuteLimitFilter.java
View File

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

10
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericFilter.java
View File

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

2
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TimeoutFilter.java
View File

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

4
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TokenFilter.java
View File

@ -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<String, String> attachments = inv.getAttachments();
String remoteToken = attachments == null ? null : attachments.get(TOKEN_KEY);
Map<String, Object> 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());
}

32
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerMethodModel.java
View File

@ -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<String, Object> attributeMap = new ConcurrentHashMap<>();
private final AsyncMethodInfo asyncInfo;
public ConsumerMethodModel(Method method) {
public ConsumerMethodModel(Method method, Map<String, Object> 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<String, Object> 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;

35
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerModel.java
View File

@ -31,20 +31,18 @@ public class ConsumerModel {
private final ServiceMetadata serviceMetadata;
private final Map<Method, ConsumerMethodModel> methodModels = new IdentityHashMap<Method, ConsumerMethodModel>();
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<String, Object> 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<ConsumerMethodModel>(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();
}

4
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java
View File

@ -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<T> implements Invoker<T> {
} catch (Throwable e) {
asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
}
RpcContext.getContext().setFuture(new FutureAdapter(asyncResult, inv));
RpcContext.getContext().setFuture(new FutureAdapter(asyncResult));
return asyncResult;
}

6
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/RpcUtils.java
View File

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

7
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/TokenFilterTest.java
View File

@ -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<String, String> attachments = new HashMap<String, String>();
Map<String, Object> 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<String, String> attachments = new HashMap<String, String>();
Map<String, Object> attachments = new HashMap<>();
attachments.put(TOKEN_KEY, "wrongToken");
Invocation invocation = Mockito.mock(Invocation.class);
when(invocation.getAttachments()).thenReturn(attachments);

807
dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/ImplicitCallBackTest.java
View File

@ -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<IDemoService> exporter = null;
protected Invoker<IDemoService> 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<String, Object> 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<String, Object> 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<String, Object> 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<Person> 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<Person> 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<Person> p1Future = RpcContext.getContext().getFuture();
int requestId2 = 1;
Person ret2 = demoProxy.get(requestId2);
Assertions.assertNull(ret2);
Future<Person> 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<Person> 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<Integer> inv = new ArrayList<Integer>();
public Map<Integer, Person> ret = new HashMap<Integer, Person>();
public Map<Integer, Throwable> errors = new HashMap<Integer, Throwable>();
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<IDemoService> exporter = null;
// protected Invoker<IDemoService> 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<String, Object> 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<String, Object> 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<String, Object> 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<Person> 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<Person> 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<Person> p1Future = RpcContext.getContext().getFuture();
//
// int requestId2 = 1;
// Person ret2 = demoProxy.get(requestId2);
// Assertions.assertNull(ret2);
// Future<Person> 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<Person> 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<Integer> inv = new ArrayList<Integer>();
// public Map<Integer, Person> ret = new HashMap<Integer, Person>();
// public Map<Integer, Throwable> errors = new HashMap<Integer, Throwable>();
// 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);
// }
// }
//}

8
dubbo-rpc/dubbo-rpc-hessian/src/main/java/org/apache/dubbo/rpc/protocol/hessian/HttpClientConnectionFactory.java
View File

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

2
dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
View File

@ -137,7 +137,7 @@ public class RestProtocol extends AbstractProxyProtocol {
}
@Override
protected <T> T getFrameworkProxy(Class<T> serviceType, URL url) throws RpcException {
protected <T> T doRefer(Class<T> serviceType, URL url) throws RpcException {
// TODO more configs to add
PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();

2
dubbo-rpc/dubbo-rpc-rmi/src/main/java/org/apache/dubbo/rpc/protocol/rmi/RmiProtocol.java
View File

@ -72,7 +72,7 @@ public class RmiProtocol extends AbstractProxyProtocol {
@Override
@SuppressWarnings("unchecked")
protected <T> T getFrameworkProxy(final Class<T> serviceType, final URL url) throws RpcException {
protected <T> T doRefer(final Class<T> 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);

123
dubbo-rpc/dubbo-rpc-rsocket/pom.xml
View File

@ -1,123 +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.
-->
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-rpc</artifactId>
<version>${revision}</version>
</parent>
<artifactId>dubbo-rpc-rsocket</artifactId>
<packaging>jar</packaging>
<name>${project.artifactId}</name>
<description>The default rpc module of dubbo project</description>
<properties>
<skip_maven_deploy>false</skip_maven_deploy>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>4.3.16.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-registry-multicast</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-core</artifactId>
<version>0.11.14</version>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-transport-netty</artifactId>
<version>0.11.14</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.54</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-rpc-api</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-remoting-api</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-config-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-config-spring</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-container-api</artifactId>
<version>${project.parent.version}</version>
<exclusions>
<exclusion>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
</exclusion>
<exclusion>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-servlet</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-serialization-hessian2</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-serialization-jdk</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
<!--<dependency>-->
<!--<groupId>javax.validation</groupId>-->
<!--<artifactId>validation-api</artifactId>-->
<!--<scope>test</scope>-->
<!--</dependency>-->
<!--<dependency>-->
<!--<groupId>org.hibernate</groupId>-->
<!--<artifactId>hibernate-validator</artifactId>-->
<!--<scope>test</scope>-->
<!--</dependency>-->
<!--<dependency>-->
<!--<groupId>org.glassfish</groupId>-->
<!--<artifactId>javax.el</artifactId>-->
<!--<scope>test</scope>-->
<!--</dependency>-->
</dependencies>
</project>

98
dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/FutureSubscriber.java
View File

@ -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<AppResponse> implements Subscriber<Payload> {
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<String, Object> 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() {
}
}

36
dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/MetadataCodec.java
View File

@ -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<String, Object> decodeMetadata(byte[] bytes) throws IOException {
return JSON.parseObject(new String(bytes, StandardCharsets.UTF_8), Map.class);
}
public static byte[] encodeMetadata(Map<String, Object> metadata) throws IOException {
String jsonStr = JSON.toJSONString(metadata);
return jsonStr.getBytes(StandardCharsets.UTF_8);
}
}

32
dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketConstants.java
View File

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

43
dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketExporter.java
View File

@ -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<T> extends AbstractExporter<T> {
private final String key;
private final Map<String, Exporter<?>> exporterMap;
public RSocketExporter(Invoker<T> invoker, String key, Map<String, Exporter<?>> exporterMap) {
super(invoker);
this.key = key;
this.exporterMap = exporterMap;
}
@Override
public void unexport() {
super.unexport();
exporterMap.remove(key);
}
}

263
dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketInvoker.java
View File

@ -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<T> extends AbstractInvoker<T> {
private final RSocket[] clients;
private final AtomicPositiveInteger index = new AtomicPositiveInteger();
private final String version;
private final ReentrantLock destroyLock = new ReentrantLock();
private final Set<Invoker<?>> invokers;
private final Serialization serialization;
public RSocketInvoker(Class<T> serviceType, URL url, RSocket[] clients, Set<Invoker<?>> 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<Payload> responseMono = currentClient.requestResponse(requestPayload);
Mono<Object> bizMono = responseMono.map(new Function<Payload, Object>() {
@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<Payload> 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<Object> requestStream(RSocket currentClient, Payload requestPayload) {
Flux<Payload> responseFlux = currentClient.requestStream(requestPayload);
Flux<Object> retFlux = responseFlux.map(new Function<Payload, Object>() {
@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<String, Object> metadataMap = new HashMap<String, Object>();
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);
}
}

727
dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketProtocol.java
View File

@ -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;
// <host:port,CloseableChannel>
private final Map<String, CloseableChannel> serverMap = new ConcurrentHashMap<String, CloseableChannel>();
// <host:port,RSocket>
private final Map<String, RSocket> referenceClientMap = new ConcurrentHashMap<String, RSocket>();
private final ConcurrentMap<String, Object> locks = new ConcurrentHashMap<String, Object>();
public RSocketProtocol() {
INSTANCE = this;
}
public static RSocketProtocol getRSocketProtocol() {
if (INSTANCE == null) {
ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(RSocketProtocol.NAME); // load
}
return INSTANCE;
}
public Collection<Exporter<?>> getExporters() {
return Collections.unmodifiableCollection(exporterMap.values());
}
Map<String, Exporter<?>> getExporterMap() {
return exporterMap;
}
Invoker<?> getInvoker(int port, Map<String, Object> 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<Invoker<?>> getInvokers() {
return Collections.unmodifiableCollection(invokers);
}
@Override
public int getDefaultPort() {
return DEFAULT_PORT;
}
@Override
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
URL url = invoker.getUrl();
// export service.
String key = serviceKey(url);
RSocketExporter<T> exporter = new RSocketExporter<T>(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 <T> Invoker<T> doRefer(Class<T> serviceType, URL url) throws RpcException {
// create rpc invoker.
RSocketInvoker<T> invoker = new RSocketInvoker<T>(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<Payload> requestResponse(Payload payload) {
try {
ByteBuffer metadataBuffer = payload.getMetadata();
byte[] metadataBytes = new byte[metadataBuffer.remaining()];
metadataBuffer.get(metadataBytes, metadataBuffer.position(), metadataBuffer.remaining());
Map<String, Object> 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<Object, Payload>() {
@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<Throwable, Publisher<Payload>>() {
@Override
public Publisher<Payload> 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<Payload> requestStream(Payload payload) {
try {
ByteBuffer metadataBuffer = payload.getMetadata();
byte[] metadataBytes = new byte[metadataBuffer.remaining()];
metadataBuffer.get(metadataBytes, metadataBuffer.position(), metadataBuffer.remaining());
Map<String, Object> 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<Object, Payload>() {
@Override
public Payload apply(Object o) {
try {
return encodeData(o, serializeId);
} catch (Throwable t) {
throw new RuntimeException(t);
}
}
}).onErrorResume(new Function<Throwable, Publisher<Payload>>() {
@Override
public Publisher<Payload> 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<String>(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<String>(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<RSocket> accept(ConnectionSetupPayload setupPayload, RSocket reactiveSocket) {
return Mono.just(
new AbstractRSocket() {
public Mono<Payload> requestResponse(Payload payload) {
try {
Map<String, Object> 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<Payload> retMono = bizMono.map(new Function<Object, Payload>() {
@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<Throwable, Publisher<Payload>>() {
@Override
public Publisher<Payload> 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<Payload> requestStream(Payload payload) {
try {
Map<String, Object> 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<Payload> retFlux = flux.map(new Function<Object, Payload>() {
@Override
public Payload apply(Object o) {
try {
return encodeData(o, serializeId);
} catch (Throwable t) {
throw new RuntimeException(t);
}
}
}).onErrorResume(new Function<Throwable, Publisher<Payload>>() {
@Override
public Publisher<Payload> 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<String, Object> 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<String, Object> 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<String, Object> map = (Map<String, Object>) 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<String, Object> metadataMap = new HashMap<String, Object>();
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<Payload> payloads = rSocket.requestResponse(payload);
Mono streamArg = payloads.map(new Function<Payload, Object>() {
@Override
public Object apply(Payload payload) {
return decodeData(serializeId, payload);
}
});
return streamArg;
}
private Flux getFluxProxy(long id, Byte serializeId, RSocket rSocket) throws IOException {
Map<String, Object> metadataMap = new HashMap<String, Object>();
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<Payload> payloads = rSocket.requestStream(payload);
Flux streamArg = payloads.map(new Function<Payload, Object>() {
@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);
}
}
}
}

62
dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/ResourceDirectory.java
View File

@ -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<Long, Object> id2ResourceMap = new ConcurrentHashMap<Long, Object>();
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);
}
}

41
dubbo-rpc/dubbo-rpc-rsocket/src/main/java/org/apache/dubbo/rpc/protocol/rsocket/ResourceInfo.java
View File

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

1
dubbo-rpc/dubbo-rpc-rsocket/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol
View File

@ -1 +0,0 @@
rsocket=org.apache.dubbo.rpc.protocol.rsocket.RSocketProtocol

51
dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/ConsumerDemo.java
View File

@ -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<String> resultMono = demoService.requestMono("world"); // call remote method
resultMono.doOnNext(new Consumer<String>() {
@Override
public void accept(String s) {
System.out.println(s); // get result
}
}).block();
} catch (Throwable throwable) {
throwable.printStackTrace();
}
}
}
}

29
dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/ProviderDemo.java
View File

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

260
dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/protocol/rsocket/RSocketProtocolTest.java
View File

@ -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<String, String> map = new HashMap<String, String>();
map.put("aa", "bb");
Set<String> 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<String> result = service.requestMono("mike");
result.doOnNext(new Consumer<String>() {
@Override
public void accept(String s) {
assertEquals(s, "hello mike");
System.out.println(s);
}
}).block();
Mono<String> result2 = service.requestMonoOnError("mike");
result2.onErrorResume(DemoException.class, new Function<DemoException, Mono<String>>() {
@Override
public Mono<String> apply(DemoException e) {
return Mono.just(e.getClass().getName());
}
}).doOnNext(new Consumer<String>() {
@Override
public void accept(String s) {
assertEquals(DemoException.class.getName(), s);
}
}).block();
Mono<String> result3 = service.requestMonoBizError("mike");
result3.onErrorResume(ArithmeticException.class, new Function<ArithmeticException, Mono<String>>() {
@Override
public Mono<String> apply(ArithmeticException e) {
return Mono.just(e.getClass().getName());
}
}).doOnNext(new Consumer<String>() {
@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<String> result = service.requestFlux("mike");
result.doOnNext(new Consumer<String>() {
@Override
public void accept(String s) {
System.out.println(s);
}
}).blockLast();
}
{
Flux<String> result2 = service.requestFluxOnError("mike");
result2.onErrorResume(new Function<Throwable, Publisher<? extends String>>() {
@Override
public Publisher<? extends String> apply(Throwable throwable) {
return Flux.just(throwable.getClass().getName());
}
}).takeLast(1).doOnNext(new Consumer<String>() {
@Override
public void accept(String s) {
assertEquals(DemoException.class.getName(), s);
}
}).blockLast();
}
{
Flux<String> result3 = service.requestFluxBizError("mike");
result3.onErrorResume(new Function<Throwable, Publisher<? extends String>>() {
@Override
public Publisher<? extends String> apply(Throwable throwable) {
return Flux.just(throwable.getClass().getName());
}
}).takeLast(1).doOnNext(new Consumer<String>() {
@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<String> result = service.requestMonoWithMonoArg(Mono.just("A"), Mono.just("B"));
result.doOnNext(new Consumer<String>() {
@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<String> result = service.requestFluxWithFluxArg(Flux.just("A","B","C"), Flux.just("1","2","3"));
result.doOnNext(new Consumer<String>() {
@Override
public void accept(String s) {
System.out.println(s);
}
}).takeLast(1).doOnNext(new Consumer<String>() {
@Override
public void accept(String s) {
assertEquals(s, "C 3");
}
}).blockLast();
}
}
}

40
dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoException.java
View File

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

69
dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoService.java
View File

@ -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<String> keys(Map<String, String> 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<String> requestMono(String name);
Mono<String> requestMonoOnError(String name);
Mono<String> requestMonoBizError(String name);
Flux<String> requestFlux(String name);
Flux<String> requestFluxOnError(String name);
Flux<String> requestFluxBizError(String name);
Mono<String> requestMonoWithMonoArg(Mono<String> m1, Mono<String> m2);
Flux<String> requestFluxWithFluxArg(Flux<String> f1, Flux<String> f2);
}

174
dubbo-rpc/dubbo-rpc-rsocket/src/test/java/org/apache/dubbo/rpc/service/DemoServiceImpl.java
View File

@ -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<String> keys(Map<String, String> 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<String> requestMono(String name) {
return Mono.just("hello " + name);
}
public Mono<String> requestMonoOnError(String name) {
return Mono.error(new DemoException(name));
}
public Mono<String> requestMonoBizError(String name) {
int a = 1 / 0;
return Mono.just("hello " + name);
}
@Override
public Flux<String> requestFlux(String name) {
return Flux.create(new Consumer<FluxSink<String>>() {
@Override
public void accept(FluxSink<String> fluxSink) {
for (int i = 0; i < 5; i++) {
fluxSink.next(name + " " + i);
}
fluxSink.complete();
}
});
}
@Override
public Flux<String> requestFluxOnError(String name) {
return Flux.create(new Consumer<FluxSink<String>>() {
@Override
public void accept(FluxSink<String> fluxSink) {
for (int i = 0; i < 5; i++) {
fluxSink.next(name + " " + i);
}
fluxSink.error(new DemoException());
}
});
}
@Override
public Flux<String> requestFluxBizError(String name) {
int a = 1 / 0;
return Flux.create(new Consumer<FluxSink<String>>() {
@Override
public void accept(FluxSink<String> fluxSink) {
for (int i = 0; i < 5; i++) {
fluxSink.next(name + " " + i);
}
fluxSink.error(new DemoException());
}
});
}
@Override
public Mono<String> requestMonoWithMonoArg(Mono<String> m1, Mono<String> m2) {
return m1.zipWith(m2, new BiFunction<String, String, String>() {
@Override
public String apply(String s, String s2) {
return s+" "+s2;
}
});
}
@Override
public Flux<String> requestFluxWithFluxArg(Flux<String> f1, Flux<String> f2) {
return f1.zipWith(f2, new BiFunction<String, String, String>() {
@Override
public String apply(String s, String s2) {
return s+" "+s2;
}
});
}
}

46
dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/log4j.xml
View File

@ -1,46 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
<!-- ===================================================================== -->
<!-- 以下是appender的定义 -->
<!-- ===================================================================== -->
<appender name="dubbo" class="org.apache.dubbo.common.utils.DubboAppender">
<param name="encoding" value="GBK"/>
<layout class="org.apache.log4j.PatternLayout">
<param name="ConversionPattern" value="%d %p [%c:%M] - %m%n"/>
</layout>
<!-- <filter class="org.apache.log4j.varia.LevelRangeFilter">
<param name="LevelMin" value="DEBUG" />
<param name="LevelMax" value="DEBUG" />
</filter> -->
</appender>
<appender name="FILE" class="org.apache.log4j.FileAppender">
<param name="File" value="dubbo.log"/>
<layout class="org.apache.log4j.PatternLayout">
<!-- <param name="ConversionPattern" value="[%t %d{dd/MM/yy HH:mm:ss:SSS
z}] %5p %c{2}: %L %m%n" /> -->
<param name="ConversionPattern" value="[%t %l %d{dd/MM/yy HH:mm:ss:SSS z}] %5p %m %n"/>
</layout>
</appender>
<root>
<level value="INFO"/>
<appender-ref ref="dubbo"/>
<appender-ref ref="FILE"/>
</root>
</log4j:configuration>

36
dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/spring/dubbo-rsocket-consumer.xml
View File

@ -1,36 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"
xmlns="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd">
<!-- consumer's application name, used for tracing dependency relationship (not a matching criterion),
don't set it same as provider -->
<dubbo:application name="demo-consumer"/>
<!-- use multicast registry center to discover service -->
<dubbo:registry address="multicast://224.5.6.7:1234"/>
<!-- generate proxy for the remote service, then demoService can be used in the same way as the
local regular interface -->
<dubbo:reference id="demoService" check="false" interface="org.apache.dubbo.rpc.service.DemoService"/>
</beans>

40
dubbo-rpc/dubbo-rpc-rsocket/src/test/resources/spring/dubbo-rsocket-provider.xml
View File

@ -1,40 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"
xmlns="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd">
<!-- provider's application name, used for tracing dependency relationship -->
<dubbo:application name="demo-provider"/>
<!-- use multicast registry center to export service -->
<dubbo:registry address="multicast://224.5.6.7:1234"/>
<!-- use rsocket protocol to export service on port 20880 -->
<dubbo:protocol name="rsocket" port="20890"/>
<!-- service implementation, as same as regular local bean -->
<bean id="demoService" class="org.apache.dubbo.rpc.service.DemoServiceImpl"/>
<!-- declare the service interface to be exported -->
<dubbo:service interface="org.apache.dubbo.rpc.service.DemoService" ref="demoService"/>
</beans>

3
dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftInvoker.java
View File

@ -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<T> extends AbstractInvoker<T> {
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;
}

14
dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftProtocol.java
View File

@ -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 <T> Exporter<T> export(Invoker<T> 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<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
if (!supportedTypes.contains(str)) {

2
dubbo-rpc/dubbo-rpc-webservice/src/main/java/org/apache/dubbo/rpc/protocol/webservice/WebServiceProtocol.java
View File

@ -110,7 +110,7 @@ public class WebServiceProtocol extends AbstractProxyProtocol {
@Override
@SuppressWarnings("unchecked")
protected <T> T getFrameworkProxy(final Class<T> serviceType, final URL url) throws RpcException {
protected <T> T doRefer(final Class<T> serviceType, final URL url) throws RpcException {
ClientProxyFactoryBean proxyFactoryBean = new ClientProxyFactoryBean();
proxyFactoryBean.setAddress(url.setProtocol("http").toIdentityString());
proxyFactoryBean.setServiceClass(serviceType);