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

fix #11033 (#11144)

This commit is contained in:
ShenFeng312 2022-12-30 16:27:16 +08:00 committed by GitHub
parent 3375f727e1
commit 854d85953d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 79 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/impl/MethodCallbackImpl.java
View File

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

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

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

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

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

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

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

42
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProxyProtocol.java
View File

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

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

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

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

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