From 96dfd20c108550cf563aec317911195ae8c31cb4 Mon Sep 17 00:00:00 2001 From: earthchen Date: Wed, 15 Mar 2023 13:14:22 +0800 Subject: [PATCH] Reactor tri diy exception api (#11829) * refactor * fix --- .../dubbo/rpc/model/PackableMethod.java | 5 +++++ .../tri/ReflectionPackableMethod.java | 12 +++++++++-- .../protocol/tri/call/AbstractServerCall.java | 20 +++++++++---------- .../tri/call/AbstractServerCallListener.java | 7 ++++++- .../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 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 responseObserver) { + ServerCallToObserverAdapter 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() {