feature: seata-server support http protocol

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

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打包后的读取

common/pom.xml

@@ -52,5 +52,11 @@
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
         </dependency>
+        <!-- https://mvnrepository.com/artifact/com.squareup.okhttp3/okhttp -->
+        <dependency>
+            <groupId>com.squareup.okhttp3</groupId>
+            <artifactId>okhttp</artifactId>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 </project>

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;

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>

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

@@ -24,8 +24,7 @@
 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 ChannelHandler[] getHandlers() {
                 @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));
                 }
             })
         };

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()
+        };
+    }
+}

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

@@ -21,6 +21,7 @@
 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

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);
+    }
+}

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);
+    }
+}

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;
+    }
+}

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
+    }
+}