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

Reactor tri diy exception api (#11829) 

* refactor

* fix
This commit is contained in:
earthchen 2023-03-15 13:14:22 +08:00 committed by GitHub
parent eeab34eb67
commit 96dfd20c10
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 43 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
dubbo-common/src/main/java/org/apache/dubbo/rpc/model/PackableMethod.java
View File

@ -62,6 +62,11 @@ public interface PackableMethod {
return getResponsePack().pack(response);
}
default boolean needWrapper() {
return false;
}
Pack getRequestPack();
Pack getResponsePack();

12
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
View File

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

20
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/AbstractServerCall.java
View File

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

7
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/AbstractServerCallListener.java
View File

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

13
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryServerCallListener.java
View File

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