diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java index 4a1ed3bffc..a931e480eb 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java @@ -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 contextAttachments = RpcContext.getContext().getObjectAttachments(); + ExtensionLoader selectorExtensionLoader = ExtensionLoader.getExtensionLoader(PenetrateAttachmentSelector.class); + Set supportedSelectors = selectorExtensionLoader.getSupportedExtensions(); + if (CollectionUtils.isNotEmpty(supportedSelectors)) { + for (String supportedSelector : supportedSelectors) { + Map selected = selectorExtensionLoader.getExtension(supportedSelector).select(); + if (CollectionUtils.isNotEmptyMap(selected)) { + ((RpcInvocation) invocation).addObjectAttachmentsIfAbsent(selected); + } + } + } else { + ((RpcInvocation) invocation).addObjectAttachmentsIfAbsent(RpcContext.getServerAttachment().getObjectAttachments()); + } + + Map 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(); } } diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ZoneAwareFilter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ZoneAwareFilter.java index cd0a7ab655..c31f01c69f 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ZoneAwareFilter.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ZoneAwareFilter.java @@ -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 loader = ExtensionLoader.getExtensionLoader(ZoneDetector.class); diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouter.java index dab69b39a5..df94f6bc99 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouter.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouter.java @@ -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; } diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java index d0544cfcc3..bd153ad8eb 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java @@ -254,7 +254,7 @@ public abstract class AbstractClusterInvoker implements ClusterInvoker { checkWhetherDestroyed(); // binding attachments into invocation. -// Map contextAttachments = RpcContext.getContext().getObjectAttachments(); +// Map contextAttachments = RpcContext.getClientAttachment().getObjectAttachments(); // if (contextAttachments != null && contextAttachments.size() != 0) { // ((RpcInvocation) invocation).addObjectAttachmentsIfAbsent(contextAttachments); // } @@ -333,7 +333,7 @@ public abstract class AbstractClusterInvoker implements ClusterInvoker { private void setContext(Invoker 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 implements ClusterInvoker { private void clearContext(Invoker invoker) { // do nothing - RpcContext context = RpcContext.getContext(); + RpcContext context = RpcContext.getServiceContext(); context.setInvoker(null); } } diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java index 92a574328b..450fa9197b 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/BroadcastClusterInvoker.java @@ -44,7 +44,7 @@ public class BroadcastClusterInvoker extends AbstractClusterInvoker { @SuppressWarnings({"unchecked", "rawtypes"}) public Result doInvoke(final Invocation invocation, List> 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 invoker : invokers) { diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvoker.java index cc3ed9c897..414deaa0fc 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvoker.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvoker.java @@ -77,7 +77,7 @@ public class FailoverClusterInvoker extends AbstractClusterInvoker { } Invoker 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()) { diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvoker.java index 85a661b3a4..664a353b3b 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvoker.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvoker.java @@ -79,7 +79,7 @@ public class ForkingClusterInvoker extends AbstractClusterInvoker { } } } - RpcContext.getContext().setInvokers((List) selected); + RpcContext.getServiceContext().setInvokers((List) selected); final AtomicInteger count = new AtomicInteger(); final BlockingQueue ref = new LinkedBlockingQueue<>(); for (final Invoker invoker : selected) { @@ -107,7 +107,7 @@ public class ForkingClusterInvoker extends AbstractClusterInvoker { } } finally { // clear attachments which is binding to current thread. - RpcContext.getContext().clearAttachments(); + RpcContext.getClientAttachment().clearAttachments(); } } } diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java index b5a235ee38..df3a47ce74 100644 --- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java @@ -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 attachments = RpcContext.getContext().getObjectAttachments(); + RpcContext.getClientAttachment().setAttachment(attachKey, attachValue); + Map attachments = RpcContext.getClientAttachment().getObjectAttachments(); Assertions.assertTrue( attachments != null && attachments.size() == 1,"set attachment failed!"); cluster = new AbstractClusterInvoker(dic) { diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailSafeClusterInvokerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailSafeClusterInvokerTest.java index f255ccbdce..c666fb8e96 100644 --- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailSafeClusterInvokerTest.java +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailSafeClusterInvokerTest.java @@ -86,7 +86,7 @@ public class FailSafeClusterInvokerTest { resetInvokerToException(); FailsafeClusterInvoker invoker = new FailsafeClusterInvoker(dic); invoker.invoke(invocation); - Assertions.assertNull(RpcContext.getContext().getInvoker()); + Assertions.assertNull(RpcContext.getServiceContext().getInvoker()); } @Test() diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailbackClusterInvokerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailbackClusterInvokerTest.java index d648943f02..84747b70da 100644 --- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailbackClusterInvokerTest.java +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailbackClusterInvokerTest.java @@ -107,7 +107,7 @@ public class FailbackClusterInvokerTest { FailbackClusterInvoker invoker = new FailbackClusterInvoker( 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); diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailfastClusterInvokerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailfastClusterInvokerTest.java index 0bdd9e6fbf..459dcfc147 100644 --- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailfastClusterInvokerTest.java +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailfastClusterInvokerTest.java @@ -86,7 +86,7 @@ public class FailfastClusterInvokerTest { resetInvoker1ToException(); FailfastClusterInvoker invoker = new FailfastClusterInvoker(dic); invoker.invoke(invocation); - Assertions.assertSame(invoker1, RpcContext.getContext().getInvoker()); + Assertions.assertSame(invoker1, RpcContext.getServiceContext().getInvoker()); }); } diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvokerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvokerTest.java index 32efeb1584..3c2699169d 100644 --- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvokerTest.java +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvokerTest.java @@ -128,9 +128,9 @@ public class ForkingClusterInvokerTest { String attachKey = "attach"; String attachValue = "value"; - RpcContext.getContext().setAttachment(attachKey, attachValue); + RpcContext.getClientAttachment().setAttachment(attachKey, attachValue); - Map attachments = RpcContext.getContext().getObjectAttachments(); + Map 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 afterInvoke = RpcContext.getContext().getObjectAttachments(); + Map afterInvoke = RpcContext.getClientAttachment().getObjectAttachments(); Assertions.assertTrue(afterInvoke != null && afterInvoke.size() == 0, "clear attachment failed!"); } diff --git a/dubbo-compatible/src/test/java/org/apache/dubbo/service/DemoServiceImpl.java b/dubbo-compatible/src/test/java/org/apache/dubbo/service/DemoServiceImpl.java index fb508c612f..ea2ac17b3e 100644 --- a/dubbo-compatible/src/test/java/org/apache/dubbo/service/DemoServiceImpl.java +++ b/dubbo-compatible/src/test/java/org/apache/dubbo/service/DemoServiceImpl.java @@ -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; } diff --git a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/bootstrap/EchoServiceImpl.java b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/bootstrap/EchoServiceImpl.java index 7c9225df40..65dfd62bf3 100644 --- a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/bootstrap/EchoServiceImpl.java +++ b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/bootstrap/EchoServiceImpl.java @@ -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); } } diff --git a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/ConfigTest.java b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/ConfigTest.java index 6dac0fc3ab..e9aefe452f 100644 --- a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/ConfigTest.java +++ b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/ConfigTest.java @@ -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(); diff --git a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/SimpleRegistryService.java b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/SimpleRegistryService.java index c9b7ac75fb..7721d9bd95 100644 --- a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/SimpleRegistryService.java +++ b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/SimpleRegistryService.java @@ -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 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 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 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 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 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()); } } diff --git a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/registry/nacos/demo/service/DefaultService.java b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/registry/nacos/demo/service/DefaultService.java index 7743adf788..405e58906a 100644 --- a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/registry/nacos/demo/service/DefaultService.java +++ b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/registry/nacos/demo/service/DefaultService.java @@ -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(), diff --git a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/samples/ZookeeperDubboSpringProviderBootstrap.java b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/samples/ZookeeperDubboSpringProviderBootstrap.java index ddd40109c3..913d0c3645 100644 --- a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/samples/ZookeeperDubboSpringProviderBootstrap.java +++ b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/samples/ZookeeperDubboSpringProviderBootstrap.java @@ -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); } diff --git a/dubbo-demo/dubbo-demo-annotation/dubbo-demo-annotation-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java b/dubbo-demo/dubbo-demo-annotation/dubbo-demo-annotation-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java index 5f223ce3d5..7b4002714f 100644 --- a/dubbo-demo/dubbo-demo-annotation/dubbo-demo-annotation-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java +++ b/dubbo-demo/dubbo-demo-annotation/dubbo-demo-annotation-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java @@ -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 diff --git a/dubbo-demo/dubbo-demo-api/dubbo-demo-api-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java b/dubbo-demo/dubbo-demo-api/dubbo-demo-api-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java index 2f13a5c671..ae1faf86ab 100644 --- a/dubbo-demo/dubbo-demo-api/dubbo-demo-api-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java +++ b/dubbo-demo/dubbo-demo-api/dubbo-demo-api-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java @@ -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 diff --git a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider-backend/src/main/java/org/apache/dubbo/demo/provider/chain/ChainServiceImpl.java b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider-backend/src/main/java/org/apache/dubbo/demo/provider/chain/ChainServiceImpl.java index bd15576a7a..39662a5634 100644 --- a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider-backend/src/main/java/org/apache/dubbo/demo/provider/chain/ChainServiceImpl.java +++ b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider-backend/src/main/java/org/apache/dubbo/demo/provider/chain/ChainServiceImpl.java @@ -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(); } } diff --git a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java index 4bccce6c36..e55170709d 100644 --- a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java +++ b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/java/org/apache/dubbo/demo/provider/DemoServiceImpl.java @@ -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 diff --git a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/java/org/apache/dubbo/demo/provider/RestDemoServiceImpl.java b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/java/org/apache/dubbo/demo/provider/RestDemoServiceImpl.java index 8936e2322b..67d523aee0 100644 --- a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/java/org/apache/dubbo/demo/provider/RestDemoServiceImpl.java +++ b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/java/org/apache/dubbo/demo/provider/RestDemoServiceImpl.java @@ -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(); } } diff --git a/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java b/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java index 47e4bc7453..9fdb7d6ccd 100644 --- a/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java +++ b/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java @@ -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 } } diff --git a/dubbo-monitor/dubbo-monitor-api/src/test/java/org/apache/dubbo/monitor/support/MonitorFilterTest.java b/dubbo-monitor/dubbo-monitor-api/src/test/java/org/apache/dubbo/monitor/support/MonitorFilterTest.java index dce7e51970..02c633cf56 100644 --- a/dubbo-monitor/dubbo-monitor-api/src/test/java/org/apache/dubbo/monitor/support/MonitorFilterTest.java +++ b/dubbo-monitor/dubbo-monitor-api/src/test/java/org/apache/dubbo/monitor/support/MonitorFilterTest.java @@ -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) { diff --git a/dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/MetricsFilter.java b/dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/MetricsFilter.java index a0e58eb2e4..abc6aa4d15 100644 --- a/dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/MetricsFilter.java +++ b/dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/MetricsFilter.java @@ -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 diff --git a/dubbo-monitor/dubbo-monitor-default/src/test/java/org/apache/dubbo/monitor/dubbo/MetricsFilterTest.java b/dubbo-monitor/dubbo-monitor-default/src/test/java/org/apache/dubbo/monitor/dubbo/MetricsFilterTest.java index 058325970d..b14673480a 100644 --- a/dubbo-monitor/dubbo-monitor-default/src/test/java/org/apache/dubbo/monitor/dubbo/MetricsFilterTest.java +++ b/dubbo-monitor/dubbo-monitor-default/src/test/java/org/apache/dubbo/monitor/dubbo/MetricsFilterTest.java @@ -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 serviceInvoker; - - @BeforeEach - void setUp() { - serviceInvoker = mock(Invoker.class); + private final Function> invokerFunction = (url)-> { + Invoker 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 invoker, AppResponse response) { + given(invoker.invoke(Mockito.any(Invocation.class))).willReturn(response); } - private final Invoker timeoutInvoker = new Invoker() { - @Override - public Class 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 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 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(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 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 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(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 serviceInvoker = invokerFunction.apply(url); + Invoker 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 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 serviceInvoker = invokerFunction.apply(url); + Invoker 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 invoker = protocol.refer(MetricsService.class, url); Invocation invocation = new RpcInvocation("getMetricsByGroup", DemoService.class.getName(), "", new Class[]{String.class}, new Object[]{DUBBO_GROUP}); try { diff --git a/dubbo-plugin/dubbo-qos/src/test/java/org/apache/dubbo/qos/legacy/service/DemoServiceImpl.java b/dubbo-plugin/dubbo-qos/src/test/java/org/apache/dubbo/qos/legacy/service/DemoServiceImpl.java index 675aebadfd..1ea3e3a139 100644 --- a/dubbo-plugin/dubbo-qos/src/test/java/org/apache/dubbo/qos/legacy/service/DemoServiceImpl.java +++ b/dubbo-plugin/dubbo-qos/src/test/java/org/apache/dubbo/qos/legacy/service/DemoServiceImpl.java @@ -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 diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java index 6b5cc02e07..368abd4aa2 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java @@ -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); } diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java index 3b9e05e943..caa3bee799 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java @@ -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 extends DynamicDirectory im return; } // Set the context of the address notification thread. - RpcContext.setRpcContext(getConsumerUrl()); + RpcServiceContext.setRpcContext(getConsumerUrl()); /** * 3.x added for extend URL address diff --git a/dubbo-registry/dubbo-registry-dns/src/test/java/org/apache/dubbo/registry/dns/util/DNSResolverTest.java b/dubbo-registry/dubbo-registry-dns/src/test/java/org/apache/dubbo/registry/dns/util/DNSResolverTest.java index 8624a7f741..17088ac5f9 100644 --- a/dubbo-registry/dubbo-registry-dns/src/test/java/org/apache/dubbo/registry/dns/util/DNSResolverTest.java +++ b/dubbo-registry/dubbo-registry-dns/src/test/java/org/apache/dubbo/registry/dns/util/DNSResolverTest.java @@ -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()); } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContextImpl.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContextImpl.java index adbc7fc2f2..1331b21e0d 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContextImpl.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContextImpl.java @@ -26,11 +26,11 @@ public class AsyncContextImpl implements AsyncContext { private CompletableFuture 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(); } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java index d61d48fdec..1e6ea783c8 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java @@ -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 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 beforeContext = (appResponse, t) -> { - tmpContext = RpcContext.getContext(); + tmpContext = RpcContext.getClientAttachment(); tmpServerContext = RpcContext.getServerContext(); RpcContext.restoreContext(storedContext); RpcContext.restoreServerContext(storedServerContext); diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/PenetrateAttachmentSelector.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/PenetrateAttachmentSelector.java new file mode 100644 index 0000000000..aee89a6ad4 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/PenetrateAttachmentSelector.java @@ -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 select(); + +} diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Protocol.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Protocol.java index 9572cb138e..dc3a0201a1 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Protocol.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Protocol.java @@ -39,7 +39,7 @@ public interface Protocol { /** * Export service for remote invocation:
* 1. Protocol should record request source address after receive a request: - * RpcContext.getContext().setRemoteAddress();
+ * RpcContext.getServerAttachment().setRemoteAddress();
* 2. export() must be idempotent, that is, there's no difference between invoking once and invoking twice when * export the same URL
* 3. Invoker instance is passed in by the framework, protocol needs not to care
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java index 36ae6a821e..f88a7a966f 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java @@ -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) *

* 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. + *

+ * There are four kinds of RpcContext, which are ServerContext, ClientAttachment, ServerAttachment and ServiceContext. + *

+ * 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 LOCAL = new InternalThreadLocal() { + private static final InternalThreadLocal SERVER_LOCAL = new InternalThreadLocal() { @Override - protected RpcContext initialValue() { - return new RpcContext(); + protected RpcContextAttachment initialValue() { + return new RpcContextAttachment(); } }; - // FIXME RESPONSE_CONTEXT - private static final InternalThreadLocal SERVER_LOCAL = new InternalThreadLocal() { + private static final InternalThreadLocal CLIENT_ATTACHMENT = new InternalThreadLocal() { @Override - protected RpcContext initialValue() { - return new RpcContext(); + protected RpcContextAttachment initialValue() { + return new RpcContextAttachment(); } }; - protected final Map attachments = new HashMap<>(); - private final Map values = new HashMap(); + private static final InternalThreadLocal SERVER_ATTACHMENT = new InternalThreadLocal() { + @Override + protected RpcContextAttachment initialValue() { + return new RpcContextAttachment(); + } + }; - private List 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> 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 SERVICE_CONTEXT = new InternalThreadLocal() { + @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 getRequest(Class 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 getResponse(Class 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 CompletableFuture getCompletableFuture() { - return FutureContext.getContext().getCompletableFuture(); + return SERVICE_CONTEXT.get().getCompletableFuture(); } /** @@ -256,7 +286,7 @@ public class RpcContext { */ @SuppressWarnings("unchecked") public Future 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 getUrls() { - return urls == null && url != null ? (List) Arrays.asList(url) : urls; + return SERVICE_CONTEXT.get().getUrls(); } public void setUrls(List 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 getObjectAttachments() { - return attachments; + Map 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 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 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 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> getInvokers() { - return invokers == null && invoker != null ? (List) Arrays.asList(invoker) : invokers; + return SERVICE_CONTEXT.get().getInvokers(); } public RpcContext setInvokers(List> invokers) { - this.invokers = invokers; - if (CollectionUtils.isNotEmpty(invokers)) { - List urls = new ArrayList(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 CompletableFuture asyncCall(Callable 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) 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 exceptionFuture = new CompletableFuture<>(); - exceptionFuture.completeExceptionally(e); - return exceptionFuture; - } - return ((CompletableFuture) 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); } -} \ No newline at end of file +} diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContextAttachment.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContextAttachment.java new file mode 100644 index 0000000000..032493ef7f --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContextAttachment.java @@ -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 attachments = new HashMap<>(); + private final Map values = new HashMap(); + + 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 getAttachments() { + return new AttachmentsAdapter.ObjectToStringMap(this.getObjectAttachments()); + } + + /** + * get attachments. + * + * @return attachments + */ + @Override + @Experimental("Experiment api for supporting Object transmission") + public Map getObjectAttachments() { + return attachments; + } + + /** + * set attachments + * + * @param attachment + * @return context + */ + @Override + public RpcContextAttachment setAttachments(Map 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 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 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); + } + +} diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java new file mode 100644 index 0000000000..8ecc3683fa --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java @@ -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 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> 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 getRequest(Class 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 getResponse(Class 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 + * @return future + */ + @Override + @SuppressWarnings("unchecked") + public CompletableFuture getCompletableFuture() { + return FutureContext.getContext().getCompletableFuture(); + } + + /** + * get future. + * + * @param + * @return future + */ + @Override + @SuppressWarnings("unchecked") + public Future getFuture() { + return FutureContext.getContext().getCompletableFuture(); + } + + /** + * set future. + * + * @param future + */ + @Override + public void setFuture(CompletableFuture future) { + FutureContext.getContext().setFuture(future); + } + + @Override + public List getUrls() { + return urls == null && url != null ? (List) Arrays.asList(url) : urls; + } + + @Override + public void setUrls(List 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> getInvokers() { + return invokers == null && invoker != null ? (List) Arrays.asList(invoker) : invokers; + } + + @Override + public RpcServiceContext setInvokers(List> invokers) { + this.invokers = invokers; + if (CollectionUtils.isNotEmpty(invokers)) { + List urls = new ArrayList(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 Future.get() is not called. + * + * @param callable + * @return get the return result from future.get() + */ + @Override + @SuppressWarnings("unchecked") + public CompletableFuture asyncCall(Callable 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) 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 exceptionFuture = new CompletableFuture<>(); + exceptionFuture.completeExceptionally(e); + return exceptionFuture; + } + return ((CompletableFuture) 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); + } + +} diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java index 291601aa42..fd39c221e3 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java @@ -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(); } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ExceptionFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ExceptionFilter.java index ca98dccb2d..c12f016e50 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ExceptionFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ExceptionFilter.java @@ -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 diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericFilter.java index 0234139e2b..ed5a01f883 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/GenericFilter.java @@ -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()) { diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TimeoutFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TimeoutFilter.java index 1d48d7be52..9d1b1b65c9 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TimeoutFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TimeoutFilter.java @@ -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()) { diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TokenFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TokenFilter.java index 5e5d86fbc2..58ada5c96f 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TokenFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/TokenFilter.java @@ -48,7 +48,7 @@ public class TokenFilter implements Filter { Map 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); diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java index c157c8d1d7..5904da3fc7 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java @@ -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 implements Invoker { invocation.addObjectAttachmentsIfAbsent(attachment); } - Map contextAttachments = RpcContext.getContext().getObjectAttachments(); + Map contextAttachments = RpcContext.getClientAttachment().getObjectAttachments(); if (contextAttachments != null && contextAttachments.size() != 0) { invocation.addObjectAttachmentsIfAbsent(contextAttachments); } + ExtensionLoader selectorExtensionLoader = ExtensionLoader.getExtensionLoader(PenetrateAttachmentSelector.class); + Set supportedSelectors = selectorExtensionLoader.getSupportedExtensions(); + if (CollectionUtils.isNotEmpty(supportedSelectors)) { + for (String supportedSelector : supportedSelectors) { + Map 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 implements Invoker { } 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; diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/AbstractProxyInvoker.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/AbstractProxyInvoker.java index 1b1d592d97..25c7ee9015 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/AbstractProxyInvoker.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/AbstractProxyInvoker.java @@ -98,7 +98,7 @@ public abstract class AbstractProxyInvoker implements Invoker { }); 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 implements Invoker { } private CompletableFuture 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) value; } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java index fb7ec5177b..e3df54faad 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java @@ -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); diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/AccessLogData.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/AccessLogData.java index e9ff35896e..4dd0051022 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/AccessLogData.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/AccessLogData.java @@ -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()); diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/RpcContextTest.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/RpcContextTest.java index 380b2c3c40..66cf0b571d 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/RpcContextTest.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/RpcContextTest.java @@ -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 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 map = new HashMap(); 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 rpcFuture = RpcContext.getContext().asyncCall(() -> { + CompletableFuture 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"); diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ContextFilterTest.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ContextFilterTest.java index ed65129ad8..cb38f78468 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ContextFilterTest.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ContextFilterTest.java @@ -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 invoker = new MyInvoker(url); Invocation invocation = new MockInvocation(); Result result = contextFilter.invoke(invoker, invocation); - assertNull(RpcContext.getContext().getInvoker()); + assertNull(RpcContext.getServiceContext().getInvoker()); } } \ No newline at end of file diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ExceptionFilterTest.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ExceptionFilterTest.java index b8ebd84a1e..80086ef7de 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ExceptionFilterTest.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ExceptionFilterTest.java @@ -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(); diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/proxy/DemoServiceImpl.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/proxy/DemoServiceImpl.java index 9395f9d9e8..a8ece47e09 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/proxy/DemoServiceImpl.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/proxy/DemoServiceImpl.java @@ -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; } diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/proxy/RemoteServiceImpl.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/proxy/RemoteServiceImpl.java index 4a60dbc8e0..ccdab45f49 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/proxy/RemoteServiceImpl.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/proxy/RemoteServiceImpl.java @@ -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(); } diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/support/DemoServiceImpl.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/support/DemoServiceImpl.java index 4667cc7bf1..e2b144b31b 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/support/DemoServiceImpl.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/support/DemoServiceImpl.java @@ -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; } diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java index d0e7f20077..037f233d6c 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java @@ -163,10 +163,10 @@ public class DubboInvoker extends AbstractInvoker { } 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 } diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java index 20dd0180a1..434c5c9daa 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java @@ -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()); } diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/filter/TraceFilter.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/filter/TraceFilter.java index 25d390ee3f..16e5feccfe 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/filter/TraceFilter.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/filter/TraceFilter.java @@ -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()) diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/support/DemoServiceImpl.java b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/support/DemoServiceImpl.java index 0d56e3e75b..ee0baed40d 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/support/DemoServiceImpl.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/support/DemoServiceImpl.java @@ -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(); } } \ No newline at end of file diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/support/RemoteServiceImpl.java b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/support/RemoteServiceImpl.java index 967bc4c7e1..fe18a3e64e 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/support/RemoteServiceImpl.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/support/RemoteServiceImpl.java @@ -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(); } diff --git a/dubbo-rpc/dubbo-rpc-grpc/src/main/java/org/apache/dubbo/rpc/protocol/grpc/interceptors/RpcContextInterceptor.java b/dubbo-rpc/dubbo-rpc-grpc/src/main/java/org/apache/dubbo/rpc/protocol/grpc/interceptors/RpcContextInterceptor.java index 86915af784..703223125c 100644 --- a/dubbo-rpc/dubbo-rpc-grpc/src/main/java/org/apache/dubbo/rpc/protocol/grpc/interceptors/RpcContextInterceptor.java +++ b/dubbo-rpc/dubbo-rpc-grpc/src/main/java/org/apache/dubbo/rpc/protocol/grpc/interceptors/RpcContextInterceptor.java @@ -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 ClientCall interceptCall(MethodDescriptor method, CallOptions callOptions, Channel next) { - RpcContext rpcContext = RpcContext.getContext(); + RpcContext rpcContext = RpcContext.getClientAttachment(); Map attachments = new HashMap<>(rpcContext.getObjectAttachments()); return new ForwardingClientCall.SimpleForwardingClientCall(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); } diff --git a/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmInvoker.java b/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmInvoker.java index 907a98e663..4f4237ed45 100644 --- a/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmInvoker.java +++ b/dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/InjvmInvoker.java @@ -60,7 +60,7 @@ class InjvmInvoker extends AbstractInvoker { 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); diff --git a/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/DemoServiceImpl.java b/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/DemoServiceImpl.java index ee7c368ecb..5725513cd9 100644 --- a/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/DemoServiceImpl.java +++ b/dubbo-rpc/dubbo-rpc-injvm/src/test/java/org/apache/dubbo/rpc/protocol/injvm/DemoServiceImpl.java @@ -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; } diff --git a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/DubboHttpProtocolServer.java b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/DubboHttpProtocolServer.java index 1ab039a2ee..71db6b963a 100644 --- a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/DubboHttpProtocolServer.java +++ b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/DubboHttpProtocolServer.java @@ -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); } } diff --git a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RpcContextFilter.java b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RpcContextFilter.java index 171118145b..20556ae4e1 100644 --- a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RpcContextFilter.java +++ b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RpcContextFilter.java @@ -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 entry : RpcContext.getContext().getObjectAttachments().entrySet()) { + for (Map.Entry entry : RpcContext.getClientAttachment().getObjectAttachments().entrySet()) { String key = entry.getKey(); String value = (String) entry.getValue(); if (illegalHttpHeaderKey(key) || illegalHttpHeaderValue(value)) { diff --git a/dubbo-rpc/dubbo-rpc-rest/src/test/java/org/apache/dubbo/rpc/protocol/rest/DemoServiceImpl.java b/dubbo-rpc/dubbo-rpc-rest/src/test/java/org/apache/dubbo/rpc/protocol/rest/DemoServiceImpl.java index c9fe82f482..3cf9622740 100644 --- a/dubbo-rpc/dubbo-rpc-rest/src/test/java/org/apache/dubbo/rpc/protocol/rest/DemoServiceImpl.java +++ b/dubbo-rpc/dubbo-rpc-rest/src/test/java/org/apache/dubbo/rpc/protocol/rest/DemoServiceImpl.java @@ -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; } diff --git a/dubbo-rpc/dubbo-rpc-rest/src/test/java/org/apache/dubbo/rpc/protocol/rest/RestProtocolTest.java b/dubbo-rpc/dubbo-rpc-rest/src/test/java/org/apache/dubbo/rpc/protocol/rest/RestProtocolTest.java index 9b207a7e0b..8004d38dce 100644 --- a/dubbo-rpc/dubbo-rpc-rest/src/test/java/org/apache/dubbo/rpc/protocol/rest/RestProtocolTest.java +++ b/dubbo-rpc/dubbo-rpc-rest/src/test/java/org/apache/dubbo/rpc/protocol/rest/RestProtocolTest.java @@ -112,7 +112,7 @@ public class RestProtocolTest { this.registerProvider(exportUrl, server, DemoService.class); - RpcContext.getContext().setAttachment("timeout", "200"); + RpcContext.getClientAttachment().setAttachment("timeout", "200"); Exporter 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)); diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java index 180d889dc6..a0d0754151 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java @@ -172,10 +172,10 @@ public class TripleInvoker extends AbstractInvoker { } 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 }