This is an automated email from the ASF dual-hosted git repository.
earthchen 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 96dfd20c10 Reactor tri diy exception api (#11829)
96dfd20c10 is described below
commit 96dfd20c108550cf563aec317911195ae8c31cb4
Author: earthchen <[email protected]>
AuthorDate: Wed Mar 15 13:14:22 2023 +0800
Reactor tri diy exception api (#11829)
* refactor
* fix
---
.../org/apache/dubbo/rpc/model/PackableMethod.java | 5 +++++
.../rpc/protocol/tri/ReflectionPackableMethod.java | 12 ++++++++++--
.../rpc/protocol/tri/call/AbstractServerCall.java | 20 ++++++++++----------
.../tri/call/AbstractServerCallListener.java | 7 ++++++-
.../protocol/tri/call/UnaryServerCallListener.java | 13 ++++++++++++-
5 files changed, 43 insertions(+), 14 deletions(-)
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/PackableMethod.java
b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/PackableMethod.java
index 779e969d4f..86bd18ad07 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/PackableMethod.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/PackableMethod.java
@@ -62,6 +62,11 @@ public interface PackableMethod {
return getResponsePack().pack(response);
}
+
+ default boolean needWrapper() {
+ return false;
+ }
+
Pack getRequestPack();
Pack getResponsePack();
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
index 989b7017c4..5316fc5452 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
@@ -53,6 +53,13 @@ public class ReflectionPackableMethod implements
PackableMethod {
private final UnPack requestUnpack;
private final UnPack responseUnpack;
+ private final boolean needWrapper;
+
+ @Override
+ public boolean needWrapper() {
+ return this.needWrapper;
+ }
+
public ReflectionPackableMethod(MethodDescriptor method, URL url, String
serializeName) {
Class<?>[] actualRequestTypes;
Class<?> actualResponseType;
@@ -79,7 +86,8 @@ public class ReflectionPackableMethod implements
PackableMethod {
}
boolean singleArgument = method.getRpcType() !=
MethodDescriptor.RpcType.UNARY;
- if (!needWrap(method, actualRequestTypes, actualResponseType)) {
+ this.needWrapper = needWrap(method, actualRequestTypes,
actualResponseType);
+ if (!needWrapper) {
requestPack = new PbArrayPacker(singleArgument);
responsePack = PB_PACK;
requestUnpack = new PbUnpack<>(actualRequestTypes[0]);
@@ -336,7 +344,6 @@ public class ReflectionPackableMethod implements
PackableMethod {
private final URL url;
private final Class<?> returnClass;
-
private WrapResponseUnpack(MultipleSerialization serialization, URL
url, Class<?> returnClass) {
this.serialization = serialization;
this.url = url;
@@ -362,6 +369,7 @@ public class ReflectionPackableMethod implements
PackableMethod {
private final URL url;
private final boolean singleArgument;
+
private WrapRequestPack(MultipleSerialization multipleSerialization,
URL url,
String serialize,
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/AbstractServerCall.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/AbstractServerCall.java
index 7780579a69..01ea3a1e79 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/AbstractServerCall.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/AbstractServerCall.java
@@ -80,13 +80,13 @@ public abstract class AbstractServerCall implements
ServerCall, ServerStream.Lis
protected Map<String, Object> requestMetadata;
AbstractServerCall(Invoker<?> invoker,
- ServerStream stream,
- FrameworkModel frameworkModel,
- ServiceDescriptor serviceDescriptor,
- String acceptEncoding,
- String serviceName,
- String methodName,
- Executor executor
+ ServerStream stream,
+ FrameworkModel frameworkModel,
+ ServiceDescriptor serviceDescriptor,
+ String acceptEncoding,
+ String serviceName,
+ String methodName,
+ Executor executor
) {
Objects.requireNonNull(serviceDescriptor,
"No service descriptor found for " + invoker.getUrl());
@@ -146,7 +146,7 @@ public abstract class AbstractServerCall implements
ServerCall, ServerStream.Lis
} catch (Exception e) {
close(TriRpcStatus.INTERNAL.withDescription("Serialize response
failed")
.withCause(e), null);
-
LOGGER.error(PROTOCOL_FAILED_SERIALIZE_TRIPLE,"","",String.format("Serialize
triple response failed, service=%s method=%s",
+ LOGGER.error(PROTOCOL_FAILED_SERIALIZE_TRIPLE, "", "",
String.format("Serialize triple response failed, service=%s method=%s",
serviceName, methodName), e);
return;
}
@@ -192,7 +192,7 @@ public abstract class AbstractServerCall implements
ServerCall, ServerStream.Lis
final TriRpcStatus status =
TriRpcStatus.UNKNOWN.withDescription("Server error")
.withCause(e);
close(status, null);
- LOGGER.error(PROTOCOL_FAILED_REQUEST,"","","Process request
failed. service=" + serviceName +
+ LOGGER.error(PROTOCOL_FAILED_REQUEST, "", "", "Process request
failed. service=" + serviceName +
" method=" + methodName, e);
} finally {
ClassLoadUtil.switchContextLoader(tccl);
@@ -373,7 +373,7 @@ public abstract class AbstractServerCall implements
ServerCall, ServerStream.Lis
ServerCall.Listener listener;
switch (methodDescriptor.getRpcType()) {
case UNARY:
- listener = new UnaryServerCallListener(invocation,
invoker, responseObserver);
+ listener = new UnaryServerCallListener(invocation,
invoker, responseObserver, packableMethod.needWrapper());
request(2);
break;
case SERVER_STREAM:
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/AbstractServerCallListener.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/AbstractServerCallListener.java
index 3bfb57f891..f645a80468 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/AbstractServerCallListener.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/AbstractServerCallListener.java
@@ -70,7 +70,7 @@ public abstract class AbstractServerCallListener implements
AbstractServerCall.L
return;
}
if (response.hasException()) {
- onReturn(response.getException());
+ doOnResponseHasException(response.getException());
return;
}
final long cost = System.currentTimeMillis() - stInMillis;
@@ -93,5 +93,10 @@ public abstract class AbstractServerCallListener implements
AbstractServerCall.L
}
}
+
+ protected void doOnResponseHasException(Throwable t) {
+ responseObserver.onError(t);
+ }
+
public abstract void onReturn(Object value);
}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryServerCallListener.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryServerCallListener.java
index 92d5398120..08f761a600 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryServerCallListener.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryServerCallListener.java
@@ -24,9 +24,12 @@ import
org.apache.dubbo.rpc.protocol.tri.observer.ServerCallToObserverAdapter;
public class UnaryServerCallListener extends AbstractServerCallListener {
+ private final boolean needWrapper;
+
public UnaryServerCallListener(RpcInvocation invocation, Invoker<?>
invoker,
- ServerCallToObserverAdapter<Object> responseObserver) {
+ ServerCallToObserverAdapter<Object>
responseObserver, boolean needWrapper) {
super(invocation, invoker, responseObserver);
+ this.needWrapper = needWrapper;
}
@Override
@@ -49,6 +52,14 @@ public class UnaryServerCallListener extends
AbstractServerCallListener {
// ignored
}
+ @Override
+ protected void doOnResponseHasException(Throwable t) {
+ if (needWrapper) {
+ onReturn(t);
+ } else {
+ super.doOnResponseHasException(t);
+ }
+ }
@Override
public void onComplete() {
