diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java index 290a2a8747..625ddfb2f8 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java @@ -626,4 +626,6 @@ public interface CommonConstants { String PAYLOAD = "payload"; String DUBBO_METRICS_CONFIGCENTER_ENABLE = "dubbo.metrics.configcenter.enable"; + + Integer TRI_EXCEPTION_CODE_NOT_EXISTS = 0; } 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 86bd18ad07..f21cc4d690 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 @@ -35,6 +35,17 @@ public interface PackableMethod { byte[] pack(Object obj) throws IOException; } + interface WrapperUnPack extends UnPack { + + default Object unpack(byte[] data) throws IOException, ClassNotFoundException { + return unpack(data, false); + } + + Object unpack(byte[] data, boolean isReturnTriException) throws IOException, ClassNotFoundException; + + + } + interface UnPack { /** @@ -44,6 +55,7 @@ public interface PackableMethod { * @throws ClassNotFoundException when no class found */ Object unpack(byte[] data) throws IOException, ClassNotFoundException; + } default Object parseRequest(byte[] data) throws IOException, ClassNotFoundException { @@ -51,7 +63,15 @@ public interface PackableMethod { } default Object parseResponse(byte[] data) throws IOException, ClassNotFoundException { - return getResponseUnpack().unpack(data); + return parseResponse(data, false); + } + + default Object parseResponse(byte[] data, boolean isReturnTriException) throws IOException, ClassNotFoundException { + UnPack unPack = getResponseUnpack(); + if (unPack instanceof WrapperUnPack) { + return ((WrapperUnPack) unPack).unpack(data, isReturnTriException); + } + return unPack.unpack(data); } default byte[] packRequest(Object request) throws IOException { diff --git a/dubbo-plugin/dubbo-security/pom.xml b/dubbo-plugin/dubbo-security/pom.xml index 8ff703e4e0..64333a8ff5 100644 --- a/dubbo-plugin/dubbo-security/pom.xml +++ b/dubbo-plugin/dubbo-security/pom.xml @@ -26,7 +26,7 @@ false - 3.21.12 + 3.22.2 1.41.0 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 622aac21ce..ae647e5ca6 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 @@ -349,7 +349,7 @@ public class ReflectionPackableMethod implements PackableMethod { } } - private static class WrapResponseUnpack implements UnPack { + private static class WrapResponseUnpack implements WrapperUnPack { private final MultipleSerialization serialization; private final URL url; @@ -367,6 +367,11 @@ public class ReflectionPackableMethod implements PackableMethod { @Override public Object unpack(byte[] data) throws IOException, ClassNotFoundException { + return unpack(data, false); + } + + + public Object unpack(byte[] data, boolean isReturnTriException) throws IOException, ClassNotFoundException { TripleCustomerProtocolWapper.TripleResponseWrapper wrapper = TripleCustomerProtocolWapper.TripleResponseWrapper .parseFrom(data); final String serializeType = convertHessianFromWrapper(wrapper.getSerializeType()); @@ -374,6 +379,9 @@ public class ReflectionPackableMethod implements PackableMethod { CodecSupport.checkSerialization(serializeType, allSerialize); ByteArrayInputStream bais = new ByteArrayInputStream(wrapper.getData()); + if (isReturnTriException) { + return serialization.deserialize(url, serializeType, Exception.class, bais); + } return serialization.deserialize(url, serializeType, returnClass, bais); } } @@ -457,7 +465,7 @@ public class ReflectionPackableMethod implements PackableMethod { } } - private class WrapRequestUnpack implements UnPack { + private class WrapRequestUnpack implements WrapperUnPack { private final MultipleSerialization serialization; private final URL url; @@ -474,8 +482,7 @@ public class ReflectionPackableMethod implements PackableMethod { this.allSerialize = allSerialize; } - @Override - public Object unpack(byte[] data) throws IOException, ClassNotFoundException { + public Object unpack(byte[] data, boolean isReturnTriException) throws IOException, ClassNotFoundException { TripleCustomerProtocolWapper.TripleRequestWrapper wrapper = TripleCustomerProtocolWapper.TripleRequestWrapper.parseFrom( data); diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHeaderEnum.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHeaderEnum.java index f32b5244b5..7fd92f125c 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHeaderEnum.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHeaderEnum.java @@ -46,6 +46,8 @@ public enum TripleHeaderEnum { TRI_HEADER_CONVERT("tri-header-convert"), + TRI_EXCEPTION_CODE("tri-exception-code"), + ; static final Map enumMap = new HashMap<>(); 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 c385f46bd1..026031c8d4 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 @@ -18,6 +18,7 @@ package org.apache.dubbo.rpc.protocol.tri.call; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.utils.StringUtils; @@ -79,6 +80,26 @@ public abstract class AbstractServerCall implements ServerCall, ServerStream.Lis protected PackableMethod packableMethod; protected Map requestMetadata; + private Integer exceptionCode = CommonConstants.TRI_EXCEPTION_CODE_NOT_EXISTS; + + public Integer getExceptionCode() { + return exceptionCode; + } + + public void setExceptionCode(Integer exceptionCode) { + this.exceptionCode = exceptionCode; + } + + private boolean isNeedReturnException = false; + + public boolean isNeedReturnException() { + return isNeedReturnException; + } + + public void setNeedReturnException(boolean needReturnException) { + isNeedReturnException = needReturnException; + } + AbstractServerCall(Invoker invoker, ServerStream stream, FrameworkModel frameworkModel, @@ -187,7 +208,7 @@ public abstract class AbstractServerCall implements ServerCall, ServerStream.Lis } @Override - public final void onMessage(byte[] message) { + public final void onMessage(byte[] message, boolean isReturnTriException) { ClassLoader tccl = Thread.currentThread() .getContextClassLoader(); try { @@ -272,6 +293,9 @@ public abstract class AbstractServerCall implements ServerCall, ServerStream.Lis headers.set(TripleHeaderEnum.GRPC_ENCODING.getHeader(), compressor.getMessageEncoding()); } + if (!exceptionCode.equals(CommonConstants.TRI_EXCEPTION_CODE_NOT_EXISTS)) { + headers.set(TripleHeaderEnum.TRI_EXCEPTION_CODE.getHeader(), String.valueOf(exceptionCode)); + } // send header failed will reset stream and close request observer cause no more data will be sent stream.sendHeader(headers) .addListener(f -> { @@ -323,7 +347,7 @@ public abstract class AbstractServerCall implements ServerCall, ServerStream.Lis return; } closed = true; - stream.complete(status, attachments); + stream.complete(status, attachments, isNeedReturnException, exceptionCode); } protected Long parseTimeoutToMills(String timeoutVal) { @@ -361,7 +385,7 @@ public abstract class AbstractServerCall implements ServerCall, ServerStream.Lis return; } closed = true; - stream.complete(status, null); + stream.complete(status, null, false, CommonConstants.TRI_EXCEPTION_CODE_NOT_EXISTS); LOGGER.error(PROTOCOL_FAILED_REQUEST, "", "", "Triple request error: service=" + serviceName + " method" + methodName, status.asException()); } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ClientCall.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ClientCall.java index 4936f54409..6f5b43b6b8 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ClientCall.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ClientCall.java @@ -53,7 +53,7 @@ public interface ClientCall { * @param status response status * @param trailers response trailers */ - void onClose(TriRpcStatus status, Map trailers); + void onClose(TriRpcStatus status, Map trailers, boolean isReturnTriException); } /** diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ObserverToClientCallListenerAdapter.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ObserverToClientCallListenerAdapter.java index 1c934ce970..5d427e05ee 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ObserverToClientCallListenerAdapter.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ObserverToClientCallListenerAdapter.java @@ -46,7 +46,7 @@ public class ObserverToClientCallListenerAdapter implements ClientCall.Listener } @Override - public void onClose(TriRpcStatus status, Map trailers) { + public void onClose(TriRpcStatus status, Map trailers, boolean isReturnTriException) { if (status.isOk()) { delegate.onCompleted(); } else { diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java index f1d0a60f6a..59aeede63b 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java @@ -68,7 +68,7 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { // stream listener start @Override - public void onMessage(byte[] message) { + public void onMessage(byte[] message, boolean isReturnTriException) { if (done) { LOGGER.warn(PROTOCOL_STREAM_LISTENER, "", "", "Received message from closed stream,connection=" + connectionClient + " service=" @@ -77,13 +77,13 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { return; } try { - final Object unpacked = requestMetadata.packableMethod.parseResponse(message); + final Object unpacked = requestMetadata.packableMethod.parseResponse(message, isReturnTriException); listener.onMessage(unpacked); } catch (Throwable t) { TriRpcStatus status = TriRpcStatus.INTERNAL.withDescription("Deserialize response failed") .withCause(t); cancelByLocal(status.asException()); - listener.onClose(status,null); + listener.onClose(status,null, false); LOGGER.error(PROTOCOL_FAILED_RESPONSE, "", "", String.format("Failed to deserialize triple response, service=%s, method=%s,connection=%s", connectionClient, requestMetadata.service, requestMetadata.method.getMethodName()), t); } @@ -98,18 +98,18 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { if (requestMetadata.cancellationContext != null) { requestMetadata.cancellationContext.cancel(status.asException()); } - onComplete(status, null, null); + onComplete(status, null, null, false); } @Override public void onComplete(TriRpcStatus status, Map attachments, - Map excludeHeaders) { + Map excludeHeaders, boolean isReturnTriException) { if (done) { return; } done = true; try { - listener.onClose(status, StreamUtils.toAttachments(attachments)); + listener.onClose(status, StreamUtils.toAttachments(attachments), isReturnTriException); } catch (Throwable t) { cancelByLocal( TriRpcStatus.INTERNAL.withDescription("Close stream error").withCause(t) @@ -191,7 +191,7 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { requestMetadata.method), t); cancelByLocal(t); listener.onClose(TriRpcStatus.INTERNAL.withDescription("Serialize request failed") - .withCause(t), null); + .withCause(t), null, false); } } // stream listener end diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryClientCallListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryClientCallListener.java index 2f00c15ac1..57dbb625f8 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryClientCallListener.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryClientCallListener.java @@ -38,11 +38,11 @@ public class UnaryClientCallListener implements ClientCall.Listener { } @Override - public void onClose(TriRpcStatus status, Map trailers) { + public void onClose(TriRpcStatus status, Map trailers, boolean isReturnTriException) { AppResponse result = new AppResponse(); result.setObjectAttachments(trailers); if (status.isOk()) { - if (appResponse instanceof Exception) { + if (isReturnTriException) { result.setException((Exception) appResponse); } else { result.setValue(appResponse); 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 08f761a600..36011c5b3f 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 @@ -18,6 +18,7 @@ package org.apache.dubbo.rpc.protocol.tri.call; import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.RpcException; import org.apache.dubbo.rpc.RpcInvocation; import org.apache.dubbo.rpc.TriRpcStatus; import org.apache.dubbo.rpc.protocol.tri.observer.ServerCallToObserverAdapter; @@ -55,12 +56,24 @@ public class UnaryServerCallListener extends AbstractServerCallListener { @Override protected void doOnResponseHasException(Throwable t) { if (needWrapper) { - onReturn(t); + onReturnException((Exception) t); } else { super.doOnResponseHasException(t); } } + + private void onReturnException(Exception value) { + TriRpcStatus status = TriRpcStatus.getStatus(value); + int exceptionCode = status.code.code; + if (exceptionCode == TriRpcStatus.UNKNOWN.code.code) { + exceptionCode = RpcException.BIZ_EXCEPTION; + } + responseObserver.setExceptionCode(exceptionCode); + responseObserver.setNeedReturnException(true); + onReturn(value); + } + @Override public void onComplete() { invoke(); diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ServerCallToObserverAdapter.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ServerCallToObserverAdapter.java index 18bb578f01..f2ae04ac8c 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ServerCallToObserverAdapter.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ServerCallToObserverAdapter.java @@ -17,6 +17,7 @@ package org.apache.dubbo.rpc.protocol.tri.observer; +import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.rpc.CancellationContext; @@ -36,6 +37,26 @@ public class ServerCallToObserverAdapter extends CancelableStreamObserver private Map attachments; private boolean terminated = false; + private boolean isNeedReturnException = false; + + private Integer exceptionCode = CommonConstants.TRI_EXCEPTION_CODE_NOT_EXISTS; + + public Integer getExceptionCode() { + return exceptionCode; + } + + public void setExceptionCode(Integer exceptionCode) { + this.exceptionCode = exceptionCode; + } + + public boolean isNeedReturnException() { + return isNeedReturnException; + } + + public void setNeedReturnException(boolean needReturnException) { + isNeedReturnException = needReturnException; + } + public ServerCallToObserverAdapter(AbstractServerCall call, CancellationContext cancellationContext) { this.call = call; @@ -61,6 +82,8 @@ public class ServerCallToObserverAdapter extends CancelableStreamObserver throw new IllegalStateException( "Stream observer has been terminated, no more data is allowed"); } + call.setExceptionCode(exceptionCode); + call.setNeedReturnException(isNeedReturnException); call.sendMessage(data); } @@ -74,6 +97,8 @@ public class ServerCallToObserverAdapter extends CancelableStreamObserver if (isTerminated()) { return; } + call.setExceptionCode(exceptionCode); + call.setNeedReturnException(isNeedReturnException); call.close(status, attachments); setTerminated(); } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/ClientStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/ClientStream.java index 67f7320d72..42842c5fe3 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/ClientStream.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/ClientStream.java @@ -52,7 +52,7 @@ public interface ClientStream extends Stream { * @param reserved triple protocol reserved data */ default void onComplete(TriRpcStatus status, Map attachments, - Map reserved) { + Map reserved, boolean isReturnTriException) { onComplete(status, attachments); } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/ServerStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/ServerStream.java index f55ab58257..0b815253d0 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/ServerStream.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/ServerStream.java @@ -51,7 +51,7 @@ public interface ServerStream extends Stream { * @param attachments response attachments * @return a future that indicates the completion of send trailers */ - Future complete(TriRpcStatus status, Map attachments); + Future complete(TriRpcStatus status, Map attachments, boolean isNeedReturnException, int exceptionCode); /** * Send message to client diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/Stream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/Stream.java index 01fcbf00ff..72b7aec844 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/Stream.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/Stream.java @@ -43,7 +43,7 @@ public interface Stream { * * @param message message received from remote peer */ - void onMessage(byte[] message); + void onMessage(byte[] message, boolean isReturnTriException); /** * Callback when receive cancel signal. diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java index d9f5dd941a..ab5816a34a 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java @@ -17,6 +17,7 @@ package org.apache.dubbo.rpc.protocol.tri.stream; +import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.rpc.TriRpcStatus; @@ -83,6 +84,8 @@ public class TripleClientStream extends AbstractStream implements ClientStream { private boolean halfClosed; private boolean rst; + private boolean isReturnTriException = false; + // for test TripleClientStream(FrameworkModel frameworkModel, Executor executor, @@ -230,10 +233,10 @@ public class TripleClientStream extends AbstractStream implements ClientStream { void handleH2TransportError(TriRpcStatus status) { writeQueue.enqueue(CancelQueueCommand.createCommand(streamChannelFuture, Http2Error.NO_ERROR)); TripleClientStream.this.rst = true; - finishProcess(status, null); + finishProcess(status, null, false); } - void finishProcess(TriRpcStatus status, Http2Headers trailers) { + void finishProcess(TriRpcStatus status, Http2Headers trailers, boolean isReturnTriException) { final Map reserved = filterReservedHeaders(trailers); final Map attachments = headersToMap(trailers, () -> { return reserved.get(TripleHeaderEnum.TRI_HEADER_CONVERT.getHeader()); @@ -245,7 +248,7 @@ public class TripleClientStream extends AbstractStream implements ClientStream { } else { detailStatus = status; } - listener.onComplete(detailStatus, attachments, reserved); + listener.onComplete(detailStatus, attachments, reserved, isReturnTriException); } private TriRpcStatus validateHeaderStatus(Http2Headers headers) { @@ -286,6 +289,13 @@ public class TripleClientStream extends AbstractStream implements ClientStream { // todo support full payload compressor CharSequence messageEncoding = headers.get(TripleHeaderEnum.GRPC_ENCODING.getHeader()); + CharSequence triExceptionCode = headers.get(TripleHeaderEnum.TRI_EXCEPTION_CODE.getHeader()); + if (triExceptionCode != null) { + Integer triExceptionCodeNum = Integer.parseInt(triExceptionCode.toString()); + if (!(triExceptionCodeNum.equals(CommonConstants.TRI_EXCEPTION_CODE_NOT_EXISTS))) { + isReturnTriException = true; + } + } if (null != messageEncoding) { String compressorStr = messageEncoding.toString(); if (!Identity.IDENTITY.getMessageEncoding().equals(compressorStr)) { @@ -303,11 +313,11 @@ public class TripleClientStream extends AbstractStream implements ClientStream { TriDecoder.Listener listener = new TriDecoder.Listener() { @Override public void onRawMessage(byte[] data) { - TripleClientStream.this.listener.onMessage(data); + TripleClientStream.this.listener.onMessage(data, isReturnTriException); } public void close() { - finishProcess(statusFromTrailers(trailers), trailers); + finishProcess(statusFromTrailers(trailers), trailers, isReturnTriException); } }; deframer = new TriDecoder(decompressor, listener); @@ -324,7 +334,7 @@ public class TripleClientStream extends AbstractStream implements ClientStream { this.trailers = trailers; TriRpcStatus status = statusFromTrailers(trailers); if (deframer == null) { - finishProcess(status, trailers); + finishProcess(status, trailers, false); } if (deframer != null) { deframer.close(); @@ -464,7 +474,7 @@ public class TripleClientStream extends AbstractStream implements ClientStream { executor.execute(() -> { transportError = TriRpcStatus.CANCELLED .withDescription("Canceled by remote peer, errorCode=" + errorCode); - finishProcess(transportError, null); + finishProcess(transportError, null, false); }); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java index 5c113ba6b2..be9bfc4951 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java @@ -18,6 +18,7 @@ package org.apache.dubbo.rpc.protocol.tri.stream; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.utils.StringUtils; @@ -157,8 +158,8 @@ public class TripleServerStream extends AbstractStream implements ServerStream { @Override - public ChannelFuture complete(TriRpcStatus status, Map attachments) { - Http2Headers trailers = getTrailers(status, attachments); + public ChannelFuture complete(TriRpcStatus status, Map attachments, boolean isNeedReturnException, int exceptionCode) { + Http2Headers trailers = getTrailers(status, attachments, isNeedReturnException, CommonConstants.TRI_EXCEPTION_CODE_NOT_EXISTS); return sendTrailers(trailers); } @@ -183,7 +184,7 @@ public class TripleServerStream extends AbstractStream implements ServerStream { }); } - private Http2Headers getTrailers(TriRpcStatus rpcStatus, Map attachments) { + private Http2Headers getTrailers(TriRpcStatus rpcStatus, Map attachments, boolean isNeedReturnException, int exceptionCode) { DefaultHttp2Headers headers = new DefaultHttp2Headers(); if (!headerSent) { headers.status(HttpResponseStatus.OK.codeAsText()); @@ -471,7 +472,7 @@ public class TripleServerStream extends AbstractStream implements ServerStream { @Override public void onRawMessage(byte[] data) { - listener.onMessage(data); + listener.onMessage(data, false); } @Override diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/ExceptionUtilsTest.java b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/ExceptionUtilsTest.java index 74c0ba63a2..598da917f2 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/ExceptionUtilsTest.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/ExceptionUtilsTest.java @@ -18,6 +18,7 @@ package org.apache.dubbo.rpc.protocol.tri; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.utils.ClassUtils; import org.apache.dubbo.common.utils.NetUtils; import org.apache.dubbo.rpc.Exporter; @@ -90,7 +91,7 @@ class ExceptionUtilsTest { ApplicationModel applicationModel = ApplicationModel.defaultModel(); URL providerUrl = URL.valueOf( - "tri://127.0.0.1:" + availablePort + "/" + IGreeter2.class.getName()); + "tri://127.0.0.1:" + availablePort + "/" + IGreeter2.class.getName()).addParameter(CommonConstants.TIMEOUT_KEY, 10000);; ModuleServiceRepository serviceRepository = applicationModel.getDefaultModule() .getServiceRepository(); @@ -111,7 +112,7 @@ class ExceptionUtilsTest { Exporter export = protocol.export(invoker); URL consumerUrl = URL.valueOf( - "tri://127.0.0.1:" + availablePort + "/" + IGreeter2.class.getName()); + "tri://127.0.0.1:" + availablePort + "/" + IGreeter2.class.getName()).addParameter(CommonConstants.TIMEOUT_KEY, 10000); ConsumerModel consumerModel = new ConsumerModel(consumerUrl.getServiceKey(), null, serviceDescriptor, null, @@ -128,6 +129,10 @@ class ExceptionUtilsTest { } catch (IGreeterException e) { Assertions.assertEquals(EXPECT_RESPONSE_MSG, e.getMessage()); } + + Exception e = greeterProxy.echoException(REQUEST_MSG); + Assertions.assertEquals(EXPECT_RESPONSE_MSG, e.getMessage()); + export.unexport(); protocol.destroy(); // resource recycle. diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/PbUnpackTest.java b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/PbUnpackTest.java index c4bf61bd85..919cfae3a1 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/PbUnpackTest.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/PbUnpackTest.java @@ -28,10 +28,10 @@ class PbUnpackTest { @Test void unpack() throws IOException { HealthCheckRequest req = HealthCheckRequest.newBuilder() - .setService("service") - .build(); - PbUnpack unpack=new PbUnpack<>(HealthCheckRequest.class); - HealthCheckRequest obj= (HealthCheckRequest) unpack.unpack(req.toByteArray()); - Assertions.assertEquals("service",obj.getService()); + .setService("service") + .build(); + PbUnpack unpack = new PbUnpack<>(HealthCheckRequest.class); + HealthCheckRequest obj = (HealthCheckRequest) unpack.unpack(req.toByteArray()); + Assertions.assertEquals("service", obj.getService()); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/call/ReflectionServerCallTest.java b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/call/ReflectionServerCallTest.java index 7b85a34ee4..7b841e5ac4 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/call/ReflectionServerCallTest.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/call/ReflectionServerCallTest.java @@ -80,7 +80,7 @@ class ReflectionServerCallTest { Collections.emptyList(), ImmediateEventExecutor.INSTANCE); call2.onHeader(Collections.emptyMap()); - call2.onMessage(new byte[0]); + call2.onMessage(new byte[0], false); call2.onComplete(); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/call/StubServerCallTest.java b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/call/StubServerCallTest.java index a4a0dc8f3c..98949a9237 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/call/StubServerCallTest.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/call/StubServerCallTest.java @@ -66,7 +66,7 @@ class StubServerCallTest { service, method, ImmediateEventExecutor.INSTANCE); call.onHeader(Collections.emptyMap()); - call.onMessage(new byte[0]); + call.onMessage(new byte[0], false); call.onComplete(); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/stream/MockClientStreamListener.java b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/stream/MockClientStreamListener.java index 62a83b56a7..ac135cf0cd 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/stream/MockClientStreamListener.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/stream/MockClientStreamListener.java @@ -39,7 +39,7 @@ public class MockClientStreamListener implements ClientStream.Listener { } @Override - public void onMessage(byte[] message) { + public void onMessage(byte[] message, boolean isNeedReturnException) { this.message = message; } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter2.java b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter2.java index f02b0571b7..ee0f67dcba 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter2.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter2.java @@ -23,4 +23,6 @@ public interface IGreeter2 { * Use request to respond */ String echo(String request) throws IGreeterException; + + Exception echoException(String request); } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter2Impl.java b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter2Impl.java index 6dca2c467b..10f35d27b1 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter2Impl.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter2Impl.java @@ -21,4 +21,9 @@ public class IGreeter2Impl implements IGreeter2{ public String echo(String request) throws IGreeterException { throw new IGreeterException("I am self define exception"); } + + @Override + public Exception echoException(String request) { + return new IGreeterException("I am self define exception"); + } }