Support HTTP and HTTP2

changes/en-us/2.x.md

@@ -6,6 +6,7 @@ Add changes here for all PR submitted to the 2.x branch.
 
 - [[#7037](https://github.com/apache/incubator-seata/pull/7037)] support fury undolog parser
 - [[#7069](https://github.com/apache/incubator-seata/pull/7069)] Raft cluster mode supports address translation
+- [[#6801](https://github.com/apache/incubator-seata/pull/6801)] seata-server support http protocol
 
 ### bugfix:
 
@@ -41,6 +42,7 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [remind](https://github.com/remind)
 - [PeppaO](https://github.com/PeppaO)
 - [funky-eyes](https://github.com/funky-eyes)
+- [PleaseGiveMeTheCoke](https://github.com/PleaseGiveMeTheCoke)
 
 
 Also, we receive many valuable issues, questions and advices from our community. Thanks for you all.

changes/zh-cn/2.x.md

@@ -6,6 +6,7 @@
 
 - [[#7037](https://github.com/apache/incubator-seata/pull/7037)] 支持UndoLog的fury序列化方式
 - [[#7069](https://github.com/apache/incubator-seata/pull/7069)] Raft集群模式支持地址转换
+- [[#6801](https://github.com/apache/incubator-seata/pull/6801)] seata-server 支持HTTP协议的解析
 
 ### bugfix:
 
@@ -41,5 +42,6 @@
 - [remind](https://github.com/remind)
 - [PeppaO](https://github.com/PeppaO)
 - [funky-eyes](https://github.com/funky-eyes)
+- [PleaseGiveMeTheCoke](https://github.com/PleaseGiveMeTheCoke)
 
 同时，我们收到了社区反馈的很多有价值的issue和建议，非常感谢大家。

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>

common/src/main/java/org/apache/seata/common/DefaultValues.java

@@ -156,6 +156,7 @@ public interface DefaultValues {
      * The constant DEFAULT_PROTOCOL.
      */
     String DEFAULT_PROTOCOL = "seata";
+    String DEFAULT_HTTP_VERSION = "http";
 
     /**
      * The constant DEFAULT_TRANSPORT_HEARTBEAT.

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/NettyServerBootstrap.java

@@ -38,6 +38,10 @@
 import org.apache.seata.common.metadata.Node;
 import org.apache.seata.common.thread.NamedThreadFactory;
 import org.apache.seata.config.ConfigurationFactory;
+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.apache.seata.core.rpc.RemotingBootstrap;
 import org.apache.seata.discovery.registry.MultiRegistryFactory;
 import org.apache.seata.discovery.registry.RegistryService;
@@ -165,7 +169,7 @@ public void start() {
                 @Override
                 public void initChannel(SocketChannel ch) {
                     ch.pipeline().addLast(new IdleStateHandler(nettyServerConfig.getChannelMaxReadIdleSeconds(), 0, 0))
-                            .addLast(new ProtocolDetectHandler(NettyServerBootstrap.this));
+                            .addLast(new ProtocolDetectHandler(new ProtocolDetector[]{new Http2Detector(getChannelHandlers()), new SeataDetector(getChannelHandlers()), new HttpDetector()}));
                 }
             });
 

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

@@ -21,21 +21,20 @@
 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;
 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())};
+    public ProtocolDetectHandler(ProtocolDetector[] supportedProtocolDetectors) {
+        this.supportedProtocolDetectors = supportedProtocolDetectors;
     }
 
     @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;
+    }
+}