This is an automated email from the ASF dual-hosted git repository.
jianbin pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/incubator-seata.git
The following commit(s) were added to refs/heads/2.x by this push:
new f83b6eabb8 feature: support grpc protocol (#6881)
f83b6eabb8 is described below
commit f83b6eabb8e33d065cf4e43e247d7f4b4c30de08
Author: yiqi <[email protected]>
AuthorDate: Sat Oct 5 13:14:21 2024 +0800
feature: support grpc protocol (#6881)
---
changes/en-us/2.x.md | 2 +
changes/zh-cn/2.x.md | 3 +-
.../org/apache/seata/common/ConfigurationKeys.java | 2 +
.../org/apache/seata/common/DefaultValues.java | 1 +
core/pom.xml | 21 +++
.../org/apache/seata/core/protocol/Protocol.java | 39 +++++
.../core/protocol/detector/Http2Detector.java | 66 ++++++++
.../core/protocol/detector/ProtocolDetector.java | 26 +++
.../core/protocol/detector/SeataDetector.java | 50 ++++++
.../seata/core/rpc/netty/NettyClientBootstrap.java | 50 +++++-
.../seata/core/rpc/netty/NettyClientConfig.java | 5 +
.../seata/core/rpc/netty/NettyServerBootstrap.java | 10 +-
.../core/rpc/netty/ProtocolDetectHandler.java | 62 +++++++
.../seata/core/rpc/netty/grpc/GrpcDecoder.java | 118 ++++++++++++++
.../seata/core/rpc/netty/grpc/GrpcEncoder.java | 99 ++++++++++++
.../seata/core/rpc/netty/grpc/GrpcHeaderEnum.java | 49 ++++++
.../seata/core/rpc/netty/v1/ProtocolEncoderV1.java | 1 -
.../core/serializer/SerializerServiceLoader.java | 28 +++-
.../seata/protocol/transcation/grpcMessage.proto | 32 ++++
dependencies/pom.xml | 6 +-
script/client/conf/file.conf | 2 +
script/client/spring/application.properties | 1 +
script/client/spring/application.yml | 1 +
script/config-center/config.txt | 1 +
.../properties/TransportProperties.java | 11 ++
test/pom.xml | 26 +++
.../seata/core/rpc/netty/mockserver/GrpcTest.java | 180 +++++++++++++++++++++
.../rpc/netty/mockserver/MockGrpcServerTest.java | 120 ++++++++++++++
.../seata/protocol/transcation/grpcMessage.proto | 32 ++++
29 files changed, 1026 insertions(+), 18 deletions(-)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index fcdb6c6523..2dd147c01b 100644
--- a/changes/en-us/2.x.md
+++ b/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)
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index 016c32c465..624db67f90 100644
--- a/changes/zh-cn/2.x.md
+++ b/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和建议，非常感谢大家。
diff --git
a/common/src/main/java/org/apache/seata/common/ConfigurationKeys.java
b/common/src/main/java/org/apache/seata/common/ConfigurationKeys.java
index dec76f51b7..ff8436b6dc 100644
--- a/common/src/main/java/org/apache/seata/common/ConfigurationKeys.java
+++ b/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
*/
diff --git a/common/src/main/java/org/apache/seata/common/DefaultValues.java
b/common/src/main/java/org/apache/seata/common/DefaultValues.java
index 8c484a1ab0..eb0d40bb30 100644
--- a/common/src/main/java/org/apache/seata/common/DefaultValues.java
+++ b/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;
diff --git a/core/pom.xml b/core/pom.xml
index 9de6107bc0..26ce3dc901 100644
--- a/core/pom.xml
+++ b/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>
diff --git a/core/src/main/java/org/apache/seata/core/protocol/Protocol.java
b/core/src/main/java/org/apache/seata/core/protocol/Protocol.java
new file mode 100644
index 0000000000..fe3cc000cf
--- /dev/null
+++ b/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;
+ }
+}
diff --git
a/core/src/main/java/org/apache/seata/core/protocol/detector/Http2Detector.java
b/core/src/main/java/org/apache/seata/core/protocol/detector/Http2Detector.java
new file mode 100644
index 0000000000..a004894f9b
--- /dev/null
+++
b/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);
+ }
+ })
+ };
+ }
+}
\ No newline at end of file
diff --git
a/core/src/main/java/org/apache/seata/core/protocol/detector/ProtocolDetector.java
b/core/src/main/java/org/apache/seata/core/protocol/detector/ProtocolDetector.java
new file mode 100644
index 0000000000..89d5d10e7b
--- /dev/null
+++
b/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();
+}
diff --git
a/core/src/main/java/org/apache/seata/core/protocol/detector/SeataDetector.java
b/core/src/main/java/org/apache/seata/core/protocol/detector/SeataDetector.java
new file mode 100644
index 0000000000..b9c30b0bc6
--- /dev/null
+++
b/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};
+ }
+}
\ No newline at end of file
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientBootstrap.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientBootstrap.java
index 4aaafc0acb..0fbd9ff079 100644
---
a/core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientBootstrap.java
+++
b/core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientBootstrap.java
@@ -18,8 +18,11 @@ package org.apache.seata.core.rpc.netty;
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.EpollChannelOption;
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 class NettyClientBootstrap implements
RemotingBootstrap {
@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 class NettyClientBootstrap implements
RemotingBootstrap {
} 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;
}
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientConfig.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientConfig.java
index f0e047ad58..6858360826 100644
--- a/core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientConfig.java
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientConfig.java
@@ -21,6 +21,7 @@ import org.apache.seata.core.constants.ConfigurationKeys;
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 class NettyClientConfig extends NettyBaseConfig {
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;
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/NettyServerBootstrap.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/NettyServerBootstrap.java
index c7b2aa57c2..b589396e5a 100644
---
a/core/src/main/java/org/apache/seata/core/rpc/netty/NettyServerBootstrap.java
+++
b/core/src/main/java/org/apache/seata/core/rpc/netty/NettyServerBootstrap.java
@@ -92,6 +92,10 @@ public class NettyServerBootstrap implements
RemotingBootstrap {
}
}
+ protected ChannelHandler[] getChannelHandlers() {
+ return channelHandlers;
+ }
+
/**
* Add channel pipeline last.
*
@@ -158,10 +162,8 @@ public class NettyServerBootstrap implements
RemotingBootstrap {
.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));
}
});
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/ProtocolDetectHandler.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/ProtocolDetectHandler.java
new file mode 100644
index 0000000000..9f1b5f8c11
--- /dev/null
+++
b/core/src/main/java/org/apache/seata/core/rpc/netty/ProtocolDetectHandler.java
@@ -0,0 +1,62 @@
+/*
+ * 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.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();
+ }
+}
\ No newline at end of file
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/grpc/GrpcDecoder.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/grpc/GrpcDecoder.java
new file mode 100644
index 0000000000..71c9caf8be
--- /dev/null
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/grpc/GrpcDecoder.java
@@ -0,0 +1,118 @@
+/*
+ * 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.rpc.netty.grpc;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http2.Http2DataFrame;
+import io.netty.handler.codec.http2.Http2HeadersFrame;
+import io.netty.util.ReferenceCountUtil;
+import io.netty.util.ReferenceCounted;
+import org.apache.commons.lang.StringUtils;
+import org.apache.seata.core.compressor.Compressor;
+import org.apache.seata.core.compressor.CompressorFactory;
+import org.apache.seata.core.protocol.HeartbeatMessage;
+import org.apache.seata.core.protocol.ProtocolConstants;
+import org.apache.seata.core.protocol.RpcMessage;
+import org.apache.seata.core.protocol.generated.GrpcMessageProto;
+import org.apache.seata.core.serializer.Serializer;
+import org.apache.seata.core.serializer.SerializerServiceLoader;
+import org.apache.seata.core.serializer.SerializerType;
+
+import java.util.Map;
+
+public class GrpcDecoder extends ChannelDuplexHandler {
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws
Exception {
+ if (msg instanceof Http2HeadersFrame) {
+ onHeadersRead(ctx, (Http2HeadersFrame) msg);
+ } else if (msg instanceof Http2DataFrame) {
+ onDataRead(ctx, (Http2DataFrame) msg);
+ } else if (msg instanceof ReferenceCounted) {
+ ReferenceCountUtil.release(msg);
+ }
+ }
+
+ public void onDataRead(ChannelHandlerContext ctx, Http2DataFrame msg)
throws Exception {
+ ByteBuf content = msg.content();
+ try {
+ int readableBytes = content.readableBytes();
+ byte[] bytes = new byte[readableBytes];
+ content.readBytes(bytes);
+ if (bytes.length < 5) {
+ return;
+ }
+
+ int srcPos = 0;
+ while (srcPos < readableBytes) {
+ // The first byte defaults to 0, indicating that no
decompression is required
+ // Read the value of the next four bytes as the length of the
body
+ int length = ((bytes[srcPos + 1] & 0xFF) << 24) |
((bytes[srcPos + 2] & 0xFF) << 16)
+ | ((bytes[srcPos + 3] & 0xFF) << 8) | (bytes[srcPos +
4] & 0xFF);
+
+ byte[] data = new byte[length];
+ System.arraycopy(bytes, srcPos + 5, data, 0, length);
+ GrpcMessageProto grpcMessageProto =
GrpcMessageProto.parseFrom(data);
+ byte[] bodyBytes = grpcMessageProto.getBody().toByteArray();
+ int messageType = grpcMessageProto.getMessageType();
+ int messageId = grpcMessageProto.getId();
+ Map<String, String> headMap = grpcMessageProto.getHeadMapMap();
+
+ RpcMessage rpcMsg = new RpcMessage();
+ if (messageType <= Byte.MAX_VALUE && messageType >=
Byte.MIN_VALUE) {
+ rpcMsg.setMessageType((byte) messageType);
+ }
+ rpcMsg.setId(messageId);
+ rpcMsg.setHeadMap(grpcMessageProto.getHeadMapMap());
+
+ if (messageType ==
ProtocolConstants.MSGTYPE_HEARTBEAT_REQUEST) {
+ rpcMsg.setBody(HeartbeatMessage.PING);
+ } else if (messageType ==
ProtocolConstants.MSGTYPE_HEARTBEAT_RESPONSE) {
+ rpcMsg.setBody(HeartbeatMessage.PONG);
+ } else {
+ String compressType =
headMap.get(GrpcHeaderEnum.COMPRESS_TYPE.header);
+ if (StringUtils.isNotBlank(compressType)) {
+ byte compress = Byte.parseByte(compressType);
+ rpcMsg.setCompressor(compress);
+ Compressor compressor =
CompressorFactory.getCompressor(compress);
+ bodyBytes = compressor.decompress(bodyBytes);
+ }
+ String codecValue =
headMap.get(GrpcHeaderEnum.CODEC_TYPE.header);
+ int codec = Integer.parseInt(codecValue);
+ SerializerType serializerType =
SerializerType.getByCode(codec);
+ rpcMsg.setCodec(serializerType.getCode());
+ Serializer serializer =
SerializerServiceLoader.load(serializerType);
+ Object messageBody = serializer.deserialize(bodyBytes);
+ rpcMsg.setBody(messageBody);
+ }
+
+ ctx.fireChannelRead(rpcMsg);
+
+ srcPos += length + 5;
+ }
+ } finally {
+ ReferenceCountUtil.release(content);
+ }
+ }
+
+
+ public void onHeadersRead(ChannelHandlerContext ctx, Http2HeadersFrame
headersFrame) throws Exception {
+ // TODO Subsequent decompression logic is possible
+ }
+}
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/grpc/GrpcEncoder.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/grpc/GrpcEncoder.java
new file mode 100644
index 0000000000..dbbbfe1be4
--- /dev/null
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/grpc/GrpcEncoder.java
@@ -0,0 +1,99 @@
+/*
+ * 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.rpc.netty.grpc;
+
+import com.google.protobuf.ByteString;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
+import io.netty.handler.codec.http2.Http2Headers;
+import org.apache.seata.core.compressor.Compressor;
+import org.apache.seata.core.compressor.CompressorFactory;
+import org.apache.seata.core.protocol.ProtocolConstants;
+import org.apache.seata.core.protocol.RpcMessage;
+import org.apache.seata.core.protocol.generated.GrpcMessageProto;
+import org.apache.seata.core.serializer.Serializer;
+import org.apache.seata.core.serializer.SerializerServiceLoader;
+import org.apache.seata.core.serializer.SerializerType;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class GrpcEncoder extends ChannelOutboundHandlerAdapter {
+ private final AtomicBoolean headerSent = new AtomicBoolean(false);
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise
promise) throws Exception {
+ if (!(msg instanceof RpcMessage)) {
+ throw new UnsupportedOperationException("GrpcEncoder not support
class:" + msg.getClass());
+ }
+
+ RpcMessage rpcMessage = (RpcMessage) msg;
+ byte messageType = rpcMessage.getMessageType();
+ Map<String, String> headMap = rpcMessage.getHeadMap();
+ Object body = rpcMessage.getBody();
+ int id = rpcMessage.getId();
+
+ if (headerSent.compareAndSet(false, true)) {
+ Http2Headers headers = new DefaultHttp2Headers();
+ headers.add(GrpcHeaderEnum.HTTP2_STATUS.header,
String.valueOf(200));
+ headers.add(GrpcHeaderEnum.GRPC_STATUS.header, String.valueOf(0));
+ headers.add(GrpcHeaderEnum.GRPC_CONTENT_TYPE.header,
"application/grpc");
+ ctx.writeAndFlush(new DefaultHttp2HeadersFrame(headers));
+ }
+
+ ByteString dataBytes;
+ if (messageType != ProtocolConstants.MSGTYPE_HEARTBEAT_REQUEST
+ && messageType !=
ProtocolConstants.MSGTYPE_HEARTBEAT_RESPONSE) {
+ Serializer serializer =
SerializerServiceLoader.load(SerializerType.getByCode(SerializerType.PROTOBUF.getCode()));
+ byte[] serializedBytes = serializer.serialize(body);
+ Compressor compressor =
CompressorFactory.getCompressor(rpcMessage.getCompressor());
+ dataBytes =
ByteString.copyFrom(compressor.compress(serializedBytes));
+ } else {
+ dataBytes = ByteString.EMPTY;
+ }
+ headMap.put(GrpcHeaderEnum.CODEC_TYPE.header,
String.valueOf(SerializerType.PROTOBUF.getCode()));
+ headMap.put(GrpcHeaderEnum.COMPRESS_TYPE.header,
String.valueOf(rpcMessage.getCompressor()));
+ GrpcMessageProto.Builder builder = GrpcMessageProto.newBuilder()
+ .putAllHeadMap(headMap)
+ .setMessageType(messageType)
+ .setId(id);
+ builder.setBody(ByteString.copyFrom(dataBytes.toByteArray()));
+ GrpcMessageProto grpcMessageProto = builder.build();
+
+ byte[] bodyBytes = grpcMessageProto.toByteArray();
+ if (bodyBytes != null) {
+ byte[] messageWithPrefix = new byte[bodyBytes.length + 5];
+ // The first byte is 0, indicating no compression
+ messageWithPrefix[0] = 0;
+ ByteBuffer buffer = ByteBuffer.allocate(4);
+ buffer.putInt(bodyBytes.length);
+ byte[] lengthBytes = buffer.array();
+ // The last four bytes indicate the length
+ System.arraycopy(lengthBytes, 0, messageWithPrefix, 1, 4);
+ // The remaining bytes are body
+ System.arraycopy(bodyBytes, 0, messageWithPrefix, 5,
bodyBytes.length);
+ ctx.writeAndFlush(new
DefaultHttp2DataFrame(Unpooled.wrappedBuffer(messageWithPrefix)));
+ }
+ }
+
+}
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/grpc/GrpcHeaderEnum.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/grpc/GrpcHeaderEnum.java
new file mode 100644
index 0000000000..ed8e729b47
--- /dev/null
+++
b/core/src/main/java/org/apache/seata/core/rpc/netty/grpc/GrpcHeaderEnum.java
@@ -0,0 +1,49 @@
+/*
+ * 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.rpc.netty.grpc;
+
+public enum GrpcHeaderEnum {
+
+ /**
+ * grpc status
+ */
+ GRPC_STATUS("grpc-status"),
+ /**
+ * http2 status
+ */
+ HTTP2_STATUS(":status"),
+ /**
+ * content-type
+ */
+ GRPC_CONTENT_TYPE("content-type"),
+
+ /**
+ * codec-type
+ */
+ CODEC_TYPE("codec-type"),
+
+ /**
+ * compress-type
+ */
+ COMPRESS_TYPE("compress-type");
+
+ public final String header;
+
+ GrpcHeaderEnum(String header) {
+ this.header = header;
+ }
+}
diff --git
a/core/src/main/java/org/apache/seata/core/rpc/netty/v1/ProtocolEncoderV1.java
b/core/src/main/java/org/apache/seata/core/rpc/netty/v1/ProtocolEncoderV1.java
index dd01b948db..39180f3bdc 100644
---
a/core/src/main/java/org/apache/seata/core/rpc/netty/v1/ProtocolEncoderV1.java
+++
b/core/src/main/java/org/apache/seata/core/rpc/netty/v1/ProtocolEncoderV1.java
@@ -66,7 +66,6 @@ public class ProtocolEncoderV1 extends MessageToByteEncoder
implements ProtocolE
public void encode(RpcMessage message, ByteBuf out) {
try {
-
ProtocolRpcMessageV1 rpcMessage = new ProtocolRpcMessageV1();
rpcMessage.rpcMsg2ProtocolMsg(message);
diff --git
a/core/src/main/java/org/apache/seata/core/serializer/SerializerServiceLoader.java
b/core/src/main/java/org/apache/seata/core/serializer/SerializerServiceLoader.java
index 63ce440edd..0aa9bd340e 100644
---
a/core/src/main/java/org/apache/seata/core/serializer/SerializerServiceLoader.java
+++
b/core/src/main/java/org/apache/seata/core/serializer/SerializerServiceLoader.java
@@ -72,8 +72,7 @@ public final class SerializerServiceLoader {
"Please manually reference
'org.apache.seata:seata-serializer-protobuf' dependency.");
}
-
- String key = serialzerKey(type, version);
+ String key = serializerKey(type, version);
Serializer serializer = SERIALIZER_MAP.get(key);
if (serializer == null) {
if (type == SerializerType.SEATA) {
@@ -86,7 +85,30 @@ public final class SerializerServiceLoader {
return serializer;
}
- private static String serialzerKey(SerializerType type, byte version) {
+ /**
+ * Load the service of {@link Serializer}
+ *
+ * @param type the serializer type
+ * @return the service of {@link Serializer}
+ * @throws EnhancedServiceNotFoundException the enhanced service not found
exception
+ */
+ public static Serializer load(SerializerType type) throws
EnhancedServiceNotFoundException {
+ if (type == SerializerType.PROTOBUF && !CONTAINS_PROTOBUF_DEPENDENCY) {
+ throw new EnhancedServiceNotFoundException("The class '" +
PROTOBUF_SERIALIZER_CLASS_NAME + "' not found. " +
+ "Please manually reference
'org.apache.seata:seata-serializer-protobuf' dependency.");
+ }
+
+ String key = type.name();
+ Serializer serializer = SERIALIZER_MAP.get(key);
+ if (serializer == null) {
+ serializer = EnhancedServiceLoader.load(Serializer.class,
type.name());
+
+ SERIALIZER_MAP.put(key, serializer);
+ }
+ return serializer;
+ }
+
+ private static String serializerKey(SerializerType type, byte version) {
if (type == SerializerType.SEATA) {
return type.name() + version;
}
diff --git
a/core/src/main/resources/protobuf/org/apache/seata/protocol/transcation/grpcMessage.proto
b/core/src/main/resources/protobuf/org/apache/seata/protocol/transcation/grpcMessage.proto
new file mode 100644
index 0000000000..dd61bd95f4
--- /dev/null
+++
b/core/src/main/resources/protobuf/org/apache/seata/protocol/transcation/grpcMessage.proto
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+syntax = "proto3";
+package org.apache.seata.protocol.protobuf;
+option java_multiple_files = true;
+option java_outer_classname = "GrpcMessage";
+option java_package = "org.apache.seata.core.protocol.generated";
+
+message GrpcMessageProto {
+ int32 id = 1;
+ int32 messageType = 2;
+ map<string, string> headMap = 3;
+ bytes body = 4;
+}
+
+service SeataService {
+ rpc sendRequest (stream GrpcMessageProto) returns (stream
GrpcMessageProto);
+}
\ No newline at end of file
diff --git a/dependencies/pom.xml b/dependencies/pom.xml
index 4bf0450d4f..15178f1416 100644
--- a/dependencies/pom.xml
+++ b/dependencies/pom.xml
@@ -76,7 +76,6 @@
<netty.version>4.1.101.Final</netty.version>
<sofa.hessian.version>4.0.3</sofa.hessian.version>
<sofa.bolt.version>1.6.7</sofa.bolt.version>
-
<protobuf.version>3.25.4</protobuf.version>
<grpc.version>1.66.0</grpc.version>
<kryo.version>5.4.0</kryo.version>
@@ -623,6 +622,11 @@
<artifactId>grpc-core</artifactId>
<version>${grpc.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-alts</artifactId>
+ <version>${grpc.version}</version>
+ </dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-api</artifactId>
diff --git a/script/client/conf/file.conf b/script/client/conf/file.conf
index 92c836e17a..4b12166092 100644
--- a/script/client/conf/file.conf
+++ b/script/client/conf/file.conf
@@ -16,6 +16,8 @@
#
transport {
+ # communication protocols, seata or grpc, default seata
+ protocol = "seata"
# tcp, unix-domain-socket
type = "TCP"
#NIO, NATIVE
diff --git a/script/client/spring/application.properties
b/script/client/spring/application.properties
index cb7d93c2ab..2a72d1e5f7 100755
--- a/script/client/spring/application.properties
+++ b/script/client/spring/application.properties
@@ -64,6 +64,7 @@ seata.log.exception-rate=100
seata.service.vgroup-mapping.default_tx_group=default
seata.service.grouplist.default=127.0.0.1:8091
seata.service.disable-global-transaction=false
+seata.transport.protocol=seata
seata.transport.shutdown.wait=3
seata.transport.thread-factory.boss-thread-prefix=NettyBoss
seata.transport.thread-factory.worker-thread-prefix=NettyServerNIOWorker
diff --git a/script/client/spring/application.yml
b/script/client/spring/application.yml
index 9eef693d33..a6100f0574 100755
--- a/script/client/spring/application.yml
+++ b/script/client/spring/application.yml
@@ -73,6 +73,7 @@ seata:
default: 127.0.0.1:8091
disable-global-transaction: false
transport:
+ protocol: seata
shutdown:
wait: 3
thread-factory:
diff --git a/script/config-center/config.txt b/script/config-center/config.txt
index 99cd7bd131..8cf986f3f9 100644
--- a/script/config-center/config.txt
+++ b/script/config-center/config.txt
@@ -17,6 +17,7 @@
#For details about configuration items, see
https://seata.apache.org/zh-cn/docs/user/configurations
#Transport configuration, for client and server
+transport.protocol=seata
transport.type=TCP
transport.server=NIO
transport.heartbeat=true
diff --git
a/seata-spring-autoconfigure/seata-spring-autoconfigure-core/src/main/java/org/apache/seata/spring/boot/autoconfigure/properties/TransportProperties.java
b/seata-spring-autoconfigure/seata-spring-autoconfigure-core/src/main/java/org/apache/seata/spring/boot/autoconfigure/properties/TransportProperties.java
index 8b4caa4e8f..64f38a385d 100644
---
a/seata-spring-autoconfigure/seata-spring-autoconfigure-core/src/main/java/org/apache/seata/spring/boot/autoconfigure/properties/TransportProperties.java
+++
b/seata-spring-autoconfigure/seata-spring-autoconfigure-core/src/main/java/org/apache/seata/spring/boot/autoconfigure/properties/TransportProperties.java
@@ -23,6 +23,7 @@ import static
org.apache.seata.common.DefaultValues.DEFAULT_ENABLE_CLIENT_BATCH_
import static
org.apache.seata.common.DefaultValues.DEFAULT_ENABLE_RM_CLIENT_BATCH_SEND_REQUEST;
import static
org.apache.seata.common.DefaultValues.DEFAULT_ENABLE_TC_SERVER_BATCH_SEND_RESPONSE;
import static
org.apache.seata.common.DefaultValues.DEFAULT_ENABLE_TM_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_TC_REQUEST_TIMEOUT;
import static
org.apache.seata.common.DefaultValues.DEFAULT_RPC_TM_REQUEST_TIMEOUT;
@@ -54,6 +55,8 @@ public class TransportProperties {
*/
private String compressor = "none";
+ private String protocol = DEFAULT_PROTOCOL;
+
/**
* enable client batch send request
*/
@@ -193,4 +196,12 @@ public class TransportProperties {
public void setRpcTcRequestTimeout(long rpcTcRequestTimeout) {
this.rpcTcRequestTimeout = rpcTcRequestTimeout;
}
+
+ public String getProtocol() {
+ return protocol;
+ }
+
+ public void setProtocol(String protocol) {
+ this.protocol = protocol;
+ }
}
diff --git a/test/pom.xml b/test/pom.xml
index 91d4c9f60f..d35f25bad5 100644
--- a/test/pom.xml
+++ b/test/pom.xml
@@ -40,6 +40,28 @@
<skip>true</skip>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.xolstice.maven.plugins</groupId>
+ <artifactId>protobuf-maven-plugin</artifactId>
+ <configuration>
+
<protoSourceRoot>${project.basedir}/src/test/resources/protobuf/org/apache/seata/protocol/transcation/</protoSourceRoot>
+ <protocArtifact>
+
com.google.protobuf:protoc:3.25.4:exe:${os.detected.classifier}
+ </protocArtifact>
+ <pluginId>grpc-java</pluginId>
+ <pluginArtifact>
+
io.grpc:protoc-gen-grpc-java:1.66.0:exe:${os.detected.classifier}
+ </pluginArtifact>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>compile</goal>
+ <goal>compile-custom</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
@@ -49,6 +71,10 @@
<artifactId>seata-tm</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-alts</artifactId>
+ </dependency>
<!-- The actual spring-related dependencies that take effect are from the
seata-dependencies module, not the seata-server module-->
<dependency>
diff --git
a/test/src/test/java/org/apache/seata/core/rpc/netty/mockserver/GrpcTest.java
b/test/src/test/java/org/apache/seata/core/rpc/netty/mockserver/GrpcTest.java
new file mode 100644
index 0000000000..0d63d2eb70
--- /dev/null
+++
b/test/src/test/java/org/apache/seata/core/rpc/netty/mockserver/GrpcTest.java
@@ -0,0 +1,180 @@
+/*
+ * 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.rpc.netty.mockserver;
+
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.stub.StreamObserver;
+import org.apache.seata.common.ConfigurationKeys;
+import org.apache.seata.common.ConfigurationTestHelper;
+import org.apache.seata.config.ConfigurationFactory;
+import org.apache.seata.core.protocol.generated.GrpcMessageProto;
+import org.apache.seata.core.rpc.netty.RmNettyRemotingClient;
+import org.apache.seata.core.rpc.netty.TmNettyRemotingClient;
+import org.apache.seata.mockserver.MockServer;
+import org.apache.seata.serializer.protobuf.generated.*;
+import org.apache.seata.core.protocol.generated.SeataServiceGrpc;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class GrpcTest {
+
+ private static ManagedChannel channel;
+
+ private static SeataServiceGrpc.SeataServiceStub seataServiceStub;
+
+ @BeforeAll
+ public static void before() {
+ ConfigurationFactory.reload();
+
ConfigurationTestHelper.putConfig(ConfigurationKeys.SERVER_SERVICE_PORT_CAMEL,
String.valueOf(ProtocolTestConstants.MOCK_SERVER_PORT));
+ MockServer.start(ProtocolTestConstants.MOCK_SERVER_PORT);
+ TmNettyRemotingClient.getInstance().destroy();
+ RmNettyRemotingClient.getInstance().destroy();
+
+ channel = ManagedChannelBuilder.forAddress("127.0.0.1",
ProtocolTestConstants.MOCK_SERVER_PORT).usePlaintext().build();
+ seataServiceStub = SeataServiceGrpc.newStub(channel);
+ }
+
+ @AfterAll
+ public static void after() {
+ //MockServer.close();
+
ConfigurationTestHelper.removeConfig(ConfigurationKeys.SERVER_SERVICE_PORT_CAMEL);
+ TmNettyRemotingClient.getInstance().destroy();
+ RmNettyRemotingClient.getInstance().destroy();
+ }
+
+ private GrpcMessageProto getRegisterTMRequest() {
+ AbstractIdentifyRequestProto abstractIdentifyRequestProto =
AbstractIdentifyRequestProto.newBuilder()
+ .setApplicationId("test-applicationId")
+ .build();
+ RegisterTMRequestProto registerTMRequestProto =
RegisterTMRequestProto.newBuilder()
+ .setAbstractIdentifyRequest(abstractIdentifyRequestProto)
+ .build();
+
+ return
GrpcMessageProto.newBuilder().setBody(registerTMRequestProto.toByteString()).build();
+ }
+
+ private GrpcMessageProto getGlobalBeginRequest() {
+ GlobalBeginRequestProto globalBeginRequestProto =
GlobalBeginRequestProto.newBuilder()
+ .setTransactionName("test-transaction")
+ .setTimeout(2000)
+ .build();
+ return
GrpcMessageProto.newBuilder().setBody(globalBeginRequestProto.toByteString()).build();
+ }
+
+ private GrpcMessageProto getBranchRegisterRequest() {
+ BranchRegisterRequestProto branchRegisterRequestProto =
BranchRegisterRequestProto.newBuilder()
+ .setXid("1")
+ .setLockKey("1")
+ .setResourceId("test-resource")
+ .setBranchType(BranchTypeProto.TCC)
+ .setApplicationData("{\"mock\":\"mock\"}")
+ .build();
+
+ return
GrpcMessageProto.newBuilder().setBody(branchRegisterRequestProto.toByteString()).build();
+ }
+
+ private GrpcMessageProto getGlobalCommitRequest() {
+ AbstractGlobalEndRequestProto globalEndRequestProto =
AbstractGlobalEndRequestProto.newBuilder()
+ .setXid("1")
+ .build();
+ GlobalCommitRequestProto globalCommitRequestProto =
GlobalCommitRequestProto.newBuilder()
+ .setAbstractGlobalEndRequest(globalEndRequestProto)
+ .build();
+
+ return
GrpcMessageProto.newBuilder().setBody(globalCommitRequestProto.toByteString()).build();
+ }
+
+ private GrpcMessageProto getGlobalRollbackRequest() {
+ AbstractGlobalEndRequestProto globalEndRequestProto =
AbstractGlobalEndRequestProto.newBuilder()
+ .setXid("1")
+ .build();
+ GlobalRollbackRequestProto globalRollbackRequestProto =
GlobalRollbackRequestProto.newBuilder()
+ .setAbstractGlobalEndRequest(globalEndRequestProto)
+ .build();
+
+ return
GrpcMessageProto.newBuilder().setBody(globalRollbackRequestProto.toByteString()).build();
+ }
+
+ @Test
+ public void testCommit() throws InterruptedException {
+ CountDownLatch countDownLatch = new CountDownLatch(4);
+ StreamObserver<GrpcMessageProto> streamObserver = new
StreamObserver<GrpcMessageProto>() {
+ @Override
+ public void onNext(GrpcMessageProto grpcMessageProto) {
+ System.out.println("receive : " + grpcMessageProto.toString());
+ countDownLatch.countDown();
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ throwable.printStackTrace();
+ }
+
+ @Override
+ public void onCompleted() {
+
+ }
+ };
+
+ StreamObserver<GrpcMessageProto> response =
seataServiceStub.sendRequest(streamObserver);
+ response.onNext(getRegisterTMRequest());
+ response.onNext(getGlobalBeginRequest());
+ response.onNext(getBranchRegisterRequest());
+ response.onNext(getGlobalCommitRequest());
+
+ response.onCompleted();
+
+ countDownLatch.await(10, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testRollback() throws InterruptedException {
+ CountDownLatch countDownLatch = new CountDownLatch(4);
+ StreamObserver<GrpcMessageProto> streamObserver = new
StreamObserver<GrpcMessageProto>() {
+ @Override
+ public void onNext(GrpcMessageProto grpcMessageProto) {
+ System.out.println("receive : " + grpcMessageProto.toString());
+ countDownLatch.countDown();
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ throwable.printStackTrace();
+ }
+
+ @Override
+ public void onCompleted() {
+
+ }
+ };
+
+ StreamObserver<GrpcMessageProto> response =
seataServiceStub.sendRequest(streamObserver);
+ response.onNext(getRegisterTMRequest());
+ response.onNext(getGlobalBeginRequest());
+ response.onNext(getBranchRegisterRequest());
+ response.onNext(getGlobalRollbackRequest());
+
+ response.onCompleted();
+
+ countDownLatch.await(10, TimeUnit.SECONDS);
+ }
+}
diff --git
a/test/src/test/java/org/apache/seata/core/rpc/netty/mockserver/MockGrpcServerTest.java
b/test/src/test/java/org/apache/seata/core/rpc/netty/mockserver/MockGrpcServerTest.java
new file mode 100644
index 0000000000..3744ddd270
--- /dev/null
+++
b/test/src/test/java/org/apache/seata/core/rpc/netty/mockserver/MockGrpcServerTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.rpc.netty.mockserver;
+
+import org.apache.seata.common.ConfigurationKeys;
+import org.apache.seata.common.ConfigurationTestHelper;
+import org.apache.seata.config.ConfigurationFactory;
+import org.apache.seata.core.exception.TransactionException;
+import org.apache.seata.core.model.BranchType;
+import org.apache.seata.core.model.GlobalStatus;
+import org.apache.seata.core.model.TransactionManager;
+import org.apache.seata.core.protocol.Protocol;
+import org.apache.seata.core.rpc.netty.RmNettyRemotingClient;
+import org.apache.seata.core.rpc.netty.TmNettyRemotingClient;
+import org.apache.seata.mockserver.MockCoordinator;
+import org.apache.seata.mockserver.MockServer;
+import org.apache.seata.rm.DefaultResourceManager;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * the type MockServerTest
+ */
+public class MockGrpcServerTest {
+
+ static String RESOURCE_ID = "mock-action";
+
+ Logger logger = LoggerFactory.getLogger(MockGrpcServerTest.class);
+
+ @BeforeAll
+ public static void before() {
+ ConfigurationFactory.reload();
+
ConfigurationTestHelper.putConfig(ConfigurationKeys.SERVER_SERVICE_PORT_CAMEL,
String.valueOf(ProtocolTestConstants.MOCK_SERVER_PORT));
+
ConfigurationTestHelper.putConfig(ConfigurationKeys.TRANSPORT_PROTOCOL,
Protocol.GPRC.value);
+ MockServer.start(ProtocolTestConstants.MOCK_SERVER_PORT);
+ TmNettyRemotingClient.getInstance().destroy();
+ RmNettyRemotingClient.getInstance().destroy();
+ }
+
+ @AfterAll
+ public static void after() {
+ //MockServer.close();
+
ConfigurationTestHelper.removeConfig(ConfigurationKeys.SERVER_SERVICE_PORT_CAMEL);
+
ConfigurationTestHelper.removeConfig(ConfigurationKeys.TRANSPORT_PROTOCOL);
+ TmNettyRemotingClient.getInstance().destroy();
+ RmNettyRemotingClient.getInstance().destroy();
+ }
+
+ @Test
+ public void testCommit() throws TransactionException {
+ String xid = doTestCommit(0);
+ Assertions.assertEquals(1, Action1Impl.getCommitTimes(xid));
+ Assertions.assertEquals(0, Action1Impl.getRollbackTimes(xid));
+ }
+
+ @Test
+ public void testCommitRetry() throws TransactionException {
+ String xid = doTestCommit(2);
+ Assertions.assertEquals(3, Action1Impl.getCommitTimes(xid));
+ Assertions.assertEquals(0, Action1Impl.getRollbackTimes(xid));
+ }
+
+ @Test
+ public void testRollback() throws TransactionException {
+ String xid = doTestRollback(0);
+ Assertions.assertEquals(0, Action1Impl.getCommitTimes(xid));
+ Assertions.assertEquals(1, Action1Impl.getRollbackTimes(xid));
+ }
+
+ @Test
+ public void testRollbackRetry() throws TransactionException {
+ String xid = doTestRollback(2);
+ Assertions.assertEquals(0, Action1Impl.getCommitTimes(xid));
+ Assertions.assertEquals(3, Action1Impl.getRollbackTimes(xid));
+ }
+
+ private String doTestCommit(int times) throws TransactionException {
+ TransactionManager tm = TmClientTest.getTm();
+ DefaultResourceManager rm = RmClientTest.getRm(RESOURCE_ID);
+
+ String xid = tm.begin(ProtocolTestConstants.APPLICATION_ID,
ProtocolTestConstants.SERVICE_GROUP, "test-commit", 60000);
+ MockCoordinator.getInstance().setExpectedRetry(xid, times);
+ Long branchId = rm.branchRegister(BranchType.TCC, RESOURCE_ID, "1",
xid, "{\"mock\":\"mock\"}", "1");
+ GlobalStatus commit = tm.commit(xid);
+ Assertions.assertEquals(GlobalStatus.Committed, commit);
+ return xid;
+ }
+
+ private String doTestRollback(int times) throws TransactionException {
+ TransactionManager tm = TmClientTest.getTm();
+ DefaultResourceManager rm = RmClientTest.getRm(RESOURCE_ID);
+
+ String xid = tm.begin(ProtocolTestConstants.APPLICATION_ID,
ProtocolTestConstants.SERVICE_GROUP, "test-rollback", 60000);
+ logger.info("doTestRollback xid:{}", xid);
+ MockCoordinator.getInstance().setExpectedRetry(xid, times);
+ Long branchId = rm.branchRegister(BranchType.TCC, RESOURCE_ID, "1",
xid, "{\"mock\":\"mock\"}", "1");
+ GlobalStatus rollback = tm.rollback(xid);
+ Assertions.assertEquals(GlobalStatus.Rollbacked, rollback);
+ return xid;
+
+ }
+}
diff --git
a/test/src/test/resources/protobuf/org/apache/seata/protocol/transcation/grpcMessage.proto
b/test/src/test/resources/protobuf/org/apache/seata/protocol/transcation/grpcMessage.proto
new file mode 100644
index 0000000000..dd61bd95f4
--- /dev/null
+++
b/test/src/test/resources/protobuf/org/apache/seata/protocol/transcation/grpcMessage.proto
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+syntax = "proto3";
+package org.apache.seata.protocol.protobuf;
+option java_multiple_files = true;
+option java_outer_classname = "GrpcMessage";
+option java_package = "org.apache.seata.core.protocol.generated";
+
+message GrpcMessageProto {
+ int32 id = 1;
+ int32 messageType = 2;
+ map<string, string> headMap = 3;
+ bytes body = 4;
+}
+
+service SeataService {
+ rpc sendRequest (stream GrpcMessageProto) returns (stream
GrpcMessageProto);
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
