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

[feat.] new triple protocol implement (#12873)

This commit is contained in:
icodening 2023-09-01 09:43:04 +08:00 committed by GitHub
parent 8e55dae8f9
commit 059b75c533
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
57 changed files with 2511 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.artifacts
View File

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

12
dubbo-distribution/dubbo-all/pom.xml
View File

@ -1261,12 +1261,24 @@
META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodec
</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>
META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory
</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>
META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory
</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>
META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListenerFactory
</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>

5
dubbo-distribution/dubbo-bom/pom.xml
View File

@ -501,6 +501,11 @@
<artifactId>dubbo-remoting-http</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-remoting-http12</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-remoting-netty</artifactId>

3
dubbo-plugin/dubbo-reactive/src/main/java/org/apache/dubbo/reactive/calls/ReactorServerCalls.java
View File

@ -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<R> response = func.apply(Mono.just(request));
ServerTripleReactorSubscriber<R> subscriber = response.subscribeWith(new ServerTripleReactorSubscriber<>());
subscriber.subscribe((ServerCallToObserverAdapter<R>) responseObserver);
subscriber.subscribe((CallStreamObserver<R>) responseObserver);
} catch (Throwable throwable) {
responseObserver.onError(throwable);
}

4
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java
View File

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

3
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1ServerTransportListener.java
View File

@ -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<RequestMetadata, HttpInputMessage>, HttpChannelHolder {
public interface Http1ServerTransportListener extends HttpTransportListener<RequestMetadata, HttpInputMessage> {
}

4
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2TransportListener.java
View File

@ -16,8 +16,6 @@
*/
package org.apache.dubbo.remoting.http12.h2;
import org.apache.dubbo.remoting.http12.HttpChannelHolder;
public interface Http2TransportListener extends CancelableTransportListener<Http2Header, Http2InputMessage>, HttpChannelHolder {
public interface Http2TransportListener extends CancelableTransportListener<Http2Header, Http2InputMessage> {
}

4
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodec.java
View File

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

39
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodecFactory.java Normal file
View File

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

35
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/JsonCodecFactory.java Normal file
View File

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

45
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java
View File

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

49
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/StreamingDecoder.java
View File

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

16
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h1/NettyHttp1ConnectionHandler.java
View File

@ -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<Htt
if (!StringUtils.hasText(contentType)) {
throw new UnsupportedMediaTypeException(contentType);
}
HttpMessageCodec codec = findSuitableCodec(contentType, frameworkModel.getExtensionLoader(HttpMessageCodec.class).getActivateExtensions());
if (codec == null) {
HttpMessageCodecFactory codecFactory = findSuitableCodec(contentType, frameworkModel.getExtensionLoader(HttpMessageCodecFactory.class).getActivateExtensions());
if (codecFactory == null) {
throw new UnsupportedMediaTypeException(contentType);
}
this.errorResponseObserver = new Http1ServerChannelObserver(new NettyHttp1Channel(ctx.channel()));
this.errorResponseObserver.setHttpMessageCodec(codec);
this.errorResponseObserver.setHttpMessageCodec(codecFactory.createCodec(url, frameworkModel));
return http1TransportListener;
}
private static HttpMessageCodec findSuitableCodec(String contentType, List<HttpMessageCodec> candidates) {
for (HttpMessageCodec codec : candidates) {
if (codec.support(contentType)) {
return codec;
private static HttpMessageCodecFactory findSuitableCodec(String contentType, List<HttpMessageCodecFactory> candidates) {
for (HttpMessageCodecFactory factory : candidates) {
if (factory.support(contentType)) {
return factory;
}
}
return null;

1
dubbo-remoting/dubbo-remoting-http12/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory Normal file
View File

@ -0,0 +1 @@
json=org.apache.dubbo.remoting.http12.message.JsonCodecFactory

5
dubbo-rpc/dubbo-rpc-triple/pom.xml
View File

@ -37,6 +37,11 @@
<artifactId>dubbo-rpc-api</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-remoting-http12</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-remoting-netty4</artifactId>

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

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

113
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
View File

@ -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<ChannelHandler> 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<ChannelHandler> 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<ChannelHandler> 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<ChannelHandler> handlers) {
Configuration config = ConfigurationUtils.getGlobalConfiguration(url.getOrDefaultApplicationModel());
final List<HeaderFilter> 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<Http2StreamChannel>() {
@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<ChannelHandler> 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<ChannelHandler> handlers = new ArrayList<>();
handlers.add(new ChannelHandlerPretender(codec));
handlers.add(new ChannelHandlerPretender(handler));
handlers.add(new ChannelHandlerPretender(new TripleTailHandler()));
operator.configChannelHandler(handlers);
}
}

14
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Bzip2.java
View File

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

4
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Compressor.java
View File

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

11
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/compressor/Gzip.java
View File

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

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

@ -17,6 +17,8 @@
package org.apache.dubbo.rpc.protocol.tri.compressor;
import java.io.OutputStream;
/**
* Default compressor
* <p>
@ -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;

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

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

27
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerCallListener.java → dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerCallListener.java
View File

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

479
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerTransportListener.java Normal file
View File

@ -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<HEADER extends RequestMetadata, MESSAGE extends HttpInputMessage> implements HttpTransportListener<HEADER, MESSAGE> {
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<HeaderFilter> 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<String, String> headers = getHttpMetadata().headers().toSingleValueMap();
Map<String, Object> 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<MethodDescriptor> 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<Response>)
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<String, Object> headersToMap(Map<String, String> headers, Supplier<Object> convertUpperHeaderSupplier) {
if (headers == null) {
return Collections.emptyMap();
}
Map<String, Object> attachments = new HashMap<>(headers.size());
for (Map.Entry<String, String> 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<String, String> map = JsonUtils.toJavaObject(json, Map.class);
for (Map.Entry<String, String> 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;
}
}

26
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AttachmentHolder.java Normal file
View File

@ -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<String, Object> attachments);
Map<String, Object> getResponseAttachments();
}

3
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/BiStreamServerCallListener.java → dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/BiStreamServerCallListener.java
View File

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

71
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/CompressibleCodec.java Normal file
View File

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

43
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/DefaultHttpMessageListener.java Normal file
View File

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

24
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/HttpMessageListener.java Normal file
View File

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

2
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ServerCallListener.java → dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerCallListener.java
View File

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

25
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerCallToObserverAdapter.java Normal file
View File

@ -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<T> extends ServerStreamObserver<T> {
void setExceptionCode(int exceptionCode);
}

2
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ServerStreamServerCallListener.java → dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerStreamServerCallListener.java
View File

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

87
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/TripleProtocolDetector.java Normal file
View File

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

28
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/UnaryServerCallListener.java → dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/UnaryServerCallListener.java
View File

@ -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<Object> 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<Object>) responseObserver).setExceptionCode(exceptionCode);
}
onReturn(t);
} else {
super.onResponseException(t);
}
}
@Override
public void onCancel(long code) {
//ignore

124
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java Normal file
View File

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

50
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodecFactory.java Normal file
View File

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

36
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHeaderNames.java Normal file
View File

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

216
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListener.java Normal file
View File

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

38
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListenerFactory.java Normal file
View File

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

61
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcStreamingDecoder.java Normal file
View File

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

54
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcUtils.java Normal file
View File

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

57
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/ProtobufHttpMessageCodec.java Normal file
View File

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

36
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
View File

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

143
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/WrapperHttpMessageCodec.java Normal file
View File

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

122
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListener.java Normal file
View File

@ -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<RequestMetadata, HttpInputMessage> 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<Object> 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<Object> responseObserver) {
super(invocation, invoker, responseObserver);
}
@Override
public void onMessage(Object message) {
super.onMessage(message);
super.onComplete();
}
}
}

33
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListenerFactory.java Normal file
View File

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

223
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListener.java Normal file
View File

@ -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<Http2Header, Http2InputMessage> 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;
}
}

40
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListenerFactory.java Normal file
View File

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

40
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2ServerCallToObserverAdapter.java Normal file
View File

@ -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<Object> {
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;
}
}

77
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2ServerStreamObserver.java Normal file
View File

@ -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<Object>, AttachmentHolder {
private final FrameworkModel frameworkModel;
private HttpMessageCodec httpMessageCodec;
private Map<String, Object> 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<String, Object> attachments) {
this.attachments = attachments;
}
@Override
public Map<String, Object> 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;
}
}

53
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtils.java
View File

@ -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<String, Object> attachments,
boolean needConvertHeaderKey) {
if (attachments == null) {
return;
}
Map<String, String> needConvertKey = new HashMap<>();
for (Map.Entry<String, Object> 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<String, Object> 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);
}
}
}

18
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleIsolationExecutorSupport.java
View File

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

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

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

1
dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory Normal file
View File

@ -0,0 +1 @@
grpc=org.apache.dubbo.rpc.protocol.tri.h12.grpc.GrpcHttp2ServerTransportListenerFactory

2
dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodec Normal file
View File

@ -0,0 +1,2 @@
grpc=org.apache.dubbo.rpc.protocol.tri.h12.grpc.GrpcCompositeCodec
protobuf=org.apache.dubbo.rpc.protocol.tri.h12.grpc.ProtobufHttpMessageCodec

1
dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory Normal file
View File

@ -0,0 +1 @@
grpc=org.apache.dubbo.rpc.protocol.tri.h12.grpc.GrpcCompositeCodecFactory

5
dubbo-test/dubbo-dependencies-all/pom.xml
View File

@ -411,6 +411,11 @@
<artifactId>dubbo-remoting-http</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-remoting-http12</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-remoting-netty</artifactId>