You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by ji...@apache.org on 2024/10/27 05:05:12 UTC
(incubator-seata) branch summer-code-http2 updated: feature: seata-server support http protocol (#6801)
This is an automated email from the ASF dual-hosted git repository.
jianbin pushed a commit to branch summer-code-http2
in repository https://gitbox.apache.org/repos/asf/incubator-seata.git
The following commit(s) were added to refs/heads/summer-code-http2 by this push:
new 84d6ce3551 feature: seata-server support http protocol (#6801)
84d6ce3551 is described below
commit 84d6ce35516f54937e268587fa8e6637dc68fd49
Author: yiqi <77...@users.noreply.github.com>
AuthorDate: Sun Oct 27 13:05:08 2024 +0800
feature: seata-server support http protocol (#6801)
---
changes/en-us/2.x.md | 1 +
changes/zh-cn/2.x.md | 2 +-
common/pom.xml | 5 +
.../org/apache/seata/common/DefaultValues.java | 1 +
core/pom.xml | 4 +
.../core/protocol/detector/Http2Detector.java | 7 +-
.../seata/core/protocol/detector/HttpDetector.java | 61 +++++++++
.../core/rpc/netty/ProtocolDetectHandler.java | 3 +-
.../core/rpc/netty/http/ControllerManager.java | 33 +++++
.../core/rpc/netty/http/HttpDispatchHandler.java | 81 +++++++++++
.../seata/core/rpc/netty/http/HttpInvocation.java | 62 +++++++++
.../seata/core/rpc/netty/http/ParamMetaData.java | 48 +++++++
.../seata/core/rpc/netty/http/ParameterParser.java | 98 ++++++++++++++
.../core/rpc/netty/http2/Http2DetectHandler.java | 52 +++++++
.../core/rpc/netty/http2/Http2DispatchHandler.java | 94 +++++++++++++
discovery/seata-discovery-raft/pom.xml | 4 +-
.../discovery/registry/enums/HttpVersion.java | 36 +++++
.../registry/raft/RaftRegistryServiceImpl.java | 112 ++++++++-------
.../discovery/registry/utils/OkHttpClientUtil.java | 150 +++++++++++++++++++++
.../registry/raft/RaftRegistryServiceImplTest.java | 148 --------------------
.../src/test/resources/file.conf | 25 ----
.../src/test/resources/registry.conf | 43 ------
.../namingserver/controller/NamingController.java | 2 +-
script/client/spring/application.properties | 1 +
script/client/spring/application.yml | 2 +
.../cluster/manager/ClusterWatcherManager.java | 24 +++-
.../server/controller/ClusterV2Controller.java | 47 +++++++
.../web/RestControllerBeanPostProcessor.java | 142 +++++++++++++++++++
28 files changed, 1011 insertions(+), 277 deletions(-)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index 2d1b8ffe1f..d9f84c9fc2 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -7,6 +7,7 @@ Add changes here for all PR submitted to the 2.x branch.
- [[#6876](https://github.com/apache/incubator-seata/pull/6876)] support kingbase
- [[#6881](https://github.com/apache/incubator-seata/pull/6881)] support grpc
- [[#6864](https://github.com/apache/incubator-seata/pull/6864)] support shentong database
+- [[#6801](https://github.com/apache/incubator-seata/pull/6801)] seata-server support http protocol
### bugfix:
- [[#6899](https://github.com/apache/incubator-seata/pull/6899)] fix file.conf read failed after package
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index 321e751c6a..a60a651907 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -6,7 +6,7 @@
- [[#6876](https://github.com/apache/incubator-seata/pull/6876)] 支持人大金仓数据库(kingbase)
- [[#6881](https://github.com/apache/incubator-seata/pull/6881)] client和server支持grpc协议
- [[#6864](https://github.com/apache/incubator-seata/pull/6864)] 支持神通数据库(oscar)
-
+- [[#6801](https://github.com/apache/incubator-seata/pull/6801)] seata-server 支持HTTP协议的解析
### bugfix:
- [[#6899](https://github.com/apache/incubator-seata/pull/6899)] 修复file.conf打包后的读取
diff --git a/common/pom.xml b/common/pom.xml
index 517aae769a..810512c9cc 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -52,5 +52,10 @@
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.squareup.okhttp3</groupId>
+ <artifactId>okhttp</artifactId>
+ <scope>provided</scope>
+ </dependency>
</dependencies>
</project>
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 2396ecd2d1..bb94403fa3 100644
--- a/common/src/main/java/org/apache/seata/common/DefaultValues.java
+++ b/common/src/main/java/org/apache/seata/common/DefaultValues.java
@@ -64,6 +64,7 @@ public interface DefaultValues {
String DEFAULT_NIO_WORKER_THREAD_PREFIX = "NettyServerNIOWorker";
String DEFAULT_EXECUTOR_THREAD_PREFIX = "NettyServerBizHandler";
String DEFAULT_PROTOCOL = "seata";
+ String DEFAULT_HTTP_VERSION = "http";
boolean DEFAULT_TRANSPORT_HEARTBEAT = true;
boolean DEFAULT_TRANSACTION_UNDO_DATA_VALIDATION = true;
diff --git a/core/pom.xml b/core/pom.xml
index 26ce3dc901..a99743e356 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -69,6 +69,10 @@
<artifactId>fastjson</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
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
index a004894f9b..5687d7e5f2 100644
--- 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
@@ -24,8 +24,7 @@ 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;
+import org.apache.seata.core.rpc.netty.http2.Http2DetectHandler;
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);
@@ -56,9 +55,7 @@ public class Http2Detector implements ProtocolDetector {
@Override
protected void initChannel(Http2StreamChannel ch) {
final ChannelPipeline p = ch.pipeline();
- p.addLast(new GrpcDecoder());
- p.addLast(new GrpcEncoder());
- p.addLast(serverHandlers);
+ p.addLast(new Http2DetectHandler(serverHandlers));
}
})
};
diff --git a/core/src/main/java/org/apache/seata/core/protocol/detector/HttpDetector.java b/core/src/main/java/org/apache/seata/core/protocol/detector/HttpDetector.java
new file mode 100644
index 0000000000..a8f5747c08
--- /dev/null
+++ b/core/src/main/java/org/apache/seata/core/protocol/detector/HttpDetector.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seata.core.protocol.detector;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpServerCodec;
+import org.apache.seata.core.rpc.netty.http.HttpDispatchHandler;
+
+public class HttpDetector implements ProtocolDetector {
+ private static final String[] HTTP_METHODS = {"GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS", "PATCH"};
+
+ @Override
+ public boolean detect(ByteBuf in) {
+ if (in.readableBytes() < 8) {
+ return false;
+ }
+
+ for (String method : HTTP_METHODS) {
+ if (startsWith(in, method)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ private boolean startsWith(ByteBuf buffer, String prefix) {
+ for (int i = 0; i < prefix.length(); i++) {
+ if (buffer.getByte(i) != (byte) prefix.charAt(i)) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ @Override
+ public ChannelHandler[] getHandlers() {
+ return new ChannelHandler[]{
+ new HttpServerCodec(),
+ new HttpObjectAggregator(1048576),
+ new HttpDispatchHandler()
+ };
+ }
+}
\ No newline at end of file
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
index 9f1b5f8c11..6787bfe758 100644
--- 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
@@ -21,6 +21,7 @@ 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.HttpDetector;
import org.apache.seata.core.protocol.detector.ProtocolDetector;
import org.apache.seata.core.protocol.detector.SeataDetector;
import org.slf4j.Logger;
@@ -35,7 +36,7 @@ public class ProtocolDetectHandler extends ByteToMessageDecoder {
public ProtocolDetectHandler(NettyServerBootstrap nettyServerBootstrap) {
this.nettyServerBootstrap = nettyServerBootstrap;
- this.supportedProtocolDetectors = new ProtocolDetector[]{new Http2Detector(nettyServerBootstrap.getChannelHandlers()), new SeataDetector(nettyServerBootstrap.getChannelHandlers())};
+ this.supportedProtocolDetectors = new ProtocolDetector[]{new Http2Detector(nettyServerBootstrap.getChannelHandlers()), new SeataDetector(nettyServerBootstrap.getChannelHandlers()), new HttpDetector()};
}
@Override
diff --git a/core/src/main/java/org/apache/seata/core/rpc/netty/http/ControllerManager.java b/core/src/main/java/org/apache/seata/core/rpc/netty/http/ControllerManager.java
new file mode 100644
index 0000000000..a4536e0c6b
--- /dev/null
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/http/ControllerManager.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seata.core.rpc.netty.http;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class ControllerManager {
+
+ private static final Map<String, HttpInvocation> HTTP_CONTROLLER_MAP = new ConcurrentHashMap<>();
+
+ public static HttpInvocation getHttpInvocation(String path) {
+ return HTTP_CONTROLLER_MAP.get(path);
+ }
+
+ public static void addHttpInvocation(HttpInvocation httpInvocation) {
+ HTTP_CONTROLLER_MAP.put(httpInvocation.getPath(), httpInvocation);
+ }
+}
\ No newline at end of file
diff --git a/core/src/main/java/org/apache/seata/core/rpc/netty/http/HttpDispatchHandler.java b/core/src/main/java/org/apache/seata/core/rpc/netty/http/HttpDispatchHandler.java
new file mode 100644
index 0000000000..5fb8507b83
--- /dev/null
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/http/HttpDispatchHandler.java
@@ -0,0 +1,81 @@
+/*
+ * 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.http;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.codec.http.multipart.Attribute;
+import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder;
+import io.netty.handler.codec.http.multipart.InterfaceHttpData;
+
+import java.lang.reflect.Method;
+
+public class HttpDispatchHandler extends SimpleChannelInboundHandler<HttpRequest> {
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, HttpRequest httpRequest) throws Exception {
+ QueryStringDecoder queryStringDecoder = new QueryStringDecoder(httpRequest.uri());
+ String path = queryStringDecoder.path();
+ ObjectMapper objectMapper = new ObjectMapper();
+ ObjectNode requestDataNode = objectMapper.createObjectNode();
+ requestDataNode.putIfAbsent("param", ParameterParser.convertParamMap(queryStringDecoder.parameters()));
+ requestDataNode.putPOJO("channel", ctx.channel());
+
+ if (httpRequest.method() == HttpMethod.POST) {
+ ObjectNode bodyDataNode = objectMapper.createObjectNode();
+ HttpPostRequestDecoder httpPostRequestDecoder = new HttpPostRequestDecoder(httpRequest);
+ for (InterfaceHttpData interfaceHttpData : httpPostRequestDecoder.getBodyHttpDatas()) {
+ if (interfaceHttpData.getHttpDataType() != InterfaceHttpData.HttpDataType.Attribute) {
+ continue;
+ }
+ Attribute attribute = (Attribute) interfaceHttpData;
+ bodyDataNode.put(attribute.getName(), attribute.getValue());
+ }
+ requestDataNode.putIfAbsent("body", bodyDataNode);
+ }
+
+ HttpInvocation httpInvocation = ControllerManager.getHttpInvocation(path);
+ if (httpInvocation != null) {
+ Object httpController = httpInvocation.getController();
+ Method handleMethod = httpInvocation.getMethod();
+ handleMethod.setAccessible(true);
+ Object[] args = ParameterParser.getArgValues(httpInvocation.getParamMetaData(), handleMethod, requestDataNode);
+ Object result = handleMethod.invoke(httpController, args);
+ if (requestDataNode.get("channel") == null) {
+ return;
+ }
+ FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer(objectMapper.writeValueAsBytes(result)));
+ response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
+ ctx.writeAndFlush(response);
+ return;
+ }
+
+ FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND, Unpooled.wrappedBuffer(Unpooled.EMPTY_BUFFER));
+ ctx.writeAndFlush(response);
+ }
+}
\ No newline at end of file
diff --git a/core/src/main/java/org/apache/seata/core/rpc/netty/http/HttpInvocation.java b/core/src/main/java/org/apache/seata/core/rpc/netty/http/HttpInvocation.java
new file mode 100644
index 0000000000..a0dc1e8c05
--- /dev/null
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/http/HttpInvocation.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.http;
+
+import java.lang.reflect.Method;
+
+public class HttpInvocation {
+
+ private Object controller;
+
+ private Method method;
+
+ private String path;
+
+ private ParamMetaData[] paramMetaData;
+
+ public Method getMethod() {
+ return method;
+ }
+
+ public void setMethod(Method method) {
+ this.method = method;
+ }
+
+ public String getPath() {
+ return path;
+ }
+
+ public void setPath(String path) {
+ this.path = path;
+ }
+
+ public Object getController() {
+ return controller;
+ }
+
+ public void setController(Object controller) {
+ this.controller = controller;
+ }
+
+ public ParamMetaData[] getParamMetaData() {
+ return paramMetaData;
+ }
+
+ public void setParamMetaData(ParamMetaData[] paramMetaData) {
+ this.paramMetaData = paramMetaData;
+ }
+}
diff --git a/core/src/main/java/org/apache/seata/core/rpc/netty/http/ParamMetaData.java b/core/src/main/java/org/apache/seata/core/rpc/netty/http/ParamMetaData.java
new file mode 100644
index 0000000000..4828e38cbf
--- /dev/null
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/http/ParamMetaData.java
@@ -0,0 +1,48 @@
+/*
+ * 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.http;
+
+public class ParamMetaData {
+
+ private ParamConvertType paramConvertType;
+
+ public ParamConvertType getParamConvertType() {
+ return paramConvertType;
+ }
+
+ public void setParamConvertType(ParamConvertType paramConvertType) {
+ this.paramConvertType = paramConvertType;
+ }
+
+ public enum ParamConvertType {
+
+ /**
+ * convert like Spring @RequestBody
+ */
+ REQUEST_BODY,
+
+ /**
+ * convert like Spring @RequestParam
+ */
+ REQUEST_PARAM,
+
+ /**
+ * convert like Spring @ModelAttribute
+ */
+ MODEL_ATTRIBUTE
+ }
+}
\ No newline at end of file
diff --git a/core/src/main/java/org/apache/seata/core/rpc/netty/http/ParameterParser.java b/core/src/main/java/org/apache/seata/core/rpc/netty/http/ParameterParser.java
new file mode 100644
index 0000000000..8805decef6
--- /dev/null
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/http/ParameterParser.java
@@ -0,0 +1,98 @@
+/*
+ * 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.http;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.Parameter;
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.netty.channel.Channel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ParameterParser {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ParameterParser.class);
+
+ public static ObjectNode convertParamMap(Map<String, List<String>> paramMap) throws JsonProcessingException {
+ ObjectMapper objectMapper = new ObjectMapper();
+ ObjectNode paramNode = objectMapper.createObjectNode();
+ for (Map.Entry<String, List<String>> entry : paramMap.entrySet()) {
+ List<String> list = entry.getValue();
+ if (list == null || list.isEmpty()) {
+ continue;
+ }
+ if (list.size() == 1) {
+ paramNode.put(entry.getKey(), list.get(0));
+ } else {
+ ArrayNode arrayNode = paramNode.putArray(entry.getKey());
+ for (String s : list) {
+ arrayNode.add(s);
+ }
+ }
+ }
+ return paramNode;
+ }
+
+ public static Object[] getArgValues(ParamMetaData[] paramMetaDatas, Method handleMethod, ObjectNode paramMap) throws JsonProcessingException {
+ Class<?>[] parameterTypes = handleMethod.getParameterTypes();
+ Parameter[] parameters = handleMethod.getParameters();
+ return getParameters(parameterTypes, paramMetaDatas, parameters, paramMap);
+ }
+
+ private static Object[] getParameters(Class<?>[] parameterTypes, ParamMetaData[] paramMetaDatas, Parameter[] parameters, ObjectNode paramMap) throws JsonProcessingException {
+ int length = parameterTypes.length;
+ Object[] ret = new Object[length];
+ for (int i = 0; i < length; i++) {
+ Class<?> parameterType = parameterTypes[i];
+ String parameterName = parameters[i].getName();
+ ParamMetaData paramMetaData = paramMetaDatas[i];
+ ret[i] = getArgValue(parameterType, parameterName, paramMetaData, paramMap);
+ if (!parameterType.isAssignableFrom(ret[i].getClass())) {
+ LOGGER.error("[HttpDispatchHandler] not compatible parameter type, expect {}, but {}", parameterType, ret[i].getClass());
+ ret[i] = null;
+ }
+ }
+
+ return ret;
+ }
+
+
+ private static Object getArgValue(Class<?> parameterType, String parameterName, ParamMetaData paramMetaData, ObjectNode paramMap) throws JsonProcessingException {
+ ParamMetaData.ParamConvertType paramConvertType = paramMetaData.getParamConvertType();
+ ObjectMapper objectMapper = new ObjectMapper();
+ if (parameterType.equals(Channel.class)) {
+ JsonNode jsonNode = paramMap.get("channel");
+ paramMap.putPOJO("channel", null);
+ return objectMapper.convertValue(jsonNode, Channel.class);
+ } else if (ParamMetaData.ParamConvertType.MODEL_ATTRIBUTE.equals(paramConvertType)) {
+ JsonNode param = paramMap.get("param");
+ return objectMapper.convertValue(param, parameterType);
+ } else if (ParamMetaData.ParamConvertType.REQUEST_BODY.equals(paramConvertType)) {
+ JsonNode body = paramMap.get("body");
+ return objectMapper.convertValue(body, parameterType);
+ } else {
+ return paramMap.get(parameterName).asText(null);
+ }
+ }
+}
diff --git a/core/src/main/java/org/apache/seata/core/rpc/netty/http2/Http2DetectHandler.java b/core/src/main/java/org/apache/seata/core/rpc/netty/http2/Http2DetectHandler.java
new file mode 100644
index 0000000000..fcae4a6b81
--- /dev/null
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/http2/Http2DetectHandler.java
@@ -0,0 +1,52 @@
+/*
+ * 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.http2;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http2.Http2HeadersFrame;
+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.grpc.GrpcHeaderEnum;
+
+public class Http2DetectHandler extends ChannelDuplexHandler {
+
+ private ChannelHandler[] serverHandlers;
+
+ public Http2DetectHandler(ChannelHandler[] serverHandlers) {
+ this.serverHandlers = serverHandlers;
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof Http2HeadersFrame) {
+ Http2HeadersFrame headersFrame = (Http2HeadersFrame) msg;
+ CharSequence contentType = headersFrame.headers().get(GrpcHeaderEnum.GRPC_CONTENT_TYPE.header);
+ if (contentType != null && "application/grpc".equals(contentType.toString())) {
+ ctx.pipeline().addLast(new GrpcDecoder());
+ ctx.pipeline().addLast(new GrpcEncoder());
+ ctx.pipeline().addLast(serverHandlers);
+ } else {
+ ctx.pipeline().addLast(new Http2DispatchHandler());
+ }
+
+ ctx.pipeline().remove(this);
+ ctx.fireChannelRead(msg);
+ }
+ }
+}
diff --git a/core/src/main/java/org/apache/seata/core/rpc/netty/http2/Http2DispatchHandler.java b/core/src/main/java/org/apache/seata/core/rpc/netty/http2/Http2DispatchHandler.java
new file mode 100644
index 0000000000..00835ff5d9
--- /dev/null
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/http2/Http2DispatchHandler.java
@@ -0,0 +1,94 @@
+/*
+ * 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.http2;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.QueryStringDecoder;
+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.Http2DataFrame;
+import io.netty.handler.codec.http2.Http2Headers;
+import io.netty.handler.codec.http2.Http2HeadersFrame;
+import org.apache.seata.core.rpc.netty.grpc.GrpcHeaderEnum;
+import org.apache.seata.core.rpc.netty.http.ControllerManager;
+import org.apache.seata.core.rpc.netty.http.HttpInvocation;
+import org.apache.seata.core.rpc.netty.http.ParameterParser;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class Http2DispatchHandler extends ChannelDuplexHandler {
+
+ private final AtomicBoolean headerSent = new AtomicBoolean(false);
+
+ private ObjectNode requestDataNode;
+
+ private String path;
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof Http2HeadersFrame) {
+ if (headerSent.compareAndSet(false, true)) {
+ Http2Headers headers = new DefaultHttp2Headers();
+ headers.add(GrpcHeaderEnum.HTTP2_STATUS.header, String.valueOf(200));
+ ctx.writeAndFlush(new DefaultHttp2HeadersFrame(headers, false));
+ }
+
+ Http2HeadersFrame http2HeadersFrame = (Http2HeadersFrame) msg;
+ path = http2HeadersFrame.headers().path().toString();
+ QueryStringDecoder queryStringDecoder = new QueryStringDecoder(path);
+ ObjectMapper objectMapper = new ObjectMapper();
+ requestDataNode = objectMapper.createObjectNode();
+ requestDataNode.putIfAbsent("param", ParameterParser.convertParamMap(queryStringDecoder.parameters()));
+ requestDataNode.putPOJO("channel", ctx.channel());
+ if (http2HeadersFrame.isEndStream()) {
+ invoke(ctx);
+ }
+ } else if (msg instanceof Http2DataFrame) {
+ ObjectMapper objectMapper = new ObjectMapper();
+ Http2DataFrame http2DataFrame = (Http2DataFrame) msg;
+ ByteBuf byteBuf = http2DataFrame.content();
+ byte[] bytes = new byte[byteBuf.readableBytes()];
+ byteBuf.readBytes(bytes);
+ requestDataNode.putIfAbsent("body", objectMapper.readTree(bytes));
+ invoke(ctx);
+ }
+ }
+
+ private void invoke(ChannelHandlerContext ctx) throws JsonProcessingException, IllegalAccessException, InvocationTargetException {
+ ObjectMapper objectMapper = new ObjectMapper();
+ HttpInvocation httpInvocation = ControllerManager.getHttpInvocation(path);
+ Object httpController = httpInvocation.getController();
+ Method handleMethod = httpInvocation.getMethod();
+ handleMethod.setAccessible(true);
+ Object[] args = ParameterParser.getArgValues(httpInvocation.getParamMetaData(), handleMethod, requestDataNode);
+ Object result = handleMethod.invoke(httpController, args);
+ if (requestDataNode.get("channel") == null) {
+ return;
+ }
+
+ ctx.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.wrappedBuffer(objectMapper.writeValueAsBytes(result)), true));
+ }
+}
diff --git a/discovery/seata-discovery-raft/pom.xml b/discovery/seata-discovery-raft/pom.xml
index 643f792e06..f1d82b14cc 100644
--- a/discovery/seata-discovery-raft/pom.xml
+++ b/discovery/seata-discovery-raft/pom.xml
@@ -36,8 +36,8 @@
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpclient</artifactId>
+ <groupId>com.squareup.okhttp3</groupId>
+ <artifactId>okhttp</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
diff --git a/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/enums/HttpVersion.java b/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/enums/HttpVersion.java
new file mode 100644
index 0000000000..292df772e5
--- /dev/null
+++ b/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/enums/HttpVersion.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seata.discovery.registry.enums;
+
+public enum HttpVersion {
+
+ /**
+ * http
+ */
+ HTTP("http"),
+
+ /**
+ * http2
+ */
+ HTTP2("http2");
+
+ public String value;
+
+ HttpVersion(String value) {
+ this.value = value;
+ }
+}
diff --git a/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java b/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java
index 8ba0d2256e..23dd0a7193 100644
--- a/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java
+++ b/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java
@@ -18,7 +18,6 @@ package org.apache.seata.discovery.registry.raft;
import java.io.IOException;
import java.net.InetSocketAddress;
-import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -36,6 +35,7 @@ import java.util.stream.Stream;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Response;
import org.apache.seata.common.ConfigurationKeys;
import org.apache.seata.common.exception.AuthenticationFailedException;
import org.apache.seata.common.exception.RetryableException;
@@ -44,18 +44,13 @@ import org.apache.seata.common.metadata.MetadataResponse;
import org.apache.seata.common.metadata.Node;
import org.apache.seata.common.thread.NamedThreadFactory;
import org.apache.seata.common.util.CollectionUtils;
-import org.apache.seata.common.util.HttpClientUtil;
import org.apache.seata.common.util.StringUtils;
import org.apache.seata.config.ConfigChangeListener;
import org.apache.seata.config.Configuration;
import org.apache.seata.config.ConfigurationFactory;
import org.apache.seata.discovery.registry.RegistryService;
-import org.apache.http.HttpStatus;
-import org.apache.http.StatusLine;
-import org.apache.http.client.methods.CloseableHttpResponse;
-import org.apache.http.entity.ContentType;
-import org.apache.http.util.EntityUtils;
-import org.apache.http.protocol.HTTP;
+import org.apache.seata.discovery.registry.enums.HttpVersion;
+import org.apache.seata.discovery.registry.utils.OkHttpClientUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -73,6 +68,8 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
private static final String PRO_USERNAME_KEY = "username";
+ private static final String PROTOCOL_KEY = "protocol";
+
private static final String PRO_PASSWORD_KEY = "password";
private static final String AUTHORIZATION_HEADER = "Authorization";
@@ -85,6 +82,8 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
private static final String PASSWORD;
+ private static final String PROTOCOL;
+
public static String jwtToken;
private static long tokenTimeStamp = -1;
@@ -118,6 +117,7 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
TOKEN_EXPIRE_TIME_IN_MILLISECONDS = CONFIG.getLong(getTokenExpireTimeInMillisecondsKey(), 29 * 60 * 1000L);
USERNAME = CONFIG.getConfig(getRaftUserNameKey());
PASSWORD = CONFIG.getConfig(getRaftPassWordKey());
+ PROTOCOL = CONFIG.getConfig(getRaftProtocolKey());
}
private RaftRegistryServiceImpl() {
@@ -263,6 +263,12 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
REGISTRY_TYPE, PRO_PASSWORD_KEY);
}
+ private static String getRaftProtocolKey() {
+ return String.join(ConfigurationKeys.FILE_CONFIG_SPLIT_CHAR, ConfigurationKeys.FILE_ROOT_REGISTRY,
+ REGISTRY_TYPE, PROTOCOL_KEY);
+ }
+
+
private static String getTokenExpireTimeInMillisecondsKey() {
return String.join(ConfigurationKeys.FILE_CONFIG_SPLIT_CHAR, ConfigurationKeys.FILE_ROOT_REGISTRY,
REGISTRY_TYPE, TOKEN_VALID_TIME_MS_KEY);
@@ -300,7 +306,7 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
private static boolean watch() throws RetryableException {
Map<String, String> header = new HashMap<>();
- header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ header.put("Content-Type", "application/x-www-form-urlencoded");
Map<String, String> param = new HashMap<>();
String clusterName = CURRENT_TRANSACTION_CLUSTER_NAME;
Map<String, Long> groupTerms = METADATA.getClusterTerm(clusterName);
@@ -313,28 +319,45 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
if (StringUtils.isNotBlank(jwtToken)) {
header.put(AUTHORIZATION_HEADER, jwtToken);
}
- try (CloseableHttpResponse response =
- HttpClientUtil.doPost("http://" + tcAddress + "/metadata/v1/watch", param, header, 30000)) {
- if (response != null) {
- StatusLine statusLine = response.getStatusLine();
- if (statusLine != null && statusLine.getStatusCode() == HttpStatus.SC_UNAUTHORIZED) {
- if (StringUtils.isNotBlank(USERNAME) && StringUtils.isNotBlank(PASSWORD)) {
- throw new RetryableException("Authentication failed!");
- } else {
- throw new AuthenticationFailedException("Authentication failed! you should configure the correct username and password.");
- }
- }
- return statusLine != null && statusLine.getStatusCode() == HttpStatus.SC_OK;
- }
- } catch (IOException e) {
- LOGGER.error("watch cluster node: {}, fail: {}", tcAddress, e.getMessage());
- throw new RetryableException(e.getMessage(), e);
+ if (PROTOCOL.equals(HttpVersion.HTTP.value)) {
+ return watchWithHttp(header, param, tcAddress);
+ } else {
+ return watchWithHttp2(header, param, tcAddress);
}
- break;
}
return false;
}
+ private static Boolean watchWithHttp2(Map<String, String> header, Map<String, String> param, String tcAddress) throws RetryableException {
+ try (Response response =
+ OkHttpClientUtil.doPostV2("http://" + tcAddress + "/metadata/v2/watch", param, header, 30000)) {
+ int code = response.code();
+ return code == 200;
+ } catch (IOException e) {
+ LOGGER.error("watch cluster node: {}, fail: {}", tcAddress, e.getMessage());
+ throw new RetryableException(e.getMessage(), e);
+ }
+ }
+
+ private static Boolean watchWithHttp(Map<String, String> header, Map<String, String> param, String tcAddress) throws RetryableException {
+ try (Response response =
+ OkHttpClientUtil.doPostV1("http://" + tcAddress + "/metadata/v1/watch", param, header, 30000)) {
+ int statusCode = response.code();
+ if (statusCode == 401) {
+ if (StringUtils.isNotBlank(USERNAME) && StringUtils.isNotBlank(PASSWORD)) {
+ throw new RetryableException("Authentication failed!");
+ } else {
+ throw new AuthenticationFailedException("Authentication failed! you should configure the correct username and password.");
+ }
+ }
+
+ return statusCode == 200;
+ } catch (IOException e) {
+ LOGGER.error("watch cluster node: {}, fail: {}", tcAddress, e.getMessage());
+ throw new RetryableException(e.getMessage(), e);
+ }
+ }
+
@Override
public List<InetSocketAddress> refreshAliveLookup(String transactionServiceGroup,
List<InetSocketAddress> aliveAddress) {
@@ -363,7 +386,7 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
private static void acquireClusterMetaData(String clusterName, String group) throws RetryableException {
String tcAddress = queryHttpAddress(clusterName, group);
Map<String, String> header = new HashMap<>();
- header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ header.put("Content-Type", "application/x-www-form-urlencoded");
if (isTokenExpired()) {
refreshToken(tcAddress);
}
@@ -374,22 +397,20 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
Map<String, String> param = new HashMap<>();
param.put("group", group);
String response = null;
- try (CloseableHttpResponse httpResponse =
- HttpClientUtil.doGet("http://" + tcAddress + "/metadata/v1/cluster", param, header, 1000)) {
- if (httpResponse != null) {
- int statusCode = httpResponse.getStatusLine().getStatusCode();
- if (statusCode == HttpStatus.SC_OK) {
- response = EntityUtils.toString(httpResponse.getEntity(), StandardCharsets.UTF_8);
- } else if (statusCode == HttpStatus.SC_UNAUTHORIZED) {
- if (StringUtils.isNotBlank(USERNAME) && StringUtils.isNotBlank(PASSWORD)) {
- refreshToken(tcAddress);
- throw new RetryableException("Token refreshed, retrying request.");
- } else {
- throw new AuthenticationFailedException("Authentication failed! you should configure the correct username and password.");
- }
+ try (Response httpResponse =
+ OkHttpClientUtil.doGet("http://" + tcAddress + "/metadata/v1/cluster", param, header, 1000)) {
+ int code = httpResponse.code();
+ if (code == 200) {
+ response = httpResponse.message();
+ } else if (code == 401) {
+ if (StringUtils.isNotBlank(USERNAME) && StringUtils.isNotBlank(PASSWORD)) {
+ refreshToken(tcAddress);
+ throw new RetryableException("Token refreshed, retrying request.");
} else {
throw new AuthenticationFailedException("Authentication failed! you should configure the correct username and password.");
}
+ } else {
+ throw new AuthenticationFailedException("Authentication failed! you should configure the correct username and password.");
}
MetadataResponse metadataResponse;
if (StringUtils.isNotBlank(response)) {
@@ -416,13 +437,14 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
param.put(PRO_USERNAME_KEY, USERNAME);
param.put(PRO_PASSWORD_KEY, PASSWORD);
Map<String, String> header = new HashMap<>();
- header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_JSON.getMimeType());
+ header.put("Content-Type", "application/json");
String response = null;
- try (CloseableHttpResponse httpResponse =
- HttpClientUtil.doPost("http://" + tcAddress + "/api/v1/auth/login", param, header, 1000)) {
+ try (Response httpResponse =
+ OkHttpClientUtil.doPostV1("http://" + tcAddress + "/api/v1/auth/login", param, header, 1000)) {
if (httpResponse != null) {
- if (httpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- response = EntityUtils.toString(httpResponse.getEntity(), StandardCharsets.UTF_8);
+ int code = httpResponse.code();
+ if (code == 200) {
+ response = httpResponse.body().string();
JsonNode jsonNode = OBJECT_MAPPER.readTree(response);
String codeStatus = jsonNode.get("code").asText();
if (!StringUtils.equals(codeStatus, "200")) {
diff --git a/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/utils/OkHttpClientUtil.java b/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/utils/OkHttpClientUtil.java
new file mode 100644
index 0000000000..49f251a6ae
--- /dev/null
+++ b/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/utils/OkHttpClientUtil.java
@@ -0,0 +1,150 @@
+/*
+ * 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.discovery.registry.utils;
+
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.netty.handler.codec.http.QueryStringEncoder;
+import okhttp3.FormBody;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Protocol;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+
+public class OkHttpClientUtil {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(OkHttpClientUtil.class);
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ private static final Map<Integer/*timeout*/, OkHttpClient> HTTP_CLIENT_MAP_V1 = new ConcurrentHashMap<>();
+ private static final Map<Integer/*timeout*/, OkHttpClient> HTTP_CLIENT_MAP_V2 = new ConcurrentHashMap<>();
+
+ public static Response doPostV1(String url, Map<String, String> params, Map<String, String> header,
+ int timeout) throws IOException {
+ OkHttpClient okHttpClient = HTTP_CLIENT_MAP_V1.computeIfAbsent(timeout, t -> new OkHttpClient()
+ .newBuilder()
+ .readTimeout(t, TimeUnit.SECONDS)
+ .protocols(Collections.singletonList(Protocol.HTTP_1_1))
+ .build());
+
+ return doPost(okHttpClient, url, params, header);
+ }
+
+ public static Response doPostV1(String url, String body, Map<String, String> header,
+ int timeout) throws IOException {
+ OkHttpClient okHttpClient = HTTP_CLIENT_MAP_V1.computeIfAbsent(timeout, t -> new OkHttpClient()
+ .newBuilder()
+ .readTimeout(t, TimeUnit.SECONDS)
+ .protocols(Collections.singletonList(Protocol.HTTP_1_1))
+ .build());
+
+ return doPost(okHttpClient, url, body, header);
+ }
+
+ public static Response doPostV2(String url, Map<String, String> params, Map<String, String> header,
+ int timeout) throws IOException {
+ OkHttpClient okHttpClient = HTTP_CLIENT_MAP_V2.computeIfAbsent(timeout, t -> new OkHttpClient()
+ .newBuilder()
+ .readTimeout(t, TimeUnit.SECONDS)
+ .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+ .build());
+
+ return doPost(okHttpClient, url, params, header);
+ }
+
+ private static Response doPost(OkHttpClient okHttpClient, String url, Map<String, String> params, Map<String, String> header) throws IOException {
+ QueryStringEncoder queryStringEncoder = new QueryStringEncoder(url);
+ params.forEach(queryStringEncoder::addParam);
+ Request.Builder builder = new Request.Builder();
+ String contentType = "";
+ if (header != null) {
+ header.forEach(builder::addHeader);
+ contentType = header.get("Content-Type");
+ }
+
+ RequestBody requestBody = null;
+ if (contentType != null && !contentType.isEmpty()) {
+ if ("application/x-www-form-urlencoded".equalsIgnoreCase(contentType)) {
+ FormBody.Builder formBuilder = new FormBody.Builder();
+ params.forEach(formBuilder::add);
+ requestBody = formBuilder.build();
+ } else if ("application/json".equalsIgnoreCase(contentType)) {
+ String json = OBJECT_MAPPER.writeValueAsString(params);
+ requestBody = RequestBody.create(json, MediaType.parse("application/json; charset=utf-8"));
+ }
+ }
+ if (requestBody != null) {
+ return okHttpClient.newCall(builder.url(queryStringEncoder.toString()).post(requestBody).build()).execute();
+ }
+
+ return okHttpClient.newCall(builder.url(queryStringEncoder.toString()).build()).execute();
+ }
+
+ private static Response doPost(OkHttpClient okHttpClient, String url, String body, Map<String, String> header) throws IOException {
+ QueryStringEncoder queryStringEncoder = new QueryStringEncoder(url);
+ Request.Builder builder = new Request.Builder();
+ String contentType = "";
+ if (header != null) {
+ header.forEach(builder::addHeader);
+ contentType = header.get("Content-Type");
+ }
+
+ RequestBody requestBody = null;
+ if (contentType != null && !contentType.isEmpty()) {
+ if ("application/json".equalsIgnoreCase(contentType)) {
+ requestBody = RequestBody.create(body, MediaType.parse("application/json; charset=utf-8"));
+ }
+ }
+ if (requestBody != null) {
+ return okHttpClient.newCall(builder.url(queryStringEncoder.toString()).post(requestBody).build()).execute();
+ }
+
+ return okHttpClient.newCall(builder.url(queryStringEncoder.toString()).build()).execute();
+ }
+
+
+ // get request
+ public static Response doGet(String url, Map<String, String> param, Map<String, String> header,
+ int timeout) throws IOException {
+ OkHttpClient okHttpClient = HTTP_CLIENT_MAP_V1.computeIfAbsent(timeout, t -> new OkHttpClient()
+ .newBuilder()
+ .readTimeout(t, TimeUnit.SECONDS)
+ .protocols(Collections.singletonList(Protocol.HTTP_1_1))
+ .build());
+
+ QueryStringEncoder queryStringEncoder = new QueryStringEncoder(url);
+ if (param != null) {
+ param.forEach(queryStringEncoder::addParam);
+ }
+ Request.Builder builder = new Request.Builder();
+ header.forEach(builder::addHeader);
+ Request request = builder.url(queryStringEncoder.toString()).build();
+ return okHttpClient.newCall(request).execute();
+ }
+
+}
diff --git a/discovery/seata-discovery-raft/src/test/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImplTest.java b/discovery/seata-discovery-raft/src/test/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImplTest.java
deleted file mode 100644
index e6c88fcc5c..0000000000
--- a/discovery/seata-discovery-raft/src/test/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImplTest.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * 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.discovery.registry.raft;
-
-
-import org.apache.seata.common.util.HttpClientUtil;
-import org.apache.seata.config.ConfigurationFactory;
-import org.apache.http.HttpStatus;
-import org.apache.http.StatusLine;
-import org.apache.http.client.methods.CloseableHttpResponse;
-import org.apache.http.entity.StringEntity;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import org.mockito.MockedStatic;
-import org.mockito.Mockito;
-
-import java.io.IOException;
-import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.util.Map;
-
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-
-class RaftRegistryServiceImplTest {
-
-
- @BeforeAll
- public static void beforeClass() {
- System.setProperty("service.vgroupMapping.tx", "default");
- System.setProperty("registry.raft.username", "seata");
- System.setProperty("registry.raft.password", "seata");
- System.setProperty("registry.raft.serverAddr", "127.0.0.1:8092");
- System.setProperty("registry.raft.tokenValidityInMilliseconds", "10000");
- ConfigurationFactory.getInstance();
- }
-
- @AfterAll
- public static void adAfterClass() throws Exception {
- System.clearProperty("service.vgroupMapping.tx");
- }
-
- /**
- * test whether throws exception when login failed
- */
- @Test
- public void testLoginFailed() throws IOException, NoSuchMethodException {
- String jwtToken = "null";
- String responseBody = "{\"code\":\"401\",\"message\":\"Login failed\",\"data\":\"" + jwtToken + "\",\"success\":false}";
-
- try (MockedStatic<HttpClientUtil> mockedStatic = Mockito.mockStatic(HttpClientUtil.class)) {
-
- CloseableHttpResponse mockResponse = mock(CloseableHttpResponse.class);
- StatusLine mockStatusLine = mock(StatusLine.class);
-
- when(mockResponse.getEntity()).thenReturn(new StringEntity(responseBody));
- when(mockResponse.getStatusLine()).thenReturn(mockStatusLine);
- when(mockStatusLine.getStatusCode()).thenReturn(HttpStatus.SC_OK);
-
-
- when(HttpClientUtil.doPost(any(String.class), any(Map.class), any(Map.class), any(int.class)))
- .thenReturn(mockResponse);
-
- // Use reflection to access and invoke the private method
- Method refreshTokenMethod = RaftRegistryServiceImpl.class.getDeclaredMethod("refreshToken", String.class);
- refreshTokenMethod.setAccessible(true);
- assertThrows(Exception.class, () -> refreshTokenMethod.invoke(RaftRegistryServiceImpl.getInstance(), "127.0.0.1:8092"));
-
- }
- }
-
- /**
- * test whether the jwtToken updated when refreshToken method invoked
- */
-
- @Test
- public void testRefreshTokenSuccess() throws IOException, NoSuchMethodException, InvocationTargetException, IllegalAccessException, NoSuchFieldException {
- String jwtToken = "newToken";
- String responseBody = "{\"code\":\"200\",\"message\":\"success\",\"data\":\"" + jwtToken + "\",\"success\":true}";
-
- try (MockedStatic<HttpClientUtil> mockedStatic = Mockito.mockStatic(HttpClientUtil.class)) {
-
- CloseableHttpResponse mockResponse = mock(CloseableHttpResponse.class);
- StatusLine mockStatusLine = mock(StatusLine.class);
-
- when(mockResponse.getEntity()).thenReturn(new StringEntity(responseBody));
- when(mockResponse.getStatusLine()).thenReturn(mockStatusLine);
- when(mockStatusLine.getStatusCode()).thenReturn(HttpStatus.SC_OK);
-
-
- when(HttpClientUtil.doPost(any(String.class), any(Map.class), any(Map.class), any(int.class)))
- .thenReturn(mockResponse);
-
-
- Method refreshTokenMethod = RaftRegistryServiceImpl.class.getDeclaredMethod("refreshToken", String.class);
- refreshTokenMethod.setAccessible(true);
- refreshTokenMethod.invoke(RaftRegistryServiceImpl.getInstance(), "127.0.0.1:8092");
- Field jwtTokenField = RaftRegistryServiceImpl.class.getDeclaredField("jwtToken");
- jwtTokenField.setAccessible(true);
- String jwtTokenAct = (String) jwtTokenField.get(null);
-
-
- assertEquals(jwtToken, jwtTokenAct);
-
- }
- }
-
-
- /**
- * test whether the jwtToken refreshed when it is expired
- */
-
- @Test
- public void testSecureTTL() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException, NoSuchFieldException, InterruptedException {
- Field tokenTimeStamp = RaftRegistryServiceImpl.class.getDeclaredField("tokenTimeStamp");
- tokenTimeStamp.setAccessible(true);
- tokenTimeStamp.setLong(RaftRegistryServiceImpl.class, System.currentTimeMillis());
- Method isExpiredMethod = RaftRegistryServiceImpl.class.getDeclaredMethod("isTokenExpired");
- isExpiredMethod.setAccessible(true);
- boolean rst = (boolean) isExpiredMethod.invoke(null);
- assertEquals(false, rst);
- Thread.sleep(10000);
- rst = (boolean) isExpiredMethod.invoke(null);
- assertEquals(true, rst);
- }
-
-}
diff --git a/discovery/seata-discovery-raft/src/test/resources/file.conf b/discovery/seata-discovery-raft/src/test/resources/file.conf
deleted file mode 100644
index 46c3e0401c..0000000000
--- a/discovery/seata-discovery-raft/src/test/resources/file.conf
+++ /dev/null
@@ -1,25 +0,0 @@
-#
-# 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.
-#
-
-service {
- #transaction service group mapping
- vgroupMapping.default_tx_group = "default"
- #only support when registry.type=file, please don't set multiple addresses
- default.grouplist = "127.0.0.1:8091"
- #disable seata
- disableGlobalTransaction = false
-}
\ No newline at end of file
diff --git a/discovery/seata-discovery-raft/src/test/resources/registry.conf b/discovery/seata-discovery-raft/src/test/resources/registry.conf
deleted file mode 100644
index 5a15b8eec3..0000000000
--- a/discovery/seata-discovery-raft/src/test/resources/registry.conf
+++ /dev/null
@@ -1,43 +0,0 @@
-#
-# 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.
-#
-
-registry {
- # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa、custom、raft
- type = "raft"
- raft {
- metadata-max-age-ms = 30000
- serverAddr = "127.0.0.1:8848"
- username = "seata"
- password = "seata"
- tokenValidityInMilliseconds = 1740000
- }
-}
-
-config {
- # file、nacos 、apollo、zk、consul、etcd3、springCloudConfig、custom
- type = "file"
- raft {
- metadata-max-age-ms = 30000
- serverAddr = "127.0.0.1:8848"
- }
- file {
- name = "file.conf"
- }
- custom {
- name = ""
- }
-}
diff --git a/namingserver/src/main/java/org/apache/seata/namingserver/controller/NamingController.java b/namingserver/src/main/java/org/apache/seata/namingserver/controller/NamingController.java
index a1630d3a46..eaa70e7975 100644
--- a/namingserver/src/main/java/org/apache/seata/namingserver/controller/NamingController.java
+++ b/namingserver/src/main/java/org/apache/seata/namingserver/controller/NamingController.java
@@ -72,7 +72,7 @@ public class NamingController {
@PostMapping("/unregister")
public Result<String> unregisterInstance(@RequestParam String namespace, @RequestParam String clusterName,
- @RequestParam String unit, @RequestBody NamingServerNode registerBody) {
+ @RequestParam String unit, @RequestBody NamingServerNode registerBody) {
Result<String> result = new Result<>();
boolean isSuccess = namingManager.unregisterInstance(namespace, clusterName, unit, registerBody);
if (isSuccess) {
diff --git a/script/client/spring/application.properties b/script/client/spring/application.properties
index 2a72d1e5f7..d656092dd4 100755
--- a/script/client/spring/application.properties
+++ b/script/client/spring/application.properties
@@ -128,6 +128,7 @@ seata.registry.raft.metadata-max-age-ms=30000
seata.registry.raft.username=seata
seata.registry.raft.password=seata
seata.registry.raft.tokenValidityInMilliseconds=1740000
+seata.registry.raft.protocol=http
seata.registry.consul.server-addr=127.0.0.1:8500
seata.registry.etcd3.server-addr=http://localhost:2379
diff --git a/script/client/spring/application.yml b/script/client/spring/application.yml
index a6100f0574..35f151953d 100755
--- a/script/client/spring/application.yml
+++ b/script/client/spring/application.yml
@@ -142,6 +142,8 @@ seata:
username: seata
password: seata
tokenValidityInMilliseconds: 1740000
+ protocol: http
+
file:
name: file.conf
consul:
diff --git a/server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java b/server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java
index cd8eb01ba4..dda663ce64 100644
--- a/server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java
+++ b/server/src/main/java/org/apache/seata/server/cluster/manager/ClusterWatcherManager.java
@@ -26,6 +26,14 @@ import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletResponse;
+
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.handler.codec.http.DefaultHttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
+import io.netty.handler.codec.http2.Http2StreamChannel;
import org.apache.seata.common.thread.NamedThreadFactory;
import org.apache.seata.server.cluster.listener.ClusterChangeEvent;
import org.apache.seata.server.cluster.listener.ClusterChangeListener;
@@ -86,14 +94,18 @@ public class ClusterWatcherManager implements ClusterChangeListener {
}
private void notify(Watcher<?> watcher) {
- AsyncContext asyncContext = (AsyncContext)watcher.getAsyncContext();
- HttpServletResponse httpServletResponse = (HttpServletResponse)asyncContext.getResponse();
- watcher.setDone(true);
+ Channel channel = (Channel) watcher.getAsyncContext();
+ if (channel instanceof Http2StreamChannel) {
+ // http2
+ channel.writeAndFlush(new DefaultHttp2DataFrame(Unpooled.wrappedBuffer(Unpooled.EMPTY_BUFFER)));
+ return;
+ } else {
+ // http
+ channel.writeAndFlush(new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK));
+ }
if (logger.isDebugEnabled()) {
- logger.debug("notify cluster change event to: {}", asyncContext.getRequest().getRemoteAddr());
+ logger.debug("notify cluster change event to: {}", channel.remoteAddress());
}
- httpServletResponse.setStatus(HttpServletResponse.SC_OK);
- asyncContext.complete();
}
public void registryWatcher(Watcher<?> watcher) {
diff --git a/server/src/main/java/org/apache/seata/server/controller/ClusterV2Controller.java b/server/src/main/java/org/apache/seata/server/controller/ClusterV2Controller.java
new file mode 100644
index 0000000000..eb111569b5
--- /dev/null
+++ b/server/src/main/java/org/apache/seata/server/controller/ClusterV2Controller.java
@@ -0,0 +1,47 @@
+/*
+ * 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.server.controller;
+
+
+import org.apache.seata.server.cluster.manager.ClusterWatcherManager;
+import org.apache.seata.server.cluster.watch.Watcher;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.annotation.Resource;
+import java.nio.channels.Channel;
+import java.util.Map;
+
+@RestController
+@RequestMapping("/metadata/v2")
+public class ClusterV2Controller {
+
+ @Resource
+ private ClusterWatcherManager clusterWatcherManager;
+
+ @PostMapping("/watch")
+ public void watchV2(Channel channel, @RequestParam Map<String, Object> groupTerms,
+ @RequestParam(defaultValue = "28000") int timeout) {
+ groupTerms.forEach((group, term) -> {
+ Watcher<Channel> watcher =
+ new Watcher<>(group, channel, timeout, Long.parseLong(String.valueOf(term)));
+ clusterWatcherManager.registryWatcher(watcher);
+ });
+ }
+}
diff --git a/server/src/main/java/org/apache/seata/server/spring/web/RestControllerBeanPostProcessor.java b/server/src/main/java/org/apache/seata/server/spring/web/RestControllerBeanPostProcessor.java
new file mode 100644
index 0000000000..49be631255
--- /dev/null
+++ b/server/src/main/java/org/apache/seata/server/spring/web/RestControllerBeanPostProcessor.java
@@ -0,0 +1,142 @@
+/*
+ * 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.server.spring.web;
+
+import org.apache.seata.core.rpc.netty.http.ControllerManager;
+import org.apache.seata.core.rpc.netty.http.HttpInvocation;
+import org.apache.seata.core.rpc.netty.http.ParamMetaData;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.stereotype.Component;
+import org.springframework.web.bind.annotation.DeleteMapping;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.ModelAttribute;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.PutMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Component
+public class RestControllerBeanPostProcessor implements BeanPostProcessor {
+
+ private static final List<Class<? extends Annotation>> MAPPING_CLASS = new ArrayList<>();
+ private static final Map<Class<? extends Annotation>, ParamMetaData.ParamConvertType> MAPPING_PARAM_TYPE = new HashMap<>();
+
+ static {
+ MAPPING_CLASS.add(GetMapping.class);
+ MAPPING_CLASS.add(PostMapping.class);
+ MAPPING_CLASS.add(RequestMapping.class);
+ MAPPING_CLASS.add(PutMapping.class);
+ MAPPING_CLASS.add(DeleteMapping.class);
+
+ MAPPING_PARAM_TYPE.put(RequestParam.class, ParamMetaData.ParamConvertType.REQUEST_PARAM);
+ MAPPING_PARAM_TYPE.put(RequestBody.class, ParamMetaData.ParamConvertType.REQUEST_BODY);
+ MAPPING_PARAM_TYPE.put(ModelAttribute.class, ParamMetaData.ParamConvertType.MODEL_ATTRIBUTE);
+ }
+
+ @Override
+ public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+ if (!bean.getClass().isAnnotationPresent(RestController.class)) {
+ return bean;
+ }
+
+ Class<?> httpControllerClass = bean.getClass();
+ RequestMapping requestMapping = httpControllerClass.getAnnotation(RequestMapping.class);
+ List<String> prePaths;
+ if (requestMapping != null) {
+ prePaths = Arrays.asList(requestMapping.value());
+ } else {
+ prePaths = new ArrayList<>();
+ }
+ Method[] methods = httpControllerClass.getDeclaredMethods();
+ for (Method method : methods) {
+ for (Class<? extends Annotation> annotationType : MAPPING_CLASS) {
+ Annotation annotation = method.getAnnotation(annotationType);
+ if (annotation != null) {
+ List<String> postPaths = getAnnotationValue(annotation);
+ addPathMapping(bean, prePaths, method, postPaths);
+ }
+ }
+ }
+
+ return bean;
+ }
+
+ private static List<String> getAnnotationValue(Annotation annotation) {
+ try {
+ Class<? extends Annotation> annotationClass = annotation.annotationType();
+ Method valueMethod = annotationClass.getMethod("value");
+ valueMethod.setAccessible(true);
+ return Arrays.asList((String[]) valueMethod.invoke(annotation));
+ } catch (Throwable e) {
+ return new ArrayList<>();
+ }
+ }
+
+ private static void addPathMapping(Object httpController, List<String> prePaths, Method method, List<String> postPaths) {
+ Class<?>[] parameterTypes = method.getParameterTypes();
+ Annotation[][] parameterAnnotations = method.getParameterAnnotations();
+ ParamMetaData[] paramMetaDatas = new ParamMetaData[parameterTypes.length];
+ for (int i = 0; i < parameterTypes.length; i++) {
+ Class<? extends Annotation> parameterAnnotationType = null;
+ if (parameterAnnotations[i] != null && parameterAnnotations[i].length > 0) {
+ parameterAnnotationType = parameterAnnotations[i][0].annotationType();
+ }
+
+ if (parameterAnnotationType == null) {
+ parameterAnnotationType = RequestParam.class;
+ }
+
+ ParamMetaData paramMetaData = new ParamMetaData();
+ ParamMetaData.ParamConvertType paramConvertType = MAPPING_PARAM_TYPE.get(parameterAnnotationType);
+ paramMetaData.setParamConvertType(paramConvertType);
+ paramMetaDatas[i] = paramMetaData;
+ }
+ int maxSize = Math.max(prePaths.size(), postPaths.size());
+ for (int i = prePaths.size(); i < maxSize; i++) {
+ prePaths.add("/");
+ }
+
+ for (int i = postPaths.size(); i < maxSize; i++) {
+ postPaths.add("/");
+ }
+
+ for (String prePath : prePaths) {
+ for (String postPath : postPaths) {
+ String fullPath = (prePath + "/" + postPath).replaceAll("(/)+", "/");
+ HttpInvocation httpInvocation = new HttpInvocation();
+ httpInvocation.setMethod(method);
+ httpInvocation.setController(httpController);
+ httpInvocation.setPath(fullPath);
+ httpInvocation.setParamMetaData(paramMetaDatas);
+ ControllerManager.addHttpInvocation(httpInvocation);
+ }
+ }
+ }
+
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org