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

[3.0] Seperate RpcContext to client and server (#7556)

This commit is contained in:
Albumen Kevin 2021-05-10 16:10:31 +08:00 committed by GitHub
parent ea0506aa65
commit 60c55b83bc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
66 changed files with 1321 additions and 479 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java
View File

@ -17,12 +17,14 @@
package org.apache.dubbo.rpc.cluster.filter.support;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.extension.ExtensionLoader;
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.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.PenetrateAttachmentSelector;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
@ -31,6 +33,7 @@ import org.apache.dubbo.rpc.TimeoutCountDown;
import org.apache.dubbo.rpc.cluster.filter.ClusterFilter;
import java.util.Map;
import java.util.Set;
import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER;
import static org.apache.dubbo.common.constants.CommonConstants.REMOTE_APPLICATION_KEY;
@ -48,15 +51,30 @@ public class ConsumerContextFilter implements ClusterFilter, ClusterFilter.Liste
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
RpcContext context = RpcContext.getContext();
context.setInvocation(invocation)
.setLocalAddress(NetUtils.getLocalHost(), 0)
.setAttachment(REMOTE_APPLICATION_KEY, invoker.getUrl().getApplication());
RpcContext.getServiceContext()
.setInvocation(invocation)
.setLocalAddress(NetUtils.getLocalHost(), 0);
RpcContext context = RpcContext.getClientAttachment();
context.setAttachment(REMOTE_APPLICATION_KEY, invoker.getUrl().getApplication());
if (invocation instanceof RpcInvocation) {
((RpcInvocation) invocation).setInvoker(invoker);
}
Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments();
ExtensionLoader<PenetrateAttachmentSelector> selectorExtensionLoader = ExtensionLoader.getExtensionLoader(PenetrateAttachmentSelector.class);
Set<String> supportedSelectors = selectorExtensionLoader.getSupportedExtensions();
if (CollectionUtils.isNotEmpty(supportedSelectors)) {
for (String supportedSelector : supportedSelectors) {
Map<String, Object> selected = selectorExtensionLoader.getExtension(supportedSelector).select();
if (CollectionUtils.isNotEmptyMap(selected)) {
((RpcInvocation) invocation).addObjectAttachmentsIfAbsent(selected);
}
}
} else {
((RpcInvocation) invocation).addObjectAttachmentsIfAbsent(RpcContext.getServerAttachment().getObjectAttachments());
}
Map<String, Object> contextAttachments = RpcContext.getClientAttachment().getObjectAttachments();
if (CollectionUtils.isNotEmptyMap(contextAttachments)) {
/**
* invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here,
@ -67,22 +85,23 @@ public class ConsumerContextFilter implements ClusterFilter, ClusterFilter.Liste
((RpcInvocation) invocation).addObjectAttachments(contextAttachments);
}
// pass default timeout set by end user (ReferenceConfig)
Object countDown = context.get(TIME_COUNTDOWN_KEY);
if (countDown != null) {
TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countDown;
if (timeoutCountDown.isExpired()) {
return AsyncRpcResult.newDefaultAsyncResult(new RpcException(RpcException.TIMEOUT_TERMINATE,
"No time left for making the following call: " + invocation.getServiceName() + "."
+ invocation.getMethodName() + ", terminate directly."), invocation);
}
}
try {
// pass default timeout set by end user (ReferenceConfig)
Object countDown = context.getObjectAttachment(TIME_COUNTDOWN_KEY);
if (countDown != null) {
TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countDown;
if (timeoutCountDown.isExpired()) {
return AsyncRpcResult.newDefaultAsyncResult(new RpcException(RpcException.TIMEOUT_TERMINATE,
"No time left for making the following call: " + invocation.getServiceName() + "."
+ invocation.getMethodName() + ", terminate directly."), invocation);
}
}
RpcContext.removeServerContext();
return invoker.invoke(invocation);
} finally {
RpcContext.removeContext();
RpcContext.removeServiceContext();
RpcContext.removeClientAttachment();
}
}

2
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ZoneAwareFilter.java
View File

@ -41,7 +41,7 @@ public class ZoneAwareFilter implements ClusterFilter {
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
RpcContext rpcContext = RpcContext.getContext();
RpcContext rpcContext = RpcContext.getClientAttachment();
String zone = (String) rpcContext.getAttachment(REGISTRY_ZONE);
String force = (String) rpcContext.getAttachment(REGISTRY_ZONE_FORCE);
ExtensionLoader<ZoneDetector> loader = ExtensionLoader.getExtensionLoader(ZoneDetector.class);

6
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouter.java
View File

@ -73,7 +73,7 @@ public class ScriptRouter extends AbstractRouter {
function = compilable.compile(rule);
} catch (ScriptException e) {
logger.error("route error, rule has been ignored. rule: " + rule +
", url: " + RpcContext.getContext().getUrl(), e);
", url: " + RpcContext.getServiceContext().getUrl(), e);
}
@ -115,7 +115,7 @@ public class ScriptRouter extends AbstractRouter {
return getRoutedInvokers(function.eval(bindings));
} catch (ScriptException e) {
logger.error("route error, rule has been ignored. rule: " + rule + ", method:" +
invocation.getMethodName() + ", url: " + RpcContext.getContext().getUrl(), e);
invocation.getMethodName() + ", url: " + RpcContext.getServiceContext().getUrl(), e);
return invokers;
}
}
@ -142,7 +142,7 @@ public class ScriptRouter extends AbstractRouter {
// create a new List of invokers
bindings.put("invokers", new ArrayList<>(invokers));
bindings.put("invocation", invocation);
bindings.put("context", RpcContext.getContext());
bindings.put("context", RpcContext.getClientAttachment());
return bindings;
}

6
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java
View File

@ -254,7 +254,7 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
checkWhetherDestroyed();
// binding attachments into invocation.
// Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments();
// Map<String, Object> contextAttachments = RpcContext.getClientAttachment().getObjectAttachments();
// if (contextAttachments != null && contextAttachments.size() != 0) {
// ((RpcInvocation) invocation).addObjectAttachmentsIfAbsent(contextAttachments);
// }
@ -333,7 +333,7 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
private void setContext(Invoker<T> invoker) {
RpcContext context = RpcContext.getContext();
RpcContext context = RpcContext.getServiceContext();
context.setInvoker(invoker)
.setRemoteAddress(invoker.getUrl().getHost(), invoker.getUrl().getPort())
.setRemoteApplicationName(invoker.getUrl().getRemoteApplication());
@ -341,7 +341,7 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
private void clearContext(Invoker<T> invoker) {
// do nothing
RpcContext context = RpcContext.getContext();
RpcContext context = RpcContext.getServiceContext();
context.setInvoker(null);
}
}

2
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java
View File

@ -44,7 +44,7 @@ public class BroadcastClusterInvoker<T> extends AbstractClusterInvoker<T> {
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
RpcContext.getContext().setInvokers((List) invokers);
RpcContext.getServiceContext().setInvokers((List) invokers);
RpcException exception = null;
Result result = null;
for (Invoker<T> invoker : invokers) {

2
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvoker.java
View File

@ -77,7 +77,7 @@ public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
}
Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
invoked.add(invoker);
RpcContext.getContext().setInvokers((List) invoked);
RpcContext.getServiceContext().setInvokers((List) invoked);
try {
Result result = invokeWithContext(invoker, invocation);
if (le != null && logger.isWarnEnabled()) {

4
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvoker.java
View File

@ -79,7 +79,7 @@ public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> {
}
}
}
RpcContext.getContext().setInvokers((List) selected);
RpcContext.getServiceContext().setInvokers((List) selected);
final AtomicInteger count = new AtomicInteger();
final BlockingQueue<Object> ref = new LinkedBlockingQueue<>();
for (final Invoker<T> invoker : selected) {
@ -107,7 +107,7 @@ public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> {
}
} finally {
// clear attachments which is binding to current thread.
RpcContext.getContext().clearAttachments();
RpcContext.getClientAttachment().clearAttachments();
}
}
}

6
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java
View File

@ -91,7 +91,7 @@ public class AbstractClusterInvokerTest {
@AfterEach
public void teardown() throws Exception {
RpcContext.getContext().clearAttachments();
RpcContext.removeContext();
}
@SuppressWarnings({"unchecked"})
@ -163,8 +163,8 @@ public class AbstractClusterInvokerTest {
final String attachValue = "value";
// setup attachment
RpcContext.getContext().setAttachment(attachKey, attachValue);
Map<String, Object> attachments = RpcContext.getContext().getObjectAttachments();
RpcContext.getClientAttachment().setAttachment(attachKey, attachValue);
Map<String, Object> attachments = RpcContext.getClientAttachment().getObjectAttachments();
Assertions.assertTrue( attachments != null && attachments.size() == 1,"set attachment failed!");
cluster = new AbstractClusterInvoker(dic) {

2
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailSafeClusterInvokerTest.java
View File

@ -86,7 +86,7 @@ public class FailSafeClusterInvokerTest {
resetInvokerToException();
FailsafeClusterInvoker<DemoService> invoker = new FailsafeClusterInvoker<DemoService>(dic);
invoker.invoke(invocation);
Assertions.assertNull(RpcContext.getContext().getInvoker());
Assertions.assertNull(RpcContext.getServiceContext().getInvoker());
}
@Test()

4
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailbackClusterInvokerTest.java
View File

@ -107,7 +107,7 @@ public class FailbackClusterInvokerTest {
FailbackClusterInvoker<FailbackClusterInvokerTest> invoker = new FailbackClusterInvoker<FailbackClusterInvokerTest>(
dic);
invoker.invoke(invocation);
Assertions.assertNull(RpcContext.getContext().getInvoker());
Assertions.assertNull(RpcContext.getServiceContext().getInvoker());
DubboAppender.clear();
}
@ -163,7 +163,7 @@ public class FailbackClusterInvokerTest {
invoker.invoke(invocation);
invoker.invoke(invocation);
invoker.invoke(invocation);
Assertions.assertNull(RpcContext.getContext().getInvoker());
Assertions.assertNull(RpcContext.getServiceContext().getInvoker());
// invoker.retryFailed();// when retry the invoker which get from failed map already is not the mocked invoker,so
//Ensure that the main thread is online
CountDownLatch countDown = new CountDownLatch(1);

2
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailfastClusterInvokerTest.java
View File

@ -86,7 +86,7 @@ public class FailfastClusterInvokerTest {
resetInvoker1ToException();
FailfastClusterInvoker<FailfastClusterInvokerTest> invoker = new FailfastClusterInvoker<FailfastClusterInvokerTest>(dic);
invoker.invoke(invocation);
Assertions.assertSame(invoker1, RpcContext.getContext().getInvoker());
Assertions.assertSame(invoker1, RpcContext.getServiceContext().getInvoker());
});
}

6
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvokerTest.java
View File

@ -128,9 +128,9 @@ public class ForkingClusterInvokerTest {
String attachKey = "attach";
String attachValue = "value";
RpcContext.getContext().setAttachment(attachKey, attachValue);
RpcContext.getClientAttachment().setAttachment(attachKey, attachValue);
Map<String, Object> attachments = RpcContext.getContext().getObjectAttachments();
Map<String, Object> attachments = RpcContext.getClientAttachment().getObjectAttachments();
Assertions.assertTrue(attachments != null && attachments.size() == 1, "set attachment failed!");
try {
invoker.invoke(invocation);
@ -139,7 +139,7 @@ public class ForkingClusterInvokerTest {
Assertions.assertTrue(expected.getMessage().contains("Failed to forking invoke provider"), "Succeeded to forking invoke provider !");
assertFalse(expected.getCause() instanceof RpcException);
}
Map<String, Object> afterInvoke = RpcContext.getContext().getObjectAttachments();
Map<String, Object> afterInvoke = RpcContext.getClientAttachment().getObjectAttachments();
Assertions.assertTrue(afterInvoke != null && afterInvoke.size() == 0, "clear attachment failed!");
}

2
dubbo-compatible/src/test/java/org/apache/dubbo/service/DemoServiceImpl.java
View File

@ -54,7 +54,7 @@ public class DemoServiceImpl implements DemoService {
}
public Object invoke(String service, String method) throws Exception {
System.out.println("RpcContext.getContext().getRemoteHost()=" + RpcContext.getContext().getRemoteHost());
System.out.println("RpcContext.getServerAttachment().getRemoteHost()=" + RpcContext.getServiceContext().getRemoteHost());
return service + ":" + method;
}

2
dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/bootstrap/EchoServiceImpl.java
View File

@ -30,7 +30,7 @@ public class EchoServiceImpl implements EchoService {
@Override
public String echo(String message) {
RpcContext rpcContext = RpcContext.getContext();
RpcContext rpcContext = RpcContext.getServiceContext();
return format("[%s:%s] ECHO - %s", rpcContext.getLocalHost(), rpcContext.getLocalPort(), message);
}
}

4
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/ConfigTest.java
View File

@ -535,7 +535,7 @@ public class ConfigTest {
assertThat(expected.getMessage(), containsString("Tried 3 times"));
}
assertEquals(3, RpcContext.getContext().getUrls().size());
assertEquals(3, RpcContext.getServiceContext().getUrls().size());
} finally {
ctx.stop();
ctx.close();
@ -566,7 +566,7 @@ public class ConfigTest {
assertThat(expected.getMessage(), containsString("Tried 1 times"));
}
assertEquals(1, RpcContext.getContext().getUrls().size());
assertEquals(1, RpcContext.getServiceContext().getUrls().size());
} finally {
ctx.stop();
ctx.close();

16
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/SimpleRegistryService.java
View File

@ -44,7 +44,7 @@ public class SimpleRegistryService extends AbstractRegistryService {
@Override
public void register(String service, URL url) {
super.register(service, url);
String client = RpcContext.getContext().getRemoteAddressString();
String client = RpcContext.getServiceContext().getRemoteAddressString();
Map<String, URL> urls = remoteRegistered.computeIfAbsent(client, k -> new ConcurrentHashMap<>());
urls.put(service, url);
notify(service, getRegistered().get(service));
@ -53,7 +53,7 @@ public class SimpleRegistryService extends AbstractRegistryService {
@Override
public void unregister(String service, URL url) {
super.unregister(service, url);
String client = RpcContext.getContext().getRemoteAddressString();
String client = RpcContext.getServiceContext().getRemoteAddressString();
Map<String, URL> urls = remoteRegistered.get(client);
if (urls != null && urls.size() > 0) {
urls.remove(service);
@ -63,7 +63,7 @@ public class SimpleRegistryService extends AbstractRegistryService {
@Override
public void subscribe(String service, URL url, NotifyListener listener) {
String client = RpcContext.getContext().getRemoteAddressString();
String client = RpcContext.getServiceContext().getRemoteAddressString();
if (logger.isInfoEnabled()) {
logger.info("[subscribe] service: " + service + ",client:" + client);
}
@ -72,7 +72,7 @@ public class SimpleRegistryService extends AbstractRegistryService {
&& CollectionUtils.isEmpty(urls)) {
register(service, new URL("dubbo",
NetUtils.getLocalHost(),
RpcContext.getContext().getLocalPort(),
RpcContext.getServiceContext().getLocalPort(),
RegistryService.class.getName(),
url.getParameters()));
List<String> rs = registries;
@ -97,7 +97,7 @@ public class SimpleRegistryService extends AbstractRegistryService {
@Override
public void unsubscribe(String service, URL url, NotifyListener listener) {
super.unsubscribe(service, url, listener);
String client = RpcContext.getContext().getRemoteAddressString();
String client = RpcContext.getServiceContext().getRemoteAddressString();
Map<String, NotifyListener> listeners = remoteListeners.get(client);
if (listeners != null && listeners.size() > 0) {
listeners.remove(service);
@ -109,7 +109,7 @@ public class SimpleRegistryService extends AbstractRegistryService {
}
public void disconnect() {
String client = RpcContext.getContext().getRemoteAddressString();
String client = RpcContext.getServiceContext().getRemoteAddressString();
if (logger.isInfoEnabled()) {
logger.info("Disconnected " + client);
}
@ -124,8 +124,8 @@ public class SimpleRegistryService extends AbstractRegistryService {
for (Map.Entry<String, NotifyListener> entry : listeners.entrySet()) {
String service = entry.getKey();
super.unsubscribe(service, new URL("subscribe",
RpcContext.getContext().getRemoteHost(),
RpcContext.getContext().getRemotePort(),
RpcContext.getServiceContext().getRemoteHost(),
RpcContext.getServiceContext().getRemotePort(),
RegistryService.class.getName(), getSubscribed(service)), entry.getValue());
}
}

2
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/registry/nacos/demo/service/DefaultService.java
View File

@ -34,7 +34,7 @@ public class DefaultService implements DemoService {
private String serviceName;
public String sayName(String name) {
RpcContext rpcContext = RpcContext.getContext();
RpcContext rpcContext = RpcContext.getServiceContext();
return String.format("Service [name :%s , protocol: %s , port : %d] %s(\"%s\") : Hello,%s",
serviceName,
rpcContext.getUrl().getProtocol(),

2
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/samples/ZookeeperDubboSpringProviderBootstrap.java
View File

@ -49,7 +49,7 @@ class DefaultDemoService implements DemoService {
@Override
public String sayName(String name) {
RpcContext rpcContext = RpcContext.getContext();
RpcContext rpcContext = RpcContext.getServiceContext();
return format("[%s:%s] Say - %s", rpcContext.getLocalHost(), rpcContext.getLocalPort(), name);
}

4
dubbo-demo/dubbo-demo-annotation/dubbo-demo-annotation-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java
View File

@ -31,8 +31,8 @@ public class DemoServiceImpl implements DemoService {
@Override
public String sayHello(String name) {
logger.info("Hello " + name + ", request from consumer: " + RpcContext.getContext().getRemoteAddress());
return "Hello " + name + ", response from provider: " + RpcContext.getContext().getLocalAddress();
logger.info("Hello " + name + ", request from consumer: " + RpcContext.getServiceContext().getRemoteAddress());
return "Hello " + name + ", response from provider: " + RpcContext.getServiceContext().getLocalAddress();
}
@Override

4
dubbo-demo/dubbo-demo-api/dubbo-demo-api-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java
View File

@ -29,8 +29,8 @@ public class DemoServiceImpl implements DemoService {
@Override
public String sayHello(String name) {
logger.info("Hello " + name + ", request from consumer: " + RpcContext.getContext().getRemoteAddress());
return "Hello " + name + ", response from provider: " + RpcContext.getContext().getLocalAddress();
logger.info("Hello " + name + ", request from consumer: " + RpcContext.getServiceContext().getRemoteAddress());
return "Hello " + name + ", response from provider: " + RpcContext.getServiceContext().getLocalAddress();
}
@Override

4
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider-backend/src/main/java/org/apache/dubbo/demo/provider/chain/ChainServiceImpl.java
View File

@ -27,7 +27,7 @@ public class ChainServiceImpl implements ChainService {
@Override
public String chain(String input) {
logger.info("Received " + input + ", request from consumer: " + RpcContext.getContext().getRemoteAddress());
return "Received " + input + ", response from provider: " + RpcContext.getContext().getLocalAddress();
logger.info("Received " + input + ", request from consumer: " + RpcContext.getServiceContext().getRemoteAddress());
return "Received " + input + ", response from provider: " + RpcContext.getServiceContext().getLocalAddress();
}
}

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

@ -29,13 +29,13 @@ public class DemoServiceImpl implements DemoService {
@Override
public String sayHello(String name) {
logger.info("Hello " + name + ", request from consumer: " + RpcContext.getContext().getRemoteAddress());
logger.info("Hello " + name + ", request from consumer: " + RpcContext.getServiceContext().getRemoteAddress());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "Hello " + name + ", response from provider: " + RpcContext.getContext().getLocalAddress();
return "Hello " + name + ", response from provider: " + RpcContext.getServiceContext().getLocalAddress();
}
@Override

4
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/java/org/apache/dubbo/demo/provider/RestDemoServiceImpl.java
View File

@ -38,7 +38,7 @@ public class RestDemoServiceImpl implements RestDemoService {
@Override
public Integer hello(Integer a, Integer b) {
context = RpcContext.getContext().getObjectAttachments();
context = RpcContext.getServerAttachment().getObjectAttachments();
return a + b;
}
@ -53,6 +53,6 @@ public class RestDemoServiceImpl implements RestDemoService {
@Override
public String getRemoteApplicationName() {
return RpcContext.getContext().getRemoteApplicationName();
return RpcContext.getServiceContext().getRemoteApplicationName();
}
}

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

@ -95,7 +95,7 @@ public class MonitorFilter implements Filter, Filter.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) invocation.get(MONITOR_FILTER_START_TIME), false);
collect(invoker, invocation, result, RpcContext.getServiceContext().getRemoteHost(), (long) invocation.get(MONITOR_FILTER_START_TIME), false);
getConcurrent(invoker, invocation).decrementAndGet(); // count down
}
}
@ -103,7 +103,7 @@ public class MonitorFilter implements Filter, Filter.Listener {
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
if (invoker.getUrl().hasParameter(MONITOR_KEY)) {
collect(invoker, invocation, null, RpcContext.getContext().getRemoteHost(), (long) invocation.get(MONITOR_FILTER_START_TIME), true);
collect(invoker, invocation, null, RpcContext.getServiceContext().getRemoteHost(), (long) invocation.get(MONITOR_FILTER_START_TIME), true);
getConcurrent(invoker, invocation).decrementAndGet(); // count down
}
}

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

@ -120,7 +120,7 @@ public class MonitorFilterTest {
MonitorFilter monitorFilter = new MonitorFilter();
monitorFilter.setMonitorFactory(monitorFactory);
Invocation invocation = new RpcInvocation("aaa", MonitorService.class.getName(), "", new Class<?>[0], new Object[0]);
RpcContext.getContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
RpcContext.getServiceContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
Result result = monitorFilter.invoke(serviceInvoker, invocation);
result.whenCompleteWithContext((r, t) -> {
if (t == null) {
@ -163,7 +163,7 @@ public class MonitorFilterTest {
MonitorFilter monitorFilter = new MonitorFilter();
monitorFilter.setMonitorFactory(monitorFactory);
Invocation invocation = new RpcInvocation("$invoke", MonitorService.class.getName(), "", new Class<?>[]{String.class, String[].class, Object[].class}, new Object[]{"xxx", new String[]{}, new Object[]{}});
RpcContext.getContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
RpcContext.getServiceContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
Result result = monitorFilter.invoke(serviceInvoker, invocation);
result.whenCompleteWithContext((r, t) -> {
if (t == null) {

5
dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/MetricsFilter.java
View File

@ -29,7 +29,6 @@ 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.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.support.RpcUtils;
@ -59,6 +58,7 @@ import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_PROTOCOL
import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_SERVICE_COMPONENT_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.METRICS_PORT;
import static org.apache.dubbo.common.constants.CommonConstants.METRICS_PROTOCOL;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER;
import static org.apache.dubbo.monitor.Constants.DUBBO_CONSUMER;
import static org.apache.dubbo.monitor.Constants.DUBBO_CONSUMER_METHOD;
import static org.apache.dubbo.monitor.Constants.DUBBO_GROUP;
@ -95,8 +95,7 @@ public class MetricsFilter implements Filter {
}
}
RpcContext context = RpcContext.getContext();
boolean isProvider = context.isProviderSide();
boolean isProvider = invoker.getUrl().getSide(PROVIDER).equalsIgnoreCase(PROVIDER);
long start = System.currentTimeMillis();
try {
Result result = invoker.invoke(invocation); // proceed invocation chain

113
dubbo-monitor/dubbo-monitor-default/src/test/java/org/apache/dubbo/monitor/dubbo/MetricsFilterTest.java
View File

@ -24,7 +24,6 @@ import org.apache.dubbo.rpc.AppResponse;
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.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcInvocation;
@ -39,17 +38,16 @@ import com.alibaba.metrics.common.MetricObject;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SIDE;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_SIDE;
import static org.apache.dubbo.common.constants.CommonConstants.SIDE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
import static org.apache.dubbo.monitor.Constants.DUBBO_CONSUMER;
@ -65,51 +63,28 @@ import static org.mockito.Mockito.mock;
public class MetricsFilterTest {
private Invoker<DemoService> serviceInvoker;
@BeforeEach
void setUp() {
serviceInvoker = mock(Invoker.class);
private final Function<URL, Invoker<DemoService>> invokerFunction = (url)-> {
Invoker<DemoService> serviceInvoker = mock(Invoker.class);
given(serviceInvoker.isAvailable()).willReturn(false);
given(serviceInvoker.getInterface()).willReturn(DemoService.class);
given(serviceInvoker.getUrl()).willReturn(getUrl());
given(serviceInvoker.getUrl()).willReturn(url);
given(serviceInvoker.invoke(Mockito.any(Invocation.class))).willReturn(null);
doNothing().when(serviceInvoker).destroy();
}
return serviceInvoker;
};
private URL getUrl() {
return URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/org.apache.dubbo.monitor.dubbo.service.DemoService");
}
private void onInvokeReturns(AppResponse response) {
given(serviceInvoker.invoke(Mockito.any(Invocation.class))).willReturn(response);
private void onInvokeReturns(Invoker<DemoService> invoker, AppResponse response) {
given(invoker.invoke(Mockito.any(Invocation.class))).willReturn(response);
}
private final Invoker<DemoService> timeoutInvoker = new Invoker<DemoService>() {
@Override
public Class<DemoService> getInterface() {
return DemoService.class;
}
public URL getUrl() {
return URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/org.apache.dubbo.monitor.dubbo.service.DemoService");
}
@Override
public boolean isAvailable() {
return false;
}
@Override
public Result invoke(Invocation invocation) throws RpcException {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION);
}
@Override
public void destroy() {
}
};
public void onInvokerThrows(Invoker<DemoService> invoker) {
given(invoker.invoke(Mockito.any(Invocation.class))).willThrow(new RpcException(RpcException.TIMEOUT_EXCEPTION));
}
@Test
public void testConsumerSuccess() throws Exception {
@ -117,13 +92,14 @@ public class MetricsFilterTest {
metricManager.clear();
MetricsFilter metricsFilter = new MetricsFilter();
Invocation invocation = new RpcInvocation("sayName", DemoService.class.getName(), "", new Class<?>[]{Integer.class}, new Object[0]);
RpcContext.getContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
RpcContext.getContext().setUrl(serviceInvoker.getUrl().addParameter(SIDE_KEY, CONSUMER_SIDE));
RpcContext.getServiceContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
URL url = getUrl().addParameter(SIDE_KEY, CONSUMER_SIDE);
Invoker<DemoService> invoker = invokerFunction.apply(url);
AppResponse response = AppResponseBuilder.create()
.build();
onInvokeReturns(response);
onInvokeReturns(invoker, response);
for (int i = 0; i < 100; i++) {
metricsFilter.invoke(serviceInvoker, invocation);
metricsFilter.invoke(invoker, invocation);
}
FastCompass dubboClient = metricManager.getFastCompass(DUBBO_GROUP, new MetricName(DUBBO_CONSUMER, MetricLevel.MAJOR));
FastCompass dubboMethod = metricManager.getFastCompass(DUBBO_GROUP, new MetricName(DUBBO_CONSUMER_METHOD, new HashMap<String, String>(4) {
@ -145,15 +121,14 @@ public class MetricsFilterTest {
metricManager.clear();
MetricsFilter metricsFilter = new MetricsFilter();
Invocation invocation = new RpcInvocation("timeoutException", DemoService.class.getName(), "", null, null);
RpcContext.getContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
RpcContext.getContext().setUrl(timeoutInvoker.getUrl().addParameter(SIDE_KEY, CONSUMER_SIDE)
.addParameter(TIMEOUT_KEY, 300));
AppResponse response = AppResponseBuilder.create()
.build();
onInvokeReturns(response);
RpcContext.getServiceContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
URL url = getUrl().addParameter(SIDE_KEY, CONSUMER_SIDE)
.addParameter(TIMEOUT_KEY, 300);
Invoker<DemoService> invoker = invokerFunction.apply(url);
onInvokerThrows(invoker);
for (int i = 0; i < 10; i++) {
try {
metricsFilter.invoke(timeoutInvoker, invocation);
metricsFilter.invoke(invoker, invocation);
} catch (RpcException e) {
//ignore
}
@ -178,13 +153,15 @@ public class MetricsFilterTest {
metricManager.clear();
MetricsFilter metricsFilter = new MetricsFilter();
Invocation invocation = new RpcInvocation("sayName", DemoService.class.getName(), "", new Class<?>[0], new Object[0]);
RpcContext.getContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
RpcContext.getContext().setUrl(serviceInvoker.getUrl().addParameter(SIDE_KEY, PROVIDER));
RpcContext.getServiceContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
URL url = getUrl().addParameter(SIDE_KEY, PROVIDER)
.addParameter(TIMEOUT_KEY, 300);
Invoker<DemoService> invoker = invokerFunction.apply(url);
AppResponse response = AppResponseBuilder.create()
.build();
onInvokeReturns(response);
.build();
onInvokeReturns(invoker, response);
for (int i = 0; i < 100; i++) {
metricsFilter.invoke(serviceInvoker, invocation);
metricsFilter.invoke(invoker, invocation);
}
FastCompass dubboClient = metricManager.getFastCompass(DUBBO_GROUP, new MetricName(DUBBO_PROVIDER, MetricLevel.MAJOR));
FastCompass dubboMethod = metricManager.getFastCompass(DUBBO_GROUP, new MetricName(DUBBO_PROVIDER_METHOD, new HashMap<String, String>(4) {
@ -205,12 +182,14 @@ public class MetricsFilterTest {
metricManager.clear();
MetricsFilter metricsFilter = new MetricsFilter();
Invocation invocation = new RpcInvocation("sayName", DemoService.class.getName(), "", new Class<?>[0], new Object[0]);
RpcContext.getContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
RpcContext.getContext().setUrl(serviceInvoker.getUrl().addParameter(SIDE_KEY, PROVIDER_SIDE)
.addParameter(TIMEOUT_KEY, 300));
AppResponse response = AppResponseBuilder.create()
.build();
onInvokeReturns(response);
RpcContext.getServiceContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
URL url = getUrl().addParameter(SIDE_KEY, PROVIDER)
.addParameter(TIMEOUT_KEY, 300);
Invoker<DemoService> serviceInvoker = invokerFunction.apply(url);
Invoker<DemoService> timeoutInvoker = invokerFunction.apply(url);
AppResponse response = AppResponseBuilder.create().build();
onInvokeReturns(serviceInvoker, response);
onInvokerThrows(timeoutInvoker);
for (int i = 0; i < 50; i++) {
try {
metricsFilter.invoke(serviceInvoker, invocation);
@ -220,7 +199,7 @@ public class MetricsFilterTest {
}
}
Protocol protocol = new DubboProtocol();
URL url = URL.valueOf("dubbo://" + NetUtils.getLocalAddress().getHostName() + ":20880/" + MetricsService.class.getName());
url = URL.valueOf("dubbo://" + NetUtils.getLocalAddress().getHostName() + ":20880/" + MetricsService.class.getName());
Invoker<MetricsService> invoker = protocol.refer(MetricsService.class, url);
invocation = new RpcInvocation("getMetricsByGroup", DemoService.class.getName(), "", new Class<?>[]{String.class}, new Object[]{DUBBO_GROUP});
try {
@ -253,12 +232,14 @@ public class MetricsFilterTest {
MetricsFilter metricsFilter = new MetricsFilter();
Invocation sayNameInvocation = new RpcInvocation("sayName", DemoService.class.getName(), "", new Class<?>[0], new Object[0]);
Invocation echoInvocation = new RpcInvocation("echo", DemoService.class.getName(), "", new Class<?>[]{Integer.class}, new Integer[]{1});
RpcContext.getContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
RpcContext.getContext().setUrl(serviceInvoker.getUrl().addParameter(SIDE_KEY, PROVIDER_SIDE)
.addParameter(TIMEOUT_KEY, 300));
AppResponse response = AppResponseBuilder.create()
.build();
onInvokeReturns(response);
RpcContext.getServiceContext().setRemoteAddress(NetUtils.getLocalHost(), 20880).setLocalAddress(NetUtils.getLocalHost(), 2345);
URL url = getUrl().addParameter(SIDE_KEY, PROVIDER)
.addParameter(TIMEOUT_KEY, 300);
Invoker<DemoService> serviceInvoker = invokerFunction.apply(url);
Invoker<DemoService> timeoutInvoker = invokerFunction.apply(url);
AppResponse response = AppResponseBuilder.create().build();
onInvokeReturns(serviceInvoker, response);
onInvokerThrows(timeoutInvoker);
for (int i = 0; i < 50; i++) {
metricsFilter.invoke(serviceInvoker, sayNameInvocation);
metricsFilter.invoke(serviceInvoker, echoInvocation);
@ -275,7 +256,7 @@ public class MetricsFilterTest {
}
Protocol protocol = new DubboProtocol();
URL url = URL.valueOf("dubbo://" + NetUtils.getLocalAddress().getHostName() + ":20880/" + MetricsService.class.getName());
url = URL.valueOf("dubbo://" + NetUtils.getLocalAddress().getHostName() + ":20880/" + MetricsService.class.getName());
Invoker<MetricsService> invoker = protocol.refer(MetricsService.class, url);
Invocation invocation = new RpcInvocation("getMetricsByGroup", DemoService.class.getName(), "", new Class<?>[]{String.class}, new Object[]{DUBBO_GROUP});
try {

4
dubbo-plugin/dubbo-qos/src/test/java/org/apache/dubbo/qos/legacy/service/DemoServiceImpl.java
View File

@ -62,7 +62,7 @@ public class DemoServiceImpl implements DemoService {
}
public Object invoke(String service, String method) throws Exception {
System.out.println("RpcContext.getContext().getRemoteHost()=" + RpcContext.getContext().getRemoteHost());
System.out.println("RpcContext.getServerAttachment().getRemoteHost()=" + RpcContext.getServiceContext().getRemoteHost());
return service + ":" + method;
}
@ -124,7 +124,7 @@ public class DemoServiceImpl implements DemoService {
@Override
public String getRemoteApplicationName() {
return RpcContext.getContext().getRemoteApplicationName();
return RpcContext.getServiceContext().getRemoteApplicationName();
}
@Override

14
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java
View File

@ -60,30 +60,30 @@ public class InstanceAddressURL extends URL {
@Override
public String getServiceInterface() {
return RpcContext.getContext().getInterfaceName();
return RpcContext.getServiceContext().getInterfaceName();
}
public String getGroup() {
return RpcContext.getContext().getGroup();
return RpcContext.getServiceContext().getGroup();
}
public String getVersion() {
return RpcContext.getContext().getVersion();
return RpcContext.getServiceContext().getVersion();
}
@Override
public String getProtocol() {
return RpcContext.getContext().getProtocol();
return RpcContext.getServiceContext().getProtocol();
}
@Override
public String getProtocolServiceKey() {
return RpcContext.getContext().getProtocolServiceKey();
return RpcContext.getServiceContext().getProtocolServiceKey();
}
@Override
public String getServiceKey() {
return RpcContext.getContext().getServiceKey();
return RpcContext.getServiceContext().getServiceKey();
}
@Override
@ -408,7 +408,7 @@ public class InstanceAddressURL extends URL {
return instance.toString();
}
String protocolServiceKey = RpcContext.getContext().getProtocolServiceKey();
String protocolServiceKey = RpcContext.getServiceContext().getProtocolServiceKey();
if (StringUtils.isNotEmpty(protocolServiceKey)) {
return instance.toString() + ", " + metadataInfo.getServiceString(protocolServiceKey);
}

4
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
View File

@ -28,7 +28,7 @@ import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.integration.DynamicDirectory;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcServiceContext;
import org.apache.dubbo.rpc.cluster.RouterChain;
import org.eclipse.collections.impl.map.mutable.UnifiedMap;
@ -81,7 +81,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends DynamicDirectory<T> im
return;
}
// Set the context of the address notification thread.
RpcContext.setRpcContext(getConsumerUrl());
RpcServiceContext.setRpcContext(getConsumerUrl());
/**
* 3.x added for extend URL address

3
dubbo-registry/dubbo-registry-dns/src/test/java/org/apache/dubbo/registry/dns/util/DNSResolverTest.java
View File

@ -26,8 +26,5 @@ public class DNSResolverTest {
DNSResolver dnsResolver = new DNSResolver("8.8.8.8", 53, 1);
ResolveResult resolve = dnsResolver.resolve("aliyun.com");
Assertions.assertTrue(resolve.getHostnameList().size() > 0);
resolve = dnsResolver.resolve("unknowhost.unknowhost.unknowhost");
Assertions.assertEquals(0, resolve.getHostnameList().size());
}
}

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

@ -26,11 +26,11 @@ public class AsyncContextImpl implements AsyncContext {
private CompletableFuture<Object> future;
private RpcContext storedContext;
private RpcContext storedServerContext;
private RpcContextAttachment storedContext;
private RpcContextAttachment storedServerContext;
public AsyncContextImpl() {
this.storedContext = RpcContext.getContext();
this.storedContext = RpcContext.getClientAttachment();
this.storedServerContext = RpcContext.getServerContext();
}

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

@ -53,8 +53,8 @@ public class AsyncRpcResult implements Result {
* RpcContext may already have been changed when callback happens, it happens when the same thread is used to execute another RPC call.
* So we should keep the reference of current RpcContext instance and restore it before callback being executed.
*/
private RpcContext storedContext;
private RpcContext storedServerContext;
private RpcContextAttachment storedContext;
private RpcContextAttachment storedServerContext;
private Executor executor;
private Invocation invocation;
@ -64,7 +64,7 @@ public class AsyncRpcResult implements Result {
public AsyncRpcResult(CompletableFuture<AppResponse> future, Invocation invocation) {
this.responseFuture = future;
this.invocation = invocation;
this.storedContext = RpcContext.getContext();
this.storedContext = RpcContext.getClientAttachment();
this.storedServerContext = RpcContext.getServerContext();
}
@ -185,7 +185,7 @@ public class AsyncRpcResult implements Result {
public Object recreate() throws Throwable {
RpcInvocation rpcInvocation = (RpcInvocation) invocation;
if (InvokeMode.FUTURE == rpcInvocation.getInvokeMode()) {
return RpcContext.getContext().getFuture();
return RpcContext.getClientAttachment().getFuture();
}
return getAppResponse().recreate();
@ -283,11 +283,11 @@ public class AsyncRpcResult implements Result {
/**
* tmp context to use when the thread switch to Dubbo thread.
*/
private RpcContext tmpContext;
private RpcContextAttachment tmpContext;
private RpcContext tmpServerContext;
private RpcContextAttachment tmpServerContext;
private BiConsumer<Result, Throwable> beforeContext = (appResponse, t) -> {
tmpContext = RpcContext.getContext();
tmpContext = RpcContext.getClientAttachment();
tmpServerContext = RpcContext.getServerContext();
RpcContext.restoreContext(storedContext);
RpcContext.restoreServerContext(storedServerContext);

34
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/PenetrateAttachmentSelector.java Normal file
View File

@ -0,0 +1,34 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc;
import org.apache.dubbo.common.extension.SPI;
import java.util.Map;
@SPI
public interface PenetrateAttachmentSelector {
/**
* Select some attachments to pass to next hop.
* These attachments can fetch from {@link RpcContext#getServerAttachment()} or user defined.
*
* @return attachment pass to next hop
*/
Map<String, Object> select();
}

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

@ -39,7 +39,7 @@ public interface Protocol {
/**
* Export service for remote invocation: <br>
* 1. Protocol should record request source address after receive a request:
* RpcContext.getContext().setRemoteAddress();<br>
* RpcContext.getServerAttachment().setRemoteAddress();<br>
* 2. export() must be idempotent, that is, there's no difference between invoking once and invoking twice when
* export the same URL<br>
* 3. Invoker instance is passed in by the framework, protocol needs not to care <br>

430
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java
View File

@ -19,13 +19,9 @@ package org.apache.dubbo.rpc;
import org.apache.dubbo.common.Experimental;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.threadlocal.InternalThreadLocal;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@ -33,93 +29,76 @@ import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
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.PROTOCOL_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_SIDE;
import static org.apache.dubbo.rpc.Constants.ASYNC_KEY;
import static org.apache.dubbo.rpc.Constants.RETURN_KEY;
/**
* Thread local context. (API, ThreadLocal, ThreadSafe)
* <p>
* Note: RpcContext is a temporary state holder. States in RpcContext changes every time when request is sent or received.
* For example: A invokes B, then B invokes C. On service B, RpcContext saves invocation info from A to B before B
* starts invoking C, and saves invocation info from B to C after B invokes C.
* <p/>
* There are four kinds of RpcContext, which are ServerContext, ClientAttachment, ServerAttachment and ServiceContext.
* <p/>
* ServiceContext: Using to pass environment parameters in the whole invocation. For example, `remotingApplicationName`,
* `remoteAddress`, etc. {@link RpcServiceContext}
* ClientAttachment, ServerAttachment and ServiceContext are using to transfer attachments.
* Imaging a situation like this, A is calling B, and B will call C, after that, B wants to return some attachments back to A.
* ClientAttachment is using to pass attachments to next hop as a consumer. ( A --> B , in A side)
* ServerAttachment is using to fetch attachments from previous hop as a provider. ( A --> B , in B side)
* ServerContext is using to return some attachments back to client as a provider. ( A <-- B , in B side)
* The reason why using `ServiceContext` is to make API compatible with previous.
*
* @export
* @see org.apache.dubbo.rpc.filter.ContextFilter
*/
public class RpcContext {
private static final RpcContext AGENT = new RpcContext();
/**
* use internal thread local to improve performance
*/
// FIXME REQUEST_CONTEXT
private static final InternalThreadLocal<RpcContext> LOCAL = new InternalThreadLocal<RpcContext>() {
private static final InternalThreadLocal<RpcContextAttachment> SERVER_LOCAL = new InternalThreadLocal<RpcContextAttachment>() {
@Override
protected RpcContext initialValue() {
return new RpcContext();
protected RpcContextAttachment initialValue() {
return new RpcContextAttachment();
}
};
// FIXME RESPONSE_CONTEXT
private static final InternalThreadLocal<RpcContext> SERVER_LOCAL = new InternalThreadLocal<RpcContext>() {
private static final InternalThreadLocal<RpcContextAttachment> CLIENT_ATTACHMENT = new InternalThreadLocal<RpcContextAttachment>() {
@Override
protected RpcContext initialValue() {
return new RpcContext();
protected RpcContextAttachment initialValue() {
return new RpcContextAttachment();
}
};
protected final Map<String, Object> attachments = new HashMap<>();
private final Map<String, Object> values = new HashMap<String, Object>();
private static final InternalThreadLocal<RpcContextAttachment> SERVER_ATTACHMENT = new InternalThreadLocal<RpcContextAttachment>() {
@Override
protected RpcContextAttachment initialValue() {
return new RpcContextAttachment();
}
};
private List<URL> urls;
private URL url;
private String methodName;
private Class<?>[] parameterTypes;
private Object[] arguments;
private InetSocketAddress localAddress;
private InetSocketAddress remoteAddress;
private String remoteApplicationName;
@Deprecated
private List<Invoker<?>> invokers;
@Deprecated
private Invoker<?> invoker;
@Deprecated
private Invocation invocation;
// now we don't use the 'values' map to hold these objects
// we want these objects to be as generic as possible
private Object request;
private Object response;
private AsyncContext asyncContext;
private static final InternalThreadLocal<RpcServiceContext> SERVICE_CONTEXT = new InternalThreadLocal<RpcServiceContext>() {
@Override
protected RpcServiceContext initialValue() {
return new RpcServiceContext();
}
};
private boolean remove = true;
protected RpcContext() {
}
/**
* get server side context.
* get server side context. ( A <-- B , in B side)
*
* @return server context
*/
public static RpcContext getServerContext() {
public static RpcContextAttachment getServerContext() {
return SERVER_LOCAL.get();
}
public static void restoreServerContext(RpcContext oldServerContext) {
public static void restoreServerContext(RpcContextAttachment oldServerContext) {
SERVER_LOCAL.set(oldServerContext);
}
@ -137,10 +116,56 @@ public class RpcContext {
*
* @return context
*/
@Deprecated
public static RpcContext getContext() {
return LOCAL.get();
return AGENT;
}
/**
* get consumer side attachment ( A --> B , in A side)
*
* @return context
*/
public static RpcContextAttachment getClientAttachment() {
return CLIENT_ATTACHMENT.get();
}
/**
* get provider side attachment from consumer ( A --> B , in B side)
*
* @return context
*/
public static RpcContextAttachment getServerAttachment() {
return SERVER_ATTACHMENT.get();
}
/**
* Using to pass environment parameters in the whole invocation. For example, `remotingApplicationName`,
* `remoteAddress`, etc. {@link RpcServiceContext}
*
* @return context
*/
public static RpcServiceContext getServiceContext() {
return SERVICE_CONTEXT.get();
}
public static void removeServiceContext() {
SERVICE_CONTEXT.remove();
}
public static void removeClientAttachment() {
if (CLIENT_ATTACHMENT.get().canRemove()) {
CLIENT_ATTACHMENT.remove();
}
}
public static void removeServerAttachment() {
if (SERVER_ATTACHMENT.get().canRemove()) {
SERVER_ATTACHMENT.remove();
}
}
public boolean canRemove() {
return remove;
}
@ -149,8 +174,8 @@ public class RpcContext {
this.remove = remove;
}
public static void restoreContext(RpcContext oldContext) {
LOCAL.set(oldContext);
public static void restoreContext(RpcContextAttachment oldContext) {
CLIENT_ATTACHMENT.set(oldContext);
}
/**
@ -168,9 +193,14 @@ public class RpcContext {
* @param checkCanRemove if need check before remove
*/
public static void removeContext(boolean checkCanRemove) {
if (LOCAL.get().canRemove()) {
LOCAL.remove();
if (CLIENT_ATTACHMENT.get().canRemove()) {
CLIENT_ATTACHMENT.remove();
}
if (SERVER_ATTACHMENT.get().canRemove()) {
SERVER_ATTACHMENT.remove();
}
SERVER_LOCAL.remove();
SERVICE_CONTEXT.remove();
}
/**
@ -179,11 +209,11 @@ public class RpcContext {
* @return null if the underlying protocol doesn't provide support for getting request
*/
public Object getRequest() {
return request;
return SERVICE_CONTEXT.get().getRequest();
}
public void setRequest(Object request) {
this.request = request;
SERVICE_CONTEXT.get().setRequest(request);
}
/**
@ -193,7 +223,7 @@ public class RpcContext {
*/
@SuppressWarnings("unchecked")
public <T> T getRequest(Class<T> clazz) {
return (request != null && clazz.isAssignableFrom(request.getClass())) ? (T) request : null;
return SERVICE_CONTEXT.get().getRequest(clazz);
}
/**
@ -202,11 +232,11 @@ public class RpcContext {
* @return null if the underlying protocol doesn't provide support for getting response
*/
public Object getResponse() {
return response;
return SERVICE_CONTEXT.get().getResponse();
}
public void setResponse(Object response) {
this.response = response;
SERVICE_CONTEXT.get().setResponse(response);
}
/**
@ -216,7 +246,7 @@ public class RpcContext {
*/
@SuppressWarnings("unchecked")
public <T> T getResponse(Class<T> clazz) {
return (response != null && clazz.isAssignableFrom(response.getClass())) ? (T) response : null;
return SERVICE_CONTEXT.get().getResponse(clazz);
}
/**
@ -225,7 +255,7 @@ public class RpcContext {
* @return provider side.
*/
public boolean isProviderSide() {
return !isConsumerSide();
return SERVICE_CONTEXT.get().isProviderSide();
}
/**
@ -234,7 +264,7 @@ public class RpcContext {
* @return consumer side.
*/
public boolean isConsumerSide() {
return getUrl().getSide(PROVIDER_SIDE).equals(CONSUMER_SIDE);
return SERVICE_CONTEXT.get().isConsumerSide();
}
/**
@ -245,7 +275,7 @@ public class RpcContext {
*/
@SuppressWarnings("unchecked")
public <T> CompletableFuture<T> getCompletableFuture() {
return FutureContext.getContext().getCompletableFuture();
return SERVICE_CONTEXT.get().getCompletableFuture();
}
/**
@ -256,7 +286,7 @@ public class RpcContext {
*/
@SuppressWarnings("unchecked")
public <T> Future<T> getFuture() {
return FutureContext.getContext().getCompletableFuture();
return SERVICE_CONTEXT.get().getFuture();
}
/**
@ -265,23 +295,23 @@ public class RpcContext {
* @param future
*/
public void setFuture(CompletableFuture<?> future) {
FutureContext.getContext().setFuture(future);
SERVICE_CONTEXT.get().setFuture(future);
}
public List<URL> getUrls() {
return urls == null && url != null ? (List<URL>) Arrays.asList(url) : urls;
return SERVICE_CONTEXT.get().getUrls();
}
public void setUrls(List<URL> urls) {
this.urls = urls;
SERVICE_CONTEXT.get().setUrls(urls);
}
public URL getUrl() {
return url;
return SERVICE_CONTEXT.get().getUrl();
}
public void setUrl(URL url) {
this.url = url;
SERVICE_CONTEXT.get().setUrl(url);
}
/**
@ -290,11 +320,11 @@ public class RpcContext {
* @return method name.
*/
public String getMethodName() {
return methodName;
return SERVICE_CONTEXT.get().getMethodName();
}
public void setMethodName(String methodName) {
this.methodName = methodName;
SERVICE_CONTEXT.get().setMethodName(methodName);
}
/**
@ -303,11 +333,11 @@ public class RpcContext {
* @serial
*/
public Class<?>[] getParameterTypes() {
return parameterTypes;
return SERVICE_CONTEXT.get().getParameterTypes();
}
public void setParameterTypes(Class<?>[] parameterTypes) {
this.parameterTypes = parameterTypes;
SERVICE_CONTEXT.get().setParameterTypes(parameterTypes);
}
/**
@ -316,11 +346,11 @@ public class RpcContext {
* @return arguments.
*/
public Object[] getArguments() {
return arguments;
return SERVICE_CONTEXT.get().getArguments();
}
public void setArguments(Object[] arguments) {
this.arguments = arguments;
SERVICE_CONTEXT.get().setArguments(arguments);
}
/**
@ -331,11 +361,7 @@ public class RpcContext {
* @return context
*/
public RpcContext setLocalAddress(String host, int port) {
if (port < 0) {
port = 0;
}
this.localAddress = InetSocketAddress.createUnresolved(host, port);
return this;
return SERVICE_CONTEXT.get().setLocalAddress(host, port);
}
/**
@ -344,7 +370,7 @@ public class RpcContext {
* @return local address
*/
public InetSocketAddress getLocalAddress() {
return localAddress;
return SERVICE_CONTEXT.get().getLocalAddress();
}
/**
@ -354,12 +380,11 @@ public class RpcContext {
* @return context
*/
public RpcContext setLocalAddress(InetSocketAddress address) {
this.localAddress = address;
return this;
return SERVICE_CONTEXT.get().setLocalAddress(address);
}
public String getLocalAddressString() {
return getLocalHost() + ":" + getLocalPort();
return SERVICE_CONTEXT.get().getLocalAddressString();
}
/**
@ -368,11 +393,7 @@ public class RpcContext {
* @return local host name
*/
public String getLocalHostName() {
String host = localAddress == null ? null : localAddress.getHostName();
if (StringUtils.isEmpty(host)) {
return getLocalHost();
}
return host;
return SERVICE_CONTEXT.get().getLocalHostName();
}
/**
@ -383,11 +404,7 @@ public class RpcContext {
* @return context
*/
public RpcContext setRemoteAddress(String host, int port) {
if (port < 0) {
port = 0;
}
this.remoteAddress = InetSocketAddress.createUnresolved(host, port);
return this;
return SERVICE_CONTEXT.get().setRemoteAddress(host, port);
}
/**
@ -396,7 +413,7 @@ public class RpcContext {
* @return remote address
*/
public InetSocketAddress getRemoteAddress() {
return remoteAddress;
return SERVICE_CONTEXT.get().getRemoteAddress();
}
/**
@ -406,17 +423,15 @@ public class RpcContext {
* @return context
*/
public RpcContext setRemoteAddress(InetSocketAddress address) {
this.remoteAddress = address;
return this;
return SERVICE_CONTEXT.get().setRemoteAddress(address);
}
public String getRemoteApplicationName() {
return remoteApplicationName;
return SERVICE_CONTEXT.get().getRemoteApplicationName();
}
public RpcContext setRemoteApplicationName(String remoteApplicationName) {
this.remoteApplicationName = remoteApplicationName;
return this;
return SERVICE_CONTEXT.get().setRemoteApplicationName(remoteApplicationName);
}
/**
@ -425,7 +440,7 @@ public class RpcContext {
* @return remote address string.
*/
public String getRemoteAddressString() {
return getRemoteHost() + ":" + getRemotePort();
return SERVICE_CONTEXT.get().getRemoteAddressString();
}
/**
@ -434,7 +449,7 @@ public class RpcContext {
* @return remote host name
*/
public String getRemoteHostName() {
return remoteAddress == null ? null : remoteAddress.getHostName();
return SERVICE_CONTEXT.get().getRemoteHostName();
}
/**
@ -443,13 +458,7 @@ public class RpcContext {
* @return local host
*/
public String getLocalHost() {
String host = localAddress == null ? null :
localAddress.getAddress() == null ? localAddress.getHostName()
: NetUtils.filterLocalHost(localAddress.getAddress().getHostAddress());
if (host == null || host.length() == 0) {
return NetUtils.getLocalHost();
}
return host;
return SERVICE_CONTEXT.get().getLocalHost();
}
/**
@ -458,7 +467,7 @@ public class RpcContext {
* @return port
*/
public int getLocalPort() {
return localAddress == null ? 0 : localAddress.getPort();
return SERVICE_CONTEXT.get().getLocalPort();
}
/**
@ -467,9 +476,7 @@ public class RpcContext {
* @return remote host
*/
public String getRemoteHost() {
return remoteAddress == null ? null :
remoteAddress.getAddress() == null ? remoteAddress.getHostName()
: NetUtils.filterLocalHost(remoteAddress.getAddress().getHostAddress());
return SERVICE_CONTEXT.get().getRemoteHost();
}
/**
@ -478,7 +485,7 @@ public class RpcContext {
* @return remote port
*/
public int getRemotePort() {
return remoteAddress == null ? 0 : remoteAddress.getPort();
return SERVICE_CONTEXT.get().getRemotePort();
}
/**
@ -488,11 +495,11 @@ public class RpcContext {
* @return attachment
*/
public String getAttachment(String key) {
Object value = attachments.get(key);
if (value instanceof String) {
return (String) value;
String client = CLIENT_ATTACHMENT.get().getAttachment(key);
if (StringUtils.isEmpty(client)) {
return SERVER_ATTACHMENT.get().getAttachment(key);
}
return null; // or JSON.toString(value);
return client;
}
/**
@ -503,7 +510,11 @@ public class RpcContext {
*/
@Experimental("Experiment api for supporting Object transmission")
public Object getObjectAttachment(String key) {
return attachments.get(key);
Object client = CLIENT_ATTACHMENT.get().getObjectAttachment(key);
if (client == null) {
return SERVER_ATTACHMENT.get().getObjectAttachment(key);
}
return client;
}
/**
@ -523,11 +534,8 @@ public class RpcContext {
@Experimental("Experiment api for supporting Object transmission")
public RpcContext setObjectAttachment(String key, Object value) {
if (value == null) {
attachments.remove(key);
} else {
attachments.put(key, value);
}
// TODO compatible with previous
CLIENT_ATTACHMENT.get().setObjectAttachment(key, value);
return this;
}
@ -538,7 +546,7 @@ public class RpcContext {
* @return context
*/
public RpcContext removeAttachment(String key) {
attachments.remove(key);
CLIENT_ATTACHMENT.get().removeAttachment(key);
return this;
}
@ -559,7 +567,10 @@ public class RpcContext {
*/
@Experimental("Experiment api for supporting Object transmission")
public Map<String, Object> getObjectAttachments() {
return attachments;
Map<String, Object> result = new HashMap<>((int) ((CLIENT_ATTACHMENT.get().attachments.size() + SERVER_ATTACHMENT.get().attachments.size()) / .75) + 1);
result.putAll(SERVER_ATTACHMENT.get().attachments);
result.putAll(CLIENT_ATTACHMENT.get().attachments);
return result;
}
/**
@ -569,9 +580,9 @@ public class RpcContext {
* @return context
*/
public RpcContext setAttachments(Map<String, String> attachment) {
this.attachments.clear();
CLIENT_ATTACHMENT.get().attachments.clear();
if (attachment != null && attachment.size() > 0) {
this.attachments.putAll(attachment);
CLIENT_ATTACHMENT.get().attachments.putAll(attachment);
}
return this;
}
@ -584,15 +595,15 @@ public class RpcContext {
*/
@Experimental("Experiment api for supporting Object transmission")
public RpcContext setObjectAttachments(Map<String, Object> attachment) {
this.attachments.clear();
CLIENT_ATTACHMENT.get().attachments.clear();
if (attachment != null && attachment.size() > 0) {
this.attachments.putAll(attachment);
CLIENT_ATTACHMENT.get().attachments.putAll(attachment);
}
return this;
}
public void clearAttachments() {
this.attachments.clear();
CLIENT_ATTACHMENT.get().attachments.clear();
}
/**
@ -600,8 +611,9 @@ public class RpcContext {
*
* @return values
*/
@Deprecated
public Map<String, Object> get() {
return values;
return CLIENT_ATTACHMENT.get().get();
}
/**
@ -611,12 +623,9 @@ public class RpcContext {
* @param value
* @return context
*/
@Deprecated
public RpcContext set(String key, Object value) {
if (value == null) {
values.remove(key);
} else {
values.put(key, value);
}
CLIENT_ATTACHMENT.get().set(key, value);
return this;
}
@ -626,8 +635,9 @@ public class RpcContext {
* @param key
* @return value
*/
@Deprecated
public RpcContext remove(String key) {
values.remove(key);
CLIENT_ATTACHMENT.get().remove(key);
return this;
}
@ -637,8 +647,9 @@ public class RpcContext {
* @param key
* @return value
*/
@Deprecated
public Object get(String key) {
return values.get(key);
return CLIENT_ATTACHMENT.get().get(key);
}
/**
@ -646,7 +657,7 @@ public class RpcContext {
*/
@Deprecated
public boolean isServerSide() {
return isProviderSide();
return SERVICE_CONTEXT.get().isServerSide();
}
/**
@ -654,7 +665,7 @@ public class RpcContext {
*/
@Deprecated
public boolean isClientSide() {
return isConsumerSide();
return SERVICE_CONTEXT.get().isClientSide();
}
/**
@ -663,19 +674,11 @@ public class RpcContext {
@Deprecated
@SuppressWarnings({"unchecked", "rawtypes"})
public List<Invoker<?>> getInvokers() {
return invokers == null && invoker != null ? (List) Arrays.asList(invoker) : invokers;
return SERVICE_CONTEXT.get().getInvokers();
}
public RpcContext setInvokers(List<Invoker<?>> invokers) {
this.invokers = invokers;
if (CollectionUtils.isNotEmpty(invokers)) {
List<URL> urls = new ArrayList<URL>(invokers.size());
for (Invoker<?> invoker : invokers) {
urls.add(invoker.getUrl());
}
setUrls(urls);
}
return this;
return SERVICE_CONTEXT.get().setInvokers(invokers);
}
/**
@ -683,15 +686,11 @@ public class RpcContext {
*/
@Deprecated
public Invoker<?> getInvoker() {
return invoker;
return SERVICE_CONTEXT.get().getInvoker();
}
public RpcContext setInvoker(Invoker<?> invoker) {
this.invoker = invoker;
if (invoker != null) {
setUrl(invoker.getUrl());
}
return this;
return SERVICE_CONTEXT.get().setInvoker(invoker);
}
/**
@ -699,17 +698,11 @@ public class RpcContext {
*/
@Deprecated
public Invocation getInvocation() {
return invocation;
return SERVICE_CONTEXT.get().getInvocation();
}
public RpcContext setInvocation(Invocation invocation) {
this.invocation = invocation;
if (invocation != null) {
setMethodName(invocation.getMethodName());
setParameterTypes(invocation.getParameterTypes());
setArguments(invocation.getArguments());
}
return this;
return SERVICE_CONTEXT.get().setInvocation(invocation);
}
/**
@ -720,30 +713,7 @@ public class RpcContext {
*/
@SuppressWarnings("unchecked")
public <T> CompletableFuture<T> asyncCall(Callable<T> callable) {
try {
try {
setAttachment(ASYNC_KEY, Boolean.TRUE.toString());
final T o = callable.call();
//local invoke will return directly
if (o != null) {
if (o instanceof CompletableFuture) {
return (CompletableFuture<T>) o;
}
return CompletableFuture.completedFuture(o);
} else {
// The service has a normal sync method signature, should get future from RpcContext.
}
} catch (Exception e) {
throw new RpcException(e);
} finally {
removeAttachment(ASYNC_KEY);
}
} catch (final RpcException e) {
CompletableFuture<T> exceptionFuture = new CompletableFuture<>();
exceptionFuture.completeExceptionally(e);
return exceptionFuture;
}
return ((CompletableFuture<T>) getContext().getFuture());
return SERVICE_CONTEXT.get().asyncCall(callable);
}
/**
@ -752,15 +722,7 @@ public class RpcContext {
* @param runnable
*/
public void asyncCall(Runnable runnable) {
try {
setAttachment(RETURN_KEY, Boolean.FALSE.toString());
runnable.run();
} catch (Throwable e) {
// FIXME should put exception in future?
throw new RpcException("oneway call error ." + e.getMessage(), e);
} finally {
removeAttachment(RETURN_KEY);
}
SERVICE_CONTEXT.get().asyncCall(runnable);
}
/**
@ -769,88 +731,58 @@ public class RpcContext {
*/
@SuppressWarnings("unchecked")
public static AsyncContext startAsync() throws IllegalStateException {
RpcContext currentContext = getContext();
if (currentContext.asyncContext == null) {
currentContext.asyncContext = new AsyncContextImpl();
}
currentContext.asyncContext.start();
return currentContext.asyncContext;
return RpcServiceContext.startAsync();
}
protected void setAsyncContext(AsyncContext asyncContext) {
this.asyncContext = asyncContext;
SERVICE_CONTEXT.get().setAsyncContext(asyncContext);
}
public boolean isAsyncStarted() {
if (this.asyncContext == null) {
return false;
}
return asyncContext.isAsyncStarted();
return SERVICE_CONTEXT.get().isAsyncStarted();
}
public boolean stopAsync() {
return asyncContext.stop();
return SERVICE_CONTEXT.get().stopAsync();
}
public AsyncContext getAsyncContext() {
return asyncContext;
return SERVICE_CONTEXT.get().getAsyncContext();
}
// RPC service context updated before each service call.
private URL consumerUrl;
public String getGroup() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getGroup();
return SERVICE_CONTEXT.get().getGroup();
}
public String getVersion() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getVersion();
return SERVICE_CONTEXT.get().getVersion();
}
public String getInterfaceName() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getServiceInterface();
return SERVICE_CONTEXT.get().getInterfaceName();
}
public String getProtocol() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getParameter(PROTOCOL_KEY, DUBBO);
return SERVICE_CONTEXT.get().getProtocol();
}
public String getServiceKey() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getServiceKey();
return SERVICE_CONTEXT.get().getServiceKey();
}
public String getProtocolServiceKey() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getProtocolServiceKey();
return SERVICE_CONTEXT.get().getProtocolServiceKey();
}
public URL getConsumerUrl() {
return consumerUrl;
return SERVICE_CONTEXT.get().getConsumerUrl();
}
public void setConsumerUrl(URL consumerUrl) {
this.consumerUrl = consumerUrl;
SERVICE_CONTEXT.get().setConsumerUrl(consumerUrl);
}
public static void setRpcContext(URL url) {
RpcContext rpcContext = RpcContext.getContext();
rpcContext.setConsumerUrl(url);
RpcServiceContext.setRpcContext(url);
}
}
}

204
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContextAttachment.java Normal file
View File

@ -0,0 +1,204 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc;
import org.apache.dubbo.common.Experimental;
import java.util.HashMap;
import java.util.Map;
public class RpcContextAttachment extends RpcContext{
protected final Map<String, Object> attachments = new HashMap<>();
private final Map<String, Object> values = new HashMap<String, Object>();
protected RpcContextAttachment() {
}
/**
* also see {@link #getObjectAttachment(String)}.
*
* @param key
* @return attachment
*/
@Override
public String getAttachment(String key) {
Object value = attachments.get(key);
if (value instanceof String) {
return (String) value;
}
return null; // or JSON.toString(value);
}
/**
* get attachment.
*
* @param key
* @return attachment
*/
@Override
@Experimental("Experiment api for supporting Object transmission")
public Object getObjectAttachment(String key) {
return attachments.get(key);
}
/**
* set attachment.
*
* @param key
* @param value
* @return context
*/
@Override
public RpcContextAttachment setAttachment(String key, String value) {
return setObjectAttachment(key, (Object) value);
}
@Override
public RpcContextAttachment setAttachment(String key, Object value) {
return setObjectAttachment(key, value);
}
@Override
@Experimental("Experiment api for supporting Object transmission")
public RpcContextAttachment setObjectAttachment(String key, Object value) {
if (value == null) {
attachments.remove(key);
} else {
attachments.put(key, value);
}
return this;
}
/**
* remove attachment.
*
* @param key
* @return context
*/
@Override
public RpcContextAttachment removeAttachment(String key) {
attachments.remove(key);
return this;
}
/**
* get attachments.
*
* @return attachments
*/
@Override
@Deprecated
public Map<String, String> getAttachments() {
return new AttachmentsAdapter.ObjectToStringMap(this.getObjectAttachments());
}
/**
* get attachments.
*
* @return attachments
*/
@Override
@Experimental("Experiment api for supporting Object transmission")
public Map<String, Object> getObjectAttachments() {
return attachments;
}
/**
* set attachments
*
* @param attachment
* @return context
*/
@Override
public RpcContextAttachment setAttachments(Map<String, String> attachment) {
this.attachments.clear();
if (attachment != null && attachment.size() > 0) {
this.attachments.putAll(attachment);
}
return this;
}
/**
* set attachments
*
* @param attachment
* @return context
*/
@Override
@Experimental("Experiment api for supporting Object transmission")
public RpcContextAttachment setObjectAttachments(Map<String, Object> attachment) {
this.attachments.clear();
if (attachment != null && attachment.size() > 0) {
this.attachments.putAll(attachment);
}
return this;
}
@Override
public void clearAttachments() {
this.attachments.clear();
}
/**
* get values.
*
* @return values
*/
@Override
@Deprecated
public Map<String, Object> get() {
return getObjectAttachments();
}
/**
* set value.
*
* @param key
* @param value
* @return context
*/
@Override
@Deprecated
public RpcContextAttachment set(String key, Object value) {
return setAttachment(key, value);
}
/**
* remove value.
*
* @param key
* @return value
*/
@Override
@Deprecated
public RpcContextAttachment remove(String key) {
return removeAttachment(key);
}
/**
* get value.
*
* @param key
* @return value
*/
@Override
@Deprecated
public Object get(String key) {
return getAttachment(key);
}
}

655
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java Normal file
View File

@ -0,0 +1,655 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
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.PROTOCOL_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_SIDE;
import static org.apache.dubbo.rpc.Constants.ASYNC_KEY;
import static org.apache.dubbo.rpc.Constants.RETURN_KEY;
public class RpcServiceContext extends RpcContext {
protected RpcServiceContext() {
}
private List<URL> urls;
private URL url;
private String methodName;
private Class<?>[] parameterTypes;
private Object[] arguments;
private InetSocketAddress localAddress;
private InetSocketAddress remoteAddress;
private String remoteApplicationName;
@Deprecated
private List<Invoker<?>> invokers;
@Deprecated
private Invoker<?> invoker;
@Deprecated
private Invocation invocation;
// now we don't use the 'values' map to hold these objects
// we want these objects to be as generic as possible
private Object request;
private Object response;
private AsyncContext asyncContext;
private boolean remove = true;
/**
* Get the request object of the underlying RPC protocol, e.g. HttpServletRequest
*
* @return null if the underlying protocol doesn't provide support for getting request
*/
@Override
public Object getRequest() {
return request;
}
@Override
public void setRequest(Object request) {
this.request = request;
}
/**
* Get the request object of the underlying RPC protocol, e.g. HttpServletRequest
*
* @return null if the underlying protocol doesn't provide support for getting request or the request is not of the specified type
*/
@Override
@SuppressWarnings("unchecked")
public <T> T getRequest(Class<T> clazz) {
return (request != null && clazz.isAssignableFrom(request.getClass())) ? (T) request : null;
}
/**
* Get the response object of the underlying RPC protocol, e.g. HttpServletResponse
*
* @return null if the underlying protocol doesn't provide support for getting response
*/
@Override
public Object getResponse() {
return response;
}
@Override
public void setResponse(Object response) {
this.response = response;
}
/**
* Get the response object of the underlying RPC protocol, e.g. HttpServletResponse
*
* @return null if the underlying protocol doesn't provide support for getting response or the response is not of the specified type
*/
@Override
@SuppressWarnings("unchecked")
public <T> T getResponse(Class<T> clazz) {
return (response != null && clazz.isAssignableFrom(response.getClass())) ? (T) response : null;
}
/**
* is provider side.
*
* @return provider side.
*/
@Override
public boolean isProviderSide() {
return !isConsumerSide();
}
/**
* is consumer side.
*
* @return consumer side.
*/
@Override
public boolean isConsumerSide() {
return getUrl().getSide(PROVIDER_SIDE).equals(CONSUMER_SIDE);
}
/**
* get CompletableFuture.
*
* @param <T>
* @return future
*/
@Override
@SuppressWarnings("unchecked")
public <T> CompletableFuture<T> getCompletableFuture() {
return FutureContext.getContext().getCompletableFuture();
}
/**
* get future.
*
* @param <T>
* @return future
*/
@Override
@SuppressWarnings("unchecked")
public <T> Future<T> getFuture() {
return FutureContext.getContext().getCompletableFuture();
}
/**
* set future.
*
* @param future
*/
@Override
public void setFuture(CompletableFuture<?> future) {
FutureContext.getContext().setFuture(future);
}
@Override
public List<URL> getUrls() {
return urls == null && url != null ? (List<URL>) Arrays.asList(url) : urls;
}
@Override
public void setUrls(List<URL> urls) {
this.urls = urls;
}
@Override
public URL getUrl() {
return url;
}
@Override
public void setUrl(URL url) {
this.url = url;
}
/**
* get method name.
*
* @return method name.
*/
@Override
public String getMethodName() {
return methodName;
}
@Override
public void setMethodName(String methodName) {
this.methodName = methodName;
}
/**
* get parameter types.
*
* @serial
*/
@Override
public Class<?>[] getParameterTypes() {
return parameterTypes;
}
@Override
public void setParameterTypes(Class<?>[] parameterTypes) {
this.parameterTypes = parameterTypes;
}
/**
* get arguments.
*
* @return arguments.
*/
@Override
public Object[] getArguments() {
return arguments;
}
@Override
public void setArguments(Object[] arguments) {
this.arguments = arguments;
}
/**
* set local address.
*
* @param host
* @param port
* @return context
*/
@Override
public RpcServiceContext setLocalAddress(String host, int port) {
if (port < 0) {
port = 0;
}
this.localAddress = InetSocketAddress.createUnresolved(host, port);
return this;
}
/**
* get local address.
*
* @return local address
*/
@Override
public InetSocketAddress getLocalAddress() {
return localAddress;
}
/**
* set local address.
*
* @param address
* @return context
*/
@Override
public RpcServiceContext setLocalAddress(InetSocketAddress address) {
this.localAddress = address;
return this;
}
@Override
public String getLocalAddressString() {
return getLocalHost() + ":" + getLocalPort();
}
/**
* get local host name.
*
* @return local host name
*/
@Override
public String getLocalHostName() {
String host = localAddress == null ? null : localAddress.getHostName();
if (StringUtils.isEmpty(host)) {
return getLocalHost();
}
return host;
}
/**
* set remote address.
*
* @param host
* @param port
* @return context
*/
@Override
public RpcServiceContext setRemoteAddress(String host, int port) {
if (port < 0) {
port = 0;
}
this.remoteAddress = InetSocketAddress.createUnresolved(host, port);
return this;
}
/**
* get remote address.
*
* @return remote address
*/
@Override
public InetSocketAddress getRemoteAddress() {
return remoteAddress;
}
/**
* set remote address.
*
* @param address
* @return context
*/
@Override
public RpcServiceContext setRemoteAddress(InetSocketAddress address) {
this.remoteAddress = address;
return this;
}
@Override
public String getRemoteApplicationName() {
return remoteApplicationName;
}
@Override
public RpcServiceContext setRemoteApplicationName(String remoteApplicationName) {
this.remoteApplicationName = remoteApplicationName;
return this;
}
/**
* get remote address string.
*
* @return remote address string.
*/
@Override
public String getRemoteAddressString() {
return getRemoteHost() + ":" + getRemotePort();
}
/**
* get remote host name.
*
* @return remote host name
*/
@Override
public String getRemoteHostName() {
return remoteAddress == null ? null : remoteAddress.getHostName();
}
/**
* get local host.
*
* @return local host
*/
@Override
public String getLocalHost() {
String host = localAddress == null ? null :
localAddress.getAddress() == null ? localAddress.getHostName()
: NetUtils.filterLocalHost(localAddress.getAddress().getHostAddress());
if (host == null || host.length() == 0) {
return NetUtils.getLocalHost();
}
return host;
}
/**
* get local port.
*
* @return port
*/
@Override
public int getLocalPort() {
return localAddress == null ? 0 : localAddress.getPort();
}
/**
* get remote host.
*
* @return remote host
*/
@Override
public String getRemoteHost() {
return remoteAddress == null ? null :
remoteAddress.getAddress() == null ? remoteAddress.getHostName()
: NetUtils.filterLocalHost(remoteAddress.getAddress().getHostAddress());
}
/**
* get remote port.
*
* @return remote port
*/
@Override
public int getRemotePort() {
return remoteAddress == null ? 0 : remoteAddress.getPort();
}
/**
* @deprecated Replace to isProviderSide()
*/
@Override
@Deprecated
public boolean isServerSide() {
return isProviderSide();
}
/**
* @deprecated Replace to isConsumerSide()
*/
@Override
@Deprecated
public boolean isClientSide() {
return isConsumerSide();
}
/**
* @deprecated Replace to getUrls()
*/
@Override
@Deprecated
@SuppressWarnings({"unchecked", "rawtypes"})
public List<Invoker<?>> getInvokers() {
return invokers == null && invoker != null ? (List) Arrays.asList(invoker) : invokers;
}
@Override
public RpcServiceContext setInvokers(List<Invoker<?>> invokers) {
this.invokers = invokers;
if (CollectionUtils.isNotEmpty(invokers)) {
List<URL> urls = new ArrayList<URL>(invokers.size());
for (Invoker<?> invoker : invokers) {
urls.add(invoker.getUrl());
}
setUrls(urls);
}
return this;
}
/**
* @deprecated Replace to getUrl()
*/
@Override
@Deprecated
public Invoker<?> getInvoker() {
return invoker;
}
@Override
public RpcServiceContext setInvoker(Invoker<?> invoker) {
this.invoker = invoker;
if (invoker != null) {
setUrl(invoker.getUrl());
}
return this;
}
/**
* @deprecated Replace to getMethodName(), getParameterTypes(), getArguments()
*/
@Override
@Deprecated
public Invocation getInvocation() {
return invocation;
}
@Override
public RpcServiceContext setInvocation(Invocation invocation) {
this.invocation = invocation;
if (invocation != null) {
setMethodName(invocation.getMethodName());
setParameterTypes(invocation.getParameterTypes());
setArguments(invocation.getArguments());
}
return this;
}
/**
* Async invocation. Timeout will be handled even if <code>Future.get()</code> is not called.
*
* @param callable
* @return get the return result from <code>future.get()</code>
*/
@Override
@SuppressWarnings("unchecked")
public <T> CompletableFuture<T> asyncCall(Callable<T> callable) {
try {
try {
setAttachment(ASYNC_KEY, Boolean.TRUE.toString());
final T o = callable.call();
//local invoke will return directly
if (o != null) {
if (o instanceof CompletableFuture) {
return (CompletableFuture<T>) o;
}
return CompletableFuture.completedFuture(o);
} else {
// The service has a normal sync method signature, should get future from RpcContext.
}
} catch (Exception e) {
throw new RpcException(e);
} finally {
removeAttachment(ASYNC_KEY);
}
} catch (final RpcException e) {
CompletableFuture<T> exceptionFuture = new CompletableFuture<>();
exceptionFuture.completeExceptionally(e);
return exceptionFuture;
}
return ((CompletableFuture<T>) getServiceContext().getFuture());
}
/**
* one way async call, send request only, and result is not required
*
* @param runnable
*/
@Override
public void asyncCall(Runnable runnable) {
try {
setAttachment(RETURN_KEY, Boolean.FALSE.toString());
runnable.run();
} catch (Throwable e) {
// FIXME should put exception in future?
throw new RpcException("oneway call error ." + e.getMessage(), e);
} finally {
removeAttachment(RETURN_KEY);
}
}
/**
* @return
* @throws IllegalStateException
*/
@SuppressWarnings("unchecked")
public static AsyncContext startAsync() throws IllegalStateException {
RpcServiceContext currentContext = getServiceContext();
if (currentContext.asyncContext == null) {
currentContext.asyncContext = new AsyncContextImpl();
}
currentContext.asyncContext.start();
return currentContext.asyncContext;
}
@Override
protected void setAsyncContext(AsyncContext asyncContext) {
this.asyncContext = asyncContext;
}
@Override
public boolean isAsyncStarted() {
if (this.asyncContext == null) {
return false;
}
return asyncContext.isAsyncStarted();
}
@Override
public boolean stopAsync() {
return asyncContext.stop();
}
@Override
public AsyncContext getAsyncContext() {
return asyncContext;
}
// RPC service context updated before each service call.
private URL consumerUrl;
@Override
public String getGroup() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getGroup();
}
@Override
public String getVersion() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getVersion();
}
@Override
public String getInterfaceName() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getServiceInterface();
}
@Override
public String getProtocol() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getParameter(PROTOCOL_KEY, DUBBO);
}
@Override
public String getServiceKey() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getServiceKey();
}
@Override
public String getProtocolServiceKey() {
if (consumerUrl == null) {
return null;
}
return consumerUrl.getProtocolServiceKey();
}
@Override
public URL getConsumerUrl() {
return consumerUrl;
}
@Override
public void setConsumerUrl(URL consumerUrl) {
this.consumerUrl = consumerUrl;
}
public static void setRpcContext(URL url) {
RpcServiceContext rpcContext = RpcContext.getServiceContext();
rpcContext.setConsumerUrl(url);
}
}

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

@ -93,21 +93,24 @@ public class ContextFilter implements Filter, Filter.Listener {
attachments = newAttach;
}
RpcContext context = RpcContext.getContext();
context.setInvoker(invoker)
.setInvocation(invocation)
RpcContext.getServiceContext().setInvoker(invoker)
.setInvocation(invocation);
RpcContext context = RpcContext.getServerAttachment();
// .setAttachments(attachments) // merged from dubbox
.setLocalAddress(invoker.getUrl().getHost(), invoker.getUrl().getPort());
context.setLocalAddress(invoker.getUrl().getHost(), invoker.getUrl().getPort());
String remoteApplication = (String) invocation.getAttachment(REMOTE_APPLICATION_KEY);
if (StringUtils.isNotEmpty(remoteApplication)) {
context.setRemoteApplicationName(remoteApplication);
RpcContext.getServiceContext().setRemoteApplicationName(remoteApplication);
} else {
context.setRemoteApplicationName((String) context.getAttachment(REMOTE_APPLICATION_KEY));
RpcContext.getServiceContext().setRemoteApplicationName((String) context.getAttachment(REMOTE_APPLICATION_KEY));
}
long timeout = RpcUtils.getTimeout(invocation, -1);
if (timeout != -1) {
context.set(TIME_COUNTDOWN_KEY, TimeoutCountDown.newCountDown(timeout, TimeUnit.MILLISECONDS));
// pass to next hop
RpcContext.getClientAttachment().setObjectAttachment(TIME_COUNTDOWN_KEY, TimeoutCountDown.newCountDown(timeout, TimeUnit.MILLISECONDS));
}
// merged from dubbox
@ -129,8 +132,10 @@ public class ContextFilter implements Filter, Filter.Listener {
return invoker.invoke(invocation);
} finally {
context.clearAfterEachInvoke(true);
RpcContext.removeServerAttachment();
RpcContext.removeServiceContext();
// IMPORTANT! For async scenario, we must remove context from current thread, so we always create a new RpcContext for the next invoke for the same thread.
RpcContext.removeContext(true);
RpcContext.getClientAttachment().removeAttachment(TIME_COUNTDOWN_KEY);
RpcContext.removeServerContext();
}
}

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

@ -76,7 +76,7 @@ public class ExceptionFilter implements Filter, Filter.Listener {
}
// 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);
logger.error("Got unchecked and undeclared exception which called by " + RpcContext.getServiceContext().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());
@ -97,14 +97,14 @@ public class ExceptionFilter implements Filter, Filter.Listener {
// 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);
logger.warn("Fail to ExceptionFilter when called by " + RpcContext.getServiceContext().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);
logger.error("Got unchecked and undeclared exception which called by " + RpcContext.getServiceContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + e.getClass().getName() + ": " + e.getMessage(), e);
}
// For test purpose

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

@ -77,7 +77,7 @@ public class GenericFilter implements Filter, Filter.Listener {
String generic = inv.getAttachment(GENERIC_KEY);
if (StringUtils.isBlank(generic)) {
generic = RpcContext.getContext().getAttachment(GENERIC_KEY);
generic = RpcContext.getClientAttachment().getAttachment(GENERIC_KEY);
}
if (StringUtils.isEmpty(generic)
@ -161,7 +161,7 @@ public class GenericFilter implements Filter, Filter.Listener {
String generic = inv.getAttachment(GENERIC_KEY);
if (StringUtils.isBlank(generic)) {
generic = RpcContext.getContext().getAttachment(GENERIC_KEY);
generic = RpcContext.getClientAttachment().getAttachment(GENERIC_KEY);
}
if (appResponse.hasException()) {

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

@ -48,7 +48,7 @@ public class TimeoutFilter implements Filter, Filter.Listener {
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
Object obj = RpcContext.getContext().get(TIME_COUNTDOWN_KEY);
Object obj = RpcContext.getClientAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY);
if (obj != null) {
TimeoutCountDown countDown = (TimeoutCountDown) obj;
if (countDown.isExpired()) {

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

@ -48,7 +48,7 @@ public class TokenFilter implements Filter {
Map<String, Object> attachments = inv.getObjectAttachments();
String remoteToken = (attachments == null ? null : (String) 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());
throw new RpcException("Invalid token! Forbid invoke remote service " + serviceType + " method " + inv.getMethodName() + "() from consumer " + RpcContext.getServiceContext().getRemoteHost() + " to provider " + RpcContext.getServiceContext().getLocalHost());
}
}
return invoker.invoke(inv);

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

@ -32,6 +32,7 @@ import org.apache.dubbo.rpc.AsyncRpcResult;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.InvokeMode;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.PenetrateAttachmentSelector;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
@ -43,6 +44,7 @@ import java.lang.reflect.InvocationTargetException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
@ -145,11 +147,24 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
invocation.addObjectAttachmentsIfAbsent(attachment);
}
Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments();
Map<String, Object> contextAttachments = RpcContext.getClientAttachment().getObjectAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
invocation.addObjectAttachmentsIfAbsent(contextAttachments);
}
ExtensionLoader<PenetrateAttachmentSelector> selectorExtensionLoader = ExtensionLoader.getExtensionLoader(PenetrateAttachmentSelector.class);
Set<String> supportedSelectors = selectorExtensionLoader.getSupportedExtensions();
if (CollectionUtils.isNotEmpty(supportedSelectors)) {
for (String supportedSelector : supportedSelectors) {
Map<String, Object> selected = selectorExtensionLoader.getExtension(supportedSelector).select();
if (CollectionUtils.isNotEmptyMap(selected)) {
((RpcInvocation) invocation).addObjectAttachmentsIfAbsent(selected);
}
}
} else {
((RpcInvocation) invocation).addObjectAttachmentsIfAbsent(RpcContext.getServerAttachment().getObjectAttachments());
}
invocation.setInvokeMode(RpcUtils.getInvokeMode(url, invocation));
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
@ -175,7 +190,7 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
} catch (Throwable e) {
asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
}
RpcContext.getContext().setFuture(new FutureAdapter<>(asyncResult.getResponseFuture()));
RpcContext.getServiceContext().setFuture(new FutureAdapter<>(asyncResult.getResponseFuture()));
waitForResultIfSync(asyncResult, invocation);
return asyncResult;

6
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/AbstractProxyInvoker.java
View File

@ -98,7 +98,7 @@ public abstract class AbstractProxyInvoker<T> implements Invoker<T> {
});
return new AsyncRpcResult(appResponseFuture, invocation);
} catch (InvocationTargetException e) {
if (RpcContext.getContext().isAsyncStarted() && !RpcContext.getContext().stopAsync()) {
if (RpcContext.getServiceContext().isAsyncStarted() && !RpcContext.getServiceContext().stopAsync()) {
logger.error("Provider async started, but got an exception from the original method, cannot write the exception back to consumer because an async result may have returned the new thread.", e);
}
return AsyncRpcResult.newDefaultAsyncResult(null, e.getTargetException(), invocation);
@ -108,8 +108,8 @@ public abstract class AbstractProxyInvoker<T> implements Invoker<T> {
}
private CompletableFuture<Object> wrapWithFuture(Object value) {
if (RpcContext.getContext().isAsyncStarted()) {
return ((AsyncContextImpl)(RpcContext.getContext().getAsyncContext())).getInternalFuture();
if (RpcContext.getServiceContext().isAsyncStarted()) {
return ((AsyncContextImpl)(RpcContext.getServiceContext().getAsyncContext())).getInternalFuture();
} else if (value instanceof CompletableFuture) {
return (CompletableFuture<Object>) value;
}

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

@ -21,8 +21,8 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.Constants;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.RpcServiceContext;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ConsumerModel;
@ -73,7 +73,7 @@ public class InvokerInvocationHandler implements InvocationHandler {
rpcInvocation.setTargetServiceUniqueName(serviceKey);
// invoker.getUrl() returns consumer url.
RpcContext.setRpcContext(url);
RpcServiceContext.setRpcContext(url);
if (consumerModel != null) {
rpcInvocation.put(Constants.CONSUMER_MODEL, consumerModel);

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

@ -61,7 +61,7 @@ public final class AccessLogData {
* Default constructor.
*/
private AccessLogData() {
RpcContext context = RpcContext.getContext();
RpcContext context = RpcContext.getServiceContext();
data = new HashMap<>();
setLocalHost(context.getLocalHost());
setLocalPort(context.getLocalPort());

22
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/RpcContextTest.java
View File

@ -31,14 +31,14 @@ public class RpcContextTest {
@Test
public void testGetContext() {
RpcContext rpcContext = RpcContext.getContext();
RpcContext rpcContext = RpcContext.getClientAttachment();
Assertions.assertNotNull(rpcContext);
RpcContext.removeContext();
RpcContext.removeClientAttachment();
// if null, will return the initialize value.
//Assertions.assertNull(RpcContext.getContext());
Assertions.assertNotNull(RpcContext.getContext());
Assertions.assertNotEquals(rpcContext, RpcContext.getContext());
Assertions.assertNotNull(RpcContext.getClientAttachment());
Assertions.assertNotEquals(rpcContext, RpcContext.getClientAttachment());
RpcContext serverRpcContext = RpcContext.getServerContext();
Assertions.assertNotNull(serverRpcContext);
@ -50,7 +50,7 @@ public class RpcContextTest {
@Test
public void testAddress() {
RpcContext context = RpcContext.getContext();
RpcContext context = RpcContext.getServiceContext();
context.setLocalAddress("127.0.0.1", 20880);
Assertions.assertEquals(20880, context.getLocalAddress().getPort());
Assertions.assertEquals("127.0.0.1:20880", context.getLocalAddressString());
@ -70,7 +70,7 @@ public class RpcContextTest {
@Test
public void testCheckSide() {
RpcContext context = RpcContext.getContext();
RpcContext context = RpcContext.getServiceContext();
//TODO fix npe
//context.isProviderSide();
@ -87,7 +87,7 @@ public class RpcContextTest {
@Test
public void testAttachments() {
RpcContext context = RpcContext.getContext();
RpcContext context = RpcContext.getClientAttachment();
Map<String, Object> map = new HashMap<>();
map.put("_11", "1111");
map.put("_22", "2222");
@ -115,7 +115,7 @@ public class RpcContextTest {
@Test
public void testObject() {
RpcContext context = RpcContext.getContext();
RpcContext context = RpcContext.getClientAttachment();
Map<String, Object> map = new HashMap<String, Object>();
map.put("_11", "1111");
map.put("_22", "2222");
@ -144,7 +144,7 @@ public class RpcContextTest {
@Test
public void testAsync() {
RpcContext rpcContext = RpcContext.getContext();
RpcContext rpcContext = RpcContext.getServiceContext();
Assertions.assertFalse(rpcContext.isAsyncStarted());
AsyncContext asyncContext = RpcContext.startAsync();
@ -160,7 +160,7 @@ public class RpcContextTest {
@Test
public void testAsyncCall() {
CompletableFuture<String> rpcFuture = RpcContext.getContext().asyncCall(() -> {
CompletableFuture<String> rpcFuture = RpcContext.getClientAttachment().asyncCall(() -> {
throw new NullPointerException();
});
@ -180,7 +180,7 @@ public class RpcContextTest {
@Test
public void testObjectAttachment() {
RpcContext rpcContext = RpcContext.getContext();
RpcContext rpcContext = RpcContext.getClientAttachment();
rpcContext.setAttachment("objectKey1", "value1");
rpcContext.setAttachment("objectKey2", "value2");

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

@ -62,7 +62,7 @@ public class ContextFilterTest {
given(invoker.getUrl()).willReturn(url);
contextFilter.invoke(invoker, invocation);
assertNull(RpcContext.getContext().getInvoker());
assertNull(RpcContext.getServiceContext().getInvoker());
}
@Test
@ -71,6 +71,6 @@ public class ContextFilterTest {
Invoker<DemoService> invoker = new MyInvoker<DemoService>(url);
Invocation invocation = new MockInvocation();
Result result = contextFilter.invoke(invoker, invocation);
assertNull(RpcContext.getContext().getInvoker());
assertNull(RpcContext.getServiceContext().getInvoker());
}
}

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

@ -47,7 +47,7 @@ public class ExceptionFilterTest {
@Test
public void testRpcException() {
Logger logger = mock(Logger.class);
RpcContext.getContext().setRemoteAddress("127.0.0.1", 1234);
RpcContext.getServiceContext().setRemoteAddress("127.0.0.1", 1234);
RpcException exception = new RpcException("TestRpcException");
ExceptionFilter exceptionFilter = new ExceptionFilter();

2
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/proxy/DemoServiceImpl.java
View File

@ -56,7 +56,7 @@ public class DemoServiceImpl implements DemoService {
}
public Object invoke(String service, String method) throws Exception {
System.out.println("RpcContext.getContext().getRemoteHost()=" + RpcContext.getContext().getRemoteHost());
System.out.println("RpcContext.getServerAttachment().getRemoteHost()=" + RpcContext.getServiceContext().getRemoteHost());
return service + ":" + method;
}

2
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/proxy/RemoteServiceImpl.java
View File

@ -22,7 +22,7 @@ import java.rmi.RemoteException;
public class RemoteServiceImpl implements RemoteService {
public String getThreadName() throws RemoteException {
System.out.println("RpcContext.getContext().getRemoteHost()=" + RpcContext.getContext().getRemoteHost());
System.out.println("RpcContext.getServerAttachment().getRemoteHost()=" + RpcContext.getServiceContext().getRemoteHost());
return Thread.currentThread().getName();
}

2
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/support/DemoServiceImpl.java
View File

@ -61,7 +61,7 @@ public class DemoServiceImpl implements DemoService {
}
public Object invoke(String service, String method) throws Exception {
System.out.println("RpcContext.getContext().getRemoteHost()=" + RpcContext.getContext().getRemoteHost());
System.out.println("RpcContext.getServerAttachment().getRemoteHost()=" + RpcContext.getServiceContext().getRemoteHost());
return service + ":" + method;
}

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

@ -163,10 +163,10 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
}
private int calculateTimeout(Invocation invocation, String methodName) {
Object countdown = RpcContext.getContext().get(TIME_COUNTDOWN_KEY);
Object countdown = RpcContext.getClientAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY);
int timeout = DEFAULT_TIMEOUT;
if (countdown == null) {
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getContext(), DEFAULT_TIMEOUT);
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getClientAttachment(), DEFAULT_TIMEOUT);
if (getUrl().getParameter(ENABLE_TIMEOUT_COUNTDOWN_KEY, false)) {
invocation.setObjectAttachment(TIMEOUT_ATTACHMENT_KEY, timeout); // pass timeout to remote server
}

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

@ -141,7 +141,7 @@ public class DubboProtocol extends AbstractProtocol {
return null;
}
}
RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
RpcContext.getServiceContext().setRemoteAddress(channel.getRemoteAddress());
Result result = invoker.invoke(inv);
return result.thenApply(Function.identity());
}

2
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/filter/TraceFilter.java
View File

@ -101,7 +101,7 @@ public class TraceFilter implements Filter {
count = c.getAndIncrement();
if (count < max) {
String prompt = channel.getUrl().getParameter(Constants.PROMPT_KEY, Constants.DEFAULT_PROMPT);
channel.send("\r\n" + RpcContext.getContext().getRemoteAddress() + " -> "
channel.send("\r\n" + RpcContext.getServiceContext().getRemoteAddress() + " -> "
+ invoker.getInterface().getName()
+ "." + invocation.getMethodName()
+ "(" + JSON.toJSONString(invocation.getArguments()) + ")" + " -> " + JSON.toJSONString(result.getValue())

4
dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/support/DemoServiceImpl.java
View File

@ -62,7 +62,7 @@ public class DemoServiceImpl implements DemoService {
}
public Object invoke(String service, String method) throws Exception {
System.out.println("RpcContext.getContext().getRemoteHost()=" + RpcContext.getContext().getRemoteHost());
System.out.println("RpcContext.getServerAttachment().getRemoteHost()=" + RpcContext.getServiceContext().getRemoteHost());
return service + ":" + method;
}
@ -124,6 +124,6 @@ public class DemoServiceImpl implements DemoService {
@Override
public String getRemoteApplicationName() {
return RpcContext.getContext().getRemoteApplicationName();
return RpcContext.getServiceContext().getRemoteApplicationName();
}
}

2
dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/support/RemoteServiceImpl.java
View File

@ -22,7 +22,7 @@ import java.rmi.RemoteException;
public class RemoteServiceImpl implements RemoteService {
public String getThreadName() throws RemoteException {
System.out.println("RpcContext.getContext().getRemoteHost()=" + RpcContext.getContext().getRemoteHost());
System.out.println("RpcContext.getServerAttachment().getRemoteHost()=" + RpcContext.getServiceContext().getRemoteHost());
return Thread.currentThread().getName();
}

11
dubbo-rpc/dubbo-rpc-grpc/src/main/java/org/apache/dubbo/rpc/protocol/grpc/interceptors/RpcContextInterceptor.java
View File

@ -16,6 +16,9 @@
*/
package org.apache.dubbo.rpc.protocol.grpc.interceptors;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.RpcContext;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
@ -25,8 +28,6 @@ import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.RpcContext;
import java.util.HashMap;
import java.util.Map;
@ -46,7 +47,7 @@ public class RpcContextInterceptor implements ClientInterceptor, ServerIntercept
@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
RpcContext rpcContext = RpcContext.getContext();
RpcContext rpcContext = RpcContext.getClientAttachment();
Map<String, Object> attachments = new HashMap<>(rpcContext.getObjectAttachments());
return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
@ -81,7 +82,7 @@ public class RpcContextInterceptor implements ClientInterceptor, ServerIntercept
public void onHalfClose() {
// the client completed all message sending and server will call the biz method if client is not the streaming
if (call.getMethodDescriptor().getType().clientSendsOneMessage()) {
RpcContext.getContext().setObjectAttachments(attachments);
RpcContext.getServerAttachment().setObjectAttachments(attachments);
}
super.onHalfClose();
}
@ -90,7 +91,7 @@ public class RpcContextInterceptor implements ClientInterceptor, ServerIntercept
public void onMessage(ReqT message) {
//server receive the request from client and call the biz method if client is streaming
if (!call.getMethodDescriptor().getType().clientSendsOneMessage()) {
RpcContext.getContext().setObjectAttachments(attachments);
RpcContext.getServerAttachment().setObjectAttachments(attachments);
}
super.onMessage(message);
}

2
dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmInvoker.java
View File

@ -60,7 +60,7 @@ class InjvmInvoker<T> extends AbstractInvoker<T> {
if (exporter == null) {
throw new RpcException("Service [" + key + "] not found.");
}
RpcContext.getContext().setRemoteAddress(LOCALHOST_VALUE, 0);
RpcContext.getServiceContext().setRemoteAddress(LOCALHOST_VALUE, 0);
// Solve local exposure, the server opens the token, and the client call fails.
URL serverURL = exporter.getInvoker().getUrl();
boolean serverHasToken = serverURL.hasParameter(Constants.TOKEN_KEY);

2
dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/DemoServiceImpl.java
View File

@ -56,7 +56,7 @@ public class DemoServiceImpl implements DemoService {
}
public Object invoke(String service, String method) throws Exception {
System.out.println("RpcContext.getContext().getRemoteHost()=" + RpcContext.getContext().getRemoteHost());
System.out.println("RpcContext.getServerAttachment().getRemoteHost()=" + RpcContext.getServiceContext().getRemoteHost());
return service + ":" + method;
}

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

@ -85,7 +85,7 @@ public class DubboHttpProtocolServer extends BaseRestProtocolServer {
@Override
public void handle(HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
RpcContext.getContext().setRemoteAddress(request.getRemoteAddr(), request.getRemotePort());
RpcContext.getServiceContext().setRemoteAddress(request.getRemoteAddr(), request.getRemotePort());
dispatcher.service(request, response);
}
}

12
dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RpcContextFilter.java
View File

@ -43,14 +43,14 @@ public class RpcContextFilter implements ContainerRequestFilter, ClientRequestFi
@Override
public void filter(ContainerRequestContext requestContext) throws IOException {
HttpServletRequest request = ResteasyProviderFactory.getContextData(HttpServletRequest.class);
RpcContext.getContext().setRequest(request);
RpcContext.getServiceContext().setRequest(request);
// this only works for servlet containers
if (request != null && RpcContext.getContext().getRemoteAddress() == null) {
RpcContext.getContext().setRemoteAddress(request.getRemoteAddr(), request.getRemotePort());
if (request != null && RpcContext.getServiceContext().getRemoteAddress() == null) {
RpcContext.getServiceContext().setRemoteAddress(request.getRemoteAddr(), request.getRemotePort());
}
RpcContext.getContext().setResponse(ResteasyProviderFactory.getContextData(HttpServletResponse.class));
RpcContext.getServiceContext().setResponse(ResteasyProviderFactory.getContextData(HttpServletResponse.class));
String headers = requestContext.getHeaderString(DUBBO_ATTACHMENT_HEADER);
if (headers != null) {
@ -60,7 +60,7 @@ public class RpcContextFilter implements ContainerRequestFilter, ClientRequestFi
String key = header.substring(0, index);
String value = header.substring(index + 1);
if (!StringUtils.isEmpty(key)) {
RpcContext.getContext().setAttachment(key.trim(), value.trim());
RpcContext.getServerAttachment().setAttachment(key.trim(), value.trim());
}
}
}
@ -70,7 +70,7 @@ public class RpcContextFilter implements ContainerRequestFilter, ClientRequestFi
@Override
public void filter(ClientRequestContext requestContext) throws IOException {
int size = 0;
for (Map.Entry<String, Object> entry : RpcContext.getContext().getObjectAttachments().entrySet()) {
for (Map.Entry<String, Object> entry : RpcContext.getClientAttachment().getObjectAttachments().entrySet()) {
String key = entry.getKey();
String value = (String) entry.getValue();
if (illegalHttpHeaderKey(key) || illegalHttpHeaderValue(value)) {

2
dubbo-rpc/dubbo-rpc-rest/src/test/java/org/apache/dubbo/rpc/protocol/rest/DemoServiceImpl.java
View File

@ -37,7 +37,7 @@ public class DemoServiceImpl implements DemoService {
@Override
public Integer hello(Integer a, Integer b) {
context = RpcContext.getContext().getObjectAttachments();
context = RpcContext.getServerAttachment().getObjectAttachments();
return a + b;
}

12
dubbo-rpc/dubbo-rpc-rest/src/test/java/org/apache/dubbo/rpc/protocol/rest/RestProtocolTest.java
View File

@ -112,7 +112,7 @@ public class RestProtocolTest {
this.registerProvider(exportUrl, server, DemoService.class);
RpcContext.getContext().setAttachment("timeout", "200");
RpcContext.getClientAttachment().setAttachment("timeout", "200");
Exporter<DemoService> exporter = protocol.export(proxy.getInvoker(server, DemoService.class, exportUrl));
DemoService demoService = this.proxy.getProxy(protocol.refer(DemoService.class, exportUrl));
@ -216,11 +216,11 @@ public class RestProtocolTest {
DemoService demoService = this.proxy.getProxy(protocol.refer(DemoService.class, nettyUrl));
// make sure null and base64 encoded string can work
RpcContext.getContext().setAttachment("key1", null);
RpcContext.getContext().setAttachment("key2", "value");
RpcContext.getContext().setAttachment("key3", "=value");
RpcContext.getContext().setAttachment("key4", "YWJjZGVmCg==");
RpcContext.getContext().setAttachment("key5", "val=ue");
RpcContext.getClientAttachment().setAttachment("key1", null);
RpcContext.getClientAttachment().setAttachment("key2", "value");
RpcContext.getClientAttachment().setAttachment("key3", "=value");
RpcContext.getClientAttachment().setAttachment("key4", "YWJjZGVmCg==");
RpcContext.getClientAttachment().setAttachment("key5", "val=ue");
Integer result = demoService.hello(1, 2);
assertThat(result, is(3));

4
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
View File

@ -172,10 +172,10 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
}
private int calculateTimeout(Invocation invocation, String methodName) {
Object countdown = RpcContext.getContext().get(TIME_COUNTDOWN_KEY);
Object countdown = RpcContext.getClientAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY);
int timeout;
if (countdown == null) {
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getContext(), DEFAULT_TIMEOUT);
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getClientAttachment(), DEFAULT_TIMEOUT);
if (getUrl().getParameter(ENABLE_TIMEOUT_COUNTDOWN_KEY, false)) {
invocation.setObjectAttachment(TIMEOUT_ATTACHMENT_KEY, timeout); // pass timeout to remote server
}