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

Optimization after merged performance tuning branch

This commit is contained in:
ken.lj 2019-09-06 21:04:58 +08:00
parent 0872090fa4
commit b89df9d8f1
45 changed files with 870 additions and 608 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouterTest.java
View File

@ -137,7 +137,7 @@ public class ConditionRouterTest {
@Test
public void testRoute_methodRoute() {
Invocation invocation = new RpcInvocation("getFoo", new Class<?>[0], new Object[0]);
Invocation invocation = new RpcInvocation("getFoo", "com.foo.BarService", new Class<?>[0], new Object[0]);
// More than one methods, mismatch
Router router = new ConditionRouterFactory().getRouter(getRouteUrl("methods=getFoo => host = 1.2.3.4"));
boolean matchWhen = ((ConditionRouter) router).matchWhen(

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

@ -166,8 +166,6 @@ public interface CommonConstants {
String REVISION_KEY = "revision";
String EXECUTOR_SERVICE_COMPONENT_KEY = ExecutorService.class.getName();
/**
* Consumer side 's proxy class
*/

2
dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepository.java
View File

@ -30,7 +30,7 @@ public interface ExecutorRepository {
/**
* Called by both Client and Server. TODO, consider separate these two parts.
* When the Client or Server starts for the first time, generate a new threadpool according to the parameters passed in usr.
* When the Client or Server starts for the first time, generate a new threadpool according to the parameters specified.
*
* @param url
* @return

21
dubbo-compatible/src/main/java/com/alibaba/dubbo/rpc/Result.java
View File

@ -17,13 +17,12 @@
package com.alibaba.dubbo.rpc;
import org.apache.dubbo.rpc.AppResponse;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import java.util.function.Function;
@Deprecated
@ -39,7 +38,7 @@ public interface Result extends org.apache.dubbo.rpc.Result {
}
abstract class AbstractResult extends org.apache.dubbo.rpc.AbstractResult implements Result {
abstract class AbstractResult implements Result {
@Override
public void setValue(Object value) {
@ -47,7 +46,7 @@ public interface Result extends org.apache.dubbo.rpc.Result {
}
@Override
public org.apache.dubbo.rpc.Result thenApplyWithContext(Function<AppResponse, AppResponse> fn) {
public org.apache.dubbo.rpc.Result whenCompleteWithContext(BiConsumer<org.apache.dubbo.rpc.Result, Throwable> fn) {
return null;
}
@ -109,32 +108,32 @@ public interface Result extends org.apache.dubbo.rpc.Result {
}
@Override
public Map<String, String> getAttachments() {
public Map<String, Object> getAttachments() {
return delegate.getAttachments();
}
@Override
public void addAttachments(Map<String, String> map) {
public void addAttachments(Map<String, Object> map) {
delegate.addAttachments(map);
}
@Override
public void setAttachments(Map<String, String> map) {
public void setAttachments(Map<String, Object> map) {
delegate.setAttachments(map);
}
@Override
public String getAttachment(String key) {
return delegate.getAttachment(key);
return (String) delegate.getAttachment(key);
}
@Override
public String getAttachment(String key, String defaultValue) {
return delegate.getAttachment(key, defaultValue);
public String getAttachment(String key, Object defaultValue) {
return (String) delegate.getAttachment(key, defaultValue);
}
@Override
public void setAttachment(String key, String value) {
public void setAttachment(String key, Object value) {
delegate.setAttachment(key, value);
}
}

5
dubbo-compatible/src/test/java/org/apache/dubbo/filter/LegacyInvocation.java
View File

@ -19,16 +19,15 @@ package org.apache.dubbo.filter;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import org.apache.dubbo.common.Constants;
import java.util.HashMap;
import java.util.Map;
import static org.apache.dubbo.remoting.Constants.DUBBO_VERSION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_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.common.constants.CommonConstants.VERSION_KEY;
import static org.apache.dubbo.remoting.Constants.DUBBO_VERSION_KEY;
import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
/**

6
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/AbstractConfig.java
View File

@ -31,7 +31,7 @@ import org.apache.dubbo.common.utils.ReflectUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.context.ConfigConfigurationAdapter;
import org.apache.dubbo.config.support.Parameter;
import org.apache.dubbo.rpc.model.ConsumerMethodModel;
import org.apache.dubbo.rpc.model.ConsumerModel;
import java.io.Serializable;
import java.lang.reflect.Method;
@ -248,7 +248,7 @@ public abstract class AbstractConfig implements Serializable {
}
}
protected static ConsumerMethodModel.AsyncMethodInfo convertMethodConfig2AsyncInfo(MethodConfig methodConfig) {
protected static ConsumerModel.AsyncMethodInfo convertMethodConfig2AyncInfo(MethodConfig methodConfig) {
if (methodConfig == null || (methodConfig.getOninvoke() == null && methodConfig.getOnreturn() == null && methodConfig.getOnthrow() == null)) {
return null;
}
@ -258,7 +258,7 @@ public abstract class AbstractConfig implements Serializable {
throw new IllegalStateException("method config error : return attribute must be set true when onreturn or onthrow has been set.");
}
ConsumerMethodModel.AsyncMethodInfo asyncMethodInfo = new ConsumerMethodModel.AsyncMethodInfo();
ConsumerModel.AsyncMethodInfo asyncMethodInfo = new ConsumerModel.AsyncMethodInfo();
asyncMethodInfo.setOninvokeInstance(methodConfig.getOninvoke());
asyncMethodInfo.setOnreturnInstance(methodConfig.getOnreturn());

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

@ -40,6 +40,7 @@ import org.apache.dubbo.rpc.cluster.support.RegistryAwareCluster;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ConsumerModel;
import org.apache.dubbo.rpc.model.ServiceMetadata;
import org.apache.dubbo.rpc.model.ServiceModel;
import org.apache.dubbo.rpc.protocol.injvm.InjvmProtocol;
import org.apache.dubbo.rpc.service.GenericService;
import org.apache.dubbo.rpc.support.ProtocolUtils;
@ -336,7 +337,7 @@ public class ReferenceConfig<T> extends AbstractReferenceConfig {
map.put(methodConfig.getName() + ".retries", "0");
}
}
attributes.put(methodConfig.getName(), convertMethodConfig2AsyncInfo(methodConfig));
attributes.put(methodConfig.getName(), convertMethodConfig2AyncInfo(methodConfig));
}
}
@ -352,7 +353,8 @@ public class ReferenceConfig<T> extends AbstractReferenceConfig {
ref = createProxy(map);
ApplicationModel.initConsumerModel(serviceMetadata.getServiceKey(), buildConsumerModel(attributes, serviceMetadata));
ServiceModel serviceModel = ApplicationModel.registerServiceModel(interfaceClass);
ApplicationModel.initConsumerModel(serviceMetadata.getServiceKey(), buildConsumerModel(attributes, serviceModel));
serviceMetadata.setTarget(ref);
serviceMetadata.addAttribute(PROXY_CLASS_REF, ref);
initialized = true;
@ -370,19 +372,14 @@ public class ReferenceConfig<T> extends AbstractReferenceConfig {
return actualInterface;
}
private ConsumerModel buildConsumerModel(Map<String, Object> attributes, ServiceMetadata metadata) {
Method[] methods = interfaceClass.getMethods();
Class serviceInterface = interfaceClass;
if (interfaceClass == GenericService.class) {
try {
serviceInterface = Class.forName(interfaceName);
methods = serviceInterface.getMethods();
} catch (ClassNotFoundException e) {
methods = interfaceClass.getMethods();
}
}
return new ConsumerModel(attributes, metadata);
return new ConsumerModel(serviceKey, interfaceClass, ref, ApplicationModel.registerServiceModel(interfaceClass), attributes);
private ConsumerModel buildConsumerModel(Map<String, Object> attributes, ServiceModel serviceModel) {
return new ConsumerModel(serviceMetadata.getServiceKey(),
interfaceClass,
ref,
serviceModel,
attributes,
serviceMetadata
);
}
@SuppressWarnings({"unchecked", "rawtypes", "deprecation"})

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

@ -475,7 +475,6 @@ public class ServiceConfig<T> extends AbstractServiceConfig {
for (ProtocolConfig protocolConfig : protocols) {
String pathKey = URL.buildKey(getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), group, version);
serviceMetadata.setServiceKey(pathKey);
ProviderModel providerModel = new ProviderModel(ref, serviceMetadata);
ProviderModel providerModel = new ProviderModel(pathKey, ref, ApplicationModel.registerServiceModel(interfaceClass));
ApplicationModel.initProviderModel(pathKey, providerModel);
doExportUrlsFor1Protocol(protocolConfig, registryURLs);

10
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java
View File

@ -41,11 +41,11 @@ public class DemoServiceImpl implements DemoService {
@Override
public CompletableFuture<String> sayHelloAsync(String name) {
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
return "async result";
});
return cf;

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

@ -24,9 +24,9 @@ import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.monitor.Monitor;
import org.apache.dubbo.monitor.MonitorFactory;
import org.apache.dubbo.monitor.MonitorService;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
@ -52,14 +52,11 @@ import static org.apache.dubbo.rpc.Constants.OUTPUT_KEY;
* MonitorFilter. (SPI, Singleton, ThreadSafe)
*/
@Activate(group = {PROVIDER, CONSUMER})
public class MonitorFilter extends ListenableFilter {
public class MonitorFilter implements Filter, Filter.Listener {
private static final Logger logger = LoggerFactory.getLogger(MonitorFilter.class);
private static final String MONITOR_FILTER_START_TIME = "monitor_filter_start_time";
public MonitorFilter() {
super.listener = new MonitorListener();
}
/**
* The Concurrent counter
*/
@ -103,93 +100,90 @@ public class MonitorFilter extends ListenableFilter {
return concurrent;
}
class MonitorListener implements Listener {
@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((String) invocation.getAttachment(MONITOR_FILTER_START_TIME)), false);
getConcurrent(invoker, invocation).decrementAndGet(); // count down
}
@Override
public void onMessage(Result result, Invoker<?> invoker, Invocation invocation) {
if (invoker.getUrl().hasParameter(MONITOR_KEY)) {
collect(invoker, invocation, result, RpcContext.getContext().getRemoteHost(), Long.valueOf((String) invocation.getAttachment(MONITOR_FILTER_START_TIME)), false);
getConcurrent(invoker, invocation).decrementAndGet(); // count down
}
@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((String) invocation.getAttachment(MONITOR_FILTER_START_TIME)), true);
getConcurrent(invoker, invocation).decrementAndGet(); // count down
}
}
/**
* The collector logic, it will be handled by the default monitor
*
* @param invoker
* @param invocation
* @param result the invoke result
* @param remoteHost the remote host address
* @param start the timestamp the invoke begin
* @param error if there is an error on the invoke
*/
private void collect(Invoker<?> invoker, Invocation invocation, Result result, String remoteHost, long start, boolean error) {
try {
URL monitorUrl = invoker.getUrl().getUrlParameter(MONITOR_KEY);
Monitor monitor = monitorFactory.getMonitor(monitorUrl);
if (monitor == null) {
return;
}
URL statisticsURL = createStatisticsUrl(invoker, invocation, result, remoteHost, start, error);
monitor.collect(statisticsURL);
} catch (Throwable t) {
logger.warn("Failed to monitor count service " + invoker.getUrl() + ", cause: " + t.getMessage(), t);
}
}
/**
* Create statistics url
*
* @param invoker
* @param invocation
* @param result
* @param remoteHost
* @param start
* @param error
* @return
*/
private URL createStatisticsUrl(Invoker<?> invoker, Invocation invocation, Result result, String remoteHost, long start, boolean error) {
// ---- service statistics ----
long elapsed = System.currentTimeMillis() - start; // invocation cost
int concurrent = getConcurrent(invoker, invocation).get(); // current concurrent count
String application = invoker.getUrl().getParameter(APPLICATION_KEY);
String service = invoker.getInterface().getName(); // service name
String method = RpcUtils.getMethodName(invocation); // method name
String group = invoker.getUrl().getParameter(GROUP_KEY);
String version = invoker.getUrl().getParameter(VERSION_KEY);
int localPort;
String remoteKey, remoteValue;
if (CONSUMER_SIDE.equals(invoker.getUrl().getParameter(SIDE_KEY))) {
// ---- for service consumer ----
localPort = 0;
remoteKey = MonitorService.PROVIDER;
remoteValue = invoker.getUrl().getAddress();
} else {
// ---- for service provider ----
localPort = invoker.getUrl().getPort();
remoteKey = MonitorService.CONSUMER;
remoteValue = remoteHost;
}
String input = "", output = "";
if (invocation.getAttachment(INPUT_KEY) != null) {
input = (String) invocation.getAttachment(INPUT_KEY);
}
if (result != null && result.getAttachment(OUTPUT_KEY) != null) {
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);
}
}
@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((String) invocation.getAttachment(MONITOR_FILTER_START_TIME)), true);
getConcurrent(invoker, invocation).decrementAndGet(); // count down
}
}
/**
* The collector logic, it will be handled by the default monitor
*
* @param invoker
* @param invocation
* @param result the invoke result
* @param remoteHost the remote host address
* @param start the timestamp the invoke begin
* @param error if there is an error on the invoke
*/
private void collect(Invoker<?> invoker, Invocation invocation, Result result, String remoteHost, long start, boolean error) {
try {
URL monitorUrl = invoker.getUrl().getUrlParameter(MONITOR_KEY);
Monitor monitor = monitorFactory.getMonitor(monitorUrl);
if (monitor == null) {
return;
}
URL statisticsURL = createStatisticsUrl(invoker, invocation, result, remoteHost, start, error);
monitor.collect(statisticsURL);
} catch (Throwable t) {
logger.warn("Failed to monitor count service " + invoker.getUrl() + ", cause: " + t.getMessage(), t);
}
}
/**
* Create statistics url
*
* @param invoker
* @param invocation
* @param result
* @param remoteHost
* @param start
* @param error
* @return
*/
private URL createStatisticsUrl(Invoker<?> invoker, Invocation invocation, Result result, String remoteHost, long start, boolean error) {
// ---- service statistics ----
long elapsed = System.currentTimeMillis() - start; // invocation cost
int concurrent = getConcurrent(invoker, invocation).get(); // current concurrent count
String application = invoker.getUrl().getParameter(APPLICATION_KEY);
String service = invoker.getInterface().getName(); // service name
String method = RpcUtils.getMethodName(invocation); // method name
String group = invoker.getUrl().getParameter(GROUP_KEY);
String version = invoker.getUrl().getParameter(VERSION_KEY);
int localPort;
String remoteKey, remoteValue;
if (CONSUMER_SIDE.equals(invoker.getUrl().getParameter(SIDE_KEY))) {
// ---- for service consumer ----
localPort = 0;
remoteKey = MonitorService.PROVIDER;
remoteValue = invoker.getUrl().getAddress();
} else {
// ---- for service provider ----
localPort = invoker.getUrl().getPort();
remoteKey = MonitorService.CONSUMER;
remoteValue = remoteHost;
}
String input = "", output = "";
if (invocation.getAttachment(INPUT_KEY) != null) {
input = (String) invocation.getAttachment(INPUT_KEY);
}
if (result != null && result.getAttachment(OUTPUT_KEY) != null) {
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);
}
}

8
dubbo-monitor/dubbo-monitor-api/src/test/java/org/apache/dubbo/monitor/support/MonitorFilterTest.java
View File

@ -124,9 +124,9 @@ public class MonitorFilterTest {
Result result = monitorFilter.invoke(serviceInvoker, invocation);
result.whenCompleteWithContext((r, t) -> {
if (t == null) {
monitorFilter.listener().onResponse(r, serviceInvoker, invocation);
monitorFilter.onMessage(r, serviceInvoker, invocation);
} else {
monitorFilter.listener().onError(t, serviceInvoker, invocation);
monitorFilter.onError(t, serviceInvoker, invocation);
}
});
while (lastStatistics == null) {
@ -167,9 +167,9 @@ public class MonitorFilterTest {
Result result = monitorFilter.invoke(serviceInvoker, invocation);
result.whenCompleteWithContext((r, t) -> {
if (t == null) {
monitorFilter.listener().onResponse(r, serviceInvoker, invocation);
monitorFilter.onMessage(r, serviceInvoker, invocation);
} else {
monitorFilter.listener().onError(t, serviceInvoker, invocation);
monitorFilter.onError(t, serviceInvoker, invocation);
}
});
while (lastStatistics == null) {

7
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AppResponse.java
View File

@ -16,13 +16,14 @@
*/
package org.apache.dubbo.rpc;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import java.util.function.Function;
/**
@ -44,7 +45,7 @@ import java.util.function.Function;
*
* @serial Do not change the class name and properties.
*/
public class AppResponse extends AbstractResult implements Serializable {
public class AppResponse implements Result {
private static final long serialVersionUID = -6925924956850004727L;
@ -158,7 +159,7 @@ public class AppResponse extends AbstractResult implements Serializable {
}
@Override
public Result thenApplyWithContext(Function<AppResponse, AppResponse> fn) {
public Result whenCompleteWithContext(BiConsumer<Result, Throwable> fn) {
throw new UnsupportedOperationException("AppResponse represents an concrete business response, there will be no status changes, you should get internal values directly.");
}

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

@ -26,6 +26,7 @@ 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;
import java.util.function.Function;
/**
@ -42,7 +43,7 @@ import java.util.function.Function;
* {@link #getValue()} and {@link #getException()} are all inherited from {@link Result} interface, implementing them are mainly
* for compatibility consideration. Because many legacy {@link Filter} implementation are most possibly to call getValue directly.
*/
public class AsyncRpcResult extends AbstractResult {
public class AsyncRpcResult implements Result {
private static final Logger logger = LoggerFactory.getLogger(AsyncRpcResult.class);
/**
@ -172,17 +173,19 @@ public class AsyncRpcResult extends AbstractResult {
@Override
public Object recreate() throws Throwable {
RpcInvocation rpcInvocation = (RpcInvocation) invocation;
FutureAdapter future = new FutureAdapter(this);
RpcContext.getContext().setFuture(future);
if (InvokeMode.FUTURE == rpcInvocation.getInvokeMode()) {
return future;
return RpcContext.getContext().getFuture();
}
return getAppResponse().recreate();
}
public Result thenApplyWithContext(Function<AppResponse, AppResponse> fn) {
this.responseFuture = responseFuture.thenApply(fn.compose(beforeContext).andThen(afterContext));
public Result whenCompleteWithContext(BiConsumer<Result, Throwable> fn) {
this.responseFuture.whenComplete((v, t) -> {
beforeContext.accept(v, t);
fn.accept(v, t);
afterContext.accept(v, t);
});
return this;
}
@ -235,18 +238,16 @@ public class AsyncRpcResult extends AbstractResult {
private RpcContext tmpContext;
private RpcContext tmpServerContext;
private Function<AppResponse, AppResponse> beforeContext = (appResponse) -> {
private BiConsumer<Result, Throwable> beforeContext = (appResponse, t) -> {
tmpContext = RpcContext.getContext();
tmpServerContext = RpcContext.getServerContext();
RpcContext.restoreContext(storedContext);
RpcContext.restoreServerContext(storedServerContext);
return appResponse;
};
private Function<AppResponse, AppResponse> afterContext = (appResponse) -> {
private BiConsumer<Result, Throwable> afterContext = (appResponse, t) -> {
RpcContext.restoreContext(tmpContext);
RpcContext.restoreServerContext(tmpServerContext);
return appResponse;
};
/**

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

@ -62,7 +62,7 @@ public interface Filter {
interface Listener {
void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation);
void onMessage(Result appResponse, Invoker<?> invoker, Invocation invocation);
void onError(Throwable t, Invoker<?> invoker, Invocation invocation);
}

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

@ -90,7 +90,7 @@ public class FutureContext {
* public final class TracingFilter implements Filter {
* public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
* Result result = invoker.invoke(invocation);
* result.whenComplete(new FinishSpanCallback(span));
* result.getResponseFuture().whenComplete(new FinishSpanCallback(span));
* ......
* }
* }

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

@ -19,6 +19,7 @@ package org.apache.dubbo.rpc;
/**
*
*/
@Deprecated
public abstract class ListenableFilter implements Filter {
protected Listener listener = null;
@ -26,4 +27,5 @@ public abstract class ListenableFilter implements Filter {
public Listener listener() {
return listener;
}
}

3
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Result.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import java.util.function.Function;
@ -130,7 +131,7 @@ public interface Result extends Serializable {
* @param fn
* @return
*/
Result thenApplyWithContext(Function<AppResponse, AppResponse> fn);
Result whenCompleteWithContext(BiConsumer<Result, Throwable> fn);
<U> CompletableFuture<U> thenApply(Function<Result, ? extends U> fn);

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

@ -101,7 +101,7 @@ public class RpcInvocation implements Invocation, Serializable {
}
public RpcInvocation(Method method, String serviceName, Object[] arguments) {
this(method, serviceName, arguments, null);
this(method, serviceName, arguments, null, null);
}
public RpcInvocation(Method method, String serviceName, Object[] arguments, Map<String, Object> attachment, Map<Object, Object> attributes) {

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

@ -18,11 +18,9 @@ package org.apache.dubbo.rpc.filter;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcStatus;
@ -44,14 +42,10 @@ import static org.apache.dubbo.rpc.Constants.ACTIVES_KEY;
* @see Filter
*/
@Activate(group = CONSUMER, value = ACTIVES_KEY)
public class ActiveLimitFilter extends ListenableFilter {
public class ActiveLimitFilter implements Filter, Filter.Listener {
private static final String ACTIVELIMIT_FILTER_START_TIME = "activelimit_filter_start_time";
public ActiveLimitFilter() {
super.listener = new ActiveLimitListener();
}
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
URL url = invoker.getUrl();
@ -82,51 +76,47 @@ public class ActiveLimitFilter extends ListenableFilter {
}
}
invocation.setAttachment(ACTIVELIMIT_FILTER_START_TIME, String.valueOf(System.currentTimeMillis()));
invocation.put(ACTIVELIMIT_FILTER_START_TIME, System.currentTimeMillis());
return invoker.invoke(invocation);
}
static class ActiveLimitListener implements Listener {
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
String methodName = invocation.getMethodName();
URL url = invoker.getUrl();
int max = invoker.getUrl().getMethodParameter(methodName, ACTIVES_KEY, 0);
@Override
public void onMessage(Result appResponse, Invoker<?> invoker, Invocation invocation) {
String methodName = invocation.getMethodName();
URL url = invoker.getUrl();
int max = invoker.getUrl().getMethodParameter(methodName, ACTIVES_KEY, 0);
RpcStatus.endCount(url, methodName, getElapsed(invocation), true);
notifyFinish(RpcStatus.getStatus(url, methodName), max);
}
RpcStatus.endCount(url, methodName, getElapsed(invocation), true);
notifyFinish(RpcStatus.getStatus(url, methodName), max);
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
String methodName = invocation.getMethodName();
URL url = invoker.getUrl();
int max = invoker.getUrl().getMethodParameter(methodName, ACTIVES_KEY, 0);
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
String methodName = invocation.getMethodName();
URL url = invoker.getUrl();
int max = invoker.getUrl().getMethodParameter(methodName, ACTIVES_KEY, 0);
if (t instanceof RpcException) {
RpcException rpcException = (RpcException)t;
if (rpcException.isLimitExceed()) {
return;
}
}
RpcStatus.endCount(url, methodName, getElapsed(invocation), false);
notifyFinish(RpcStatus.getStatus(url, methodName), max);
}
private long getElapsed(Invocation invocation) {
String beginTime = (String) invocation.getAttachment(ACTIVELIMIT_FILTER_START_TIME);
return StringUtils.isNotEmpty(beginTime) ? System.currentTimeMillis() - Long.parseLong(beginTime) : 0;
}
private void notifyFinish(final RpcStatus rpcStatus, int max) {
if (max > 0) {
synchronized (rpcStatus) {
rpcStatus.notifyAll();
}
if (t instanceof RpcException) {
RpcException rpcException = (RpcException) t;
if (rpcException.isLimitExceed()) {
return;
}
}
RpcStatus.endCount(url, methodName, getElapsed(invocation), false);
notifyFinish(RpcStatus.getStatus(url, methodName), max);
}
private long getElapsed(Invocation invocation) {
Object beginTime = invocation.get(ACTIVELIMIT_FILTER_START_TIME);
return beginTime != null ? System.currentTimeMillis() - (Long) beginTime : 0;
}
private void notifyFinish(final RpcStatus rpcStatus, int max) {
if (max > 0) {
synchronized (rpcStatus) {
rpcStatus.notifyAll();
}
}
}
}

63
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/CompatibleFilter.java
View File

@ -23,7 +23,6 @@ import org.apache.dubbo.common.utils.PojoUtils;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
@ -46,54 +45,48 @@ import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
* @see Filter
*
*/
public class CompatibleFilter extends ListenableFilter {
public class CompatibleFilter implements Filter, Filter.Listener {
private static Logger logger = LoggerFactory.getLogger(CompatibleFilter.class);
public CompatibleFilter() {
super.listener = new CompatibleListener();
}
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
return invoker.invoke(invocation);
}
static class CompatibleListener implements Listener {
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
if (!invocation.getMethodName().startsWith("$") && !appResponse.hasException()) {
Object value = appResponse.getValue();
if (value != null) {
try {
Method method = invoker.getInterface().getMethod(invocation.getMethodName(), invocation.getParameterTypes());
Class<?> type = method.getReturnType();
Object newValue;
String serialization = invoker.getUrl().getParameter(SERIALIZATION_KEY);
if ("json".equals(serialization) || "fastjson".equals(serialization)) {
// If the serialization key is json or fastjson
Type gtype = method.getGenericReturnType();
newValue = PojoUtils.realize(value, type, gtype);
} else if (!type.isInstance(value)) {
//if local service interface's method's return type is not instance of return value
newValue = PojoUtils.isPojo(type) ? PojoUtils.realize(value, type) : CompatibleTypeUtils.compatibleTypeConvert(value, type);
@Override
public void onMessage(Result appResponse, Invoker<?> invoker, Invocation invocation) {
if (!invocation.getMethodName().startsWith("$") && !appResponse.hasException()) {
Object value = appResponse.getValue();
if (value != null) {
try {
Method method = invoker.getInterface().getMethod(invocation.getMethodName(), invocation.getParameterTypes());
Class<?> type = method.getReturnType();
Object newValue;
String serialization = invoker.getUrl().getParameter(SERIALIZATION_KEY);
if ("json".equals(serialization) || "fastjson".equals(serialization)) {
// If the serialization key is json or fastjson
Type gtype = method.getGenericReturnType();
newValue = PojoUtils.realize(value, type, gtype);
} else if (!type.isInstance(value)) {
//if local service interface's method's return type is not instance of return value
newValue = PojoUtils.isPojo(type) ? PojoUtils.realize(value, type) : CompatibleTypeUtils.compatibleTypeConvert(value, type);
} else {
newValue = value;
}
if (newValue != value) {
appResponse.setValue(newValue);
}
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
} else {
newValue = value;
}
if (newValue != value) {
appResponse.setValue(newValue);
}
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
}
}
}

22
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ConsumerContextFilter.java
View File

@ -18,9 +18,9 @@ package org.apache.dubbo.rpc.filter;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
@ -36,11 +36,7 @@ import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER;
* @see RpcContext
*/
@Activate(group = CONSUMER, order = -10000)
public class ConsumerContextFilter extends ListenableFilter {
public ConsumerContextFilter() {
super.listener = new ConsumerContextListener();
}
public class ConsumerContextFilter implements Filter, Filter.Listener {
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
@ -61,15 +57,13 @@ public class ConsumerContextFilter extends ListenableFilter {
}
}
static class ConsumerContextListener implements Listener {
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
RpcContext.getServerContext().setAttachments(appResponse.getAttachments());
}
@Override
public void onMessage(Result appResponse, Invoker<?> invoker, Invocation invocation) {
RpcContext.getServerContext().setAttachments(appResponse.getAttachments());
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
}
}
}

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

@ -17,9 +17,9 @@
package org.apache.dubbo.rpc.filter;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
@ -48,13 +48,9 @@ import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
* @see RpcContext
*/
@Activate(group = PROVIDER, order = -10000)
public class ContextFilter extends ListenableFilter {
public class ContextFilter implements Filter, Filter.Listener {
private static final String TAG_KEY = "dubbo.tag";
public ContextFilter() {
super.listener = new ContextListener();
}
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
Map<String, Object> attachments = invocation.getAttachments();
@ -101,16 +97,14 @@ public class ContextFilter extends ListenableFilter {
}
}
static class ContextListener implements Listener {
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
// pass attachments to result
appResponse.addAttachments(RpcContext.getServerContext().getAttachments());
}
@Override
public void onMessage(Result appResponse, Invoker<?> invoker, Invocation invocation) {
// pass attachments to result
appResponse.addAttachments(RpcContext.getServerContext().getAttachments());
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
}
}
}

115
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ExceptionFilter.java
View File

@ -22,9 +22,9 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ReflectUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
@ -44,82 +44,67 @@ import java.lang.reflect.Method;
* </ol>
*/
@Activate(group = CommonConstants.PROVIDER)
public class ExceptionFilter extends ListenableFilter {
public ExceptionFilter() {
super.listener = new ExceptionListener();
}
public class ExceptionFilter implements Filter, Filter.Listener {
private Logger logger = LoggerFactory.getLogger(ExceptionFilter.class);
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
return invoker.invoke(invocation);
}
static class ExceptionListener implements Listener {
@Override
public void onMessage(Result appResponse, Invoker<?> invoker, Invocation invocation) {
if (appResponse.hasException() && GenericService.class != invoker.getInterface()) {
try {
Throwable exception = appResponse.getException();
private Logger logger = LoggerFactory.getLogger(ExceptionListener.class);
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
if (appResponse.hasException() && GenericService.class != invoker.getInterface()) {
try {
Throwable exception = appResponse.getException();
// directly throw if it's checked exception
if (!(exception instanceof RuntimeException) && (exception instanceof Exception)) {
return;
}
// directly throw if the exception appears in the signature
try {
Method method = invoker.getInterface().getMethod(invocation.getMethodName(), invocation.getParameterTypes());
Class<?>[] exceptionClassses = method.getExceptionTypes();
for (Class<?> exceptionClass : exceptionClassses) {
if (exception.getClass().equals(exceptionClass)) {
return;
}
}
} catch (NoSuchMethodException e) {
return;
}
// for the exception not found in method's signature, print ERROR message in server's log.
logger.error("Got unchecked and undeclared exception which called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + exception.getClass().getName() + ": " + exception.getMessage(), exception);
// directly throw if exception class and interface class are in the same jar file.
String serviceFile = ReflectUtils.getCodeBase(invoker.getInterface());
String exceptionFile = ReflectUtils.getCodeBase(exception.getClass());
if (serviceFile == null || exceptionFile == null || serviceFile.equals(exceptionFile)) {
return;
}
// directly throw if it's JDK exception
String className = exception.getClass().getName();
if (className.startsWith("java.") || className.startsWith("javax.")) {
return;
}
// directly throw if it's dubbo exception
if (exception instanceof RpcException) {
return;
}
// otherwise, wrap with RuntimeException and throw back to the client
appResponse.setException(new RuntimeException(StringUtils.toString(exception)));
return;
} catch (Throwable e) {
logger.warn("Fail to ExceptionFilter when called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + e.getClass().getName() + ": " + e.getMessage(), e);
// directly throw if it's checked exception
if (!(exception instanceof RuntimeException) && (exception instanceof Exception)) {
return;
}
// directly throw if the exception appears in the signature
try {
Method method = invoker.getInterface().getMethod(invocation.getMethodName(), invocation.getParameterTypes());
Class<?>[] exceptionClassses = method.getExceptionTypes();
for (Class<?> exceptionClass : exceptionClassses) {
if (exception.getClass().equals(exceptionClass)) {
return;
}
}
} catch (NoSuchMethodException e) {
return;
}
// for the exception not found in method's signature, print ERROR message in server's log.
logger.error("Got unchecked and undeclared exception which called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + exception.getClass().getName() + ": " + exception.getMessage(), exception);
// directly throw if exception class and interface class are in the same jar file.
String serviceFile = ReflectUtils.getCodeBase(invoker.getInterface());
String exceptionFile = ReflectUtils.getCodeBase(exception.getClass());
if (serviceFile == null || exceptionFile == null || serviceFile.equals(exceptionFile)) {
return;
}
// directly throw if it's JDK exception
String className = exception.getClass().getName();
if (className.startsWith("java.") || className.startsWith("javax.")) {
return;
}
// directly throw if it's dubbo exception
if (exception instanceof RpcException) {
return;
}
// otherwise, wrap with RuntimeException and throw back to the client
appResponse.setException(new RuntimeException(StringUtils.toString(exception)));
} catch (Throwable e) {
logger.warn("Fail to ExceptionFilter when called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + e.getClass().getName() + ": " + e.getMessage(), e);
}
}
}
@Override
public void onError(Throwable e, Invoker<?> invoker, Invocation invocation) {
logger.error("Got unchecked and undeclared exception which called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + e.getClass().getName() + ": " + e.getMessage(), e);
}
// For test purpose
public void setLogger(Logger logger) {
this.logger = logger;
}
@Override
public void onError(Throwable e, Invoker<?> invoker, Invocation invocation) {
logger.error("Got unchecked and undeclared exception which called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + e.getClass().getName() + ": " + e.getMessage(), e);
}
}

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

@ -19,10 +19,9 @@ package org.apache.dubbo.rpc.filter;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcStatus;
@ -38,14 +37,10 @@ import static org.apache.dubbo.rpc.Constants.EXECUTES_KEY;
*
*/
@Activate(group = CommonConstants.PROVIDER, value = EXECUTES_KEY)
public class ExecuteLimitFilter extends ListenableFilter {
public class ExecuteLimitFilter implements Filter, Filter.Listener {
private static final String EXECUTELIMIT_FILTER_START_TIME = "execugtelimit_filter_start_time";
public ExecuteLimitFilter() {
super.listener = new ExecuteLimitListener();
}
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
URL url = invoker.getUrl();
@ -58,7 +53,7 @@ public class ExecuteLimitFilter extends ListenableFilter {
"\" /> limited.");
}
invocation.setAttachment(EXECUTELIMIT_FILTER_START_TIME, String.valueOf(System.currentTimeMillis()));
invocation.put(EXECUTELIMIT_FILTER_START_TIME, System.currentTimeMillis());
try {
return invoker.invoke(invocation);
} catch (Throwable t) {
@ -70,26 +65,24 @@ public class ExecuteLimitFilter extends ListenableFilter {
}
}
static class ExecuteLimitListener implements Listener {
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
RpcStatus.endCount(invoker.getUrl(), invocation.getMethodName(), getElapsed(invocation), true);
}
@Override
public void onMessage(Result appResponse, Invoker<?> invoker, Invocation invocation) {
RpcStatus.endCount(invoker.getUrl(), invocation.getMethodName(), getElapsed(invocation), true);
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
if (t instanceof RpcException) {
RpcException rpcException = (RpcException)t;
if (rpcException.isLimitExceed()) {
return;
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
if (t instanceof RpcException) {
RpcException rpcException = (RpcException) t;
if (rpcException.isLimitExceed()) {
return;
}
RpcStatus.endCount(invoker.getUrl(), invocation.getMethodName(), getElapsed(invocation), false);
}
RpcStatus.endCount(invoker.getUrl(), invocation.getMethodName(), getElapsed(invocation), false);
}
private long getElapsed(Invocation invocation) {
String beginTime = (String) invocation.getAttachment(EXECUTELIMIT_FILTER_START_TIME);
return StringUtils.isNotEmpty(beginTime) ? System.currentTimeMillis() - Long.parseLong(beginTime) : 0;
}
private long getElapsed(Invocation invocation) {
Object beginTime = invocation.get(EXECUTELIMIT_FILTER_START_TIME);
return beginTime != null ? System.currentTimeMillis() - (Long) beginTime : 0;
}
}

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

@ -28,9 +28,9 @@ import org.apache.dubbo.common.serialize.Serialization;
import org.apache.dubbo.common.utils.PojoUtils;
import org.apache.dubbo.common.utils.ReflectUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
@ -53,11 +53,7 @@ import static org.apache.dubbo.rpc.Constants.GENERIC_SERIALIZATION_PROTOBUF;
* GenericInvokerFilter.
*/
@Activate(group = CommonConstants.PROVIDER, order = -20000)
public class GenericFilter extends ListenableFilter {
public GenericFilter() {
super.listener = new GenericListener();
}
public class GenericFilter implements Filter, Filter.Listener {
@Override
public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
@ -148,57 +144,54 @@ public class GenericFilter extends ListenableFilter {
return invoker.invoke(inv);
}
static class GenericListener implements Listener {
@Override
public void onMessage(Result appResponse, Invoker<?> invoker, Invocation inv) {
if ((inv.getMethodName().equals($INVOKE) || inv.getMethodName().equals($INVOKE_ASYNC))
&& inv.getArguments() != null
&& inv.getArguments().length == 3
&& !GenericService.class.isAssignableFrom(invoker.getInterface())) {
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation inv) {
if ((inv.getMethodName().equals($INVOKE) || inv.getMethodName().equals($INVOKE_ASYNC))
&& inv.getArguments() != null
&& inv.getArguments().length == 3
&& !GenericService.class.isAssignableFrom(invoker.getInterface())) {
String generic = (String) inv.getAttachment(GENERIC_KEY);
if (StringUtils.isBlank(generic)) {
generic = (String) RpcContext.getContext().getAttachment(GENERIC_KEY);
}
String generic = (String) inv.getAttachment(GENERIC_KEY);
if (StringUtils.isBlank(generic)) {
generic = (String) RpcContext.getContext().getAttachment(GENERIC_KEY);
if (appResponse.hasException() && !(appResponse.getException() instanceof GenericException)) {
appResponse.setException(new GenericException(appResponse.getException()));
}
if (ProtocolUtils.isJavaGenericSerialization(generic)) {
try {
UnsafeByteArrayOutputStream os = new UnsafeByteArrayOutputStream(512);
ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(GENERIC_SERIALIZATION_NATIVE_JAVA).serialize(null, os).writeObject(appResponse.getValue());
appResponse.setValue(os.toByteArray());
} catch (IOException e) {
throw new RpcException(
"Generic serialization [" +
GENERIC_SERIALIZATION_NATIVE_JAVA +
"] serialize result failed.", e);
}
if (appResponse.hasException() && !(appResponse.getException() instanceof GenericException)) {
appResponse.setException(new GenericException(appResponse.getException()));
}
if (ProtocolUtils.isJavaGenericSerialization(generic)) {
try {
UnsafeByteArrayOutputStream os = new UnsafeByteArrayOutputStream(512);
ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(GENERIC_SERIALIZATION_NATIVE_JAVA).serialize(null, os).writeObject(appResponse.getValue());
appResponse.setValue(os.toByteArray());
} catch (IOException e) {
throw new RpcException(
"Generic serialization [" +
GENERIC_SERIALIZATION_NATIVE_JAVA +
"] serialize result failed.", e);
}
} else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
appResponse.setValue(JavaBeanSerializeUtil.serialize(appResponse.getValue(), JavaBeanAccessor.METHOD));
} else if (ProtocolUtils.isProtobufGenericSerialization(generic)) {
try {
UnsafeByteArrayOutputStream os = new UnsafeByteArrayOutputStream(512);
ExtensionLoader.getExtensionLoader(Serialization.class)
.getExtension(GENERIC_SERIALIZATION_PROTOBUF)
.serialize(null, os).writeObject(appResponse.getValue());
appResponse.setValue(os.toString());
} catch (IOException e) {
throw new RpcException("Generic serialization [" +
GENERIC_SERIALIZATION_PROTOBUF +
"] serialize result failed.", e);
}
} else {
appResponse.setValue(PojoUtils.generalize(appResponse.getValue()));
} else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
appResponse.setValue(JavaBeanSerializeUtil.serialize(appResponse.getValue(), JavaBeanAccessor.METHOD));
} else if (ProtocolUtils.isProtobufGenericSerialization(generic)) {
try {
UnsafeByteArrayOutputStream os = new UnsafeByteArrayOutputStream(512);
ExtensionLoader.getExtensionLoader(Serialization.class)
.getExtension(GENERIC_SERIALIZATION_PROTOBUF)
.serialize(null, os).writeObject(appResponse.getValue());
appResponse.setValue(os.toString());
} catch (IOException e) {
throw new RpcException("Generic serialization [" +
GENERIC_SERIALIZATION_PROTOBUF +
"] serialize result failed.", e);
}
} else {
appResponse.setValue(PojoUtils.generalize(appResponse.getValue()));
}
}
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
}
}
}

130
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericImplFilter.java
View File

@ -25,9 +25,9 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.PojoUtils;
import org.apache.dubbo.common.utils.ReflectUtils;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcInvocation;
@ -48,7 +48,7 @@ import static org.apache.dubbo.rpc.Constants.GENERIC_KEY;
* GenericImplInvokerFilter
*/
@Activate(group = CommonConstants.CONSUMER, value = GENERIC_KEY, order = 20000)
public class GenericImplFilter extends ListenableFilter {
public class GenericImplFilter implements Filter, Filter.Listener {
private static final Logger logger = LoggerFactory.getLogger(GenericImplFilter.class);
@ -56,10 +56,6 @@ public class GenericImplFilter extends ListenableFilter {
private static final String GENERIC_PARAMETER_DESC = "Ljava/lang/String;[Ljava/lang/String;[Ljava/lang/Object;";
public GenericImplFilter() {
super.listener = new GenericImplListener();
}
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
String generic = invoker.getUrl().getParameter(GENERIC_KEY);
@ -126,79 +122,77 @@ public class GenericImplFilter extends ListenableFilter {
throw new RpcException("Generic serialization [" + generic + "] only support message type " + expected + " and your message type is " + actual);
}
static class GenericImplListener implements Listener {
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
String generic = invoker.getUrl().getParameter(GENERIC_KEY);
String methodName = invocation.getMethodName();
Class<?>[] parameterTypes = invocation.getParameterTypes();
if (ProtocolUtils.isGeneric(generic)
&& (!$INVOKE.equals(invocation.getMethodName()) && !$INVOKE_ASYNC.equals(invocation.getMethodName()))
&& invocation instanceof RpcInvocation) {
if (!appResponse.hasException()) {
Object value = appResponse.getValue();
try {
Method method = invoker.getInterface().getMethod(methodName, parameterTypes);
if (ProtocolUtils.isBeanGenericSerialization(generic)) {
if (value == null) {
appResponse.setValue(value);
} else if (value instanceof JavaBeanDescriptor) {
appResponse.setValue(JavaBeanSerializeUtil.deserialize((JavaBeanDescriptor) value));
} else {
throw new RpcException("The type of result value is " + value.getClass().getName() + " other than " + JavaBeanDescriptor.class.getName() + ", and the result is " + value);
}
@Override
public void onMessage(Result appResponse, Invoker<?> invoker, Invocation invocation) {
String generic = invoker.getUrl().getParameter(GENERIC_KEY);
String methodName = invocation.getMethodName();
Class<?>[] parameterTypes = invocation.getParameterTypes();
if (ProtocolUtils.isGeneric(generic)
&& (!$INVOKE.equals(invocation.getMethodName()) && !$INVOKE_ASYNC.equals(invocation.getMethodName()))
&& invocation instanceof RpcInvocation) {
if (!appResponse.hasException()) {
Object value = appResponse.getValue();
try {
Method method = invoker.getInterface().getMethod(methodName, parameterTypes);
if (ProtocolUtils.isBeanGenericSerialization(generic)) {
if (value == null) {
appResponse.setValue(value);
} else if (value instanceof JavaBeanDescriptor) {
appResponse.setValue(JavaBeanSerializeUtil.deserialize((JavaBeanDescriptor) value));
} else {
Type[] types = ReflectUtils.getReturnTypes(method);
appResponse.setValue(PojoUtils.realize(value, (Class<?>) types[0], types[1]));
throw new RpcException("The type of result value is " + value.getClass().getName() + " other than " + JavaBeanDescriptor.class.getName() + ", and the result is " + value);
}
} catch (NoSuchMethodException e) {
throw new RpcException(e.getMessage(), e);
} else {
Type[] types = ReflectUtils.getReturnTypes(method);
appResponse.setValue(PojoUtils.realize(value, (Class<?>) types[0], types[1]));
}
} else if (appResponse.getException() instanceof GenericException) {
GenericException exception = (GenericException) appResponse.getException();
} catch (NoSuchMethodException e) {
throw new RpcException(e.getMessage(), e);
}
} else if (appResponse.getException() instanceof GenericException) {
GenericException exception = (GenericException) appResponse.getException();
try {
String className = exception.getExceptionClass();
Class<?> clazz = ReflectUtils.forName(className);
Throwable targetException = null;
Throwable lastException = null;
try {
String className = exception.getExceptionClass();
Class<?> clazz = ReflectUtils.forName(className);
Throwable targetException = null;
Throwable lastException = null;
try {
targetException = (Throwable) clazz.newInstance();
} catch (Throwable e) {
lastException = e;
for (Constructor<?> constructor : clazz.getConstructors()) {
try {
targetException = (Throwable) constructor.newInstance(new Object[constructor.getParameterTypes().length]);
break;
} catch (Throwable e1) {
lastException = e1;
}
}
}
if (targetException != null) {
try {
Field field = Throwable.class.getDeclaredField("detailMessage");
if (!field.isAccessible()) {
field.setAccessible(true);
}
field.set(targetException, exception.getExceptionMessage());
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
appResponse.setException(targetException);
} else if (lastException != null) {
throw lastException;
}
targetException = (Throwable) clazz.newInstance();
} catch (Throwable e) {
throw new RpcException("Can not deserialize exception " + exception.getExceptionClass() + ", message: " + exception.getExceptionMessage(), e);
lastException = e;
for (Constructor<?> constructor : clazz.getConstructors()) {
try {
targetException = (Throwable) constructor.newInstance(new Object[constructor.getParameterTypes().length]);
break;
} catch (Throwable e1) {
lastException = e1;
}
}
}
if (targetException != null) {
try {
Field field = Throwable.class.getDeclaredField("detailMessage");
if (!field.isAccessible()) {
field.setAccessible(true);
}
field.set(targetException, exception.getExceptionMessage());
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
appResponse.setException(targetException);
} else if (lastException != null) {
throw lastException;
}
} catch (Throwable e) {
throw new RpcException("Can not deserialize exception " + exception.getExceptionClass() + ", message: " + exception.getExceptionMessage(), e);
}
}
}
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
}
}
}

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

@ -20,9 +20,9 @@ import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
@ -32,40 +32,33 @@ import java.util.Arrays;
* Log any invocation timeout, but don't stop server from running
*/
@Activate(group = CommonConstants.PROVIDER)
public class TimeoutFilter extends ListenableFilter {
public class TimeoutFilter implements Filter, Filter.Listener {
private static final Logger logger = LoggerFactory.getLogger(TimeoutFilter.class);
private static final String TIMEOUT_FILTER_START_TIME = "timeout_filter_start_time";
public TimeoutFilter() {
super.listener = new TimeoutListener();
}
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
invocation.setAttachment(TIMEOUT_FILTER_START_TIME, String.valueOf(System.currentTimeMillis()));
invocation.put(TIMEOUT_FILTER_START_TIME, System.currentTimeMillis());
return invoker.invoke(invocation);
}
static class TimeoutListener implements Listener {
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
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)) {
if (logger.isWarnEnabled()) {
logger.warn("invoke time out. method: " + invocation.getMethodName() + " arguments: " + Arrays.toString(invocation.getArguments()) + " , url is " + invoker.getUrl() + ", invoke elapsed " + elapsed + " ms.");
}
@Override
public void onMessage(Result appResponse, Invoker<?> invoker, Invocation invocation) {
Object startTime = invocation.get(TIMEOUT_FILTER_START_TIME);
if (startTime != null) {
long elapsed = System.currentTimeMillis() - (Long) startTime;
if (invoker.getUrl() != null && elapsed > invoker.getUrl().getMethodParameter(invocation.getMethodName(), "timeout", Integer.MAX_VALUE)) {
if (logger.isWarnEnabled()) {
logger.warn("invoke time out. method: " + invocation.getMethodName() + " arguments: " + Arrays.toString(invocation.getArguments()) + " , url is " + invoker.getUrl() + ", invoke elapsed " + elapsed + " ms.");
}
}
}
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
}
}
}

147
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ConsumerMethodModel.java Normal file
View File

@ -0,0 +1,147 @@
/*
* 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.model;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE;
/**
* Replaced with {@link MethodModel}
*/
@Deprecated
public class ConsumerMethodModel {
private final Method method;
// private final boolean isCallBack;
// private final boolean isFuture;
private final String[] parameterTypes;
private final Class<?>[] parameterClasses;
private final Class<?> returnClass;
private final String methodName;
private final boolean generic;
private final AsyncMethodInfo asyncInfo;
private final ConcurrentMap<String, Object> attributeMap = new ConcurrentHashMap<>();
public ConsumerMethodModel(Method method, Map<String, Object> attributes) {
this.method = method;
this.parameterClasses = method.getParameterTypes();
this.returnClass = method.getReturnType();
this.parameterTypes = this.createParamSignature(parameterClasses);
this.methodName = method.getName();
this.generic = methodName.equals($INVOKE) && parameterTypes != null && parameterTypes.length == 3;
if (attributes != null) {
ConsumerModel.AsyncMethodInfo consumerAsyncInfo = (ConsumerModel.AsyncMethodInfo) attributes.get(methodName);
this.asyncInfo = new AsyncMethodInfo(consumerAsyncInfo);
} 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 asyncInfo;
}
public String getMethodName() {
return methodName;
}
public String[] getParameterTypes() {
return parameterTypes;
}
private String[] createParamSignature(Class<?>[] args) {
if (args == null || args.length == 0) {
return new String[]{};
}
String[] paramSig = new String[args.length];
for (int x = 0; x < args.length; x++) {
paramSig[x] = args[x].getName();
}
return paramSig;
}
public boolean isGeneric() {
return generic;
}
public Class<?>[] getParameterClasses() {
return parameterClasses;
}
public static class AsyncMethodInfo {
private ConsumerModel.AsyncMethodInfo delegate;
public AsyncMethodInfo(ConsumerModel.AsyncMethodInfo methodInfo) {
this.delegate = methodInfo;
}
public Object getOninvokeInstance() {
return delegate.getOninvokeInstance();
}
public Method getOninvokeMethod() {
return delegate.getOninvokeMethod();
}
public Object getOnreturnInstance() {
return delegate.getOnreturnInstance();
}
public Method getOnreturnMethod() {
return delegate.getOnreturnMethod();
}
public Object getOnthrowInstance() {
return delegate.getOnthrowInstance();
}
public Method getOnthrowMethod() {
return delegate.getOnthrowMethod();
}
}
}

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

@ -20,8 +20,13 @@ import org.apache.dubbo.common.utils.Assert;
import org.apache.dubbo.common.utils.CollectionUtils;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
/**
@ -31,6 +36,7 @@ public class ConsumerModel {
private final String serviceKey;
private final Object proxyObject;
private final ServiceModel serviceModel;
private final Map<String, AsyncMethodInfo> methodConfigs = new HashMap<>();
/**
@ -163,4 +169,80 @@ public class ConsumerModel {
this.onthrowMethod = onthrowMethod;
}
}
/* *************** Start, metadata compatible **************** */
private ServiceMetadata serviceMetadata;
private final Map<Method, ConsumerMethodModel> methodModels = new IdentityHashMap<Method, ConsumerMethodModel>();
public ConsumerModel(String serviceKey
, Class<?> serviceInterfaceClass
, Object proxyObject
, ServiceModel serviceModel
, Map<String, Object> attributes
, ServiceMetadata metadata) {
this(serviceKey, serviceInterfaceClass, proxyObject, serviceModel, attributes);
for (Method method : metadata.getServiceType().getMethods()) {
methodModels.put(method, new ConsumerMethodModel(method, attributes));
}
}
/**
* @return serviceMetadata
*/
public ServiceMetadata getServiceMetadata() {
return serviceMetadata;
}
/**
* Return method model for the given method on consumer side
*
* @param method method object
* @return method model
*/
public ConsumerMethodModel getMethodModel(Method method) {
return methodModels.get(method);
}
/**
* Return method model for the given method on consumer side
*
* @param method method object
* @return method model
*/
public ConsumerMethodModel getMethodModel(String method) {
Optional<Map.Entry<Method, ConsumerMethodModel>> consumerMethodModelEntry = methodModels.entrySet().stream().filter(entry -> entry.getKey().getName().equals(method)).findFirst();
return consumerMethodModelEntry.map(Map.Entry::getValue).orElse(null);
}
/**
* @param method metodName
* @param argsType method arguments type
* @return
*/
public ConsumerMethodModel getMethodModel(String method, String[] argsType) {
Optional<ConsumerMethodModel> consumerMethodModel = methodModels.entrySet().stream()
.filter(entry -> entry.getKey().getName().equals(method))
.map(Map.Entry::getValue).filter(methodModel -> Arrays.equals(argsType, methodModel.getParameterTypes()))
.findFirst();
return consumerMethodModel.orElse(null);
}
/**
* Return all method models for the current service
*
* @return method model list
*/
public List<ConsumerMethodModel> getAllMethodModels() {
return new ArrayList<>(methodModels.values());
}
public String getServiceName() {
return this.serviceMetadata.getServiceKey();
}
}

80
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ProviderMethodModel.java Normal file
View File

@ -0,0 +1,80 @@
/*
* 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.model;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* Replaced with {@link MethodModel}
*/
@Deprecated
public class ProviderMethodModel {
private final Method method;
private final String methodName;
private final Class<?>[] parameterClasses;
private final String[] methodArgTypes;
private final Type[] genericParameterTypes;
private final ConcurrentMap<String, Object> attributeMap = new ConcurrentHashMap<>();
public ProviderMethodModel(Method method) {
this.method = method;
this.methodName = method.getName();
this.parameterClasses = method.getParameterTypes();
this.methodArgTypes = getArgTypes(method);
this.genericParameterTypes = method.getGenericParameterTypes();
}
public Method getMethod() {
return method;
}
public String getMethodName() {
return methodName;
}
public String[] getMethodArgTypes() {
return methodArgTypes;
}
public ConcurrentMap<String, Object> getAttributeMap() {
return attributeMap;
}
private static String[] getArgTypes(Method method) {
String[] methodArgTypes = new String[0];
Class<?>[] parameterTypes = method.getParameterTypes();
if (parameterTypes.length > 0) {
methodArgTypes = new String[parameterTypes.length];
int index = 0;
for (Class<?> paramType : parameterTypes) {
methodArgTypes[index++] = paramType.getName();
}
}
return methodArgTypes;
}
public Class<?>[] getParameterClasses() {
return parameterClasses;
}
public Type[] getGenericParameterTypes() {
return genericParameterTypes;
}
}

76
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/model/ProviderModel.java
View File

@ -16,6 +16,13 @@
*/
package org.apache.dubbo.rpc.model;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
@ -24,7 +31,7 @@ import java.util.Set;
public class ProviderModel {
private final String serviceKey;
private final Object serviceInstance;
private final ServiceModel serviceModel;
private ServiceModel serviceModel;
public ProviderModel(String serviceKey, Object serviceInstance, ServiceModel serviceModel) {
if (null == serviceInstance) {
@ -55,4 +62,71 @@ public class ProviderModel {
public ServiceModel getServiceModel() {
return serviceModel;
}
/* *************** Start, metadata compatible **************** */
private ServiceMetadata serviceMetadata;
private final Map<String, List<ProviderMethodModel>> methods = new HashMap<String, List<ProviderMethodModel>>();
public ProviderModel(String serviceKey,
Object serviceInstance,
ServiceModel serviceModel,
ServiceMetadata serviceMetadata) {
this(serviceKey, serviceInstance, serviceModel);
this.serviceMetadata = serviceMetadata;
}
public String getServiceName() {
return this.serviceMetadata.getServiceKey();
}
public List<ProviderMethodModel> getAllMethodModels() {
List<ProviderMethodModel> result = new ArrayList<ProviderMethodModel>();
for (List<ProviderMethodModel> models : methods.values()) {
result.addAll(models);
}
return result;
}
public ProviderMethodModel getMethodModel(String methodName, String[] argTypes) {
List<ProviderMethodModel> methodModels = methods.get(methodName);
if (methodModels != null) {
for (ProviderMethodModel methodModel : methodModels) {
if (Arrays.equals(argTypes, methodModel.getMethodArgTypes())) {
return methodModel;
}
}
}
return null;
}
public List<ProviderMethodModel> getMethodModelList(String methodName) {
List<ProviderMethodModel> resultList = methods.get(methodName);
return resultList == null ? Collections.emptyList() : resultList;
}
private void initMethod(Class<?> serviceInterfaceClass) {
Method[] methodsToExport;
methodsToExport = serviceInterfaceClass.getMethods();
for (Method method : methodsToExport) {
method.setAccessible(true);
List<ProviderMethodModel> methodModels = methods.get(method.getName());
if (methodModels == null) {
methodModels = new ArrayList<ProviderMethodModel>();
methods.put(method.getName(), methodModels);
}
methodModels.add(new ProviderMethodModel(method));
}
}
/**
* @return serviceMetadata
*/
public ServiceMetadata getServiceMetadata() {
return serviceMetadata;
}
}

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

@ -26,6 +26,10 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
/**
* ServiceModel and ServiceMetadata are to some extend duplicated with each other.
* We should merge them in the future.
*/
public class ServiceModel {
private final String serviceName;
private final Class<?> serviceInterfaceClass;

16
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;
@ -55,21 +55,21 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
private final URL url;
private final Map<String, String> attachment;
private final Map<String, Object> attachment;
private volatile boolean available = true;
private AtomicBoolean destroyed = new AtomicBoolean(false);
public AbstractInvoker(Class<T> type, URL url) {
this(type, url, (Map<String, String>) null);
this(type, url, (Map<String, Object>) null);
}
public AbstractInvoker(Class<T> type, URL url, String[] keys) {
this(type, url, convertAttachment(url, keys));
}
public AbstractInvoker(Class<T> type, URL url, Map<String, String> attachment) {
public AbstractInvoker(Class<T> type, URL url, Map<String, Object> attachment) {
if (type == null) {
throw new IllegalArgumentException("service type == null");
}
@ -81,11 +81,11 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
this.attachment = attachment == null ? null : Collections.unmodifiableMap(attachment);
}
private static Map<String, String> convertAttachment(URL url, String[] keys) {
private static Map<String, Object> convertAttachment(URL url, String[] keys) {
if (ArrayUtils.isEmpty(keys)) {
return null;
}
Map<String, String> attachment = new HashMap<String, String>();
Map<String, Object> attachment = new HashMap<String, Object>();
for (String key : keys) {
String value = url.getParameter(key);
if (value != null && value.length() > 0) {
@ -143,11 +143,11 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
if (CollectionUtils.isNotEmptyMap(attachment)) {
invocation.addAttachmentsIfAbsent(attachment);
}
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
Map<String, Object> contextAttachments = RpcContext.getContext().getAttachments();
if (CollectionUtils.isNotEmptyMap(contextAttachments)) {
/**
* invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here,
* because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered
* because the {@link RpcContext#setAttachment(String, Object)} is passed in the Filter when the call is triggered
* by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is
* a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information).
*/

92
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/ProtocolFilterWrapper.java
View File

@ -81,16 +81,38 @@ public class ProtocolFilterWrapper implements Protocol {
try {
asyncResult = filter.invoke(next, invocation);
} catch (Exception e) {
// onError callback
if (filter instanceof ListenableFilter) {
if (filter instanceof ListenableFilter) {// Deprecated!
Filter.Listener listener = ((ListenableFilter) filter).listener();
if (listener != null) {
listener.onError(e, invoker, invocation);
}
} else if (filter instanceof Filter.Listener) {
Filter.Listener listener = (Filter.Listener) filter;
listener.onError(e, invoker, invocation);
}
throw e;
}
return asyncResult;
return asyncResult.whenCompleteWithContext((r, t) -> {
if (filter instanceof ListenableFilter) {// Deprecated!
Filter.Listener listener = ((ListenableFilter) filter).listener();
if (listener != null) {
if (t == null) {
listener.onMessage(r, invoker, invocation);
} else {
listener.onError(t, invoker, invocation);
}
}
} else if (filter instanceof Filter.Listener) {
Filter.Listener listener = (Filter.Listener) filter;
if (t == null) {
listener.onMessage(r, invoker, invocation);
} else {
listener.onError(t, invoker, invocation);
}
} else {
filter.onResponse(r, invoker, invocation);
}
});
}
@Override
@ -106,7 +128,7 @@ public class ProtocolFilterWrapper implements Protocol {
}
}
return new CallbackRegistrationInvoker<>(last, filters);
return last;
}
@Override
@ -135,66 +157,4 @@ public class ProtocolFilterWrapper implements Protocol {
protocol.destroy();
}
/**
* Register callback for each filter may be better, just like {@link java.util.concurrent.CompletionStage}, each callback
* registration generates a new CompletionStage whose status is determined by the original CompletionStage.
*
* If bridging status between filters is proved to not has significant performance drop, consider revert to the following commit:
* https://github.com/apache/dubbo/pull/4127
*/
static class CallbackRegistrationInvoker<T> implements Invoker<T> {
private final Invoker<T> filterInvoker;
private final List<Filter> filters;
public CallbackRegistrationInvoker(Invoker<T> filterInvoker, List<Filter> filters) {
this.filterInvoker = filterInvoker;
this.filters = filters;
}
@Override
public Result invoke(Invocation invocation) throws RpcException {
Result asyncResult = filterInvoker.invoke(invocation);
asyncResult = asyncResult.whenCompleteWithContext((r, t) -> {
for (int i = filters.size() - 1; i >= 0; i--) {
Filter filter = filters.get(i);
// onResponse callback
if (filter instanceof ListenableFilter) {
Filter.Listener listener = ((ListenableFilter) filter).listener();
if (listener != null) {
if (t == null) {
listener.onResponse(r, filterInvoker, invocation);
} else {
listener.onError(t, filterInvoker, invocation);
}
}
} else {
filter.onResponse(r, filterInvoker, invocation);
}
}
});
return asyncResult;
}
@Override
public Class<T> getInterface() {
return filterInvoker.getInterface();
}
@Override
public URL getUrl() {
return filterInvoker.getUrl();
}
@Override
public boolean isAvailable() {
return filterInvoker.isAvailable();
}
@Override
public void destroy() {
filterInvoker.destroy();
}
}
}

5
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/FutureAdapter.java → dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/FutureAdapter.java
View File

@ -14,7 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc;
package org.apache.dubbo.rpc.protocol.dubbo;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.RpcException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;

7
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AbstractResult.java → dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/package-info.java
View File

@ -14,10 +14,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc;
/**
*
* {@link org.apache.dubbo.rpc.protocol.dubbo.FutureAdapter} was in dubbo-rpc-dubbo module,
* considering some users will use this class directly, keep the package not changed.
*/
public abstract class AbstractResult implements Result {
}
package org.apache.dubbo.rpc.protocol.dubbo;

8
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ActiveLimitFilterTest.java
View File

@ -123,7 +123,7 @@ public class ActiveLimitFilterTest {
try {
Result asyncResult = activeLimitFilter.invoke(invoker, invocation);
Result result = asyncResult.get();
activeLimitFilter.listener().onResponse(result, invoker, invocation);
activeLimitFilter.onMessage(result, invoker, invocation);
} catch (RpcException expected) {
count.incrementAndGet();
// activeLimitFilter.listener().onError(expected, invoker, invocation);
@ -171,10 +171,10 @@ public class ActiveLimitFilterTest {
try {
Result asyncResult = activeLimitFilter.invoke(invoker, invocation);
Result result = asyncResult.get();
activeLimitFilter.listener().onResponse(result, invoker, invocation);
activeLimitFilter.onMessage(result, invoker, invocation);
} catch (RpcException expected) {
count.incrementAndGet();
activeLimitFilter.listener().onError(expected, invoker, invocation);
activeLimitFilter.onError(expected, invoker, invocation);
} catch (Exception e) {
fail();
}
@ -219,7 +219,7 @@ public class ActiveLimitFilterTest {
try {
activeLimitFilter.invoke(invoker, invocation);
} catch (RuntimeException ex) {
activeLimitFilter.listener().onError(ex, invoker, invocation);
activeLimitFilter.onError(ex, invoker, invocation);
int afterExceptionActiveCount = count.getActive();
assertEquals(beforeExceptionActiveCount, afterExceptionActiveCount, "After exception active count should be same");
}

4
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/CompatibleFilterFilterTest.java
View File

@ -106,7 +106,7 @@ public class CompatibleFilterFilterTest {
Result asyncResult = compatibleFilter.invoke(invoker, invocation);
AppResponse appResponse = (AppResponse) asyncResult.get();
compatibleFilter.listener().onResponse(appResponse, invoker, invocation);
compatibleFilter.onMessage(appResponse, invoker, invocation);
assertEquals(Type.High, appResponse.getValue());
}
@ -128,7 +128,7 @@ public class CompatibleFilterFilterTest {
Result asyncResult = compatibleFilter.invoke(invoker, invocation);
AppResponse appResponse = (AppResponse) asyncResult.get();
compatibleFilter.listener().onResponse(appResponse, invoker, invocation);
compatibleFilter.onMessage(appResponse, invoker, invocation);
assertEquals(Type.High, appResponse.getValue());
}

5
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ExceptionFilterTest.java
View File

@ -60,8 +60,7 @@ public class ExceptionFilterTest {
exceptionFilter.invoke(invoker, invocation);
} catch (RpcException e) {
assertEquals("TestRpcException", e.getMessage());
((ExceptionFilter.ExceptionListener) exceptionFilter.listener()).setLogger(logger);
exceptionFilter.listener().onError(e, invoker, invocation);
exceptionFilter.onError(e, invoker, invocation);
}
Mockito.verify(logger).error(eq("Got unchecked and undeclared exception which called by 127.0.0.1. service: "
@ -129,7 +128,7 @@ public class ExceptionFilterTest {
Result asyncResult = exceptionFilter.invoke(invoker, invocation);
AppResponse appResponse = (AppResponse) asyncResult.get();
exceptionFilter.listener().onResponse(appResponse, invoker, invocation);
exceptionFilter.onMessage(appResponse, invoker, invocation);
Assertions.assertFalse(appResponse.getException() instanceof HessianException);

2
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ExecuteLimitFilterTest.java
View File

@ -82,7 +82,7 @@ public class ExecuteLimitFilterTest {
executeLimitFilter.invoke(invoker, invocation);
} catch (Exception e) {
Assertions.assertTrue(e instanceof RpcException);
executeLimitFilter.listener().onError(e, invoker, invocation);
executeLimitFilter.onError(e, invoker, invocation);
}
Assertions.assertEquals(1, RpcStatus.getStatus(url, invocation.getMethodName()).getFailed());
}

2
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/GenericFilterTest.java
View File

@ -67,7 +67,7 @@ public class GenericFilterTest {
Result asyncResult = genericFilter.invoke(invoker, invocation);
AppResponse appResponse = (AppResponse) asyncResult.get();
genericFilter.listener().onResponse(appResponse, invoker, invocation);
genericFilter.onMessage(appResponse, invoker, invocation);
Assertions.assertEquals(HashMap.class, appResponse.getValue().getClass());
Assertions.assertEquals(10, ((HashMap) appResponse.getValue()).get("age"));

4
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/GenericImplFilterTest.java
View File

@ -67,7 +67,7 @@ public class GenericImplFilterTest {
Result asyncResult = genericImplFilter.invoke(invoker, invocation);
Result result = asyncResult.get();
genericImplFilter.listener().onResponse(result, invoker, invocation);
genericImplFilter.onMessage(result, invoker, invocation);
Assertions.assertEquals(Person.class, result.getValue().getClass());
Assertions.assertEquals(10, ((Person) result.getValue()).getAge());
@ -90,7 +90,7 @@ public class GenericImplFilterTest {
Result asyncResult = genericImplFilter.invoke(invoker, invocation);
Result result = asyncResult.get();
genericImplFilter.listener().onResponse(result, invoker, invocation);
genericImplFilter.onMessage(result, invoker, invocation);
Assertions.assertEquals(RuntimeException.class, result.getException().getClass());
}

7
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java
View File

@ -23,6 +23,7 @@ import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.TimeoutException;
import org.apache.dubbo.remoting.exchange.ExchangeClient;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.AsyncRpcResult;
import org.apache.dubbo.rpc.FutureContext;
import org.apache.dubbo.rpc.Invocation;
@ -93,10 +94,10 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
currentClient.send(inv, isSent);
return AsyncRpcResult.newDefaultAsyncResult(invocation);
} else {
CompletableFuture<AppResponse> appResponseFuture = currentClient.request(inv, timeout).thenApply(obj -> (AppResponse) obj);
RpcContext.getContext().setFuture(new FutureAdapter(appResponseFuture));
CompletableFuture<AppResponse> appResponseFuture =
currentClient.request(inv, timeout, getCallbackExecutor(getUrl(), inv)).thenApply(obj -> (AppResponse) obj);
// save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter
FutureContext.getContext().setCompatibleFuture(responseFuture);
FutureContext.getContext().setCompatibleFuture(appResponseFuture);
return new AsyncRpcResult(appResponseFuture, inv);
}
} catch (TimeoutException e) {

37
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/filter/FutureFilter.java
View File

@ -20,9 +20,9 @@ import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.model.ApplicationModel;
@ -37,14 +37,10 @@ import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE;
* EventFilter
*/
@Activate(group = CommonConstants.CONSUMER)
public class FutureFilter extends ListenableFilter {
public class FutureFilter implements Filter, Filter.Listener {
protected static final Logger logger = LoggerFactory.getLogger(FutureFilter.class);
public FutureFilter() {
super.listener = new FutureListener();
}
@Override
public Result invoke(final Invoker<?> invoker, final Invocation invocation) throws RpcException {
fireInvokeCallback(invoker, invocation);
@ -53,6 +49,20 @@ public class FutureFilter extends ListenableFilter {
return invoker.invoke(invocation);
}
@Override
public void onMessage(Result result, Invoker<?> invoker, Invocation invocation) {
if (result.hasException()) {
fireThrowCallback(invoker, invocation, result.getException());
} else {
fireReturnCallback(invoker, invocation, result.getValue());
}
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
}
private void fireInvokeCallback(final Invoker<?> invoker, final Invocation invocation) {
final ConsumerModel.AsyncMethodInfo asyncMethodInfo = getAsyncMethodInfo(invoker, invocation);
if (asyncMethodInfo == null) {
@ -193,19 +203,4 @@ public class FutureFilter extends ListenableFilter {
return asyncMethodInfo;
}
class FutureListener implements Listener {
@Override
public void onResponse(Result result, Invoker<?> invoker, Invocation invocation) {
if (result.hasException()) {
fireThrowCallback(invoker, invocation, result.getException());
} else {
fireReturnCallback(invoker, invocation, result.getValue());
}
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
}
}
}

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

@ -23,7 +23,6 @@ import org.apache.dubbo.remoting.TimeoutException;
import org.apache.dubbo.remoting.exchange.ExchangeClient;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.AsyncRpcResult;
import org.apache.dubbo.rpc.FutureAdapter;
import org.apache.dubbo.rpc.FutureContext;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@ -96,7 +95,6 @@ public class ThriftInvoker<T> extends AbstractInvoker<T> {
int timeout = getUrl().getMethodParameter(methodName, TIMEOUT_KEY, DEFAULT_TIMEOUT);
CompletableFuture<AppResponse> appResponseFuture = currentClient.request(inv, timeout).thenApply(obj -> (AppResponse) obj);
RpcContext.getContext().setFuture(new FutureAdapter(appResponseFuture));
// save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter
FutureContext.getContext().setCompatibleFuture(appResponseFuture);
return new AsyncRpcResult(appResponseFuture, invocation);