feature: support grpc protocol

changes/en-us/2.x.md

@@ -5,6 +5,7 @@ Add changes here for all PR submitted to the 2.x branch.
 ### feature:
 
 - [[#6876](https://github.com/apache/incubator-seata/pull/6876)]support kingbase
+- [[#6881](https://github.com/apache/incubator-seata/pull/6881)]support grpc
 
 ### bugfix:
 
@@ -35,6 +36,7 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [dk2k](https://github.com/dk2k)
 - [MaoMaoandSnail](https://github.com/MaoMaoandSnail)
 - [yougecn](https://github.com/yougecn)
+- [PleaseGiveMeTheCoke](https://github.com/PleaseGiveMeTheCoke)
 
 
 

changes/zh-cn/2.x.md

@@ -4,6 +4,7 @@
 
 ### feature:
 [[#6876](https://github.com/apache/incubator-seata/pull/6876)]支持人大金仓数据库(kingbase)
+[[#6881](https://github.com/apache/incubator-seata/pull/6881)]全链路支持grpc
 
 ### bugfix:
 
@@ -35,7 +36,7 @@
 - [dk2k](https://github.com/dk2k)
 - [MaoMaoandSnail](https://github.com/MaoMaoandSnail)
 - [yougecn](https://github.com/yougecn)
-
+- [PleaseGiveMeTheCoke](https://github.com/PleaseGiveMeTheCoke)
 
 
 同时，我们收到了社区反馈的很多有价值的issue和建议，非常感谢大家。

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 TRANSPORT_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,6 +69,10 @@
             <artifactId>fastjson</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+        </dependency>
     </dependencies>
 
     <build>
@@ -90,6 +94,23 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <groupId>org.xolstice.maven.plugins</groupId>
+                <artifactId>protobuf-maven-plugin</artifactId>
+                <configuration>
+                    <protoSourceRoot>${project.basedir}/src/main/resources/protobuf/org/apache/seata/protocol/transcation/</protoSourceRoot>
+                    <protocArtifact>
+                        com.google.protobuf:protoc:3.25.4:exe:${os.detected.classifier}
+                    </protocArtifact>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>compile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 </project>

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

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seata.core.protocol;
+
+/**
+ * seata transport protocol
+ */
+public enum Protocol {
+
+    /**
+     * grpc
+     */
+    GPRC("grpc"),
+
+    /**
+     * seata
+     */
+    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,66 @@
+/*
+ * 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.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.rpc.netty.grpc.GrpcDecoder;
+import org.apache.seata.core.rpc.netty.grpc.GrpcEncoder;
+
+public class Http2Detector implements ProtocolDetector {
+    private static 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,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.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,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.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 static 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,14 +139,18 @@ public void start() {
                 @Override
                 public void initChannel(SocketChannel ch) {
                     ChannelPipeline pipeline = ch.pipeline();
-                    pipeline
-                        .addLast(new IdleStateHandler(nettyClientConfig.getChannelMaxReadIdleSeconds(),
-                            nettyClientConfig.getChannelMaxWriteIdleSeconds(),
-                            nettyClientConfig.getChannelMaxAllIdleSeconds()))
-                        .addLast(new ProtocolDecoderV1())
-                        .addLast(new ProtocolEncoderV1());
-                    if (channelHandlers != null) {
-                        addChannelPipelineLast(ch, channelHandlers);
+                    if (nettyClientConfig.getProtocol().equals(Protocol.GPRC.value)) {
+                        pipeline.addLast(Http2FrameCodecBuilder.forClient().build())
+                                .addLast(new Http2MultiplexHandler(new ChannelDuplexHandler()));
+                    } else {
+                        pipeline.addLast(new IdleStateHandler(nettyClientConfig.getChannelMaxReadIdleSeconds(),
+                                nettyClientConfig.getChannelMaxWriteIdleSeconds(),
+                                nettyClientConfig.getChannelMaxAllIdleSeconds()));
+                        pipeline.addLast(new ProtocolDecoderV1())
+                                .addLast(new ProtocolEncoderV1());
+                        if (channelHandlers != null) {
+                            addChannelPipelineLast(ch, channelHandlers);
+                        }
                     }
                 }
             });
@@ -177,9 +190,30 @@ public Channel getNewChannel(InetSocketAddress address) {
             } else {
                 channel = f.channel();
             }
+
+            if (nettyClientConfig.getProtocol().equals(Protocol.GPRC.value)) {
+                Http2StreamChannelBootstrap bootstrap = new Http2StreamChannelBootstrap(channel);
+                bootstrap.handler(new ChannelInboundHandlerAdapter() {
+                    @Override
+                    public void handlerAdded(ChannelHandlerContext ctx) {
+                        Channel channel = ctx.channel();
+                        channel.pipeline().addLast(new IdleStateHandler(nettyClientConfig.getChannelMaxReadIdleSeconds(),
+                                nettyClientConfig.getChannelMaxWriteIdleSeconds(),
+                                nettyClientConfig.getChannelMaxAllIdleSeconds()));
+                        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.TRANSPORT_PROTOCOL, DEFAULT_PROTOCOL);
+    }
+
     @Deprecated
     public static boolean isEnableClientBatchSendRequest() {
         return ENABLE_CLIENT_BATCH_SEND_REQUEST;