feature: support grpc protocol

common/src/main/java/org/apache/seata/common/ConfigurationKeys.java

@@ -628,6 +628,8 @@ public interface ConfigurationKeys {
     @Deprecated
     String ENABLE_CLIENT_BATCH_SEND_REQUEST = TRANSPORT_PREFIX + "enableClientBatchSendRequest";
 
+    String PROTOCOL = TRANSPORT_PREFIX + "protocol";
+
     /**
      * The constant ENABLE_TM_CLIENT_BATCH_SEND_REQUEST
      */

common/src/main/java/org/apache/seata/common/DefaultValues.java

@@ -63,6 +63,7 @@ public interface DefaultValues {
     String DEFAULT_BOSS_THREAD_PREFIX = "NettyBoss";
     String DEFAULT_NIO_WORKER_THREAD_PREFIX = "NettyServerNIOWorker";
     String DEFAULT_EXECUTOR_THREAD_PREFIX = "NettyServerBizHandler";
+    String DEFAULT_PROTOCOL = "seata";
 
     boolean DEFAULT_TRANSPORT_HEARTBEAT = true;
     boolean DEFAULT_TRANSACTION_UNDO_DATA_VALIDATION = true;

core/pom.xml

@@ -69,9 +69,20 @@
             <artifactId>fastjson</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+        </dependency>
     </dependencies>
 
     <build>
+        <extensions>
+            <extension>
+                <groupId>kr.motd.maven</groupId>
+                <artifactId>os-maven-plugin</artifactId>
+                <version>1.5.0.Final</version>
+            </extension>
+        </extensions>
         <plugins>
             <plugin>
                 <groupId>icu.easyj.maven.plugins</groupId>
@@ -90,6 +101,28 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <groupId>org.xolstice.maven.plugins</groupId>
+                <artifactId>protobuf-maven-plugin</artifactId>
+                <version>0.6.1</version>
+                <configuration>
+                    <protocArtifact>
+                        com.google.protobuf:protoc:3.1.0:exe:${os.detected.classifier}
+                    </protocArtifact>
+                    <pluginId>grpc-java</pluginId>
+                    <pluginArtifact>
+                        io.grpc:protoc-gen-grpc-java:1.11.0:exe:${os.detected.classifier}
+                    </pluginArtifact>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>compile</goal>
+                            <goal>compile-custom</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 </project>

core/src/main/java/org/apache/seata/core/protocol/Protocol.java

@@ -0,0 +1,17 @@
+package org.apache.seata.core.protocol;
+
+/**
+ *
+ */
+public enum Protocol {
+
+    GPRC("grpc"),
+
+    SEATA("seata");
+
+    public final String value;
+
+    Protocol(String value) {
+        this.value = value;
+    }
+}

core/src/main/java/org/apache/seata/core/protocol/detector/Http2Detector.java

@@ -0,0 +1,51 @@
+package org.apache.seata.core.protocol.detector;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
+import io.netty.handler.codec.http2.Http2MultiplexHandler;
+import io.netty.handler.codec.http2.Http2StreamChannel;
+import io.netty.util.CharsetUtil;
+import org.apache.seata.core.protocol.detector.ProtocolDetector;
+import org.apache.seata.core.rpc.netty.grpc.GrpcDecoder;
+import org.apache.seata.core.rpc.netty.grpc.GrpcEncoder;
+
+public class Http2Detector implements ProtocolDetector {
+    private final byte[] HTTP2_PREFIX_BYTES = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n".getBytes(CharsetUtil.UTF_8);
+    private ChannelHandler[] serverHandlers;
+
+    public Http2Detector(ChannelHandler[] serverHandlers) {
+        this.serverHandlers = serverHandlers;
+    }
+
+    @Override
+    public boolean detect(ByteBuf in) {
+        if (in.readableBytes() < HTTP2_PREFIX_BYTES.length) {
+            return false;
+        }
+        for (int i = 0; i < HTTP2_PREFIX_BYTES.length; i++) {
+            if (in.getByte(i) != HTTP2_PREFIX_BYTES[i]) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public ChannelHandler[] getHandlers() {
+        return new ChannelHandler[]{
+                Http2FrameCodecBuilder.forServer().build(),
+                new Http2MultiplexHandler(new ChannelInitializer<Http2StreamChannel>() {
+                    @Override
+                    protected void initChannel(Http2StreamChannel ch) {
+                        final ChannelPipeline p = ch.pipeline();
+                        p.addLast(new GrpcDecoder());
+                        p.addLast(new GrpcEncoder());
+                        p.addLast(serverHandlers);
+                    }
+                })
+        };
+    }
+}

core/src/main/java/org/apache/seata/core/protocol/detector/ProtocolDetector.java

@@ -0,0 +1,10 @@
+package org.apache.seata.core.protocol.detector;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+
+public interface ProtocolDetector {
+    boolean detect(ByteBuf in);
+
+    ChannelHandler[] getHandlers();
+}

core/src/main/java/org/apache/seata/core/protocol/detector/SeataDetector.java

@@ -0,0 +1,34 @@
+package org.apache.seata.core.protocol.detector;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+import org.apache.seata.core.rpc.netty.MultiProtocolDecoder;
+
+public class SeataDetector implements ProtocolDetector {
+    private final byte[] MAGIC_CODE_BYTES = {(byte) 0xda, (byte) 0xda};
+    private ChannelHandler[] serverHandlers;
+
+    public SeataDetector(ChannelHandler[] serverHandlers) {
+        this.serverHandlers = serverHandlers;
+    }
+
+    @Override
+    public boolean detect(ByteBuf in) {
+        if (in.readableBytes() < MAGIC_CODE_BYTES.length) {
+            return false;
+        }
+        for (int i = 0; i < MAGIC_CODE_BYTES.length; i++) {
+            if (in.getByte(i) != MAGIC_CODE_BYTES[i]) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public ChannelHandler[] getHandlers() {
+        MultiProtocolDecoder multiProtocolDecoder = new MultiProtocolDecoder(serverHandlers);
+
+        return new ChannelHandler[]{multiProtocolDecoder};
+    }
+}

core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientBootstrap.java

@@ -18,8 +18,11 @@
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelPipeline;
@@ -28,13 +31,19 @@
 import io.netty.channel.epoll.EpollMode;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
+import io.netty.handler.codec.http2.Http2MultiplexHandler;
+import io.netty.handler.codec.http2.Http2StreamChannelBootstrap;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.DefaultEventExecutorGroup;
 import io.netty.util.concurrent.EventExecutorGroup;
 import io.netty.util.internal.PlatformDependent;
 import org.apache.seata.common.exception.FrameworkException;
 import org.apache.seata.common.thread.NamedThreadFactory;
+import org.apache.seata.core.protocol.Protocol;
 import org.apache.seata.core.rpc.RemotingBootstrap;
+import org.apache.seata.core.rpc.netty.grpc.GrpcDecoder;
+import org.apache.seata.core.rpc.netty.grpc.GrpcEncoder;
 import org.apache.seata.core.rpc.netty.v1.ProtocolDecoderV1;
 import org.apache.seata.core.rpc.netty.v1.ProtocolEncoderV1;
 import org.slf4j.Logger;
@@ -130,12 +139,19 @@ public void start() {
                 @Override
                 public void initChannel(SocketChannel ch) {
                     ChannelPipeline pipeline = ch.pipeline();
-                    pipeline
-                        .addLast(new IdleStateHandler(nettyClientConfig.getChannelMaxReadIdleSeconds(),
+                    pipeline.addLast(new IdleStateHandler(nettyClientConfig.getChannelMaxReadIdleSeconds(),
                             nettyClientConfig.getChannelMaxWriteIdleSeconds(),
-                            nettyClientConfig.getChannelMaxAllIdleSeconds()))
-                        .addLast(new ProtocolDecoderV1())
-                        .addLast(new ProtocolEncoderV1());
+                            nettyClientConfig.getChannelMaxAllIdleSeconds()));
+                    if (nettyClientConfig.getProtocol().equals(Protocol.GPRC.value)) {
+                        pipeline.addLast(Http2FrameCodecBuilder.forClient().build())
+                                .addLast(new Http2MultiplexHandler(new ChannelDuplexHandler()))
+                                .addLast(new GrpcDecoder())
+                                .addLast(new GrpcEncoder());
+                    } else {
+                        pipeline.addLast(new ProtocolDecoderV1())
+                                .addLast(new ProtocolEncoderV1());
+                    }
+
                     if (channelHandlers != null) {
                         addChannelPipelineLast(ch, channelHandlers);
                     }
@@ -177,9 +193,28 @@ public Channel getNewChannel(InetSocketAddress address) {
             } else {
                 channel = f.channel();
             }
+
+            // TODO tmp only for grpc
+            if (nettyClientConfig.getProtocol().equals(Protocol.GPRC.value)) {
+                Http2StreamChannelBootstrap bootstrap = new Http2StreamChannelBootstrap(channel);
+                bootstrap.handler(new ChannelInboundHandlerAdapter() {
+                    @Override
+                    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+                        Channel channel = ctx.channel();
+                        channel.pipeline().addLast(new GrpcDecoder());
+                        channel.pipeline().addLast(new GrpcEncoder());
+                        if (channelHandlers != null) {
+                            addChannelPipelineLast(channel, channelHandlers);
+                        }
+                    }
+                });
+                channel = bootstrap.open().get();
+            }
+
         } catch (Exception e) {
             throw new FrameworkException(e, "can not connect to services-server.");
         }
+
         return channel;
     }
 

core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientConfig.java

@@ -21,6 +21,7 @@
 import org.apache.seata.core.rpc.TransportServerType;
 
 import static org.apache.seata.common.DefaultValues.DEFAULT_ENABLE_CLIENT_BATCH_SEND_REQUEST;
+import static org.apache.seata.common.DefaultValues.DEFAULT_PROTOCOL;
 import static org.apache.seata.common.DefaultValues.DEFAULT_RPC_RM_REQUEST_TIMEOUT;
 import static org.apache.seata.common.DefaultValues.DEFAULT_RPC_TM_REQUEST_TIMEOUT;
 import static org.apache.seata.common.DefaultValues.DEFAULT_SELECTOR_THREAD_PREFIX;
@@ -451,6 +452,10 @@ public String getRmDispatchThreadPrefix() {
         return RPC_DISPATCH_THREAD_PREFIX + "_" + NettyPoolKey.TransactionRole.RMROLE.name();
     }
 
+    public String getProtocol() {
+        return CONFIG.getConfig(org.apache.seata.common.ConfigurationKeys.PROTOCOL, DEFAULT_PROTOCOL);
+    }
+
     @Deprecated
     public static boolean isEnableClientBatchSendRequest() {
         return ENABLE_CLIENT_BATCH_SEND_REQUEST;

core/src/main/java/org/apache/seata/core/rpc/netty/NettyServerBootstrap.java

@@ -92,6 +92,10 @@ protected void setChannelHandlers(final ChannelHandler... handlers) {
         }
     }
 
+    protected ChannelHandler[] getChannelHandlers() {
+        return channelHandlers;
+    }
+
     /**
      * Add channel pipeline last.
      *
@@ -158,10 +162,8 @@ public void start() {
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) {
-                    MultiProtocolDecoder multiProtocolDecoder = new MultiProtocolDecoder(channelHandlers);
-                    ch.pipeline()
-                        .addLast(new IdleStateHandler(nettyServerConfig.getChannelMaxReadIdleSeconds(), 0, 0))
-                        .addLast(multiProtocolDecoder);
+                    ch.pipeline().addLast(new IdleStateHandler(nettyServerConfig.getChannelMaxReadIdleSeconds(), 0, 0))
+                            .addLast(new ProtocolDetectHandler(NettyServerBootstrap.this));
                 }
             });
 

core/src/main/java/org/apache/seata/core/rpc/netty/ProtocolDetectHandler.java

@@ -0,0 +1,46 @@
+package org.apache.seata.core.rpc.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import org.apache.seata.core.protocol.detector.Http2Detector;
+import org.apache.seata.core.protocol.detector.ProtocolDetector;
+import org.apache.seata.core.protocol.detector.SeataDetector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+public class ProtocolDetectHandler extends ByteToMessageDecoder {
+    private static final Logger LOGGER = LoggerFactory.getLogger(ProtocolDetectHandler.class);
+    private NettyServerBootstrap nettyServerBootstrap;
+    private ProtocolDetector[] supportedProtocolDetectors;
+
+    public ProtocolDetectHandler(NettyServerBootstrap nettyServerBootstrap) {
+        this.nettyServerBootstrap = nettyServerBootstrap;
+        this.supportedProtocolDetectors = new ProtocolDetector[]{new Http2Detector(nettyServerBootstrap.getChannelHandlers()), new SeataDetector(nettyServerBootstrap.getChannelHandlers())};
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+        for (ProtocolDetector protocolDetector : supportedProtocolDetectors) {
+            if (protocolDetector.detect(in)) {
+                ChannelHandler[] protocolHandlers = protocolDetector.getHandlers();
+                ctx.pipeline().addLast(protocolHandlers);
+                ctx.pipeline().remove(this);
+
+                in.resetReaderIndex();
+                return;
+            }
+
+            in.resetReaderIndex();
+        }
+
+        byte[] preface = new byte[in.readableBytes()];
+        in.readBytes(preface);
+        LOGGER.error("Can not recognize protocol from remote {}, preface = {}", ctx.channel().remoteAddress(), preface);
+        in.clear();
+        ctx.close();
+    }
+}