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 2cb12ff9a6 Fix AbstractProxyProtocol compatibility (#12507)
2cb12ff9a6 is described below
commit 2cb12ff9a6444b69f6b7f1b3cac7532be93e6083
Author: Albumen Kevin <[email protected]>
AuthorDate: Mon Jun 12 20:51:13 2023 +0800
Fix AbstractProxyProtocol compatibility (#12507)
---
.../dubbo/rpc/protocol/AbstractProtocol.java | 3 +-
.../dubbo/rpc/protocol/AbstractProxyProtocol.java | 87 ++++++++++++++--------
.../dubbo/rpc/protocol/rest/RestProtocol.java | 15 ++--
3 files changed, 65 insertions(+), 40 deletions(-)
diff --git
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java
index fd49e60298..25a3068693 100644
---
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java
+++
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java
@@ -42,7 +42,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_VALUE;
@@ -63,7 +62,7 @@ public abstract class AbstractProtocol implements Protocol,
ScopeModelAware {
/**
* <host:port, ProtocolServer>
*/
- protected final ConcurrentMap<String, ProtocolServer> serverMap = new
ConcurrentHashMap<>();
+ protected final Map<String, ProtocolServer> serverMap = new
ConcurrentHashMap<>();
// TODO SoftReference
protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<>();
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 db55424ed6..e1727c38ca 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
@@ -19,9 +19,11 @@ package org.apache.dubbo.rpc.protocol;
import org.apache.dubbo.common.Parameters;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.Channel;
import org.apache.dubbo.remoting.ChannelHandler;
+import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.RemotingServer;
import org.apache.dubbo.rpc.Exporter;
@@ -30,7 +32,6 @@ 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;
@@ -41,6 +42,8 @@ import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
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.LoggerCodeConstants.PROTOCOL_UNSUPPORTED;
/**
@@ -48,7 +51,7 @@ import static
org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_UNS
*/
public abstract class AbstractProxyProtocol extends AbstractProtocol {
- protected final List<Class<?>> rpcExceptions = new
CopyOnWriteArrayList<Class<?>>();
+ private final List<Class<?>> rpcExceptions = new
CopyOnWriteArrayList<Class<?>>();
protected ProxyFactory proxyFactory;
@@ -84,35 +87,7 @@ public abstract class AbstractProxyProtocol extends
AbstractProtocol {
return exporter;
}
}
- 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());
+ final Runnable runnable = doExport(proxyFactory.getProxy(invoker,
true), invoker.getInterface(), invoker.getUrl());
exporter = new AbstractExporter<T>(invoker) {
@Override
public void afterUnExport() {
@@ -130,6 +105,46 @@ public abstract class AbstractProxyProtocol extends
AbstractProtocol {
return exporter;
}
+ @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) {
+ @Override
+ protected Result doInvoke(Invocation invocation) throws Throwable {
+ try {
+ Result result = target.invoke(invocation);
+ // FIXME result is an AsyncRpcResult instance.
+ Throwable e = result.getException();
+ if (e != null) {
+ for (Class<?> rpcException : rpcExceptions) {
+ if (rpcException.isAssignableFrom(e.getClass())) {
+ throw getRpcException(type, url, invocation,
e);
+ }
+ }
+ }
+ return result;
+ } catch (RpcException e) {
+ if (e.getCode() == RpcException.UNKNOWN_EXCEPTION) {
+ e.setCode(getErrorCode(e.getCause()));
+ }
+ throw e;
+ } catch (Throwable e) {
+ throw getRpcException(type, url, invocation, e);
+ }
+ }
+
+ @Override
+ public void destroy() {
+ super.destroy();
+ target.destroy();
+ invokers.remove(this);
+ AbstractProxyProtocol.this.destroyInternal(url);
+ }
+ };
+ invokers.add(invoker);
+ return invoker;
+ }
+
// used to destroy unused clients and other resource
protected void destroyInternal(URL url) {
// subclass override
@@ -142,12 +157,22 @@ public abstract class AbstractProxyProtocol extends
AbstractProtocol {
return re;
}
+ protected String getAddr(URL url) {
+ String bindIp = url.getParameter(Constants.BIND_IP_KEY, url.getHost());
+ if (url.getParameter(ANYHOST_KEY, false)) {
+ bindIp = ANYHOST_VALUE;
+ }
+ return NetUtils.getIpByHost(bindIp) + ":" +
url.getParameter(Constants.BIND_PORT_KEY, url.getPort());
+ }
+
protected int getErrorCode(Throwable e) {
return RpcException.UNKNOWN_EXCEPTION;
}
protected abstract <T> Runnable doExport(T impl, Class<T> type, URL url)
throws RpcException;
+ protected abstract <T> T doRefer(Class<T> type, URL url) throws
RpcException;
+
protected class ProxyProtocolServer implements ProtocolServer {
private RemotingServer server;
diff --git
a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
index ad273efda3..f42a2e7266 100644
---
a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
+++
b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
@@ -31,7 +31,6 @@ import org.apache.dubbo.rpc.protocol.AbstractProtocol;
import
org.apache.dubbo.rpc.protocol.rest.annotation.consumer.HttpConnectionPreBuildIntercept;
import org.apache.dubbo.rpc.protocol.rest.annotation.metadata.MetadataResolver;
-
import java.util.Map;
import java.util.Objects;
import java.util.Set;
@@ -91,12 +90,14 @@ public class RestProtocol extends AbstractProtocol {
// TODO add Extension filter
// create rest server
- RestProtocolServer server = (RestProtocolServer)
ConcurrentHashMapUtils.computeIfAbsent(serverMap, getAddr(url), restServer -> {
- RestProtocolServer s =
serverFactory.createServer(url.getParameter(SERVER_KEY, DEFAULT_SERVER));
- s.setAddress(url.getAddress());
- s.start(url);
- return s;
- });
+ RestProtocolServer server = (RestProtocolServer)
ConcurrentHashMapUtils.computeIfAbsent(
+ (ConcurrentMap<? super String, ? super RestProtocolServer>)
serverMap,
+ getAddr(url), restServer -> {
+ RestProtocolServer s =
serverFactory.createServer(url.getParameter(SERVER_KEY, DEFAULT_SERVER));
+ s.setAddress(url.getAddress());
+ s.start(url);
+ return s;
+ });
server.deploy(serviceRestMetadata, invoker);
