perf[net]: 支持http协议

net/src/main/java/com/zfoo/net/core/AbstractServer.java

@@ -94,7 +94,7 @@ public abstract class AbstractServer implements IServer {
 
         allServers.add(this);
 
-        logger.info("TcpServer started at [{}:{}]", hostAddress, port);
+        logger.info("{} started at [{}:{}]", this.getClass().getSimpleName(), hostAddress, port);
     }
 
 

net/src/main/java/com/zfoo/net/core/http/HttpServer.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright (C) 2020 The zfoo Authors
+ * Licensed 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 com.zfoo.net.core.http;
+
+import com.zfoo.net.core.AbstractServer;
+import com.zfoo.net.handler.ServerDispatcherHandler;
+import com.zfoo.net.handler.codec.http.HttpCodecHandler;
+import com.zfoo.protocol.IPacket;
+import com.zfoo.util.net.HostAndPort;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.stream.ChunkedWriteHandler;
+
+import java.util.function.Function;
+
+/**
+ * @author jaysunxiao
+ * @version 3.0
+ */
+public class HttpServer extends AbstractServer {
+
+    /**
+     * http的地址解析器
+     */
+    private Function<String, IPacket> uriResolver;
+
+    public HttpServer(HostAndPort host, Function<String, IPacket> uriResolver) {
+        super(host);
+        this.uriResolver = uriResolver;
+    }
+
+    @Override
+    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
+        return new ChannelHandlerInitializer();
+    }
+
+
+    private class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
+        @Override
+        protected void initChannel(SocketChannel channel) {
+            channel.pipeline().addLast(new HttpServerCodec());
+            channel.pipeline().addLast(new ChunkedWriteHandler());
+            channel.pipeline().addLast(new HttpObjectAggregator(64 * 1024));
+            channel.pipeline().addLast(new HttpCodecHandler(uriResolver));
+            channel.pipeline().addLast(new ServerDispatcherHandler());
+        }
+    }
+}

net/src/main/java/com/zfoo/net/core/udp/UdpServer.java

@@ -26,6 +26,8 @@ import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioDatagramChannel;
 import io.netty.util.concurrent.DefaultThreadFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * @author jaysunxiao
@@ -33,6 +35,8 @@ import io.netty.util.concurrent.DefaultThreadFactory;
  */
 public class UdpServer extends AbstractServer {
 
+    private static final Logger logger = LoggerFactory.getLogger(UdpServer.class);
+
     public UdpServer(HostAndPort host) {
         super(host);
     }
@@ -58,6 +62,8 @@ public class UdpServer extends AbstractServer {
         channel = channelFuture.channel();
 
         allServers.add(this);
+
+        logger.info("{} started at [{}:{}]", this.getClass().getSimpleName(), hostAddress, port);
     }
 
     @Override

net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java

@@ -18,7 +18,6 @@ import com.zfoo.net.handler.ServerDispatcherHandler;
 import com.zfoo.net.handler.codec.websocket.WebSocketCodecHandler;
 import com.zfoo.util.net.HostAndPort;
 import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelPipeline;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.HttpServerCodec;
@@ -45,19 +44,18 @@ public class WebsocketServer extends AbstractServer {
 
         @Override
         public void initChannel(SocketChannel channel) {
-            ChannelPipeline pipeline = channel.pipeline();
             // 编解码 http 请求
-            pipeline.addLast(new HttpServerCodec());
+            channel.pipeline().addLast(new HttpServerCodec());
             // 写文件内容，支持异步发送大的码流，一般用于发送文件流
-            pipeline.addLast(new ChunkedWriteHandler());
+            channel.pipeline().addLast(new ChunkedWriteHandler());
             // 聚合解码 HttpRequest/HttpContent/LastHttpContent 到 FullHttpRequest
             // 保证接收的 Http 请求的完整性
-            pipeline.addLast(new HttpObjectAggregator(64 * 1024));
+            channel.pipeline().addLast(new HttpObjectAggregator(64 * 1024));
             // 处理其他的 WebSocketFrame
-            pipeline.addLast(new WebSocketServerProtocolHandler("/websocket"));
+            channel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket"));
             // 编解码WebSocketFrame二进制协议
-            pipeline.addLast(new WebSocketCodecHandler());
-            pipeline.addLast(new ServerDispatcherHandler());
+            channel.pipeline().addLast(new WebSocketCodecHandler());
+            channel.pipeline().addLast(new ServerDispatcherHandler());
         }
     }
 

net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java

@@ -129,9 +129,6 @@ public class PacketDispatcher implements IPacketDispatcher {
                         }
                         return;
                     }
-                    break;
-                case NORMAL_PACKET:
-
                     break;
                 default:
                     break;

net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright (C) 2020 The zfoo Authors
+ * Licensed 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 com.zfoo.net.handler.codec.http;
+
+import com.zfoo.net.NetContext;
+import com.zfoo.net.packet.model.DecodedPacketInfo;
+import com.zfoo.net.packet.model.EncodedPacketInfo;
+import com.zfoo.net.packet.model.HttpPacketAttachment;
+import com.zfoo.net.packet.service.PacketService;
+import com.zfoo.net.util.SessionUtils;
+import com.zfoo.protocol.IPacket;
+import com.zfoo.protocol.util.JsonUtils;
+import com.zfoo.protocol.util.StringUtils;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageCodec;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.util.ReferenceCountUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.function.Function;
+
+/**
+ * @author jaysunxiao
+ * @version 3.0
+ */
+public class HttpCodecHandler extends MessageToMessageCodec<FullHttpRequest, EncodedPacketInfo> {
+
+    private static final Logger logger = LoggerFactory.getLogger(HttpCodecHandler.class);
+
+    private Function<String, IPacket> uriResolver;
+
+    public HttpCodecHandler(Function<String, IPacket> uriResolver) {
+        super();
+        this.uriResolver = uriResolver;
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext channelHandlerContext, FullHttpRequest fullHttpRequest, List<Object> list) {
+        var uri = fullHttpRequest.uri();
+
+        ByteBuf in = fullHttpRequest.content();
+
+        // 不够读一个int
+        if (in.readableBytes() <= PacketService.PACKET_HEAD_LENGTH) {
+            return;
+        }
+
+        in.markReaderIndex();
+        var length = in.readInt();
+
+        // 如果长度非法，则抛出异常断开连接
+        if (length < 0) {
+            throw new IllegalArgumentException(StringUtils.format("[session:{}]的包头长度[length:{}]非法"
+                    , SessionUtils.sessionInfo(channelHandlerContext), length));
+        }
+
+        // ByteBuf里的数据太小
+        if (in.readableBytes() < length) {
+            in.resetReaderIndex();
+            return;
+        }
+
+        ByteBuf tmpByteBuf = null;
+        try {
+            tmpByteBuf = in.readRetainedSlice(length);
+            DecodedPacketInfo packetInfo = NetContext.getPacketService().read(tmpByteBuf);
+
+            packetInfo.setPacketAttachment(HttpPacketAttachment.valueOf());
+            list.add(packetInfo);
+        } catch (Exception e) {
+            logger.error("exception异常", e);
+            throw e;
+        } catch (Throwable t) {
+            logger.error("throwable错误", t);
+            throw t;
+        } finally {
+            ReferenceCountUtil.release(tmpByteBuf);
+        }
+    }
+
+    @Override
+    protected void encode(ChannelHandlerContext channelHandlerContext, EncodedPacketInfo out, List<Object> list) {
+        try {
+            var byteBuf = channelHandlerContext.alloc().ioBuffer();
+            var httpPacketAttachment = (HttpPacketAttachment) out.getPacketAttachment();
+
+            var fullHttpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer("I am ok".getBytes()));
+
+            list.add(fullHttpResponse);
+        } catch (Exception e) {
+            logger.error("[{}]编码exception异常", JsonUtils.object2String(out), e);
+            throw e;
+        } catch (Throwable t) {
+            logger.error("[{}]编码throwable错误", JsonUtils.object2String(out), t);
+            throw t;
+        }
+    }
+}

net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright (C) 2020 The zfoo Authors
+ *
+ * Licensed 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 com.zfoo.net.packet.model;
+
+import com.zfoo.util.math.RandomUtils;
+
+/**
+ * @author jaysunxiao
+ * @version 3.0
+ */
+public class HttpPacketAttachment implements IPacketAttachment {
+
+    public static final transient short PROTOCOL_ID = 3;
+
+
+    public static HttpPacketAttachment valueOf() {
+        var attachment = new HttpPacketAttachment();
+        return attachment;
+    }
+
+    @Override
+    public PacketAttachmentType packetType() {
+        return PacketAttachmentType.HTTP_PACKET;
+    }
+
+    @Override
+    public int executorConsistentHash() {
+        return RandomUtils.randomInt();
+    }
+
+    @Override
+    public short protocolId() {
+        return PROTOCOL_ID;
+    }
+
+}

net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java

@@ -21,7 +21,7 @@ package com.zfoo.net.packet.model;
  */
 public class NoAnswerAttachment implements IPacketAttachment {
 
-    public static final transient short PROTOCOL_ID = 3;
+    public static final transient short PROTOCOL_ID = 4;
 
     /**
      * 用来在TaskManage中计算一致性hash的参数

net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java

@@ -22,32 +22,33 @@ import java.util.Map;
  */
 public enum PacketAttachmentType {
 
-    /**
-     * 正常的附加包
-     */
-    NORMAL_PACKET((byte) 0, null),
-
     /**
      * 带有同步或者异步信息的附加包
      */
-    SIGNAL_PACKET((byte) 1, SignalPacketAttachment.class),
+    SIGNAL_PACKET((byte) 0, SignalPacketAttachment.class),
 
     /**
      * 带有网关信息的附加包
      */
-    GATEWAY_PACKET((byte) 2, GatewayPacketAttachment.class),
+    GATEWAY_PACKET((byte) 1, GatewayPacketAttachment.class),
 
     /**
      * udp消息的附加包
      */
-    UDP_PACKET((byte) 3, UdpPacketAttachment.class),
+    UDP_PACKET((byte) 2, UdpPacketAttachment.class),
+
+
+    /**
+     * http消息的附加包
+     */
+    HTTP_PACKET((byte) 3, HttpPacketAttachment.class),
 
     /**
      * 无返回消息的附加包
      */
     NO_ANSWER_PACKET((byte) 4, NoAnswerAttachment.class),
 
-    
+
     ;
 
 
@@ -60,7 +61,7 @@ public enum PacketAttachmentType {
     }
 
     public static PacketAttachmentType getPacketType(byte packetType) {
-        return map.getOrDefault(packetType, PacketAttachmentType.NORMAL_PACKET);
+        return map.getOrDefault(packetType, PacketAttachmentType.NO_ANSWER_PACKET);
     }
 
     public byte getPacketType() {

net/src/test/resources/protocol.xml

@@ -7,7 +7,8 @@
         <protocol id="0" location="com.zfoo.net.packet.model.SignalPacketAttachment"/>
         <protocol id="1" location="com.zfoo.net.packet.model.GatewayPacketAttachment"/>
         <protocol id="2" location="com.zfoo.net.packet.model.UdpPacketAttachment"/>
-        <protocol id="3" location="com.zfoo.net.packet.model.NoAnswerAttachment"/>
+        <protocol id="3" location="com.zfoo.net.packet.model.HttpPacketAttachment"/>
+        <protocol id="4" location="com.zfoo.net.packet.model.NoAnswerAttachment"/>
 
 
         <protocol id="20" location="com.zfoo.net.core.gateway.model.AuthUidToGatewayCheck"/>