自定义protobuf编解码器

pom.xml

@@ -15,7 +15,6 @@
         <module>logic</module>
         <module>protobuf</module>
         <module>redis</module>
-        <module>serv2serv</module>
         <module>thrift</module>
         <module>thirdparty</module>
     </modules>

protobuf/pom.xml

@@ -20,5 +20,10 @@
             <artifactId>protobuf-java</artifactId>
             <version>2.6.1</version>
         </dependency>
+        <dependency>
+            <groupId>thirdparty</groupId>
+            <artifactId>thirdparty</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
     </dependencies>
 </project>

protobuf/protobuf.iml

@@ -12,6 +12,7 @@
     <orderEntry type="jdk" jdkName="1.8" jdkType="JavaSDK" />
     <orderEntry type="sourceFolder" forTests="false" />
     <orderEntry type="library" name="Maven: com.google.protobuf:protobuf-java:2.6.1" level="project" />
+    <orderEntry type="module" module-name="thirdparty" />
     <orderEntry type="library" name="Maven: org.slf4j:slf4j-api:1.7.7" level="project" />
     <orderEntry type="library" name="Maven: ch.qos.logback:logback-classic:1.1.3" level="project" />
     <orderEntry type="library" name="Maven: ch.qos.logback:logback-core:1.1.3" level="project" />

protobuf/src/main/java/protobuf/analysis/ParseMap.java

@@ -19,6 +19,7 @@ public interface Parsing{
     private static final Logger logger = LoggerFactory.getLogger(ParseMap.class);
 
     public static HashMap<Integer, ParseMap.Parsing> parseMap = new HashMap<Integer, Parsing>();
+    public static HashMap<Message, Integer> msg2ptoNum = new HashMap<Message, Integer>();
 
     public static void register(int ptoNum, ParseMap.Parsing cla) {
         if (parseMap.get(ptoNum) == null)
@@ -29,13 +30,16 @@ public static void register(int ptoNum, ParseMap.Parsing cla) {
         }
     }
 
-    public static Message parse(int ptoNum, byte[] bytes) throws IOException {
+    public static Message getMessage(int ptoNum, byte[] bytes) throws IOException {
         Parsing parser = parseMap.get(ptoNum);
         if(parser == null) {
             logger.error("UnKnown Protocol Num: {}", ptoNum);
         }
+        Message msg = parser.process(bytes);
 
-        return parser.process(bytes);
+        if(msg2ptoNum.get(msg) != null) {
+            msg2ptoNum.put(msg, ptoNum);
+        }
+        return msg;
     }
-
 }

protobuf/src/main/java/protobuf/code/ProtobufDecoder.java

@@ -1,11 +1,13 @@
 package protobuf.code;
 
+import com.google.protobuf.Message;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import protobuf.analysis.ParseMap;
 
 import java.util.List;
 
@@ -27,25 +29,17 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in,
         in.readBytes(byteBuf);
 
         try {
-            byte[] inByte = byteBuf.array();
+            /* 解密消息体
+            ThreeDES des = ctx.channel().attr(ClientAttr.ENCRYPT).get();
+            byte[] bareByte = des.decrypt(inByte);*/
 
-            // 解密消息体
-            //ThreeDES des = ctx.channel().attr(ClientAttr.ENCRYPT).get();
-            //byte[] bareByte = des.decrypt(inByte);
+            int ptoNum = in.readInt();
+            byte[] body= byteBuf.array();
 
-            /*-----------------------------------------------------------------------
-            Message msg = Message.parseFrom(bareByte);
-            logger.info("[APP-SERVER][RECV][remoteAddress:"
-                    + ctx.channel().remoteAddress() + "][total length:"
-                    + length + "][bare length:" + msg.getSerializedSize()
-                    + "]:\r\n" + msg.toString());
+            Message msg = ParseMap.getMessage(ptoNum, body);
+            out.add(msg);
+            logger.info("GateServer Received Message: length {}, ptoNum: {}", length, ptoNum);
 
-            if (msg != null) {
-                // 获取业务消息头
-                out.add(msg);
-
-            }
-            -----------------------------------------------------------------------*/
         } catch (Exception e) {
             logger.error(ctx.channel().remoteAddress() + ",decode failed.", e);
         }
@@ -59,7 +53,7 @@ int checkLength(ChannelHandlerContext ctx, ByteBuf in){
             return 0;
         }
 
-        int length = in.readUnsignedShort();
+        int length = in.readInt();
 
         if (length < 0) {
             ctx.close();

protobuf/src/main/java/protobuf/code/ProtobufEncoder.java

@@ -1,7 +1,40 @@
 package protobuf.code;
 
+import com.google.protobuf.Message;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import protobuf.analysis.ParseMap;
+
 /**
  * Created by Administrator on 2016/1/29.
  */
-public class ProtobufEncoder {
+public class ProtobufEncoder extends MessageToByteEncoder<Message> {
+    private static final Logger logger = LoggerFactory.getLogger(ProtobufEncoder.class);
+
+    @Override
+    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out)
+            throws Exception {
+
+        byte[] bytes = msg.toByteArray();// 将对象转换为byte
+        int ptoNum = ParseMap.msg2ptoNum.get(msg);
+        int length = bytes.length;
+
+        /* 加密消息体
+        ThreeDES des = ctx.channel().attr(ClientAttr.ENCRYPT).get();
+        byte[] encryptByte = des.encrypt(bytes);
+        int length = encryptByte.length;*/
+
+        ByteBuf buf = Unpooled.buffer(8 + length);
+        buf.writeInt(length);//
+        buf.writeInt(ptoNum);
+        buf.writeBytes(bytes);
+        out.writeBytes(buf);
+
+        logger.info("GateServer Send Message, remoteAddress: {}, length {}, ptoNum: {}", ctx.channel().remoteAddress(), length, ptoNum);
+
+    }
 }