调通protobuf通讯

Author: Peng Hu

src/main/java/info/xiaomo/gengine/network/AbstractHandler.java

@@ -14,7 +14,7 @@ public abstract class AbstractHandler<T> implements IQueueDriverCommand {
 
     protected T message;
 
-    protected Object session;
+    protected ISession session;
 
     protected int queueId;
 
@@ -50,7 +50,7 @@ public Object getParam() {
 
     @Override
     public void setParam(Object session) {
-        this.session = session;
+        this.session = (ISession) session;
     }
 
     @Override

src/main/java/info/xiaomo/gengine/network/IMessageAndHandler.java

@@ -25,10 +25,9 @@ public interface IMessageAndHandler {
 
     /**
      * 获取handler
-     * @param  handlerName handlerName
      * @return AbstractHandler
      */
-    AbstractHandler getHandler(String handlerName);
+    AbstractHandler getHandler(int messageId);
 
     /**
      * 注册

src/main/java/info/xiaomo/gengine/network/INetworkConsumer.java

@@ -1,6 +1,5 @@
 package info.xiaomo.gengine.network;
 
-import com.google.protobuf.AbstractMessage;
 import io.netty.channel.Channel;
 
 /**
@@ -11,10 +10,10 @@ public interface INetworkConsumer {
     /**
      * 执行具体的指令
      *
-     * @param message message
+     * @param msg msg
      * @param channel channel
      */
-    void consume(AbstractMessage message, Channel channel);
+    void consume(Message msg, Channel channel);
 
 
 }

…/info/xiaomo/gengine/network/Packet.java …info/xiaomo/gengine/network/Message.javasrc/main/java/info/xiaomo/gengine/network/Packet.java renamed to src/main/java/info/xiaomo/gengine/network/Message.java src/main/java/info/xiaomo/gengine/network/Packet.java renamed to src/main/java/info/xiaomo/gengine/network/Message.java

@@ -3,7 +3,7 @@
 import lombok.Data;
 
 @Data
-public class Packet {
+public class Message {
     public static final byte HEAD_TCP = -128;
     public static final byte HEAD_UDP = 0;
     public static final byte HEAD_NEED_ACK = 64;
@@ -13,15 +13,15 @@ public class Packet {
     public static final byte PROTOCOL_JSON = 1;
     private final byte head;
     private final short sid;
-    private final int cmd;
+    private final int msgId;
     private final byte[] bytes;
 
-    public Packet(byte head, int cmd, byte[] bytes) {
+    public Message(byte head, int cmd, byte[] bytes) {
         this(head, (short) 0, cmd, bytes);
     }
 
-    public Packet(byte head, short sid, int cmd, byte[] bytes) {
-        this.cmd = cmd;
+    public Message(byte head, short sid, int msgId, byte[] bytes) {
+        this.msgId = msgId;
         this.bytes = bytes;
         this.head = head;
         this.sid = sid;

src/main/java/info/xiaomo/gengine/network/client/Client.java

@@ -2,11 +2,9 @@
 
 import com.google.protobuf.AbstractMessage;
 import com.google.protobuf.Descriptors;
-import com.google.protobuf.Message;
-import info.xiaomo.gengine.network.Packet;
+import info.xiaomo.gengine.network.Message;
 import info.xiaomo.gengine.network.handler.MessageDecoder;
 import info.xiaomo.gengine.network.handler.MessageEncoder;
-import info.xiaomo.gengine.network.pool.MessageAndHandlerPool;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.*;
 import io.netty.channel.nio.NioEventLoopGroup;
@@ -20,7 +18,6 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.*;
 
 /**
@@ -152,14 +149,14 @@ public boolean sendMsg(AbstractMessage message) {
         Channel channel = getChannel(Thread.currentThread().getId());
         if (channel != null && channel.isActive()) {
             int cmd = getMessageID(message);
-            Packet packet = new Packet(Packet.HEAD_TCP, cmd, message.toByteArray());
+            Message packet = new Message(Message.HEAD_TCP, cmd, message.toByteArray());
             channel.writeAndFlush(packet);
             return true;
         }
         return false;
     }
 
-    public static int getMessageID(Message msg) {
+    public static int getMessageID(com.google.protobuf.Message msg) {
         for (Map.Entry<Descriptors.FieldDescriptor, Object> fieldDescriptorObjectEntry : msg.getAllFields().entrySet()) {
             if (fieldDescriptorObjectEntry.getKey().getName().equals("msgId")) {
                 return ((Descriptors.EnumValueDescriptor) fieldDescriptorObjectEntry.getValue()).getNumber();

src/main/java/info/xiaomo/gengine/network/client/NetworkConsumerAdapter.java

@@ -1,7 +1,7 @@
 package info.xiaomo.gengine.network.client;
 
-import com.google.protobuf.AbstractMessage;
 import info.xiaomo.gengine.network.INetworkConsumer;
+import info.xiaomo.gengine.network.Message;
 import io.netty.channel.Channel;
 
 /**
@@ -13,7 +13,7 @@
 public class NetworkConsumerAdapter implements INetworkConsumer {
 
     @Override
-    public void consume(AbstractMessage msg, Channel channel) {
+    public void consume(Message msg, Channel channel) {
         // Nothing to do
     }
 }

src/main/java/info/xiaomo/gengine/network/handler/MessageDecoder.java

@@ -1,6 +1,6 @@
 package info.xiaomo.gengine.network.handler;
 
-import info.xiaomo.gengine.network.Packet;
+import info.xiaomo.gengine.network.Message;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ByteToMessageDecoder;
@@ -38,6 +38,6 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
         }
         byte[] bytes = new byte[length - 4];
         in.readBytes(bytes);
-        out.add(new Packet(head, cmd, bytes));
+        out.add(new Message(head, cmd, bytes));
     }
 }

src/main/java/info/xiaomo/gengine/network/handler/MessageEncoder.java

@@ -1,6 +1,6 @@
 package info.xiaomo.gengine.network.handler;
 
-import info.xiaomo.gengine.network.Packet;
+import info.xiaomo.gengine.network.Message;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToByteEncoder;
@@ -13,7 +13,7 @@
  * 比如客户端请求登录的Protobuf协议如下：
  */
 @Slf4j
-public class MessageEncoder extends MessageToByteEncoder<Packet> {
+public class MessageEncoder extends MessageToByteEncoder<Message> {
 
 
     private final int downLimit;
@@ -24,14 +24,14 @@ public MessageEncoder(int downLimit) {
 
 
     @Override
-    protected void encode(ChannelHandlerContext ctx, Packet packet, ByteBuf buf) {
-        if ((packet.getBytes().length > this.downLimit) && (log.isWarnEnabled()))
-            log.warn("packet size[" + packet.getBytes().length + "] is over limit[" + this.downLimit + "]");
-
-        buf.writeByte(packet.getHead());
-        buf.writeShort(packet.getBytes().length + 4);
-        buf.writeInt(packet.getCmd());
-        buf.writeBytes(packet.getBytes());
+    protected void encode(ChannelHandlerContext ctx, Message message, ByteBuf buf) {
+        if ((message.getBytes().length > this.downLimit) && (log.isWarnEnabled()))
+            log.warn("packet size[" + message.getBytes().length + "] is over limit[" + this.downLimit + "]");
+
+        buf.writeByte(message.getHead());
+        buf.writeShort(message.getBytes().length + 4);
+        buf.writeInt(message.getMsgId());
+        buf.writeBytes(message.getBytes());
     }
 
     @Override

src/main/java/info/xiaomo/gengine/network/handler/MessageExecutor.java

@@ -1,8 +1,8 @@
 package info.xiaomo.gengine.network.handler;
 
-import com.google.protobuf.AbstractMessage;
 import info.xiaomo.gengine.network.INetworkConsumer;
 import info.xiaomo.gengine.network.INetworkEventListener;
+import info.xiaomo.gengine.network.Message;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 
@@ -29,7 +29,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        consumer.consume((AbstractMessage) msg, ctx.channel());
+        consumer.consume((Message) msg, ctx.channel());
     }
 
     @Override

src/main/java/info/xiaomo/gengine/network/pool/MessageAndHandlerPool.java

@@ -25,7 +25,7 @@ public class MessageAndHandlerPool implements IMessageAndHandler {
      */
     private final Map<String, Integer> ids = new HashMap<>(10);
 
-    private final Map<String, Class<? extends AbstractHandler>> handlers = new HashMap<>(10);
+    private final Map<Integer, Class<? extends AbstractHandler>> handlers = new HashMap<>(10);
 
     @Override
     public AbstractMessage getMessage(int messageId) {
@@ -39,8 +39,8 @@ public int getMessageId(AbstractMessage message) {
 
 
     @Override
-    public AbstractHandler getHandler(String handlerName) {
-        Class<? extends AbstractHandler> clazz = handlers.get(handlerName);
+    public AbstractHandler getHandler(int messageId) {
+        Class<? extends AbstractHandler> clazz = handlers.get(messageId);
         if (clazz != null) {
             try {
                 return clazz.getDeclaredConstructor().newInstance();
@@ -54,7 +54,7 @@ public AbstractHandler getHandler(String handlerName) {
     @Override
     public void register(int messageId, AbstractMessage messageClazz, Class<? extends AbstractHandler> handler) {
         messages.put(messageId, messageClazz);
-        handlers.put(messageClazz.getClass().getName(), handler);
+        handlers.put(messageId, handler);
         ids.put(messageClazz.getClass().getName(), messageId);
     }