perf[net]: http server的消息派发使用统一的规则

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

@@ -14,7 +14,7 @@
 package com.zfoo.net.core.http;
 
 import com.zfoo.net.core.AbstractServer;
-import com.zfoo.net.handler.HttpDispatcherHandler;
+import com.zfoo.net.handler.ServerDispatcherHandler;
 import com.zfoo.net.handler.codec.http.HttpCodecHandler;
 import com.zfoo.net.packet.model.DecodedPacketInfo;
 import com.zfoo.util.net.HostAndPort;
@@ -56,7 +56,7 @@ public class HttpServer extends AbstractServer {
             channel.pipeline().addLast(new ChunkedWriteHandler());
             channel.pipeline().addLast(new HttpObjectAggregator(64 * 1024));
             channel.pipeline().addLast(new HttpCodecHandler(uriResolver));
-            channel.pipeline().addLast(new HttpDispatcherHandler());
+            channel.pipeline().addLast(new ServerDispatcherHandler());
         }
     }
 }

net/src/main/java/com/zfoo/net/handler/HttpDispatcherHandler.java

@@ -1,45 +0,0 @@
-/*
- * 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;
-
-import com.zfoo.net.NetContext;
-import com.zfoo.net.packet.model.DecodedPacketInfo;
-import com.zfoo.net.packet.model.HttpPacketAttachment;
-import com.zfoo.net.util.SessionUtils;
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.ChannelHandlerContext;
-
-/**
- * @author jaysunxiao
- * @version 3.0
- */
-@ChannelHandler.Sharable
-public class HttpDispatcherHandler extends ServerDispatcherHandler {
-
-    @Override
-    public void channelRead(ChannelHandlerContext ctx, Object msg) {
-        var session = SessionUtils.getSession(ctx);
-        if (session == null) {
-            return;
-        }
-        DecodedPacketInfo decodedPacketInfo = (DecodedPacketInfo) msg;
-
-        var httpPacketAttachment = (HttpPacketAttachment) decodedPacketInfo.getPacketAttachment();
-        if (httpPacketAttachment.getUid() <= 0) {
-            httpPacketAttachment.useExecutorConsistentHash(session.getSid());
-        }
-
-        NetContext.getDispatcher().receive(session, decodedPacketInfo.getPacket(), decodedPacketInfo.getPacketAttachment());
-    }
-}

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

@@ -21,10 +21,7 @@ import com.zfoo.protocol.util.JsonUtils;
 import com.zfoo.protocol.util.StringUtils;
 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.HttpHeaderNames;
-import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -71,6 +68,10 @@ public class HttpCodecHandler extends MessageToMessageCodec<FullHttpRequest, Enc
             var httpResponseStatus = attachment.getHttpResponseStatus();
             if (packet.protocolId() == Message.PROTOCOL_ID) {
                 var message = (Message) packet;
+                if (message.fail()) {
+                    httpResponseStatus = HttpResponseStatus.BAD_REQUEST;
+                }
+
                 if (StringUtils.isEmpty(message.getMessage())) {
                     var fullHttpResponse = new DefaultFullHttpResponse(protocolVersion, httpResponseStatus);
                     fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);

net/src/main/java/com/zfoo/net/packet/common/Message.java

@@ -26,7 +26,8 @@ public class Message implements IPacket {
 
     public static final transient short PROTOCOL_ID = 100;
 
-    public static final Message DEFAULT = new Message();
+    public static final Message SUCCESS = valueSuccess(null);
+    public static final Message FAIL = valueSuccess(null);
 
     private byte module;
 
@@ -37,6 +38,14 @@ public class Message implements IPacket {
 
     private String message;
 
+    public boolean success() {
+        return code == 1;
+    }
+
+    public boolean fail() {
+        return code == 0;
+    }
+
     public static Message valueOf(IPacket packet, int code, String message) {
         var mess = new Message();
         mess.module = ProtocolManager.moduleByProtocolId(packet.protocolId()).getId();
@@ -55,15 +64,25 @@ public class Message implements IPacket {
         return mess;
     }
 
+    public static Message valueFail(String message) {
+        var mess = new Message();
+        mess.code = 0;
+        mess.message = message;
+        return mess;
+    }
+
+    public static Message valueSuccess(String message) {
+        var mess = new Message();
+        mess.code = 1;
+        mess.message = message;
+        return mess;
+    }
+
     @Override
     public short protocolId() {
         return PROTOCOL_ID;
     }
 
-    public boolean success() {
-        return code == 1;
-    }
-
     public byte getModule() {
         return module;
     }