From 059b75c53389bc0d4a0af2b611fa861baee31921 Mon Sep 17 00:00:00 2001 From: icodening Date: Fri, 1 Sep 2023 09:43:04 +0800 Subject: [PATCH] [feat.] new triple protocol implement (#12873) --- .artifacts | 1 + dubbo-distribution/dubbo-all/pom.xml | 12 + dubbo-distribution/dubbo-bom/pom.xml | 5 + .../reactive/calls/ReactorServerCalls.java | 3 +- .../AbstractServerHttpChannelObserver.java | 4 + .../h1/Http1ServerTransportListener.java | 3 +- .../http12/h2/Http2TransportListener.java | 4 +- .../http12/message/HttpMessageCodec.java | 4 +- .../message/HttpMessageCodecFactory.java | 39 ++ .../http12/message/JsonCodecFactory.java | 35 ++ .../message/LengthFieldStreamingDecoder.java | 45 +- .../http12/message/StreamingDecoder.java | 49 +- .../h1/NettyHttp1ConnectionHandler.java | 16 +- ...ing.http12.message.HttpMessageCodecFactory | 1 + dubbo-rpc/dubbo-rpc-triple/pom.xml | 5 + .../tri/ReflectionPackableMethod.java | 7 +- .../rpc/protocol/tri/TripleHttp2Protocol.java | 113 +++-- .../rpc/protocol/tri/compressor/Bzip2.java | 14 +- .../protocol/tri/compressor/Compressor.java | 4 + .../rpc/protocol/tri/compressor/Gzip.java | 11 + .../rpc/protocol/tri/compressor/Identity.java | 7 + .../rpc/protocol/tri/compressor/Snappy.java | 7 +- .../tri/h12}/AbstractServerCallListener.java | 27 +- .../h12/AbstractServerTransportListener.java | 479 ++++++++++++++++++ .../protocol/tri/h12/AttachmentHolder.java | 26 + .../tri/h12}/BiStreamServerCallListener.java | 3 +- .../protocol/tri/h12/CompressibleCodec.java | 71 +++ .../tri/h12/DefaultHttpMessageListener.java | 43 ++ .../protocol/tri/h12/HttpMessageListener.java | 24 + .../protocol/tri/h12}/ServerCallListener.java | 2 +- .../tri/h12/ServerCallToObserverAdapter.java | 25 + .../h12}/ServerStreamServerCallListener.java | 2 +- .../tri/h12/TripleProtocolDetector.java | 87 ++++ .../tri/h12}/UnaryServerCallListener.java | 28 +- .../tri/h12/grpc/GrpcCompositeCodec.java | 124 +++++ .../h12/grpc/GrpcCompositeCodecFactory.java | 50 ++ .../tri/h12/grpc/GrpcHeaderNames.java | 36 ++ .../GrpcHttp2ServerTransportListener.java | 216 ++++++++ ...pcHttp2ServerTransportListenerFactory.java | 38 ++ .../tri/h12/grpc/GrpcStreamingDecoder.java | 61 +++ .../rpc/protocol/tri/h12/grpc/GrpcUtils.java | 54 ++ .../h12/grpc/ProtobufHttpMessageCodec.java | 57 +++ .../grpc/StreamingHttpMessageListener.java | 36 +- .../tri/h12/grpc/WrapperHttpMessageCodec.java | 143 ++++++ .../DefaultHttp11ServerTransportListener.java | 122 +++++ ...tHttp11ServerTransportListenerFactory.java | 33 ++ .../GenericHttp2ServerTransportListener.java | 223 ++++++++ ...icHttp2ServerTransportListenerFactory.java | 40 ++ .../Http2ServerCallToObserverAdapter.java | 40 ++ .../h12/http2/Http2ServerStreamObserver.java | 77 +++ .../rpc/protocol/tri/stream/StreamUtils.java | 53 +- .../TripleIsolationExecutorSupport.java | 18 +- .../TripleServerConnectionHandler.java | 13 +- ...p12.h2.Http2ServerTransportListenerFactory | 1 + ...o.remoting.http12.message.HttpMessageCodec | 2 + ...ing.http12.message.HttpMessageCodecFactory | 1 + dubbo-test/dubbo-dependencies-all/pom.xml | 5 + 57 files changed, 2511 insertions(+), 138 deletions(-) create mode 100644 dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodecFactory.java create mode 100644 dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/JsonCodecFactory.java create mode 100644 dubbo-remoting/dubbo-remoting-http12/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory rename {dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12 => dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12}/AbstractServerCallListener.java (70%) create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerTransportListener.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AttachmentHolder.java rename {dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12 => dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12}/BiStreamServerCallListener.java (95%) create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/CompressibleCodec.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/DefaultHttpMessageListener.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/HttpMessageListener.java rename {dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12 => dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12}/ServerCallListener.java (96%) create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerCallToObserverAdapter.java rename {dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12 => dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12}/ServerStreamServerCallListener.java (97%) create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/TripleProtocolDetector.java rename {dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12 => dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12}/UnaryServerCallListener.java (64%) create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodecFactory.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHeaderNames.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListener.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListenerFactory.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcStreamingDecoder.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcUtils.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/ProtobufHttpMessageCodec.java rename dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/DefaultHttp2StreamingDecoder.java => dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/StreamingHttpMessageListener.java (50%) create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/WrapperHttpMessageCodec.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListener.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListenerFactory.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListener.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListenerFactory.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2ServerCallToObserverAdapter.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2ServerStreamObserver.java create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodec create mode 100644 dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory diff --git a/.artifacts b/.artifacts index be912a56f7..7b113ce8ab 100644 --- a/.artifacts +++ b/.artifacts @@ -85,6 +85,7 @@ dubbo-registry-zookeeper dubbo-remoting dubbo-remoting-api dubbo-remoting-http +dubbo-remoting-http12 dubbo-remoting-netty dubbo-remoting-netty4 dubbo-remoting-zookeeper-api diff --git a/dubbo-distribution/dubbo-all/pom.xml b/dubbo-distribution/dubbo-all/pom.xml index d0ff531c7e..be8bdbf11b 100644 --- a/dubbo-distribution/dubbo-all/pom.xml +++ b/dubbo-distribution/dubbo-all/pom.xml @@ -1261,12 +1261,24 @@ META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodec + + + META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory + + META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory + + + META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListenerFactory + + diff --git a/dubbo-distribution/dubbo-bom/pom.xml b/dubbo-distribution/dubbo-bom/pom.xml index 9757c56352..0978c711e1 100644 --- a/dubbo-distribution/dubbo-bom/pom.xml +++ b/dubbo-distribution/dubbo-bom/pom.xml @@ -501,6 +501,11 @@ dubbo-remoting-http ${project.version} + + org.apache.dubbo + dubbo-remoting-http12 + ${project.version} + org.apache.dubbo dubbo-remoting-netty diff --git a/dubbo-plugin/dubbo-reactive/src/main/java/org/apache/dubbo/reactive/calls/ReactorServerCalls.java b/dubbo-plugin/dubbo-reactive/src/main/java/org/apache/dubbo/reactive/calls/ReactorServerCalls.java index 33c54d1496..e9ce9bd41d 100644 --- a/dubbo-plugin/dubbo-reactive/src/main/java/org/apache/dubbo/reactive/calls/ReactorServerCalls.java +++ b/dubbo-plugin/dubbo-reactive/src/main/java/org/apache/dubbo/reactive/calls/ReactorServerCalls.java @@ -21,7 +21,6 @@ import org.apache.dubbo.common.stream.StreamObserver; import org.apache.dubbo.reactive.ServerTripleReactorPublisher; import org.apache.dubbo.reactive.ServerTripleReactorSubscriber; import org.apache.dubbo.rpc.protocol.tri.observer.CallStreamObserver; -import org.apache.dubbo.rpc.protocol.tri.observer.ServerCallToObserverAdapter; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -72,7 +71,7 @@ public final class ReactorServerCalls { try { Flux response = func.apply(Mono.just(request)); ServerTripleReactorSubscriber subscriber = response.subscribeWith(new ServerTripleReactorSubscriber<>()); - subscriber.subscribe((ServerCallToObserverAdapter) responseObserver); + subscriber.subscribe((CallStreamObserver) responseObserver); } catch (Throwable throwable) { responseObserver.onError(throwable); } diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java index 5af04e629f..3e0b2a9b75 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java +++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java @@ -41,6 +41,10 @@ public abstract class AbstractServerHttpChannelObserver implements CustomizableH this.httpMessageCodec = httpMessageCodec; } + protected HttpMessageCodec getHttpMessageCodec() { + return httpMessageCodec; + } + @Override public void setHeadersCustomizer(HeadersCustomizer headersCustomizer) { this.headersCustomizer = headersCustomizer; diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1ServerTransportListener.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1ServerTransportListener.java index eb816afb02..9e41598f42 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1ServerTransportListener.java +++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1ServerTransportListener.java @@ -16,11 +16,10 @@ */ package org.apache.dubbo.remoting.http12.h1; -import org.apache.dubbo.remoting.http12.HttpChannelHolder; import org.apache.dubbo.remoting.http12.HttpInputMessage; import org.apache.dubbo.remoting.http12.HttpTransportListener; import org.apache.dubbo.remoting.http12.RequestMetadata; -public interface Http1ServerTransportListener extends HttpTransportListener, HttpChannelHolder { +public interface Http1ServerTransportListener extends HttpTransportListener { } diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2TransportListener.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2TransportListener.java index 00bfbc6554..19faa32e4d 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2TransportListener.java +++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2TransportListener.java @@ -16,8 +16,6 @@ */ package org.apache.dubbo.remoting.http12.h2; -import org.apache.dubbo.remoting.http12.HttpChannelHolder; - -public interface Http2TransportListener extends CancelableTransportListener, HttpChannelHolder { +public interface Http2TransportListener extends CancelableTransportListener { } diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodec.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodec.java index 5437d9af47..65c3ad65e9 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodec.java +++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodec.java @@ -34,14 +34,14 @@ public interface HttpMessageCodec { default void encode(OutputStream outputStream, Object[] data) throws EncodeException { //default encode first data - this.encode(outputStream, data[0]); + this.encode(outputStream, data == null || data.length == 0 ? null : data[0]); } Object decode(InputStream inputStream, Class targetType) throws DecodeException; default Object[] decode(InputStream inputStream, Class[] targetTypes) throws DecodeException { //default decode first target type - return new Object[]{this.decode(inputStream, targetTypes[0])}; + return new Object[]{this.decode(inputStream, targetTypes == null || targetTypes.length == 0 ? null : targetTypes[0])}; } MediaType contentType(); diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodecFactory.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodecFactory.java new file mode 100644 index 0000000000..80978b6a98 --- /dev/null +++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodecFactory.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.remoting.http12.message; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.extension.ExtensionScope; +import org.apache.dubbo.common.extension.SPI; +import org.apache.dubbo.rpc.model.FrameworkModel; + +/** + * for http body codec + */ +@SPI(scope = ExtensionScope.FRAMEWORK) +public interface HttpMessageCodecFactory { + + HttpMessageCodec createCodec(URL url, FrameworkModel frameworkModel); + + MediaType contentType(); + + default boolean support(String contentType) { + MediaType mediaType = this.contentType(); + return mediaType.getName().startsWith(contentType); + } + +} diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/JsonCodecFactory.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/JsonCodecFactory.java new file mode 100644 index 0000000000..8f7d2f356c --- /dev/null +++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/JsonCodecFactory.java @@ -0,0 +1,35 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.remoting.http12.message; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.extension.Activate; +import org.apache.dubbo.rpc.model.FrameworkModel; + +@Activate +public class JsonCodecFactory implements HttpMessageCodecFactory { + + @Override + public HttpMessageCodec createCodec(URL url, FrameworkModel frameworkModel) { + return new JsonCodec(); + } + + @Override + public MediaType contentType() { + return MediaType.APPLICATION_JSON_VALUE; + } +} diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java index dc2d57d948..67da0dea90 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java +++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java @@ -20,6 +20,7 @@ import org.apache.dubbo.remoting.http12.CompositeInputStream; import org.apache.dubbo.remoting.http12.exception.DecodeException; import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; @@ -37,11 +38,7 @@ public class LengthFieldStreamingDecoder implements StreamingDecoder { private final CompositeInputStream accumulate = new CompositeInputStream(); - private final Class[] targetTypes; - - private HttpMessageCodec httpMessageCodec; - - private Listener listener; + private FragmentListener listener; private final int lengthFieldOffset; @@ -49,25 +46,22 @@ public class LengthFieldStreamingDecoder implements StreamingDecoder { private int requiredLength; - public LengthFieldStreamingDecoder(Class[] targetTypes) { - this(4, targetTypes); + private InputStream dataHeader = new ByteArrayInputStream(new byte[0]); + + public LengthFieldStreamingDecoder() { + this(4); } - public LengthFieldStreamingDecoder(int lengthFieldLength, Class[] targetTypes) { - this(0, lengthFieldLength, targetTypes); + public LengthFieldStreamingDecoder(int lengthFieldLength) { + this(0, lengthFieldLength); } - public LengthFieldStreamingDecoder(int lengthFieldOffset, int lengthFieldLength, Class[] targetTypes) { + public LengthFieldStreamingDecoder(int lengthFieldOffset, int lengthFieldLength) { this.lengthFieldOffset = lengthFieldOffset; this.lengthFieldLength = lengthFieldLength; - this.targetTypes = targetTypes; this.requiredLength = lengthFieldOffset + lengthFieldLength; } - public void setHttpMessageCodec(HttpMessageCodec httpMessageCodec) { - this.httpMessageCodec = httpMessageCodec; - } - @Override public final void decode(InputStream inputStream) throws DecodeException { if (closing || closed) { @@ -91,7 +85,7 @@ public class LengthFieldStreamingDecoder implements StreamingDecoder { } @Override - public final void setListener(Listener listener) { + public final void setFragmentListener(FragmentListener listener) { this.listener = listener; } @@ -136,10 +130,16 @@ public class LengthFieldStreamingDecoder implements StreamingDecoder { } private void processHeader() throws IOException { - processOffset(accumulate, lengthFieldOffset); + ByteArrayOutputStream bos = new ByteArrayOutputStream(lengthFieldOffset + lengthFieldLength); + byte[] offsetData = new byte[lengthFieldOffset]; + int ignore = accumulate.read(offsetData); + bos.write(offsetData); + processOffset(new ByteArrayInputStream(offsetData), lengthFieldOffset); byte[] lengthBytes = new byte[lengthFieldLength]; - accumulate.read(lengthBytes); + ignore = accumulate.read(lengthBytes); + bos.write(lengthBytes); requiredLength = bytesToInt(lengthBytes); + this.dataHeader = new ByteArrayInputStream(bos.toByteArray()); // Continue reading the frame body. state = DecodeState.PAYLOAD; @@ -154,20 +154,23 @@ public class LengthFieldStreamingDecoder implements StreamingDecoder { if (lengthFieldOffset != 0) { return; } - inputStream.read(new byte[lengthFieldOffset]); + int ignore = inputStream.read(new byte[lengthFieldOffset]); } private void processBody() throws IOException { byte[] rawMessage = readRawMessage(accumulate, requiredLength); InputStream inputStream = new ByteArrayInputStream(rawMessage); - Object[] decodeParameters = httpMessageCodec.decode(inputStream, targetTypes); - this.listener.onMessage(decodeParameters); + invokeListener(inputStream); // Done with this frame, begin processing the next header. state = DecodeState.HEADER; requiredLength = lengthFieldOffset + lengthFieldLength; } + protected void invokeListener(InputStream inputStream) { + this.listener.onFragmentMessage(dataHeader, inputStream); + } + protected byte[] readRawMessage(InputStream inputStream, int length) throws IOException { byte[] data = new byte[length]; inputStream.read(data, 0, length); diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/StreamingDecoder.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/StreamingDecoder.java index 212b68a6ac..f1b1b2e046 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/StreamingDecoder.java +++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/StreamingDecoder.java @@ -16,8 +16,55 @@ */ package org.apache.dubbo.remoting.http12.message; -public interface StreamingDecoder extends ListeningDecoder { +import org.apache.dubbo.remoting.http12.exception.DecodeException; + +import java.io.InputStream; + +public interface StreamingDecoder { void request(int numMessages); + void decode(InputStream inputStream) throws DecodeException; + + void close(); + + void setFragmentListener(FragmentListener listener); + + interface FragmentListener { + + /** + * @param rawMessage raw message + */ + void onFragmentMessage(InputStream rawMessage); + + /** + * @param rawMessage raw message + */ + default void onFragmentMessage(InputStream dataHeader, InputStream rawMessage){ + onFragmentMessage(rawMessage); + } + + default void onClose() { + + } + } + + class DefaultFragmentListener implements FragmentListener { + + private final ListeningDecoder listeningDecoder; + + public DefaultFragmentListener(ListeningDecoder listeningDecoder) { + this.listeningDecoder = listeningDecoder; + } + + @Override + public void onFragmentMessage(InputStream rawMessage) { + listeningDecoder.decode(rawMessage); + } + + @Override + public void onClose() { + this.listeningDecoder.close(); + } + } } diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h1/NettyHttp1ConnectionHandler.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h1/NettyHttp1ConnectionHandler.java index 84ddeda1bb..9385a30564 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h1/NettyHttp1ConnectionHandler.java +++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h1/NettyHttp1ConnectionHandler.java @@ -29,7 +29,7 @@ import org.apache.dubbo.remoting.http12.h1.Http1Request; import org.apache.dubbo.remoting.http12.h1.Http1ServerChannelObserver; import org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListener; import org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListenerFactory; -import org.apache.dubbo.remoting.http12.message.HttpMessageCodec; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory; import org.apache.dubbo.rpc.model.FrameworkModel; import java.util.List; @@ -90,19 +90,19 @@ public class NettyHttp1ConnectionHandler extends SimpleChannelInboundHandler candidates) { - for (HttpMessageCodec codec : candidates) { - if (codec.support(contentType)) { - return codec; + private static HttpMessageCodecFactory findSuitableCodec(String contentType, List candidates) { + for (HttpMessageCodecFactory factory : candidates) { + if (factory.support(contentType)) { + return factory; } } return null; diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory b/dubbo-remoting/dubbo-remoting-http12/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory new file mode 100644 index 0000000000..f72f51f4b6 --- /dev/null +++ b/dubbo-remoting/dubbo-remoting-http12/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory @@ -0,0 +1 @@ +json=org.apache.dubbo.remoting.http12.message.JsonCodecFactory diff --git a/dubbo-rpc/dubbo-rpc-triple/pom.xml b/dubbo-rpc/dubbo-rpc-triple/pom.xml index 1c28453239..a09ac23906 100644 --- a/dubbo-rpc/dubbo-rpc-triple/pom.xml +++ b/dubbo-rpc/dubbo-rpc-triple/pom.xml @@ -37,6 +37,11 @@ dubbo-rpc-api ${project.parent.version} + + org.apache.dubbo + dubbo-remoting-http12 + ${project.parent.version} + org.apache.dubbo dubbo-remoting-netty4 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 da3e67dfa4..161c19f1a3 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 @@ -17,18 +17,17 @@ package org.apache.dubbo.rpc.protocol.tri; +import com.google.protobuf.Message; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.serialize.MultipleSerialization; import org.apache.dubbo.common.stream.StreamObserver; import org.apache.dubbo.config.Constants; -import org.apache.dubbo.remoting.utils.UrlUtils; import org.apache.dubbo.remoting.transport.CodecSupport; +import org.apache.dubbo.remoting.utils.UrlUtils; import org.apache.dubbo.rpc.model.MethodDescriptor; import org.apache.dubbo.rpc.model.Pack; import org.apache.dubbo.rpc.model.PackableMethod; - -import com.google.protobuf.Message; import org.apache.dubbo.rpc.model.UnPack; import org.apache.dubbo.rpc.model.WrapperUnPack; @@ -140,7 +139,7 @@ public class ReflectionPackableMethod implements PackableMethod { * * @return true if the request and response object is not generated by protobuf */ - static boolean needWrap(MethodDescriptor methodDescriptor, Class[] parameterClasses, + public static boolean needWrap(MethodDescriptor methodDescriptor, Class[] parameterClasses, Class returnClass) { String methodName = methodDescriptor.getMethodName(); // generic call must be wrapped diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java index 8daa1f600f..3937921e5e 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java @@ -19,7 +19,10 @@ package org.apache.dubbo.rpc.protocol.tri; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.codec.http2.Http2FrameCodec; +import io.netty.handler.codec.http2.Http2FrameCodecBuilder; import io.netty.handler.codec.http2.Http2FrameLogger; import io.netty.handler.codec.http2.Http2MultiplexHandler; import io.netty.handler.codec.http2.Http2Settings; @@ -31,22 +34,25 @@ import org.apache.dubbo.common.config.Configuration; import org.apache.dubbo.common.config.ConfigurationUtils; import org.apache.dubbo.common.extension.Activate; import org.apache.dubbo.common.extension.ExtensionLoader; -import org.apache.dubbo.common.threadpool.manager.ExecutorRepository; import org.apache.dubbo.remoting.ChannelHandler; import org.apache.dubbo.remoting.api.AbstractWireProtocol; import org.apache.dubbo.remoting.api.pu.ChannelHandlerPretender; import org.apache.dubbo.remoting.api.pu.ChannelOperator; import org.apache.dubbo.remoting.api.ssl.ContextOperator; +import org.apache.dubbo.remoting.http12.netty4.HttpWriteQueueHandler; +import org.apache.dubbo.remoting.http12.netty4.h1.NettyHttp1Codec; +import org.apache.dubbo.remoting.http12.netty4.h1.NettyHttp1ConnectionHandler; +import org.apache.dubbo.remoting.http12.netty4.h2.NettyHttp2FrameCodec; +import org.apache.dubbo.remoting.http12.netty4.h2.NettyHttp2ProtocolSelectorHandler; import org.apache.dubbo.rpc.HeaderFilter; -import org.apache.dubbo.rpc.executor.ExecutorSupport; import org.apache.dubbo.rpc.model.FrameworkModel; import org.apache.dubbo.rpc.model.ScopeModelAware; +import org.apache.dubbo.rpc.protocol.tri.h12.TripleProtocolDetector; +import org.apache.dubbo.rpc.protocol.tri.h12.http1.DefaultHttp11ServerTransportListenerFactory; +import org.apache.dubbo.rpc.protocol.tri.h12.http2.GenericHttp2ServerTransportListenerFactory; import org.apache.dubbo.rpc.protocol.tri.transport.TripleClientHandler; -import org.apache.dubbo.rpc.protocol.tri.transport.TripleCommandOutBoundHandler; -import org.apache.dubbo.rpc.protocol.tri.transport.TripleHttp2FrameServerHandler; import org.apache.dubbo.rpc.protocol.tri.transport.TripleServerConnectionHandler; import org.apache.dubbo.rpc.protocol.tri.transport.TripleTailHandler; -import org.apache.dubbo.rpc.protocol.tri.transport.TripleWriteQueue; import java.util.ArrayList; import java.util.Collections; @@ -80,7 +86,7 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo private FrameworkModel frameworkModel; public TripleHttp2Protocol() { - super(new Http2ProtocolDetector()); + super(new TripleProtocolDetector()); } @Override @@ -94,8 +100,61 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo super.close(); } + @Override + public void configClientPipeline(URL url, ChannelOperator operator, ContextOperator contextOperator) { + Configuration config = ConfigurationUtils.getGlobalConfiguration(url.getOrDefaultApplicationModel()); + final Http2FrameCodec codec = Http2FrameCodecBuilder.forClient() + .gracefulShutdownTimeoutMillis(10000) + .initialSettings(new Http2Settings().headerTableSize( + config.getInt(H2_SETTINGS_HEADER_TABLE_SIZE_KEY, DEFAULT_SETTING_HEADER_LIST_SIZE)) + .pushEnabled(config.getBoolean(H2_SETTINGS_ENABLE_PUSH_KEY, false)) + .maxConcurrentStreams( + config.getInt(H2_SETTINGS_MAX_CONCURRENT_STREAMS_KEY, Integer.MAX_VALUE)) + .initialWindowSize( + config.getInt(H2_SETTINGS_INITIAL_WINDOW_SIZE_KEY, DEFAULT_WINDOW_INIT_SIZE)) + .maxFrameSize(config.getInt(H2_SETTINGS_MAX_FRAME_SIZE_KEY, DEFAULT_MAX_FRAME_SIZE)) + .maxHeaderListSize(config.getInt(H2_SETTINGS_MAX_HEADER_LIST_SIZE_KEY, + DEFAULT_MAX_HEADER_LIST_SIZE))) + .frameLogger(CLIENT_LOGGER) + .build(); +// codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter()); + final Http2MultiplexHandler handler = new Http2MultiplexHandler( + new TripleClientHandler(frameworkModel)); + List handlers = new ArrayList<>(); + handlers.add(new ChannelHandlerPretender(codec)); + handlers.add(new ChannelHandlerPretender(handler)); + handlers.add(new ChannelHandlerPretender(new TripleTailHandler())); + operator.configChannelHandler(handlers); + } + @Override public void configServerProtocolHandler(URL url, ChannelOperator operator) { + String httpVersion = operator.detectResult().getAttribute(TripleProtocolDetector.HTTP_VERSION); + List channelHandlerPretenders = new ArrayList<>(); + try { + //h1 + if (TripleProtocolDetector.HttpVersion.HTTP1.getVersion().equals(httpVersion)) { + configurerHttp1Handlers(url, channelHandlerPretenders); + return; + } + + //h2 + if (TripleProtocolDetector.HttpVersion.HTTP2.getVersion().equals(httpVersion)) { + configurerHttp2Handlers(url, channelHandlerPretenders); + } + } finally { + operator.configChannelHandler(channelHandlerPretenders); + } + } + + private void configurerHttp1Handlers(URL url, List handlers) { + handlers.add(new ChannelHandlerPretender(new HttpServerCodec())); + handlers.add(new ChannelHandlerPretender(new HttpObjectAggregator(Integer.MAX_VALUE))); + handlers.add(new ChannelHandlerPretender(new NettyHttp1Codec())); + handlers.add(new ChannelHandlerPretender(new NettyHttp1ConnectionHandler(url, frameworkModel, DefaultHttp11ServerTransportListenerFactory.INSTANCE))); + } + + private void configurerHttp2Handlers(URL url, List handlers) { Configuration config = ConfigurationUtils.getGlobalConfiguration(url.getOrDefaultApplicationModel()); final List headFilters; if (filtersLoader != null) { @@ -117,55 +176,19 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo DEFAULT_MAX_HEADER_LIST_SIZE))) .frameLogger(SERVER_LOGGER) .build(); - ExecutorSupport executorSupport = ExecutorRepository.getInstance(url.getOrDefaultApplicationModel()).getExecutorSupport(url); - codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter()); - TripleWriteQueue writeQueue = new TripleWriteQueue(); final Http2MultiplexHandler handler = new Http2MultiplexHandler( new ChannelInitializer() { @Override protected void initChannel(Http2StreamChannel ch) { final ChannelPipeline p = ch.pipeline(); - p.addLast(new TripleCommandOutBoundHandler()); - p.addLast(new TripleHttp2FrameServerHandler(frameworkModel, executorSupport, - headFilters, ch, writeQueue)); + p.addLast(new NettyHttp2FrameCodec()); + p.addLast(new NettyHttp2ProtocolSelectorHandler(url, frameworkModel, GenericHttp2ServerTransportListenerFactory.INSTANCE)); } }); - List handlers = new ArrayList<>(); + handlers.add(new ChannelHandlerPretender(new HttpWriteQueueHandler())); handlers.add(new ChannelHandlerPretender(codec)); handlers.add(new ChannelHandlerPretender(new FlushConsolidationHandler(64, true))); handlers.add(new ChannelHandlerPretender(new TripleServerConnectionHandler())); handlers.add(new ChannelHandlerPretender(handler)); - handlers.add(new ChannelHandlerPretender(new TripleTailHandler())); - operator.configChannelHandler(handlers); - - - } - - @Override - public void configClientPipeline(URL url, ChannelOperator operator, ContextOperator contextOperator) { - Configuration config = ConfigurationUtils.getGlobalConfiguration(url.getOrDefaultApplicationModel()); - final Http2FrameCodec codec = TripleHttp2FrameCodecBuilder.forClient() - .customizeConnection((connection) -> connection.remote().flowController(new TriHttp2RemoteFlowController(connection, url.getOrDefaultApplicationModel()))) - .gracefulShutdownTimeoutMillis(10000) - .initialSettings(new Http2Settings().headerTableSize( - config.getInt(H2_SETTINGS_HEADER_TABLE_SIZE_KEY, DEFAULT_SETTING_HEADER_LIST_SIZE)) - .pushEnabled(config.getBoolean(H2_SETTINGS_ENABLE_PUSH_KEY, false)) - .maxConcurrentStreams( - config.getInt(H2_SETTINGS_MAX_CONCURRENT_STREAMS_KEY, Integer.MAX_VALUE)) - .initialWindowSize( - config.getInt(H2_SETTINGS_INITIAL_WINDOW_SIZE_KEY, DEFAULT_WINDOW_INIT_SIZE)) - .maxFrameSize(config.getInt(H2_SETTINGS_MAX_FRAME_SIZE_KEY, DEFAULT_MAX_FRAME_SIZE)) - .maxHeaderListSize(config.getInt(H2_SETTINGS_MAX_HEADER_LIST_SIZE_KEY, - DEFAULT_MAX_HEADER_LIST_SIZE))) - .frameLogger(CLIENT_LOGGER) - .build(); - codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter()); - final Http2MultiplexHandler handler = new Http2MultiplexHandler( - new TripleClientHandler(frameworkModel)); - List handlers = new ArrayList<>(); - handlers.add(new ChannelHandlerPretender(codec)); - handlers.add(new ChannelHandlerPretender(handler)); - handlers.add(new ChannelHandlerPretender(new TripleTailHandler())); - operator.configChannelHandler(handlers); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Bzip2.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Bzip2.java index dbc54be92c..5a53cb3f71 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Bzip2.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Bzip2.java @@ -17,13 +17,14 @@ package org.apache.dubbo.rpc.protocol.tri.compressor; -import org.apache.dubbo.rpc.RpcException; - import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream; import org.apache.commons.compress.compressors.bzip2.BZip2CompressorOutputStream; import org.apache.commons.io.output.ByteArrayOutputStream; +import org.apache.dubbo.rpc.RpcException; import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.OutputStream; /** @@ -59,6 +60,15 @@ public class Bzip2 implements Compressor, DeCompressor { return out.toByteArray(); } + @Override + public OutputStream decorate(OutputStream outputStream) { + try { + return new BZip2CompressorOutputStream(outputStream); + } catch (IOException e) { + throw new IllegalStateException(e); + } + } + @Override public byte[] decompress(byte[] payloadByteArr) { if (null == payloadByteArr || 0 == payloadByteArr.length) { diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Compressor.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Compressor.java index 35d7ab80fb..73a33f3fd9 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Compressor.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Compressor.java @@ -22,6 +22,8 @@ import org.apache.dubbo.common.extension.SPI; import org.apache.dubbo.rpc.Constants; import org.apache.dubbo.rpc.model.FrameworkModel; +import java.io.OutputStream; + /** * compress payload for grpc request， and decompress response payload Configure it in files, * pictures or other configurations that exist in the system properties Configure {@link @@ -51,4 +53,6 @@ public interface Compressor extends MessageEncoding { */ byte[] compress(byte[] payloadByteArr); + OutputStream decorate(OutputStream outputStream); + } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Gzip.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Gzip.java index 371b650bae..168bc627aa 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Gzip.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Gzip.java @@ -21,6 +21,8 @@ import org.apache.dubbo.rpc.RpcException; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStream; import java.util.zip.GZIPInputStream; import java.util.zip.GZIPOutputStream; @@ -52,6 +54,15 @@ public class Gzip implements Compressor, DeCompressor { return byteOutStream.toByteArray(); } + @Override + public OutputStream decorate(OutputStream outputStream) { + try { + return new GZIPOutputStream(outputStream); + } catch (IOException e) { + throw new IllegalStateException(e); + } + } + @Override public byte[] decompress(byte[] payloadByteArr) throws RpcException { if (null == payloadByteArr || 0 == payloadByteArr.length) { diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Identity.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Identity.java index 3d3d32d788..ee6d16a671 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Identity.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Identity.java @@ -17,6 +17,8 @@ package org.apache.dubbo.rpc.protocol.tri.compressor; +import java.io.OutputStream; + /** * Default compressor *

@@ -38,6 +40,11 @@ public class Identity implements Compressor, DeCompressor { return payloadByteArr; } + @Override + public OutputStream decorate(OutputStream outputStream) { + return outputStream; + } + @Override public byte[] decompress(byte[] payloadByteArr) { return payloadByteArr; diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Snappy.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Snappy.java index be5b11fe81..2d1e9b08c2 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Snappy.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Snappy.java @@ -20,6 +20,7 @@ package org.apache.dubbo.rpc.protocol.tri.compressor; import org.apache.dubbo.rpc.RpcException; import java.io.IOException; +import java.io.OutputStream; /** * snappy compressor, Provide high-speed compression speed and reasonable compression ratio @@ -40,7 +41,6 @@ public class Snappy implements Compressor, DeCompressor { if (null == payloadByteArr || 0 == payloadByteArr.length) { return new byte[0]; } - try { return org.xerial.snappy.Snappy.compress(payloadByteArr); } catch (IOException e) { @@ -48,6 +48,11 @@ public class Snappy implements Compressor, DeCompressor { } } + @Override + public OutputStream decorate(OutputStream outputStream) { + return outputStream; + } + @Override public byte[] decompress(byte[] payloadByteArr) { if (null == payloadByteArr || 0 == payloadByteArr.length) { diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerCallListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerCallListener.java similarity index 70% rename from dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerCallListener.java rename to dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerCallListener.java index dfa8ac877e..2ae592501c 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerCallListener.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerCallListener.java @@ -14,17 +14,23 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.dubbo.remoting.http12; +package org.apache.dubbo.rpc.protocol.tri.h12; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.stream.StreamObserver; import org.apache.dubbo.remoting.http12.exception.HttpRequestTimeout; +import org.apache.dubbo.remoting.http12.h2.Http2CancelableStreamObserver; import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Result; import org.apache.dubbo.rpc.RpcContext; import org.apache.dubbo.rpc.RpcInvocation; +import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum; +import org.apache.dubbo.rpc.protocol.tri.call.AbstractServerCall; +import java.net.InetSocketAddress; + +import static org.apache.dubbo.common.constants.CommonConstants.REMOTE_APPLICATION_KEY; import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_TIMEOUT_SERVER; public abstract class AbstractServerCallListener implements ServerCallListener { @@ -46,16 +52,31 @@ public abstract class AbstractServerCallListener implements ServerCallListener { } public void invoke() { + if (responseObserver instanceof Http2CancelableStreamObserver) { + RpcContext.restoreCancellationContext(((Http2CancelableStreamObserver) responseObserver).getCancellationContext()); + } + InetSocketAddress remoteAddress = (InetSocketAddress) invocation.getAttributes() + .remove(AbstractServerCall.REMOTE_ADDRESS_KEY); + RpcContext.getServiceContext().setRemoteAddress(remoteAddress); + String remoteApp = (String) invocation.getAttributes() + .remove(TripleHeaderEnum.CONSUMER_APP_NAME_KEY); + if (null != remoteApp) { + RpcContext.getServiceContext().setRemoteApplicationName(remoteApp); + invocation.setAttachmentIfAbsent(REMOTE_APPLICATION_KEY, remoteApp); + } try { final long stInMillis = System.currentTimeMillis(); final Result response = invoker.invoke(invocation); response.whenCompleteWithContext((r, t) -> { + if (responseObserver instanceof AttachmentHolder) { + ((AttachmentHolder) responseObserver).setResponseAttachments(response.getObjectAttachments()); + } if (t != null) { responseObserver.onError(t); return; } if (response.hasException()) { - doOnResponseHasException(response.getException()); + onResponseException(response.getException()); return; } final long cost = System.currentTimeMillis() - stInMillis; @@ -80,7 +101,7 @@ public abstract class AbstractServerCallListener implements ServerCallListener { } } - protected void doOnResponseHasException(Throwable t) { + protected void onResponseException(Throwable t) { responseObserver.onError(t); } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerTransportListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerTransportListener.java new file mode 100644 index 0000000000..a294d61e2a --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerTransportListener.java @@ -0,0 +1,479 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12; + +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.CollectionUtils; +import org.apache.dubbo.common.utils.JsonUtils; +import org.apache.dubbo.remoting.http12.HttpChannel; +import org.apache.dubbo.remoting.http12.HttpHeaderNames; +import org.apache.dubbo.remoting.http12.HttpHeaders; +import org.apache.dubbo.remoting.http12.HttpInputMessage; +import org.apache.dubbo.remoting.http12.HttpStatus; +import org.apache.dubbo.remoting.http12.HttpTransportListener; +import org.apache.dubbo.remoting.http12.RequestMetadata; +import org.apache.dubbo.remoting.http12.exception.HttpStatusException; +import org.apache.dubbo.remoting.http12.exception.IllegalPathException; +import org.apache.dubbo.remoting.http12.exception.UnimplementedException; +import org.apache.dubbo.remoting.http12.exception.UnsupportedMediaTypeException; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodec; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory; +import org.apache.dubbo.remoting.http12.message.MethodMetadata; +import org.apache.dubbo.rpc.HeaderFilter; +import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.PathResolver; +import org.apache.dubbo.rpc.RpcInvocation; +import org.apache.dubbo.rpc.TriRpcStatus; +import org.apache.dubbo.rpc.model.FrameworkModel; +import org.apache.dubbo.rpc.model.MethodDescriptor; +import org.apache.dubbo.rpc.model.ProviderModel; +import org.apache.dubbo.rpc.model.ServiceDescriptor; +import org.apache.dubbo.rpc.protocol.tri.TripleConstant; +import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum; +import org.apache.dubbo.rpc.protocol.tri.TripleProtocol; +import org.apache.dubbo.rpc.protocol.tri.stream.StreamUtils; +import org.apache.dubbo.rpc.service.ServiceDescriptorInternalCache; +import org.apache.dubbo.rpc.stub.StubSuppliers; + +import java.io.InputStream; +import java.lang.reflect.InvocationTargetException; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.Executor; +import java.util.function.Supplier; + +import static org.apache.dubbo.common.constants.CommonConstants.HEADER_FILTER_KEY; +import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR; +import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_PARSE; + +public abstract class AbstractServerTransportListener
implements HttpTransportListener { + + private static final ErrorTypeAwareLogger LOGGER = LoggerFactory.getErrorTypeAwareLogger(AbstractServerTransportListener.class); + + private final PathResolver pathResolver; + + private final FrameworkModel frameworkModel; + + private final URL url; + + private final HttpChannel httpChannel; + + private final List headerFilters; + + private HttpMessageCodec httpMessageCodec; + + private Invoker invoker; + + private ServiceDescriptor serviceDescriptor; + + private MethodDescriptor methodDescriptor; + + private RpcInvocation rpcInvocation; + + private MethodMetadata methodMetadata; + + private HEADER httpMetadata; + + private Executor executor; + + private boolean hasStub; + + private HttpMessageListener httpMessageListener; + + public AbstractServerTransportListener(FrameworkModel frameworkModel, URL url, HttpChannel httpChannel) { + this.frameworkModel = frameworkModel; + this.url = url; + this.httpChannel = httpChannel; + this.pathResolver = frameworkModel.getExtensionLoader(PathResolver.class).getDefaultExtension(); + this.headerFilters = frameworkModel.getExtensionLoader(HeaderFilter.class).getActivateExtension(url, HEADER_FILTER_KEY); + } + + protected Executor initializeExecutor(HEADER metadata) { + //default direct executor + return Runnable::run; + } + + @Override + public void onMetadata(HEADER metadata) { + try { + this.executor = initializeExecutor(metadata); + } catch (Throwable throwable) { + onError(throwable); + return; + } + if (this.executor == null) { + onError(new NullPointerException("initializeExecutor return null")); + return; + } + executor.execute(() -> { + try { + doOnMetadata(metadata); + } catch (Throwable throwable) { + onError(throwable); + } + }); + } + + protected void doOnMetadata(HEADER metadata) { + onPrepareMetadata(metadata); + this.httpMetadata = metadata; + String path = metadata.path(); + HttpHeaders headers = metadata.headers(); + //1.check necessary header + String contentType = headers.getFirst(HttpHeaderNames.CONTENT_TYPE.getName()); + if (contentType == null) { + throw new UnsupportedMediaTypeException("'" + HttpHeaderNames.CONTENT_TYPE.getName() + "' must be not null."); + } + + //2. check service + String[] parts = path.split("/"); + if (parts.length != 3) { + throw new IllegalPathException(path); + } + String serviceName = parts[1]; + this.hasStub = pathResolver.hasNativeStub(path); + this.invoker = getInvoker(metadata, serviceName); + if (invoker == null) { + throw new UnimplementedException(serviceName); + } + HttpMessageCodec httpMessageCodec = determineHttpMessageCodec(contentType); + if (httpMessageCodec == null) { + throw new UnsupportedMediaTypeException(contentType); + } + this.httpMessageCodec = httpMessageCodec; + this.serviceDescriptor = findServiceDescriptor(invoker, serviceName, hasStub); + setHttpMessageListener(newHttpMessageListener()); + onMetadataCompletion(metadata); + } + + protected abstract HttpMessageListener newHttpMessageListener(); + + @Override + public void onData(MESSAGE message) { + this.executor.execute(() -> { + try { + doOnData(message); + } catch (Throwable e) { + onError(e); + } + }); + } + + protected void doOnData(MESSAGE message) { + //decode message + onPrepareData(message); + InputStream body = message.getBody(); + httpMessageListener.onMessage(body); + onDataCompletion(message); + } + + protected void onPrepareMetadata(HEADER header) { + //default no op + } + + protected void onMetadataCompletion(HEADER metadata) { + //default no op + } + + protected void onPrepareData(MESSAGE message) { + //default no op + } + + protected void onDataCompletion(MESSAGE message) { + //default no op + } + + protected void onError(Throwable throwable) { + //default rethrow + if (throwable instanceof RuntimeException) { + throw ((RuntimeException) throwable); + } + if (throwable instanceof InvocationTargetException) { + Throwable targetException = ((InvocationTargetException) throwable).getTargetException(); + if (targetException instanceof RuntimeException) { + throw (RuntimeException) targetException; + } else if (targetException instanceof Error) { + throw (Error) targetException; + } + } + throw new HttpStatusException(HttpStatus.INTERNAL_SERVER_ERROR.getCode(), throwable); + } + + private Invoker getInvoker(HEADER metadata, String serviceName) { + HttpHeaders headers = metadata.headers(); + final String version = + headers.containsKey(TripleHeaderEnum.SERVICE_VERSION.getHeader()) ? headers.get( + TripleHeaderEnum.SERVICE_VERSION.getHeader()).toString() : null; + final String group = + headers.containsKey(TripleHeaderEnum.SERVICE_GROUP.getHeader()) ? headers.get( + TripleHeaderEnum.SERVICE_GROUP.getHeader()).toString() : null; + final String key = URL.buildKey(serviceName, group, version); + Invoker invoker = pathResolver.resolve(key); + if (invoker == null && TripleProtocol.RESOLVE_FALLBACK_TO_DEFAULT) { + invoker = pathResolver.resolve(URL.buildKey(serviceName, group, "1.0.0")); + } + if (invoker == null && TripleProtocol.RESOLVE_FALLBACK_TO_DEFAULT) { + invoker = pathResolver.resolve(serviceName); + } + return invoker; + } + + + protected HttpMessageCodec determineHttpMessageCodec(String contentType) { + for (HttpMessageCodecFactory httpMessageCodecFactory : frameworkModel.getExtensionLoader(HttpMessageCodecFactory.class).getActivateExtensions()) { + if (httpMessageCodecFactory.support(contentType)) { + return httpMessageCodecFactory.createCodec(invoker.getUrl(), frameworkModel); + } + } + return null; + } + + private static ServiceDescriptor findServiceDescriptor(Invoker invoker, String serviceName, boolean hasStub) throws UnimplementedException { + ServiceDescriptor result; + if (hasStub) { + result = getStubServiceDescriptor(invoker.getUrl(), serviceName); + } else { + result = getReflectionServiceDescriptor(invoker.getUrl()); + } + if (result == null) { + throw new UnimplementedException("service:" + serviceName); + } + return result; + } + + protected static MethodDescriptor findMethodDescriptor(ServiceDescriptor serviceDescriptor, String originalMethodName, boolean hasStub) throws UnimplementedException { + MethodDescriptor result; + if (hasStub) { + result = serviceDescriptor.getMethods(originalMethodName).get(0); + } else { + result = findReflectionMethodDescriptor(serviceDescriptor, originalMethodName); + } + return result; + } + + protected RpcInvocation buildRpcInvocation(Invoker invoker, + ServiceDescriptor serviceDescriptor, + MethodDescriptor methodDescriptor) { + final URL url = invoker.getUrl(); + RpcInvocation inv = new RpcInvocation(url.getServiceModel(), + methodDescriptor.getMethodName(), + serviceDescriptor.getInterfaceName(), url.getProtocolServiceKey(), + methodDescriptor.getParameterClasses(), + new Object[0]); + inv.setTargetServiceUniqueName(url.getServiceKey()); + inv.setReturnTypes(methodDescriptor.getReturnTypes()); + Map headers = getHttpMetadata().headers().toSingleValueMap(); + Map requestMetadata = headersToMap(headers, () -> { + return Optional.ofNullable(headers.get(TripleHeaderEnum.TRI_HEADER_CONVERT.getHeader())) + .map(CharSequence::toString) + .orElse(null); + }); + inv.setObjectAttachments(StreamUtils.toAttachments(requestMetadata)); + + inv.put("tri.remote.address", httpChannel.remoteAddress()); + //customizer RpcInvocation + headerFilters.forEach(f -> f.invoke(invoker, inv)); + return inv; + } + + protected static ServiceDescriptor getStubServiceDescriptor(URL url, String serviceName) { + ServiceDescriptor serviceDescriptor; + if (url.getServiceModel() != null) { + serviceDescriptor = url + .getServiceModel() + .getServiceModel(); + } else { + serviceDescriptor = StubSuppliers.getServiceDescriptor(serviceName); + } + return serviceDescriptor; + } + + protected static ServiceDescriptor getReflectionServiceDescriptor(URL url) { + ProviderModel providerModel = (ProviderModel) url.getServiceModel(); + if (providerModel == null || providerModel.getServiceModel() == null) { + return null; + } + return providerModel.getServiceModel(); + } + + protected static boolean isEcho(String methodName) { + return CommonConstants.$ECHO.equals(methodName); + } + + protected static boolean isGeneric(String methodName) { + return CommonConstants.$INVOKE.equals(methodName) || CommonConstants.$INVOKE_ASYNC.equals( + methodName); + } + + protected static MethodDescriptor findReflectionMethodDescriptor(ServiceDescriptor serviceDescriptor, String methodName) { + MethodDescriptor methodDescriptor = null; + if (isGeneric(methodName)) { + // There should be one and only one + methodDescriptor = ServiceDescriptorInternalCache.genericService() + .getMethods(methodName).get(0); + } else if (isEcho(methodName)) { + // There should be one and only one + return ServiceDescriptorInternalCache.echoService().getMethods(methodName) + .get(0); + } else { + List methodDescriptors = serviceDescriptor.getMethods(methodName); + // try lower-case method + if (CollectionUtils.isEmpty(methodDescriptors)) { + final String lowerMethod = + Character.toLowerCase(methodName.charAt(0)) + methodName.substring(1); + methodDescriptors = serviceDescriptor.getMethods(lowerMethod); + } + if (CollectionUtils.isEmpty(methodDescriptors)) { + return null; + } + // In most cases there is only one method + if (methodDescriptors.size() == 1) { + methodDescriptor = methodDescriptors.get(0); + } + // generated unary method ,use unary type + // Response foo(Request) + // void foo(Request,StreamObserver) + if (methodDescriptors.size() == 2) { + if (methodDescriptors.get(1).getRpcType() == MethodDescriptor.RpcType.SERVER_STREAM) { + methodDescriptor = methodDescriptors.get(0); + } else if (methodDescriptors.get(0).getRpcType() == MethodDescriptor.RpcType.SERVER_STREAM) { + methodDescriptor = methodDescriptors.get(1); + } + } + } + return methodDescriptor; + } + + protected FrameworkModel getFrameworkModel() { + return frameworkModel; + } + + protected HEADER getHttpMetadata() { + return httpMetadata; + } + + protected Invoker getInvoker() { + return invoker; + } + + protected ServiceDescriptor getServiceDescriptor() { + return serviceDescriptor; + } + + protected MethodDescriptor getMethodDescriptor() { + return methodDescriptor; + } + + public void setServiceDescriptor(ServiceDescriptor serviceDescriptor) { + this.serviceDescriptor = serviceDescriptor; + } + + public void setMethodDescriptor(MethodDescriptor methodDescriptor) { + this.methodDescriptor = methodDescriptor; + } + + public void setMethodMetadata(MethodMetadata methodMetadata) { + this.methodMetadata = methodMetadata; + } + + protected RpcInvocation getRpcInvocation() { + return rpcInvocation; + } + + public void setRpcInvocation(RpcInvocation rpcInvocation) { + this.rpcInvocation = rpcInvocation; + } + + protected MethodMetadata getMethodMetadata() { + return methodMetadata; + } + + protected HttpMessageCodec getHttpMessageCodec() { + return httpMessageCodec; + } + + protected void setHttpMessageListener(HttpMessageListener httpMessageListener) { + this.httpMessageListener = httpMessageListener; + } + + protected HttpMessageListener getHttpMessageListener() { + return httpMessageListener; + } + + protected PathResolver getPathResolver() { + return pathResolver; + } + + protected final URL getUrl() { + return url; + } + + public boolean isHasStub() { + return hasStub; + } + + protected Map headersToMap(Map headers, Supplier convertUpperHeaderSupplier) { + if (headers == null) { + return Collections.emptyMap(); + } + Map attachments = new HashMap<>(headers.size()); + for (Map.Entry header : headers.entrySet()) { + String key = header.getKey(); + if (key.endsWith(TripleConstant.HEADER_BIN_SUFFIX) + && key.length() > TripleConstant.HEADER_BIN_SUFFIX.length()) { + try { + String realKey = key.substring(0, + key.length() - TripleConstant.HEADER_BIN_SUFFIX.length()); + byte[] value = StreamUtils.decodeASCIIByte(header.getValue()); + attachments.put(realKey, value); + } catch (Exception e) { + LOGGER.error(PROTOCOL_FAILED_PARSE, "", "", "Failed to parse response attachment key=" + key, e); + } + } else { + attachments.put(key, header.getValue()); + } + } + + // try converting upper key + Object obj = convertUpperHeaderSupplier.get(); + if (obj == null) { + return attachments; + } + if (obj instanceof String) { + String json = TriRpcStatus.decodeMessage((String) obj); + Map map = JsonUtils.toJavaObject(json, Map.class); + for (Map.Entry entry : map.entrySet()) { + Object val = attachments.remove(entry.getKey()); + if (val != null) { + attachments.put(entry.getValue(), val); + } + } + } else { + // If convertUpperHeaderSupplier does not return String, just fail... + // Internal invocation, use INTERNAL_ERROR instead. + + LOGGER.error(INTERNAL_ERROR, "wrong internal invocation", "", "Triple convertNoLowerCaseHeader error, obj is not String"); + } + return attachments; + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AttachmentHolder.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AttachmentHolder.java new file mode 100644 index 0000000000..be0e488096 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AttachmentHolder.java @@ -0,0 +1,26 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12; + +import java.util.Map; + +public interface AttachmentHolder { + + void setResponseAttachments(Map attachments); + + Map getResponseAttachments(); +} diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/BiStreamServerCallListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/BiStreamServerCallListener.java similarity index 95% rename from dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/BiStreamServerCallListener.java rename to dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/BiStreamServerCallListener.java index 705031f98f..0ffd8d7ca7 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/BiStreamServerCallListener.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/BiStreamServerCallListener.java @@ -15,9 +15,10 @@ * limitations under the License. */ -package org.apache.dubbo.remoting.http12; +package org.apache.dubbo.rpc.protocol.tri.h12; import org.apache.dubbo.common.stream.StreamObserver; +import org.apache.dubbo.remoting.http12.FlowControlStreamObserver; import org.apache.dubbo.remoting.http12.exception.HttpStatusException; import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.RpcInvocation; diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/CompressibleCodec.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/CompressibleCodec.java new file mode 100644 index 0000000000..7da92bd17f --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/CompressibleCodec.java @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12; + +import org.apache.dubbo.remoting.http12.exception.DecodeException; +import org.apache.dubbo.remoting.http12.exception.EncodeException; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodec; +import org.apache.dubbo.remoting.http12.message.MediaType; +import org.apache.dubbo.rpc.protocol.tri.compressor.Compressor; + +import java.io.InputStream; +import java.io.OutputStream; + +public class CompressibleCodec implements HttpMessageCodec { + + private final HttpMessageCodec delegate; + + private Compressor compressor = Compressor.NONE; + + public CompressibleCodec(HttpMessageCodec delegate) { + this.delegate = delegate; + } + + public void setCompressor(Compressor compressor) { + this.compressor = compressor; + } + + @Override + public void encode(OutputStream outputStream, Object data) throws EncodeException { + delegate.encode(compressor.decorate(outputStream), data); + } + + @Override + public Object decode(InputStream inputStream, Class targetType) throws DecodeException { + return delegate.decode(inputStream, targetType); + } + + @Override + public void encode(OutputStream outputStream, Object[] data) throws EncodeException { + delegate.encode(outputStream, data); + } + + @Override + public Object[] decode(InputStream inputStream, Class[] targetTypes) throws DecodeException { + return delegate.decode(inputStream, targetTypes); + } + + @Override + public boolean support(String contentType) { + return delegate.support(contentType); + } + + @Override + public MediaType contentType() { + return delegate.contentType(); + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/DefaultHttpMessageListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/DefaultHttpMessageListener.java new file mode 100644 index 0000000000..0ecab54e4f --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/DefaultHttpMessageListener.java @@ -0,0 +1,43 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12; + +import org.apache.dubbo.remoting.http12.message.ListeningDecoder; + +import java.io.InputStream; + +public class DefaultHttpMessageListener implements HttpMessageListener { + + private ListeningDecoder listeningDecoder; + + public DefaultHttpMessageListener() { + + } + + public DefaultHttpMessageListener(ListeningDecoder listeningDecoder) { + this.listeningDecoder = listeningDecoder; + } + + public void setListeningDecoder(ListeningDecoder listeningDecoder) { + this.listeningDecoder = listeningDecoder; + } + + @Override + public void onMessage(InputStream inputStream) { + listeningDecoder.decode(inputStream); + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/HttpMessageListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/HttpMessageListener.java new file mode 100644 index 0000000000..810af9bded --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/HttpMessageListener.java @@ -0,0 +1,24 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12; + +import java.io.InputStream; + +public interface HttpMessageListener { + + void onMessage(InputStream inputStream); +} diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ServerCallListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerCallListener.java similarity index 96% rename from dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ServerCallListener.java rename to dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerCallListener.java index d5d076029f..55616bdd82 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ServerCallListener.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerCallListener.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.dubbo.remoting.http12; +package org.apache.dubbo.rpc.protocol.tri.h12; public interface ServerCallListener { diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerCallToObserverAdapter.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerCallToObserverAdapter.java new file mode 100644 index 0000000000..b463971087 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerCallToObserverAdapter.java @@ -0,0 +1,25 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12; + +import org.apache.dubbo.rpc.protocol.tri.ServerStreamObserver; + +public interface ServerCallToObserverAdapter extends ServerStreamObserver { + + void setExceptionCode(int exceptionCode); + +} diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ServerStreamServerCallListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerStreamServerCallListener.java similarity index 97% rename from dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ServerStreamServerCallListener.java rename to dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerStreamServerCallListener.java index e65478eb1f..5a0e60969a 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ServerStreamServerCallListener.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerStreamServerCallListener.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.dubbo.remoting.http12; +package org.apache.dubbo.rpc.protocol.tri.h12; import org.apache.dubbo.common.stream.StreamObserver; import org.apache.dubbo.remoting.http12.exception.HttpStatusException; diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/TripleProtocolDetector.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/TripleProtocolDetector.java new file mode 100644 index 0000000000..c6c7128ab1 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/TripleProtocolDetector.java @@ -0,0 +1,87 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12; + +import io.netty.handler.codec.http2.Http2CodecUtil; +import org.apache.dubbo.remoting.api.ProtocolDetector; +import org.apache.dubbo.remoting.buffer.ByteBufferBackedChannelBuffer; +import org.apache.dubbo.remoting.buffer.ChannelBuffer; +import org.apache.dubbo.remoting.buffer.ChannelBuffers; + +import static java.lang.Math.min; + +public class TripleProtocolDetector implements ProtocolDetector { + + public static final String HTTP_VERSION = "HTTP_VERSION"; + + private final ChannelBuffer clientPrefaceString = new ByteBufferBackedChannelBuffer( + Http2CodecUtil.connectionPrefaceBuf().nioBuffer()); + + @Override + public Result detect(ChannelBuffer in) { + //http1 + if (in.readableBytes() < 2) { + return Result.needMoreData(); + } + byte[] magics = new byte[5]; + in.getBytes(in.readerIndex(), magics, 0, 5); + if (isHttp(magics)) { + Result recognized = Result.recognized(); + recognized.setAttribute(HTTP_VERSION, HttpVersion.HTTP1.getVersion()); + return recognized; + } + in.resetReaderIndex(); + + //http2 + int prefaceLen = clientPrefaceString.readableBytes(); + int bytesRead = min(in.readableBytes(), prefaceLen); + if (bytesRead == 0 || !ChannelBuffers.prefixEquals(in, clientPrefaceString, bytesRead)) { + return Result.unrecognized(); + } + if (bytesRead == prefaceLen) { + Result recognized = Result.recognized(); + recognized.setAttribute(HTTP_VERSION, HttpVersion.HTTP2.getVersion()); + return recognized; + } + return Result.needMoreData(); + } + + private static boolean isHttp(byte[] magic) { + if (magic[0] == 'G' && magic[1] == 'E' && magic[2] == 'T') { + return true; + } + if (magic[0] == 'P' && magic[1] == 'O' && magic[2] == 'S' && magic[3] == 'T') { + return true; + } + return false; + } + + public static enum HttpVersion { + HTTP1("http1"), + HTTP2("http2"); + + private final String version; + + HttpVersion(String version) { + this.version = version; + } + + public String getVersion() { + return version; + } + } +} diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/UnaryServerCallListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/UnaryServerCallListener.java similarity index 64% rename from dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/UnaryServerCallListener.java rename to dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/UnaryServerCallListener.java index e1a985803f..f933020ddd 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/UnaryServerCallListener.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/UnaryServerCallListener.java @@ -14,20 +14,28 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.dubbo.remoting.http12; +package org.apache.dubbo.rpc.protocol.tri.h12; import org.apache.dubbo.common.stream.StreamObserver; import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.RpcException; import org.apache.dubbo.rpc.RpcInvocation; +import org.apache.dubbo.rpc.TriRpcStatus; public class UnaryServerCallListener extends AbstractServerCallListener { + private boolean applyCustomizeException = false; + public UnaryServerCallListener(RpcInvocation invocation, Invoker invoker, StreamObserver responseObserver) { super(invocation, invoker, responseObserver); } + public void setApplyCustomizeException(boolean applyCustomizeException) { + this.applyCustomizeException = applyCustomizeException; + } + @Override public void onReturn(Object value) { responseObserver.onNext(value); @@ -43,6 +51,24 @@ public class UnaryServerCallListener extends AbstractServerCallListener { } } + @Override + protected void onResponseException(Throwable t) { + if (applyCustomizeException) { + TriRpcStatus status = TriRpcStatus.getStatus(t); + int exceptionCode = status.code.code; + if (exceptionCode == TriRpcStatus.UNKNOWN.code.code) { + exceptionCode = RpcException.BIZ_EXCEPTION; + } + if (responseObserver instanceof ServerCallToObserverAdapter) { + ((ServerCallToObserverAdapter) responseObserver).setExceptionCode(exceptionCode); + } + onReturn(t); + } else { + super.onResponseException(t); + } + + } + @Override public void onCancel(long code) { //ignore diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java new file mode 100644 index 0000000000..5ff1b1da21 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java @@ -0,0 +1,124 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.grpc; + +import com.google.protobuf.Message; +import org.apache.dubbo.common.extension.Activate; +import org.apache.dubbo.remoting.http12.exception.DecodeException; +import org.apache.dubbo.remoting.http12.exception.EncodeException; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodec; +import org.apache.dubbo.remoting.http12.message.MediaType; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +/** + * compatible low version. + * version < 3.3 + * + * @since 3.3 + */ +@Activate +public class GrpcCompositeCodec implements HttpMessageCodec { + + private static final MediaType MEDIA_TYPE = new MediaType("application", "grpc"); + + private final ProtobufHttpMessageCodec protobufHttpMessageCodec; + + private final WrapperHttpMessageCodec wrapperHttpMessageCodec; + + public GrpcCompositeCodec(ProtobufHttpMessageCodec protobufHttpMessageCodec, + WrapperHttpMessageCodec wrapperHttpMessageCodec) { + this.protobufHttpMessageCodec = protobufHttpMessageCodec; + this.wrapperHttpMessageCodec = wrapperHttpMessageCodec; + } + + public void setEncodeTypes(Class[] encodeTypes) { + this.wrapperHttpMessageCodec.setEncodeTypes(encodeTypes); + } + + public void setDecodeTypes(Class[] decodeTypes) { + this.wrapperHttpMessageCodec.setDecodeTypes(decodeTypes); + } + + + @Override + public void encode(OutputStream outputStream, Object data) throws EncodeException { + //protobuf + //TODO int compressed = Identity.MESSAGE_ENCODING.equals(requestMetadata.compressor.getMessageEncoding()) ? 0 : 1; + try { + int compressed = 0; + outputStream.write(compressed); + if (data instanceof Message) { + int serializedSize = ((Message) data).getSerializedSize(); + //write length + writeLength(outputStream, serializedSize); + protobufHttpMessageCodec.encode(outputStream, data); + return; + } + //wrapper + wrapperHttpMessageCodec.encode(outputStream, data); + } catch (IOException e) { + throw new EncodeException(e); + } + } + + @Override + public Object decode(InputStream inputStream, Class targetType) throws DecodeException { + if (isProtobuf(targetType)) { + return protobufHttpMessageCodec.decode(inputStream, targetType); + } + return wrapperHttpMessageCodec.decode(inputStream, targetType); + } + + @Override + public Object[] decode(InputStream inputStream, Class[] targetTypes) throws DecodeException { + if (targetTypes.length > 1) { + return wrapperHttpMessageCodec.decode(inputStream, targetTypes); + } + return HttpMessageCodec.super.decode(inputStream, targetTypes); + } + + private boolean isProtobuf(Class targetType) { + if (targetType == null) { + return false; + } + return Message.class.isAssignableFrom(targetType); + } + + private static void writeLength(OutputStream outputStream, int length) { + try { + outputStream.write(((length >> 24) & 0xFF)); + outputStream.write(((length >> 16) & 0xFF)); + outputStream.write(((length >> 8) & 0xFF)); + outputStream.write((length & 0xFF)); + } catch (IOException e) { + throw new EncodeException(e); + } + } + + @Override + public MediaType contentType() { + return MEDIA_TYPE; + } + + @Override + public boolean support(String contentType) { + return contentType.startsWith(MEDIA_TYPE.getName()); + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodecFactory.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodecFactory.java new file mode 100644 index 0000000000..792fde774d --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodecFactory.java @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.grpc; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.extension.Activate; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodec; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory; +import org.apache.dubbo.remoting.http12.message.MediaType; +import org.apache.dubbo.remoting.utils.UrlUtils; +import org.apache.dubbo.rpc.model.FrameworkModel; + +@Activate +public class GrpcCompositeCodecFactory implements HttpMessageCodecFactory { + + private static final MediaType MEDIA_TYPE = new MediaType("application", "grpc"); + + @Override + public HttpMessageCodec createCodec(URL url, FrameworkModel frameworkModel) { + final String serializeName = UrlUtils.serializationOrDefault(url); + WrapperHttpMessageCodec wrapperHttpMessageCodec = new WrapperHttpMessageCodec(url, frameworkModel); + wrapperHttpMessageCodec.setSerializeType(serializeName); + ProtobufHttpMessageCodec protobufHttpMessageCodec = new ProtobufHttpMessageCodec(); + return new GrpcCompositeCodec(protobufHttpMessageCodec, wrapperHttpMessageCodec); + } + + @Override + public MediaType contentType() { + return MEDIA_TYPE; + } + + @Override + public boolean support(String contentType) { + return contentType.startsWith(MEDIA_TYPE.getName()); + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHeaderNames.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHeaderNames.java new file mode 100644 index 0000000000..030cb53259 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHeaderNames.java @@ -0,0 +1,36 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.grpc; + +public enum GrpcHeaderNames { + + GRPC_STATUS("grpc-status"), + GRPC_MESSAGE("grpc-message"), + GRPC_ENCODING("grpc-encoding"), + GRPC_TIMEOUT("grpc-timeout"), + ; + + private final String name; + + GrpcHeaderNames(String name) { + this.name = name; + } + + public String getName() { + return name; + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListener.java new file mode 100644 index 0000000000..19f1680ef1 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListener.java @@ -0,0 +1,216 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.grpc; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.remoting.http12.HttpHeaders; +import org.apache.dubbo.remoting.http12.exception.DecodeException; +import org.apache.dubbo.remoting.http12.exception.UnimplementedException; +import org.apache.dubbo.remoting.http12.h2.H2StreamChannel; +import org.apache.dubbo.remoting.http12.h2.Http2Header; +import org.apache.dubbo.remoting.http12.h2.Http2TransportListener; +import org.apache.dubbo.remoting.http12.message.MethodMetadata; +import org.apache.dubbo.remoting.http12.message.StreamingDecoder; +import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.RpcInvocation; +import org.apache.dubbo.rpc.model.FrameworkModel; +import org.apache.dubbo.rpc.model.MethodDescriptor; +import org.apache.dubbo.rpc.model.ServiceDescriptor; +import org.apache.dubbo.rpc.protocol.tri.TripleCustomerProtocolWapper; +import org.apache.dubbo.rpc.protocol.tri.call.AbstractServerCall; +import org.apache.dubbo.rpc.protocol.tri.compressor.DeCompressor; +import org.apache.dubbo.rpc.protocol.tri.compressor.Identity; +import org.apache.dubbo.rpc.protocol.tri.h12.HttpMessageListener; +import org.apache.dubbo.rpc.protocol.tri.h12.http2.GenericHttp2ServerTransportListener; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; + +import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_PARSE; + +public class GrpcHttp2ServerTransportListener extends GenericHttp2ServerTransportListener implements Http2TransportListener { + + private static final ErrorTypeAwareLogger LOGGER = LoggerFactory.getErrorTypeAwareLogger(AbstractServerCall.class); + + public GrpcHttp2ServerTransportListener(H2StreamChannel h2StreamChannel, URL url, FrameworkModel frameworkModel) { + super(h2StreamChannel, url, frameworkModel); + initialize(); + } + + private void initialize() { + getServerChannelObserver().setTrailersCustomizer(this::grpcTrailersCustomize); + } + + private void grpcTrailersCustomize(HttpHeaders httpHeaders, Throwable throwable) { + httpHeaders.set(GrpcHeaderNames.GRPC_STATUS.getName(), "0"); + if (throwable != null) { + httpHeaders.set(GrpcHeaderNames.GRPC_MESSAGE.getName(), throwable.getMessage()); + } + } + + @Override + protected RpcInvocation buildRpcInvocation(Invoker invoker, ServiceDescriptor serviceDescriptor, MethodDescriptor methodDescriptor) { + RpcInvocation rpcInvocation = super.buildRpcInvocation(invoker, serviceDescriptor, methodDescriptor); + HttpHeaders headers = getHttpMetadata().headers(); + String timeoutString = headers.getFirst(GrpcHeaderNames.GRPC_TIMEOUT.getName()); + try { + if (Objects.nonNull(timeoutString)) { + Long timeout = GrpcUtils.parseTimeoutToMills(timeoutString); + rpcInvocation.put("timeout", timeout); + } + } catch (Throwable t) { + LOGGER.warn(PROTOCOL_FAILED_PARSE, "", "", String.format("Failed to parse request timeout set from:%s, service=%s " + + "method=%s", timeoutString, serviceDescriptor.getInterfaceName(), getMethodDescriptor().getMethodName())); + } + return rpcInvocation; + } + + @Override + protected StreamingDecoder newStreamingDecoder() { + return new GrpcStreamingDecoder(); + } + + @Override + protected HttpMessageListener newHttpMessageListener() { + Http2Header httpMetadata = getHttpMetadata(); + boolean hasStub = getPathResolver().hasNativeStub(httpMetadata.path()); + if (hasStub) { + return GrpcHttp2ServerTransportListener.super.newHttpMessageListener(); + } + return new LazyFindMethodListener(); + } + + @Override + protected void onMetadataCompletion(Http2Header metadata) { + super.onMetadataCompletion(metadata); + processGrpcHeaders(metadata); + } + + private void processGrpcHeaders(Http2Header metadata) { + String messageEncoding = metadata.headers().getFirst(GrpcHeaderNames.GRPC_ENCODING.getName()); + if (null != messageEncoding) { + if (!Identity.MESSAGE_ENCODING.equals(messageEncoding)) { + DeCompressor compressor = DeCompressor.getCompressor(getFrameworkModel(), + messageEncoding); + if (null == compressor) { + throw new UnimplementedException(GrpcHeaderNames.GRPC_ENCODING.getName() + " '" + messageEncoding + "'"); + } + getStreamingDecoder().setDeCompressor(compressor); + } + } + } + + @Override + protected GrpcStreamingDecoder getStreamingDecoder() { + return (GrpcStreamingDecoder) super.getStreamingDecoder(); + } + + private class LazyFindMethodListener implements HttpMessageListener { + + private final StreamingDecoder streamingDecoder; + + private LazyFindMethodListener() { + this.streamingDecoder = new GrpcStreamingDecoder(); + this.streamingDecoder.setFragmentListener(new DetermineMethodDescriptorListener()); + this.streamingDecoder.request(Integer.MAX_VALUE); + } + + @Override + public void onMessage(InputStream inputStream) { + streamingDecoder.decode(inputStream); + } + } + + private class DetermineMethodDescriptorListener implements StreamingDecoder.FragmentListener { + + @Override + public void onFragmentMessage(InputStream rawMessage) { + + } + + @Override + public void onClose() { + getStreamingDecoder().close(); + } + + @Override + public void onFragmentMessage(InputStream dataHeader, InputStream rawMessage) { + try { + ByteArrayOutputStream merge = new ByteArrayOutputStream(dataHeader.available() + rawMessage.available()); + transferToOutputStream(merge, dataHeader); + ByteArrayOutputStream bos = new ByteArrayOutputStream(rawMessage.available()); + transferToOutputStream(bos, rawMessage); + byte[] data = bos.toByteArray(); + MethodDescriptor methodDescriptor = getMethodDescriptor(); + if (methodDescriptor == null) { + Http2Header httpMetadata = getHttpMetadata(); + String path = httpMetadata.path(); + String[] parts = path.split("/"); + String originalMethodName = parts[2]; + methodDescriptor = findReflectionMethodDescriptor(getServiceDescriptor(), originalMethodName); + if (methodDescriptor == null) { + List methodDescriptors = getServiceDescriptor().getMethods(originalMethodName); + final TripleCustomerProtocolWapper.TripleRequestWrapper request; + request = TripleCustomerProtocolWapper.TripleRequestWrapper.parseFrom(data); + final String[] paramTypes = request.getArgTypes() + .toArray(new String[request.getArgs().size()]); + // wrapper mode the method can overload so maybe list + for (MethodDescriptor descriptor : methodDescriptors) { + // params type is array + if (Arrays.equals(descriptor.getCompatibleParamSignatures(), paramTypes)) { + methodDescriptor = descriptor; + break; + } + } + if (methodDescriptor == null) { + throw new UnimplementedException("method:" + originalMethodName); + } + } + setMethodDescriptor(methodDescriptor); + setMethodMetadata(MethodMetadata.fromMethodDescriptor(methodDescriptor)); + setRpcInvocation(buildRpcInvocation(getInvoker(), getServiceDescriptor(), methodDescriptor)); + //replace decoder + HttpMessageListener httpMessageListener = GrpcHttp2ServerTransportListener.super.newHttpMessageListener(); + GrpcCompositeCodec grpcCompositeCodec = (GrpcCompositeCodec) getHttpMessageCodec(); + grpcCompositeCodec.setEncodeTypes(new Class[]{getMethodMetadata().getActualResponseType()}); + grpcCompositeCodec.setDecodeTypes(getMethodMetadata().getActualRequestTypes()); + setHttpMessageListener(httpMessageListener); + } + transferToOutputStream(merge, new ByteArrayInputStream(data)); + getHttpMessageListener().onMessage(new ByteArrayInputStream(merge.toByteArray())); + } catch (IOException e) { + throw new DecodeException(e); + } + } + + private void transferToOutputStream(OutputStream out, InputStream inputStream) throws IOException { + byte[] bytes = new byte[1024]; + int len; + while ((len = inputStream.read(bytes)) != -1) { + out.write(bytes, 0, len); + } + } + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListenerFactory.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListenerFactory.java new file mode 100644 index 0000000000..cdcd42489e --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListenerFactory.java @@ -0,0 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.grpc; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.remoting.http12.h2.H2StreamChannel; +import org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory; +import org.apache.dubbo.remoting.http12.h2.Http2TransportListener; +import org.apache.dubbo.rpc.model.FrameworkModel; + +public class GrpcHttp2ServerTransportListenerFactory implements Http2ServerTransportListenerFactory { + + public static final String CONTENT_TYPE = "application/grpc"; + + @Override + public Http2TransportListener newInstance(H2StreamChannel streamChannel, URL url, FrameworkModel frameworkModel) { + return new GrpcHttp2ServerTransportListener(streamChannel, url, frameworkModel); + } + + @Override + public boolean supportContentType(String contentType) { + return contentType.startsWith(CONTENT_TYPE); + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcStreamingDecoder.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcStreamingDecoder.java new file mode 100644 index 0000000000..d0bc805afc --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcStreamingDecoder.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.grpc; + +import org.apache.dubbo.remoting.http12.message.LengthFieldStreamingDecoder; +import org.apache.dubbo.rpc.RpcException; +import org.apache.dubbo.rpc.protocol.tri.compressor.DeCompressor; + +import java.io.IOException; +import java.io.InputStream; + +public class GrpcStreamingDecoder extends LengthFieldStreamingDecoder { + + private static final int COMPRESSED_FLAG_MASK = 1; + private static final int RESERVED_MASK = 0xFE; + + private boolean compressedFlag; + + private DeCompressor deCompressor = DeCompressor.NONE; + + public GrpcStreamingDecoder() { + super(1, 4); + } + + public void setDeCompressor(DeCompressor deCompressor) { + this.deCompressor = deCompressor; + } + + @Override + protected void processOffset(InputStream inputStream, int lengthFieldOffset) throws IOException { + int type = inputStream.read(); + if ((type & RESERVED_MASK) != 0) { + throw new RpcException("gRPC frame header malformed: reserved bits not zero"); + } + compressedFlag = (type & COMPRESSED_FLAG_MASK) != 0; + } + + @Override + protected byte[] readRawMessage(InputStream inputStream, int length) throws IOException { + byte[] rawMessage = super.readRawMessage(inputStream, length); + return compressedFlag ? deCompressedMessage(rawMessage) : rawMessage; + } + + private byte[] deCompressedMessage(byte[] rawMessage) { + return deCompressor.decompress(rawMessage); + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcUtils.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcUtils.java new file mode 100644 index 0000000000..2689824a6f --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcUtils.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.grpc; + +import org.apache.dubbo.common.utils.StringUtils; + +import java.util.concurrent.TimeUnit; + +public class GrpcUtils { + + private GrpcUtils() { + + } + + public static Long parseTimeoutToMills(String timeoutVal) { + if (StringUtils.isEmpty(timeoutVal) || StringUtils.isContains(timeoutVal, "null")) { + return null; + } + long value = Long.parseLong(timeoutVal.substring(0, timeoutVal.length() - 1)); + char unit = timeoutVal.charAt(timeoutVal.length() - 1); + switch (unit) { + case 'n': + return TimeUnit.NANOSECONDS.toMillis(value); + case 'u': + return TimeUnit.MICROSECONDS.toMillis(value); + case 'm': + return value; + case 'S': + return TimeUnit.SECONDS.toMillis(value); + case 'M': + return TimeUnit.MINUTES.toMillis(value); + case 'H': + return TimeUnit.HOURS.toMillis(value); + default: + // invalid timeout config + return null; + } + } + +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/ProtobufHttpMessageCodec.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/ProtobufHttpMessageCodec.java new file mode 100644 index 0000000000..1d6fce91df --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/ProtobufHttpMessageCodec.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.grpc; + +import org.apache.dubbo.common.extension.Activate; +import org.apache.dubbo.remoting.http12.exception.DecodeException; +import org.apache.dubbo.remoting.http12.exception.EncodeException; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodec; +import org.apache.dubbo.remoting.http12.message.MediaType; +import org.apache.dubbo.rpc.protocol.tri.SingleProtobufUtils; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +@Activate(onClass = "com.google.protobuf.Message") +public class ProtobufHttpMessageCodec implements HttpMessageCodec { + + private static final MediaType MEDIA_TYPE = new MediaType("application", "x-protobuf"); + + @Override + public void encode(OutputStream outputStream, Object data) throws EncodeException { + try { + SingleProtobufUtils.serialize(data, outputStream); + } catch (IOException e) { + throw new EncodeException(e); + } + } + + @Override + public Object decode(InputStream inputStream, Class targetType) throws DecodeException { + try { + return SingleProtobufUtils.deserialize(inputStream, targetType); + } catch (IOException e) { + throw new DecodeException(e); + } + } + + @Override + public MediaType contentType() { + return MEDIA_TYPE; + } +} diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/DefaultHttp2StreamingDecoder.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/StreamingHttpMessageListener.java similarity index 50% rename from dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/DefaultHttp2StreamingDecoder.java rename to dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/StreamingHttpMessageListener.java index b43d2d1732..bbcac87d43 100644 --- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/DefaultHttp2StreamingDecoder.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/StreamingHttpMessageListener.java @@ -14,40 +14,26 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.dubbo.remoting.http12.h2; +package org.apache.dubbo.rpc.protocol.tri.h12.grpc; -import org.apache.dubbo.remoting.http12.exception.DecodeException; -import org.apache.dubbo.remoting.http12.message.DefaultListeningDecoder; -import org.apache.dubbo.remoting.http12.message.HttpMessageCodec; import org.apache.dubbo.remoting.http12.message.StreamingDecoder; +import org.apache.dubbo.rpc.protocol.tri.h12.HttpMessageListener; import java.io.InputStream; -public class DefaultHttp2StreamingDecoder implements StreamingDecoder { +public class StreamingHttpMessageListener implements HttpMessageListener { - private final DefaultListeningDecoder delegate; + private StreamingDecoder streamingDecoder; - public DefaultHttp2StreamingDecoder(HttpMessageCodec httpMessageCodec, Class[] targetTypes) { - this.delegate = new DefaultListeningDecoder(httpMessageCodec, targetTypes); + public StreamingHttpMessageListener() { + } + + public StreamingHttpMessageListener(StreamingDecoder streamingDecoder) { + this.streamingDecoder = streamingDecoder; } @Override - public void decode(InputStream inputStream) throws DecodeException { - delegate.decode(inputStream); - } - - @Override - public void close() { - delegate.close(); - } - - @Override - public void setListener(Listener listener) { - delegate.setListener(listener); - } - - @Override - public void request(int numMessages) { - //no op + public void onMessage(InputStream inputStream) { + streamingDecoder.decode(inputStream); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/WrapperHttpMessageCodec.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/WrapperHttpMessageCodec.java new file mode 100644 index 0000000000..21a7a428f6 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/WrapperHttpMessageCodec.java @@ -0,0 +1,143 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.grpc; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.constants.CommonConstants; +import org.apache.dubbo.common.serialize.MultipleSerialization; +import org.apache.dubbo.config.Constants; +import org.apache.dubbo.remoting.http12.exception.DecodeException; +import org.apache.dubbo.remoting.http12.exception.EncodeException; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodec; +import org.apache.dubbo.remoting.http12.message.MediaType; +import org.apache.dubbo.rpc.model.FrameworkModel; +import org.apache.dubbo.rpc.protocol.tri.TripleCustomerProtocolWapper; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +public class WrapperHttpMessageCodec implements HttpMessageCodec { + + private static final MediaType MEDIA_TYPE = new MediaType("application", "triple+wrapper"); + + private static final String DEFAULT_SERIALIZE_TYPE = "fastjson2"; + + private final MultipleSerialization serialization; + + private final URL url; + + private Class[] encodeTypes; + + private Class[] decodeTypes; + + private String serializeType = DEFAULT_SERIALIZE_TYPE; + + public WrapperHttpMessageCodec(URL url, FrameworkModel frameworkModel) { + this.url = url; + this.serialization = frameworkModel + .getExtensionLoader(MultipleSerialization.class) + .getExtension(url.getParameter(Constants.MULTI_SERIALIZATION_KEY, + CommonConstants.DEFAULT_KEY)); + } + + public void setSerializeType(String serializeType) { + this.serializeType = serializeType; + } + + public void setEncodeTypes(Class[] encodeTypes) { + this.encodeTypes = encodeTypes; + } + + public void setDecodeTypes(Class[] decodeTypes) { + this.decodeTypes = decodeTypes; + } + + @Override + public void encode(OutputStream outputStream, Object data) throws EncodeException { + try { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + serialization.serialize(url, serializeType, encodeTypes[0], data, bos); + byte[] encoded = TripleCustomerProtocolWapper.TripleResponseWrapper.Builder.newBuilder() + .setSerializeType(serializeType) + .setType(encodeTypes[0].getName()) + .setData(bos.toByteArray()) + .build() + .toByteArray(); + writeLength(outputStream, encoded.length); + outputStream.write(encoded); + } catch (IOException e) { + throw new EncodeException(e); + } + } + + @Override + public void encode(OutputStream outputStream, Object[] data) throws EncodeException { + //TODO + } + + @Override + public Object decode(InputStream inputStream, Class targetType) throws DecodeException { + Object[] decode = this.decode(inputStream, new Class[]{targetType}); + if (decode == null || decode.length == 0) { + return null; + } + return decode[0]; + } + + @Override + public Object[] decode(InputStream inputStream, Class[] targetTypes) throws DecodeException { + try { + int len; + byte[] data = new byte[4096]; + ByteArrayOutputStream bos = new ByteArrayOutputStream(4096); + while ((len = inputStream.read(data)) != -1) { + bos.write(data, 0, len); + } + TripleCustomerProtocolWapper.TripleRequestWrapper wrapper = TripleCustomerProtocolWapper.TripleRequestWrapper.parseFrom( + bos.toByteArray()); + setSerializeType(wrapper.getSerializeType()); + Object[] ret = new Object[wrapper.getArgs().size()]; + for (int i = 0; i < wrapper.getArgs().size(); i++) { + ByteArrayInputStream in = new ByteArrayInputStream( + wrapper.getArgs().get(i)); + try { + ret[i] = this.serialization.deserialize(url, wrapper.getSerializeType(), targetTypes[i], in); + } catch (ClassNotFoundException e) { + throw new DecodeException(e); + } + } + return ret; + } catch (IOException e) { + throw new DecodeException(e); + } + } + + @Override + public MediaType contentType() { + return MEDIA_TYPE; + } + + private static void writeLength(OutputStream outputStream, int length) throws IOException { + outputStream.write(((length >> 24) & 0xFF)); + outputStream.write(((length >> 16) & 0xFF)); + outputStream.write(((length >> 8) & 0xFF)); + outputStream.write((length & 0xFF)); + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListener.java new file mode 100644 index 0000000000..e8c880753e --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListener.java @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.http1; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.stream.StreamObserver; +import org.apache.dubbo.remoting.http12.HttpChannel; +import org.apache.dubbo.remoting.http12.HttpHeaderNames; +import org.apache.dubbo.remoting.http12.HttpInputMessage; +import org.apache.dubbo.remoting.http12.RequestMetadata; +import org.apache.dubbo.remoting.http12.h1.Http1ServerChannelObserver; +import org.apache.dubbo.remoting.http12.h1.Http1ServerStreamChannelObserver; +import org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListener; +import org.apache.dubbo.remoting.http12.message.DefaultListeningDecoder; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodec; +import org.apache.dubbo.remoting.http12.message.ListeningDecoder; +import org.apache.dubbo.remoting.http12.message.MediaType; +import org.apache.dubbo.remoting.http12.message.MethodMetadata; +import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.RpcInvocation; +import org.apache.dubbo.rpc.model.FrameworkModel; +import org.apache.dubbo.rpc.model.MethodDescriptor; +import org.apache.dubbo.rpc.protocol.tri.h12.AbstractServerTransportListener; +import org.apache.dubbo.rpc.protocol.tri.h12.DefaultHttpMessageListener; +import org.apache.dubbo.rpc.protocol.tri.h12.HttpMessageListener; +import org.apache.dubbo.rpc.protocol.tri.h12.ServerCallListener; +import org.apache.dubbo.rpc.protocol.tri.h12.ServerStreamServerCallListener; +import org.apache.dubbo.rpc.protocol.tri.h12.UnaryServerCallListener; + +public class DefaultHttp11ServerTransportListener extends AbstractServerTransportListener implements Http1ServerTransportListener { + + private final HttpChannel httpChannel; + + private final URL url; + + public DefaultHttp11ServerTransportListener(HttpChannel httpChannel, URL url, FrameworkModel frameworkModel) { + super(frameworkModel, url, httpChannel); + this.url = url; + this.httpChannel = httpChannel; + } + + private ServerCallListener startListener(RpcInvocation invocation, + MethodDescriptor methodDescriptor, + Invoker invoker) { + switch (methodDescriptor.getRpcType()) { + case UNARY: + Http1ServerChannelObserver http1ChannelObserver = new Http1ServerChannelObserver(httpChannel); + http1ChannelObserver.setHttpMessageCodec(getHttpMessageCodec()); + return new AutoCompleteUnaryServerCallListener(invocation, invoker, http1ChannelObserver); + case SERVER_STREAM: + Http1ServerChannelObserver serverStreamChannelObserver = new Http1ServerStreamChannelObserver(httpChannel); + serverStreamChannelObserver.setHttpMessageCodec(getHttpMessageCodec()); + serverStreamChannelObserver.setHeadersCustomizer((headers) -> headers.set(HttpHeaderNames.CONTENT_TYPE.getName(), MediaType.TEXT_EVENT_STREAM_VALUE.getName())); + return new AutoCompleteServerStreamServerCallListener(invocation, invoker, serverStreamChannelObserver); + default: + throw new UnsupportedOperationException("HTTP1.x only support unary and server-stream"); + } + } + + @Override + protected HttpMessageListener newHttpMessageListener() { + RequestMetadata httpMetadata = getHttpMetadata(); + String path = httpMetadata.path(); + String[] parts = path.split("/"); + String originalMethodName = parts[2]; + boolean hasStub = getPathResolver().hasNativeStub(path); + MethodDescriptor methodDescriptor = findMethodDescriptor(getServiceDescriptor(), originalMethodName, hasStub); + MethodMetadata methodMetadata = MethodMetadata.fromMethodDescriptor(methodDescriptor); + RpcInvocation rpcInvocation = buildRpcInvocation(getInvoker(), getServiceDescriptor(), methodDescriptor); + setRpcInvocation(rpcInvocation); + HttpMessageCodec httpMessageCodec = getHttpMessageCodec(); + ListeningDecoder listeningDecoder = newListeningDecoder(httpMessageCodec, methodMetadata.getActualRequestTypes()); + return new DefaultHttpMessageListener(listeningDecoder); + } + + private ListeningDecoder newListeningDecoder(HttpMessageCodec codec, Class[] actualRequestTypes) { + DefaultListeningDecoder defaultListeningDecoder = new DefaultListeningDecoder(codec, actualRequestTypes); + ServerCallListener serverCallListener = startListener(getRpcInvocation(), getMethodDescriptor(), getInvoker()); + defaultListeningDecoder.setListener(serverCallListener::onMessage); + return defaultListeningDecoder; + } + + private static class AutoCompleteUnaryServerCallListener extends UnaryServerCallListener { + + public AutoCompleteUnaryServerCallListener(RpcInvocation invocation, Invoker invoker, StreamObserver responseObserver) { + super(invocation, invoker, responseObserver); + } + + @Override + public void onMessage(Object message) { + super.onMessage(message); + super.onComplete(); + } + } + + private static class AutoCompleteServerStreamServerCallListener extends ServerStreamServerCallListener { + + public AutoCompleteServerStreamServerCallListener(RpcInvocation invocation, Invoker invoker, StreamObserver responseObserver) { + super(invocation, invoker, responseObserver); + } + + @Override + public void onMessage(Object message) { + super.onMessage(message); + super.onComplete(); + } + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListenerFactory.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListenerFactory.java new file mode 100644 index 0000000000..cddeb4182e --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListenerFactory.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.http1; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.remoting.http12.HttpChannel; +import org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListener; +import org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListenerFactory; +import org.apache.dubbo.rpc.model.FrameworkModel; + +public class DefaultHttp11ServerTransportListenerFactory implements Http1ServerTransportListenerFactory { + + public static final Http1ServerTransportListenerFactory INSTANCE = new DefaultHttp11ServerTransportListenerFactory(); + + @Override + public Http1ServerTransportListener newInstance(HttpChannel httpChannel, URL url, FrameworkModel frameworkModel) { + return new DefaultHttp11ServerTransportListener(httpChannel, url, frameworkModel); + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListener.java new file mode 100644 index 0000000000..8c325fab2a --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListener.java @@ -0,0 +1,223 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.http2; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.threadpool.manager.ExecutorRepository; +import org.apache.dubbo.common.threadpool.serial.SerializingExecutor; +import org.apache.dubbo.remoting.http12.RequestMetadata; +import org.apache.dubbo.remoting.http12.exception.HttpStatusException; +import org.apache.dubbo.remoting.http12.h2.H2StreamChannel; +import org.apache.dubbo.remoting.http12.h2.Http2Header; +import org.apache.dubbo.remoting.http12.h2.Http2InputMessage; +import org.apache.dubbo.remoting.http12.h2.Http2ServerChannelObserver; +import org.apache.dubbo.remoting.http12.h2.Http2TransportListener; +import org.apache.dubbo.remoting.http12.message.DefaultListeningDecoder; +import org.apache.dubbo.remoting.http12.message.JsonCodec; +import org.apache.dubbo.remoting.http12.message.LengthFieldStreamingDecoder; +import org.apache.dubbo.remoting.http12.message.ListeningDecoder; +import org.apache.dubbo.remoting.http12.message.MethodMetadata; +import org.apache.dubbo.remoting.http12.message.StreamingDecoder; +import org.apache.dubbo.rpc.CancellationContext; +import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.RpcContext; +import org.apache.dubbo.rpc.RpcInvocation; +import org.apache.dubbo.rpc.executor.ExecutorSupport; +import org.apache.dubbo.rpc.model.FrameworkModel; +import org.apache.dubbo.rpc.model.MethodDescriptor; +import org.apache.dubbo.rpc.protocol.tri.ReflectionPackableMethod; +import org.apache.dubbo.rpc.protocol.tri.h12.AbstractServerTransportListener; +import org.apache.dubbo.rpc.protocol.tri.h12.BiStreamServerCallListener; +import org.apache.dubbo.rpc.protocol.tri.h12.HttpMessageListener; +import org.apache.dubbo.rpc.protocol.tri.h12.ServerCallListener; +import org.apache.dubbo.rpc.protocol.tri.h12.ServerStreamServerCallListener; +import org.apache.dubbo.rpc.protocol.tri.h12.UnaryServerCallListener; +import org.apache.dubbo.rpc.protocol.tri.h12.grpc.StreamingHttpMessageListener; + +import java.util.concurrent.Executor; + +public class GenericHttp2ServerTransportListener extends AbstractServerTransportListener implements Http2TransportListener { + + private final Http2ServerChannelObserver serverChannelObserver; + + private final H2StreamChannel h2StreamChannel; + + private final ExecutorSupport executorSupport; + + private final StreamingDecoder streamingDecoder; + + private ServerCallListener serverCallListener; + + public GenericHttp2ServerTransportListener(H2StreamChannel h2StreamChannel, URL url, FrameworkModel frameworkModel) { + super(frameworkModel, url, h2StreamChannel); + this.h2StreamChannel = h2StreamChannel; + this.executorSupport = ExecutorRepository.getInstance(url.getOrDefaultApplicationModel()).getExecutorSupport(url); + this.streamingDecoder = newStreamingDecoder(); + this.serverChannelObserver = new Http2ServerCallToObserverAdapter(frameworkModel, h2StreamChannel); + this.serverChannelObserver.setHttpMessageCodec(JsonCodec.INSTANCE); + this.serverChannelObserver.setStreamingDecoder(streamingDecoder); + } + + @Override + protected Executor initializeExecutor(Http2Header metadata) { + Executor executor = executorSupport.getExecutor(metadata); + return new SerializingExecutor(executor); + } + + private ServerCallListener startListener(RpcInvocation invocation, + MethodDescriptor methodDescriptor, + Invoker invoker) { + Http2ServerChannelObserver responseObserver = getServerChannelObserver(); + CancellationContext cancellationContext = RpcContext.getCancellationContext(); + responseObserver.setCancellationContext(cancellationContext); + switch (methodDescriptor.getRpcType()) { + case UNARY: + Http2Header httpMetadata = getHttpMetadata(); + boolean hasStub = getPathResolver().hasNativeStub(httpMetadata.path()); + boolean applyCustomizeException = false; + if (!hasStub) { + applyCustomizeException = ReflectionPackableMethod.needWrap(methodDescriptor, getMethodMetadata().getActualRequestTypes(), getMethodMetadata().getActualResponseType()); + } + UnaryServerCallListener unaryServerCallListener = startUnary(invocation, invoker, responseObserver); + unaryServerCallListener.setApplyCustomizeException(applyCustomizeException); + return unaryServerCallListener; + case SERVER_STREAM: + return startServerStreaming(invocation, invoker, responseObserver); + case BI_STREAM: + case CLIENT_STREAM: + return startBiStreaming(invocation, invoker, responseObserver); + default: + throw new IllegalStateException("Can not reach here"); + } + } + + public Http2ServerChannelObserver getServerChannelObserver() { + return serverChannelObserver; + } + + @Override + public void cancelByRemote(long errorCode) { + this.serverChannelObserver.cancel(new HttpStatusException((int) errorCode)); + this.serverCallListener.onCancel(errorCode); + } + + protected StreamingDecoder newStreamingDecoder() { + //default lengthFieldLength = 4 + return new LengthFieldStreamingDecoder(); + } + + protected void doOnMetadata(Http2Header metadata) { + if (metadata.isEndStream()) { + return; + } + super.doOnMetadata(metadata); + } + + @Override + protected HttpMessageListener newHttpMessageListener() { + RequestMetadata httpMetadata = getHttpMetadata(); + String path = httpMetadata.path(); + String[] parts = path.split("/"); + String originalMethodName = parts[2]; + MethodDescriptor methodDescriptor = getMethodDescriptor(); + if (methodDescriptor == null) { + methodDescriptor = findMethodDescriptor(getServiceDescriptor(), originalMethodName, isHasStub()); + setMethodDescriptor(methodDescriptor); + } + MethodMetadata methodMetadata = getMethodMetadata(); + if (methodMetadata == null) { + methodMetadata = MethodMetadata.fromMethodDescriptor(getMethodDescriptor()); + setMethodMetadata(methodMetadata); + } + RpcInvocation rpcInvocation = getRpcInvocation(); + if (rpcInvocation == null) { + setRpcInvocation(buildRpcInvocation(getInvoker(), getServiceDescriptor(), methodDescriptor)); + } + initializeServerCallListener(); + DefaultListeningDecoder defaultListeningDecoder = new DefaultListeningDecoder(getHttpMessageCodec(), getMethodMetadata().getActualRequestTypes()); + defaultListeningDecoder.setListener(new Http2StreamingDecodeListener(serverCallListener)); + streamingDecoder.setFragmentListener(new StreamingDecoder.DefaultFragmentListener(defaultListeningDecoder)); + getServerChannelObserver().setStreamingDecoder(streamingDecoder); + return new StreamingHttpMessageListener(streamingDecoder); + } + + @Override + protected void onMetadataCompletion(Http2Header metadata) { + super.onMetadataCompletion(metadata); + this.serverChannelObserver.setHttpMessageCodec(getHttpMessageCodec()); + this.serverChannelObserver.request(1); + } + + @Override + protected void onDataCompletion(Http2InputMessage message) { + if (message.isEndStream()) { + serverCallListener.onComplete(); + } + } + + @Override + protected void onError(Throwable throwable) { + serverChannelObserver.onError(throwable); + } + + protected StreamingDecoder getStreamingDecoder() { + return this.streamingDecoder; + } + + private static class Http2StreamingDecodeListener implements ListeningDecoder.Listener { + + private final ServerCallListener serverCallListener; + + private Http2StreamingDecodeListener(ServerCallListener serverCallListener) { + this.serverCallListener = serverCallListener; + } + + @Override + public void onMessage(Object message) { + this.serverCallListener.onMessage(message); + } + + @Override + public void onClose() { + this.serverCallListener.onComplete(); + } + } + + private void initializeServerCallListener() { + if (serverCallListener == null) { + this.serverCallListener = startListener(getRpcInvocation(), getMethodDescriptor(), getInvoker()); + } + } + + private UnaryServerCallListener startUnary(RpcInvocation invocation, + Invoker invoker, + Http2ServerChannelObserver responseObserver) { + return new UnaryServerCallListener(invocation, invoker, responseObserver); + } + + private ServerStreamServerCallListener startServerStreaming(RpcInvocation invocation, Invoker invoker, Http2ServerChannelObserver responseObserver) { + return new ServerStreamServerCallListener(invocation, invoker, responseObserver); + } + + private BiStreamServerCallListener startBiStreaming(RpcInvocation invocation, Invoker invoker, Http2ServerChannelObserver responseObserver) { + return new BiStreamServerCallListener(invocation, invoker, responseObserver); + } + + protected ServerCallListener getServerCallListener() { + return serverCallListener; + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListenerFactory.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListenerFactory.java new file mode 100644 index 0000000000..f6e6841311 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListenerFactory.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.http2; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.remoting.http12.h2.H2StreamChannel; +import org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory; +import org.apache.dubbo.remoting.http12.h2.Http2TransportListener; +import org.apache.dubbo.rpc.model.FrameworkModel; + +public class GenericHttp2ServerTransportListenerFactory implements Http2ServerTransportListenerFactory { + + public static final Http2ServerTransportListenerFactory INSTANCE = new GenericHttp2ServerTransportListenerFactory(); + + @Override + public Http2TransportListener newInstance(H2StreamChannel streamChannel, + URL url, + FrameworkModel frameworkModel) { + return new GenericHttp2ServerTransportListener(streamChannel, url, frameworkModel); + } + + @Override + public boolean supportContentType(String contentType) { + return true; + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2ServerCallToObserverAdapter.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2ServerCallToObserverAdapter.java new file mode 100644 index 0000000000..932f94563b --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2ServerCallToObserverAdapter.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.http2; + +import org.apache.dubbo.remoting.http12.h2.H2StreamChannel; +import org.apache.dubbo.rpc.model.FrameworkModel; +import org.apache.dubbo.rpc.protocol.tri.h12.ServerCallToObserverAdapter; + +public class Http2ServerCallToObserverAdapter extends Http2ServerStreamObserver implements ServerCallToObserverAdapter { + + private int exceptionCode; + + public Http2ServerCallToObserverAdapter(FrameworkModel frameworkModel, H2StreamChannel h2StreamChannel) { + super(frameworkModel, h2StreamChannel); + setHeadersCustomizer((headers) -> { + if (exceptionCode != 0) { + headers.set("tri-exception-code", String.valueOf(exceptionCode)); + } + }); + } + + @Override + public void setExceptionCode(int exceptionCode) { + this.exceptionCode = exceptionCode; + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2ServerStreamObserver.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2ServerStreamObserver.java new file mode 100644 index 0000000000..03dab446f3 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2ServerStreamObserver.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.rpc.protocol.tri.h12.http2; + +import org.apache.dubbo.remoting.http12.HttpHeaders; +import org.apache.dubbo.remoting.http12.HttpMetadata; +import org.apache.dubbo.remoting.http12.h2.H2StreamChannel; +import org.apache.dubbo.remoting.http12.h2.Http2ServerChannelObserver; +import org.apache.dubbo.remoting.http12.message.HttpMessageCodec; +import org.apache.dubbo.rpc.model.FrameworkModel; +import org.apache.dubbo.rpc.protocol.tri.ServerStreamObserver; +import org.apache.dubbo.rpc.protocol.tri.TripleProtocol; +import org.apache.dubbo.rpc.protocol.tri.compressor.Compressor; +import org.apache.dubbo.rpc.protocol.tri.h12.AttachmentHolder; +import org.apache.dubbo.rpc.protocol.tri.h12.CompressibleCodec; +import org.apache.dubbo.rpc.protocol.tri.stream.StreamUtils; + +import java.util.Map; + +public class Http2ServerStreamObserver extends Http2ServerChannelObserver implements ServerStreamObserver, AttachmentHolder { + + private final FrameworkModel frameworkModel; + + private HttpMessageCodec httpMessageCodec; + + private Map attachments; + + public Http2ServerStreamObserver(FrameworkModel frameworkModel, H2StreamChannel h2StreamChannel) { + super(h2StreamChannel); + this.frameworkModel = frameworkModel; + } + + @Override + public void setCompression(String compression) { + CompressibleCodec compressibleCodec = new CompressibleCodec(httpMessageCodec); + compressibleCodec.setCompressor(Compressor.getCompressor(frameworkModel, compression)); + super.setHttpMessageCodec(compressibleCodec); + } + + @Override + public void setHttpMessageCodec(HttpMessageCodec httpMessageCodec) { + super.setHttpMessageCodec(httpMessageCodec); + this.httpMessageCodec = httpMessageCodec; + } + + @Override + public void setResponseAttachments(Map attachments) { + this.attachments = attachments; + } + + @Override + public Map getResponseAttachments() { + return this.attachments; + } + + @Override + protected HttpMetadata encodeTrailers(Throwable throwable) { + HttpMetadata httpMetadata = super.encodeTrailers(throwable); + HttpHeaders headers = httpMetadata.headers(); + StreamUtils.convertAttachment(headers, attachments, TripleProtocol.CONVERT_NO_LOWER_HEADER); + return httpMetadata; + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtils.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtils.java index 07def88303..8df63eab78 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtils.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtils.java @@ -17,16 +17,16 @@ package org.apache.dubbo.rpc.protocol.tri.stream; +import io.netty.handler.codec.http2.DefaultHttp2Headers; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.utils.JsonUtils; import org.apache.dubbo.common.utils.LRU2Cache; +import org.apache.dubbo.remoting.http12.HttpHeaders; import org.apache.dubbo.rpc.TriRpcStatus; import org.apache.dubbo.rpc.protocol.tri.TripleConstant; import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum; -import io.netty.handler.codec.http2.DefaultHttp2Headers; - import java.nio.charset.StandardCharsets; import java.util.Base64; import java.util.Collections; @@ -115,6 +115,38 @@ public class StreamUtils { } } + public static void convertAttachment(HttpHeaders headers, + Map attachments, + boolean needConvertHeaderKey) { + if (attachments == null) { + return; + } + Map needConvertKey = new HashMap<>(); + for (Map.Entry entry : attachments.entrySet()) { + String key = lruHeaderMap.get(entry.getKey()); + if (key == null) { + final String lowerCaseKey = entry.getKey().toLowerCase(Locale.ROOT); + lruHeaderMap.put(entry.getKey(), lowerCaseKey); + key = lowerCaseKey; + } + if (TripleHeaderEnum.containsExcludeAttachments(key)) { + continue; + } + final Object v = entry.getValue(); + if (v == null) { + continue; + } + if (needConvertHeaderKey && !key.equals(entry.getKey())) { + needConvertKey.put(key, entry.getKey()); + } + convertSingleAttachment(headers, key, v); + } + if (!needConvertKey.isEmpty()) { + String needConvertJson = JsonUtils.toJson(needConvertKey); + headers.set(TripleHeaderEnum.TRI_HEADER_CONVERT.getHeader(), TriRpcStatus.encodeMessage(needConvertJson)); + } + } + public static void convertAttachment(DefaultHttp2Headers headers, Map attachments) { @@ -145,5 +177,22 @@ public class StreamUtils { } } + private static void convertSingleAttachment(HttpHeaders headers, String key, Object v) { + try { + if (v instanceof String || v instanceof Number || v instanceof Boolean) { + String str = v.toString(); + headers.set(key, str); + } else if (v instanceof byte[]) { + String str = encodeBase64ASCII((byte[]) v); + headers.set(key + TripleConstant.HEADER_BIN_SUFFIX, str); + } else { + LOGGER.warn(PROTOCOL_UNSUPPORTED, "", "", "Unsupported attachment k: " + key + " class: " + v.getClass().getName()); + } + } catch (Throwable t) { + LOGGER.warn(PROTOCOL_UNSUPPORTED, "", "", "Meet exception when convert single attachment key:" + key + " value=" + v, + t); + } + } + } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleIsolationExecutorSupport.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleIsolationExecutorSupport.java index a2f08ee9f5..a00de30f5a 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleIsolationExecutorSupport.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleIsolationExecutorSupport.java @@ -20,10 +20,11 @@ import org.apache.dubbo.common.ServiceKey; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.remoting.http12.HttpHeaders; +import org.apache.dubbo.remoting.http12.h2.Http2Header; import org.apache.dubbo.rpc.executor.AbstractIsolationExecutorSupport; import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum; -import io.netty.handler.codec.http2.Http2Headers; public class TripleIsolationExecutorSupport extends AbstractIsolationExecutorSupport { private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(TripleIsolationExecutorSupport.class); @@ -34,18 +35,19 @@ public class TripleIsolationExecutorSupport extends AbstractIsolationExecutorSup @Override protected ServiceKey getServiceKey(Object data) { - if (!(data instanceof Http2Headers)) { + if (!(data instanceof Http2Header)) { return null; } - Http2Headers headers = (Http2Headers) data; - String path = headers.path().toString(); + Http2Header http2Metadata = (Http2Header) data; + HttpHeaders headers = http2Metadata.headers(); + String path = http2Metadata.path(); String[] parts = path.split("/"); // path like /{interfaceName}/{methodName} String interfaceName = parts[1]; - String version = headers.contains(TripleHeaderEnum.SERVICE_VERSION.getHeader()) ? - headers.get(TripleHeaderEnum.SERVICE_VERSION.getHeader()).toString() : null; - String group = headers.contains(TripleHeaderEnum.SERVICE_GROUP.getHeader()) ? - headers.get(TripleHeaderEnum.SERVICE_GROUP.getHeader()).toString() : null; + String version = headers.containsKey(TripleHeaderEnum.SERVICE_VERSION.getHeader()) ? + headers.getFirst(TripleHeaderEnum.SERVICE_VERSION.getHeader()) : null; + String group = headers.containsKey(TripleHeaderEnum.SERVICE_GROUP.getHeader()) ? + headers.getFirst(TripleHeaderEnum.SERVICE_GROUP.getHeader()) : null; return new ServiceKey(interfaceName, version, group); } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleServerConnectionHandler.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleServerConnectionHandler.java index 9703cb287b..5750470384 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleServerConnectionHandler.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleServerConnectionHandler.java @@ -17,17 +17,16 @@ package org.apache.dubbo.rpc.protocol.tri.transport; -import io.netty.handler.codec.http2.DefaultHttp2ResetFrame; -import io.netty.handler.codec.http2.Http2Error; -import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; -import org.apache.dubbo.common.logger.LoggerFactory; - import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; +import io.netty.handler.codec.http2.DefaultHttp2ResetFrame; import io.netty.handler.codec.http2.Http2ChannelDuplexHandler; +import io.netty.handler.codec.http2.Http2Error; import io.netty.handler.codec.http2.Http2GoAwayFrame; import io.netty.handler.codec.http2.Http2PingFrame; import io.netty.util.ReferenceCountUtil; +import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; +import org.apache.dubbo.common.logger.LoggerFactory; import java.io.IOException; import java.net.SocketException; @@ -74,6 +73,10 @@ public class TripleServerConnectionHandler extends Http2ChannelDuplexHandler { super.channelInactive(ctx); //reset all active stream on connection close forEachActiveStream(stream -> { + //ignore remote side close + if (!stream.state().remoteSideOpen()) { + return true; + } DefaultHttp2ResetFrame resetFrame = new DefaultHttp2ResetFrame(Http2Error.NO_ERROR).stream(stream); ctx.fireChannelRead(resetFrame); return true; diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory new file mode 100644 index 0000000000..1ea7a85f72 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory @@ -0,0 +1 @@ +grpc=org.apache.dubbo.rpc.protocol.tri.h12.grpc.GrpcHttp2ServerTransportListenerFactory diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodec b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodec new file mode 100644 index 0000000000..388cbd7e3e --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodec @@ -0,0 +1,2 @@ +grpc=org.apache.dubbo.rpc.protocol.tri.h12.grpc.GrpcCompositeCodec +protobuf=org.apache.dubbo.rpc.protocol.tri.h12.grpc.ProtobufHttpMessageCodec diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory new file mode 100644 index 0000000000..39350a2c21 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory @@ -0,0 +1 @@ +grpc=org.apache.dubbo.rpc.protocol.tri.h12.grpc.GrpcCompositeCodecFactory diff --git a/dubbo-test/dubbo-dependencies-all/pom.xml b/dubbo-test/dubbo-dependencies-all/pom.xml index 2a5f1a3e1d..84ce1ca939 100644 --- a/dubbo-test/dubbo-dependencies-all/pom.xml +++ b/dubbo-test/dubbo-dependencies-all/pom.xml @@ -411,6 +411,11 @@ dubbo-remoting-http ${project.version} + + org.apache.dubbo + dubbo-remoting-http12 + ${project.version} + org.apache.dubbo dubbo-remoting-netty