This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.2
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.2 by this push:
new 854d85953d fix #11033 (#11144)
854d85953d is described below
commit 854d85953d6421d19fa3e4bfc288d3f3a154010a
Author: ShenFeng312 <[email protected]>
AuthorDate: Fri Dec 30 16:27:16 2022 +0800
fix #11033 (#11144)
---
.../config/spring/impl/MethodCallbackImpl.java | 27 +++++++-------
.../java/org/apache/dubbo/rpc/RpcInvocation.java | 3 ++
.../org/apache/dubbo/rpc/filter/ContextFilter.java | 5 +--
.../apache/dubbo/rpc/protocol/AbstractInvoker.java | 10 +++++-
.../dubbo/rpc/protocol/AbstractProxyProtocol.java | 42 ++++++++++++++++++++--
.../dubbo/rpc/protocol/dubbo/DubboInvoker.java | 4 ---
.../dubbo/rpc/protocol/rest/RpcContextFilter.java | 11 +++++-
7 files changed, 79 insertions(+), 23 deletions(-)
diff --git
a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/impl/MethodCallbackImpl.java
b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/impl/MethodCallbackImpl.java
index a005471040..a8fe9bc23a 100644
---
a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/impl/MethodCallbackImpl.java
+++
b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/impl/MethodCallbackImpl.java
@@ -29,6 +29,9 @@ import java.util.concurrent.atomic.AtomicInteger;
public class MethodCallbackImpl implements MethodCallback {
private String onInvoke1 = "";
+
+ private final Object lock = new Object();
+
private String onReturn1 = "";
private String onThrow1 = "";
@@ -55,11 +58,11 @@ public class MethodCallbackImpl implements MethodCallback {
try {
checkInjection();
checkTranscation();
- synchronized (this.onInvoke1) {
+ synchronized (lock) {
this.onInvoke1 += "dubbo invoke success!";
}
} catch (Exception e) {
- synchronized (this.onInvoke1) {
+ synchronized (lock) {
this.onInvoke1 += e.toString();
}
throw e;
@@ -72,11 +75,11 @@ public class MethodCallbackImpl implements MethodCallback {
try {
checkInjection();
checkTranscation();
- synchronized (this.onInvoke2) {
+ synchronized (lock) {
this.onInvoke2 += "dubbo invoke success(2)!";
}
} catch (Exception e) {
- synchronized (this.onInvoke2) {
+ synchronized (lock) {
this.onInvoke2 += e.toString();
}
throw e;
@@ -89,11 +92,11 @@ public class MethodCallbackImpl implements MethodCallback {
try {
checkInjection();
checkTranscation();
- synchronized (this.onReturn1) {
+ synchronized (lock) {
this.onReturn1 += "dubbo return success!";
}
} catch (Exception e) {
- synchronized (this.onReturn1) {
+ synchronized (lock) {
this.onReturn1 += e.toString();
}
throw e;
@@ -108,11 +111,11 @@ public class MethodCallbackImpl implements MethodCallback
{
try {
checkInjection();
checkTranscation();
- synchronized (this.onReturn2) {
+ synchronized (lock) {
this.onReturn2 += "dubbo return success(2)!";
}
} catch (Exception e) {
- synchronized (this.onReturn2) {
+ synchronized (lock) {
this.onReturn2 += e.toString();
}
throw e;
@@ -127,11 +130,11 @@ public class MethodCallbackImpl implements MethodCallback
{
try {
checkInjection();
checkTranscation();
- synchronized (this.onThrow1) {
+ synchronized (lock) {
this.onThrow1 += "dubbo throw exception!";
}
} catch (Exception e) {
- synchronized (this.onThrow1) {
+ synchronized (lock) {
this.onThrow1 += e.toString();
}
throw e;
@@ -144,11 +147,11 @@ public class MethodCallbackImpl implements MethodCallback
{
try {
checkInjection();
checkTranscation();
- synchronized (this.onThrow2) {
+ synchronized (lock) {
this.onThrow2 += "dubbo throw exception(2)!";
}
} catch (Exception e) {
- synchronized (this.onThrow2) {
+ synchronized (lock) {
this.onThrow2 += e.toString();
}
throw e;
diff --git
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
index b4aa155890..4ac08dbb1e 100644
---
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
+++
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
@@ -437,6 +437,9 @@ public class RpcInvocation implements Invocation,
Serializable {
public Map<String, String> getAttachments() {
try {
attachmentLock.lock();
+ if (attachments == null) {
+ attachments = new HashMap<>();
+ }
return new AttachmentsAdapter.ObjectToStringMap(attachments);
} finally {
attachmentLock.unlock();
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 487e0238cc..88ef7d6ede 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
@@ -126,7 +126,7 @@ public class ContextFilter implements Filter,
Filter.Listener {
// merged from dubbox
// we may already add some attachments into RpcContext before this
filter (e.g. in rest protocol)
- if (attachments != null) {
+ if (CollectionUtils.isNotEmptyMap(attachments)) {
if (context.getObjectAttachments().size() > 0) {
context.getObjectAttachments().putAll(attachments);
} else {
@@ -135,7 +135,8 @@ public class ContextFilter implements Filter,
Filter.Listener {
}
if (invocation instanceof RpcInvocation) {
- ((RpcInvocation) invocation).setInvoker(invoker);
+ RpcInvocation rpcInvocation = (RpcInvocation) invocation;
+ rpcInvocation.setInvoker(invoker);
}
try {
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 c2d7b31699..e9e21902d5 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
@@ -49,6 +49,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
+import static
org.apache.dubbo.common.constants.CommonConstants.DEFAULT_VERSION;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_REQUEST;
import static org.apache.dubbo.rpc.Constants.SERIALIZATION_ID_KEY;
@@ -75,6 +76,8 @@ public abstract class AbstractInvoker<T> implements
Invoker<T> {
*/
private final Map<String, Object> attachment;
+ protected final String version;
+
/**
* {@link Node} available
*/
@@ -109,7 +112,12 @@ public abstract class AbstractInvoker<T> implements
Invoker<T> {
}
this.type = type;
this.url = url;
- this.attachment = attachment == null ? null :
Collections.unmodifiableMap(attachment);
+
+ this.attachment = attachment == null
+ ? null
+ : Collections.unmodifiableMap(attachment);
+ this.version = url.getVersion(DEFAULT_VERSION);
+
}
private static Map<String, Object> convertAttachment(URL url, String[]
keys) {
diff --git
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProxyProtocol.java
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProxyProtocol.java
index e1727c38ca..9328598081 100644
---
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProxyProtocol.java
+++
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProxyProtocol.java
@@ -32,6 +32,7 @@ import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ProtocolServer;
import org.apache.dubbo.rpc.ProxyFactory;
import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import java.net.InetSocketAddress;
@@ -44,7 +45,12 @@ import java.util.concurrent.CopyOnWriteArrayList;
import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_VALUE;
+import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_UNSUPPORTED;
+import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
/**
* AbstractProxyProtocol
@@ -87,7 +93,35 @@ public abstract class AbstractProxyProtocol extends
AbstractProtocol {
return exporter;
}
}
- final Runnable runnable = doExport(proxyFactory.getProxy(invoker,
true), invoker.getInterface(), invoker.getUrl());
+ final Runnable runnable = doExport(proxyFactory.getProxy(
+ new Invoker<T>() {
+ @Override
+ public Class<T> getInterface() {
+ return invoker.getInterface();
+ }
+
+ @Override
+ public Result invoke(Invocation invocation) throws
RpcException {
+
RpcContext.getServiceContext().getObjectAttachments().forEach(invocation::setObjectAttachment);
+ return invoker.invoke(invocation);
+ }
+
+ @Override
+ public URL getUrl() {
+ return invoker.getUrl();
+ }
+
+ @Override
+ public boolean isAvailable() {
+ return invoker.isAvailable();
+ }
+
+ @Override
+ public void destroy() {
+ invoker.destroy();
+ }
+ }, true), invoker.getInterface(),
+ invoker.getUrl());
exporter = new AbstractExporter<T>(invoker) {
@Override
public void afterUnExport() {
@@ -108,10 +142,12 @@ public abstract class AbstractProxyProtocol extends
AbstractProtocol {
@Override
protected <T> Invoker<T> protocolBindingRefer(final Class<T> type, final
URL url) throws RpcException {
final Invoker<T> target = proxyFactory.getInvoker(doRefer(type, url),
type, url);
- Invoker<T> invoker = new AbstractInvoker<T>(type, url) {
+ Invoker<T> invoker = new AbstractInvoker<T>(type, url, new
String[]{INTERFACE_KEY, GROUP_KEY, TOKEN_KEY}) {
@Override
- protected Result doInvoke(Invocation invocation) throws Throwable {
+ protected Result doInvoke(Invocation invocation) {
try {
+ invocation.setAttachment(PATH_KEY, getUrl().getPath());
+ invocation.setAttachment(VERSION_KEY, version);
Result result = target.invoke(invocation);
// FIXME result is an AsyncRpcResult instance.
Throwable e = result.getException();
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 9be288b14b..d32042d97d 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
@@ -43,7 +43,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import static
org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT;
-import static
org.apache.dubbo.common.constants.CommonConstants.DEFAULT_VERSION;
import static
org.apache.dubbo.common.constants.CommonConstants.ENABLE_TIMEOUT_COUNTDOWN_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
@@ -64,7 +63,6 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
private final AtomicPositiveInteger index = new AtomicPositiveInteger();
- private final String version;
private final ReentrantLock destroyLock = new ReentrantLock();
@@ -79,8 +77,6 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
public DubboInvoker(Class<T> serviceType, URL url, ExchangeClient[]
clients, Set<Invoker<?>> invokers) {
super(serviceType, url, new String[]{INTERFACE_KEY, GROUP_KEY,
TOKEN_KEY});
this.clients = clients;
- // get version.
- this.version = url.getVersion(DEFAULT_VERSION);
this.invokers = invokers;
this.serverShutdownTimeout =
ConfigurationUtils.getServerShutdownTimeout(getUrl().getScopeModel());
}
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 20556ae4e1..70edad5bcc 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
@@ -16,7 +16,9 @@
*/
package org.apache.dubbo.rpc.protocol.rest;
+import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.RpcContext;
import org.jboss.resteasy.spi.ResteasyProviderFactory;
@@ -30,6 +32,7 @@ import javax.ws.rs.container.ContainerRequestContext;
import javax.ws.rs.container.ContainerRequestFilter;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
import java.util.Map;
@Priority(Integer.MIN_VALUE + 1)
@@ -70,7 +73,13 @@ public class RpcContextFilter implements
ContainerRequestFilter, ClientRequestFi
@Override
public void filter(ClientRequestContext requestContext) throws IOException
{
int size = 0;
- for (Map.Entry<String, Object> entry :
RpcContext.getClientAttachment().getObjectAttachments().entrySet()) {
+ Map<String, Object> objectAttachments = new
HashMap<>(RpcContext.getClientAttachment().getObjectAttachments());
+ Invocation invocation = RpcContext.getServiceContext().getInvocation();
+ //should merge attachments from invocation and
RpcContext.getClientAttachment()
+ if(invocation != null &&
CollectionUtils.isNotEmptyMap(invocation.getObjectAttachments()) ){
+ objectAttachments.putAll(invocation.getObjectAttachments());
+ }
+ for (Map.Entry<String, Object> entry : objectAttachments.entrySet()) {
String key = entry.getKey();
String value = (String) entry.getValue();
if (illegalHttpHeaderKey(key) || illegalHttpHeaderValue(value)) {
