diff --git a/README.md b/README.md
index 5b7d3ae5d34..359190a0c39 100644
--- a/README.md
+++ b/README.md
@@ -7,7 +7,7 @@
[![Build Status](https://github.com/apache/incubator-seata/workflows/build/badge.svg?branch=develop)](https://github.com/apache/incubator-seata/actions)
[![codecov](https://codecov.io/gh/apache/incubator-seata/graph/badge.svg?token=tbmHt2ZfxO)](https://codecov.io/gh/apache/incubator-seata)
[![license](https://img.shields.io/github/license/apache/incubator-seata.svg)](https://www.apache.org/licenses/LICENSE-2.0.html)
-[![maven](https://img.shields.io/maven-central/v/io.apache/incubator-seata-parent?versionSuffix=2.0.0)](https://search.maven.org/search?q=io.seata)
+[![maven](https://img.shields.io/maven-central/v/org.apache.seata/seata-all?versionSuffix=2.1.0)](https://central.sonatype.com/search?q=org.apache.seata%3Aseata-all)
## What is Seata?
@@ -82,16 +82,16 @@ For more details about principle and design, please go to [Seata wiki page](http
## Maven dependency
-Depending on the scenario, choose one of the two dependencies: `io.seata:seata-all` or `io.seata:seata-spring-boot-starter`.
+Depending on the scenario, choose one of the two dependencies: `org.apache.seata:seata-all` or `org.apache.seata:seata-spring-boot-starter`.
```xml
- 2.0.0
+ 2.1.0
- io.seata
+ org.apache.seataseata-all${seata.version}
@@ -99,7 +99,7 @@ Depending on the scenario, choose one of the two dependencies: `io.seata:seata-a
- io.seata
+ org.apache.seataseata-spring-boot-starter${seata.version}
@@ -142,7 +142,7 @@ Contributors are welcomed to join the Seata project. Please check [CONTRIBUTING]
* [Seata Website](https://github.com/apache/incubator-seata.github.io) - Seata official website
* [Seata GoLang](https://github.com/apache/incubator-seata-go) - Seata GoLang client and server
* [Seata Samples](https://github.com/apache/incubator-seata-samples) - Samples for Seata
-* [Seata GoLang Simples](https://github.com/apache/incubator-seata-go-samples) - Samples for Seata GoLang
+* [Seata GoLang Samples](https://github.com/apache/incubator-seata-go-samples) - Samples for Seata GoLang
* [Seata K8s](https://github.com/apache/incubator-seata-k8s) - Seata integration with k8s
* [Seata CLI](https://github.com/apache/incubator-seata-ctl) - CLI tool for Seata
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index fcdb6c6523a..96d8512a2dd 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -5,9 +5,11 @@ 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:
-
+- [[#6899](https://github.com/apache/incubator-seata/pull/6899)] fix file.conf read failed after package
+- [[#6890](https://github.com/apache/incubator-seata/pull/6890)] fix designerJson to standardJson: subStateMachine compensateState cannot be recognized
### optimize:
- [[#6826](https://github.com/apache/incubator-seata/pull/6826)] remove the branch registration operation of the XA read-only transaction
@@ -17,6 +19,7 @@ Add changes here for all PR submitted to the 2.x branch.
- [[#6892](https://github.com/apache/incubator-seata/pull/6892)] upgrade npmjs version
- [[#6889](https://github.com/apache/incubator-seata/pull/6889)] Correct word spelling errors
- [[#6898](https://github.com/apache/incubator-seata/pull/6898)] upgrade npmjs version in saga module
+- [[#6902](https://github.com/apache/incubator-seata/pull/6900)] optimize readme docs
### refactor:
@@ -35,7 +38,8 @@ 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)
-
+- [xjlgod](https://github.com/xjlgod)
+- [PleaseGiveMeTheCoke](https://github.com/PleaseGiveMeTheCoke)
Also, we receive many valuable issues, questions and advices from our community. Thanks for you all.
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index 016c32c4651..b100d18320c 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -4,9 +4,11 @@
### feature:
[[#6876](https://github.com/apache/incubator-seata/pull/6876)]支持人大金仓数据库(kingbase)
+[[#6881](https://github.com/apache/incubator-seata/pull/6881)]全链路支持grpc
### bugfix:
-
+- [[#6899](https://github.com/apache/incubator-seata/pull/6899)] 修复file.conf打包后的读取
+- [[#6890](https://github.com/apache/incubator-seata/pull/6890)] 修复saga设计json转标准json过程中: 子状态机补偿节点无法被识别
### optimize:
- [[#6826](https://github.com/apache/incubator-seata/pull/6826)] 移除只读XA事务的分支注册操作
@@ -16,7 +18,8 @@
- [[#6883](https://github.com/apache/incubator-seata/pull/6874)] 删除代码中无用对象的创建
- [[#6892](https://github.com/apache/incubator-seata/pull/6892)] 升级 npmjs 版本
- [[#6889](https://github.com/apache/incubator-seata/pull/6889)] 修正单词拼写错误
-- [[#6898](https://github.com/apache/incubator-seata/pull/6898)] 升级 saga 模块 npmjs 版本
+- [[#6898](https://github.com/apache/incubator-seata/pull/6898)] 升级 saga 模块 npmjs 版本
+- [[#6902](https://github.com/apache/incubator-seata/pull/6900)] 优化 readme 文档
### refactor:
@@ -35,7 +38,8 @@
- [dk2k](https://github.com/dk2k)
- [MaoMaoandSnail](https://github.com/MaoMaoandSnail)
- [yougecn](https://github.com/yougecn)
-
+- [xjlgod](https://github.com/xjlgod)
+- [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 dec76f51b7c..ff8436b6dc9 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 8c484a1ab0f..eb0d40bb308 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/config/seata-config-core/src/main/java/org/apache/seata/config/FileConfiguration.java b/config/seata-config-core/src/main/java/org/apache/seata/config/FileConfiguration.java
index 8c775fe2c93..91b2a290c04 100644
--- a/config/seata-config-core/src/main/java/org/apache/seata/config/FileConfiguration.java
+++ b/config/seata-config-core/src/main/java/org/apache/seata/config/FileConfiguration.java
@@ -16,13 +16,22 @@
*/
package org.apache.seata.config;
+import org.apache.commons.lang.ObjectUtils;
+import org.apache.seata.common.thread.NamedThreadFactory;
+import org.apache.seata.common.util.CollectionUtils;
+import org.apache.seata.common.util.StringUtils;
+import org.apache.seata.config.ConfigFuture.ConfigOperation;
+import org.apache.seata.config.file.FileConfig;
+
import java.io.File;
import java.io.UnsupportedEncodingException;
import java.net.URL;
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -32,15 +41,8 @@
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import org.apache.seata.common.thread.NamedThreadFactory;
-import org.apache.seata.common.util.CollectionUtils;
-import org.apache.seata.common.util.StringUtils;
-import org.apache.seata.config.ConfigFuture.ConfigOperation;
-import org.apache.seata.config.file.FileConfig;
-import org.apache.commons.lang.ObjectUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
/**
* The type FileConfiguration.
*
@@ -134,7 +136,6 @@ private File getConfigFile(String name) {
boolean filePathCustom = name.startsWith(SYS_FILE_RESOURCE_PREFIX);
String filePath = filePathCustom ? name.substring(SYS_FILE_RESOURCE_PREFIX.length()) : name;
String decodedPath = URLDecoder.decode(filePath, StandardCharsets.UTF_8.name());
-
File targetFile = getFileFromFileSystem(decodedPath);
if (targetFile != null) {
return targetFile;
@@ -157,21 +158,18 @@ private File getFileFromFileSystem(String decodedPath) {
// run with jar file and not package third lib into jar file, this.getClass().getClassLoader() will be null
URL resourceUrl = this.getClass().getClassLoader().getResource("");
- String[] tryPaths = null;
+ // try to get log dir (spring.config.additional-location) after package and run sh or bat in bin dir
+ String configLocation = System.getProperty("spring.config.additional-location");
+ List tryPathsList = new ArrayList<>();
+ tryPathsList.add(decodedPath);
if (resourceUrl != null) {
- tryPaths = new String[]{
- // first: project dir
- resourceUrl.getPath() + decodedPath,
- // second: system path
- decodedPath
- };
- } else {
- tryPaths = new String[]{
- decodedPath
- };
+ tryPathsList.add(resourceUrl.getPath() + decodedPath);
+ }
+ if (configLocation != null) {
+ tryPathsList.add(configLocation + decodedPath);
}
-
+ String[] tryPaths = tryPathsList.toArray(new String[0]);
for (String tryPath : tryPaths) {
File targetFile = new File(tryPath);
if (targetFile.exists()) {
diff --git a/core/pom.xml b/core/pom.xml
index 9de6107bc03..26ce3dc9018 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -69,6 +69,10 @@
fastjsontest
+
+ com.google.protobuf
+ protobuf-java
+
@@ -90,6 +94,23 @@
+
+ org.xolstice.maven.plugins
+ protobuf-maven-plugin
+
+ ${project.basedir}/src/main/resources/protobuf/org/apache/seata/protocol/transcation/
+
+ com.google.protobuf:protoc:3.25.4:exe:${os.detected.classifier}
+
+
+
+
+
+ compile
+
+
+
+
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 00000000000..fe3cc000cfc
--- /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 00000000000..a004894f9b1
--- /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() {
+ @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 00000000000..89d5d10e7be
--- /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 00000000000..b9c30b0bc6d
--- /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 4aaafc0acb0..0fbd9ff0795 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 @@
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;
}
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 f0e047ad58d..68583608262 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.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;
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 c7b2aa57c21..b589396e5ab 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 @@ 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() {
@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 00000000000..9f1b5f8c113
--- /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