init project

mpush-api/src/main/java/com/shinemo/mpush/api/Connection.java

@@ -6,7 +6,12 @@
  * Created by ohun on 2015/12/22.
  */
 public interface Connection {
+
     String getId();
 
     void send(Packet packet);
+
+    boolean isClosed();
+
+    boolean isOpen();
 }

mpush-api/src/main/java/com/shinemo/mpush/api/ConnectionInfo.java

@@ -4,16 +4,14 @@
  * Created by ohun on 2015/12/22.
  */
 public class ConnectionInfo {
-    public final String clientIp;
-    public final int clientPort;
-    public final String clientVersion;
+    public final String os;
+    public final String clientVer;
     public final String deviceId;
     public final String desKey;
 
-    public ConnectionInfo(String clientIp, int clientPort, String clientVersion, String deviceId, String desKey) {
-        this.clientIp = clientIp;
-        this.clientPort = clientPort;
-        this.clientVersion = clientVersion;
+    public ConnectionInfo(String os, String clientVer, String deviceId, String desKey) {
+        this.os = os;
+        this.clientVer = clientVer;
         this.deviceId = deviceId;
         this.desKey = desKey;
     }

mpush-api/src/main/java/com/shinemo/mpush/api/Constants.java

@@ -7,4 +7,8 @@
  */
 public interface Constants {
     Charset UTF_8 = Charset.forName("UTF-8");
+    int MAX_PACKET_SIZE = 1024;
+    int HEADER_LEN = 13;
+    byte MAGIC_NUM1 = (byte) 33;
+    byte MAGIC_NUM2 = (byte) 99;
 }

mpush-api/src/main/java/com/shinemo/mpush/api/RouterInfo.java

@@ -5,11 +5,29 @@
  */
 public class RouterInfo {
     private String ip;
+    private String os;
+    private String clientVer;
 
     public RouterInfo(String ip) {
         this.ip = ip;
     }
 
+    public String getOs() {
+        return os;
+    }
+
+    public void setOs(String os) {
+        this.os = os;
+    }
+
+    public String getClientVer() {
+        return clientVer;
+    }
+
+    public void setClientVer(String clientVer) {
+        this.clientVer = clientVer;
+    }
+
     public String getIp() {
         return ip;
     }

mpush-api/src/main/java/com/shinemo/mpush/api/exception/DecodeException.java

@@ -0,0 +1,10 @@
+package com.shinemo.mpush.api.exception;
+
+/**
+ * Created by ohun on 2015/12/23.
+ */
+public class DecodeException extends RuntimeException {
+    public DecodeException(String message) {
+        super(message);
+    }
+}

mpush-api/src/main/java/com/shinemo/mpush/api/protocol/Packet.java

@@ -2,15 +2,16 @@
 
 /**
  * Created by ohun on 2015/12/19.
- * +-----------+-------+-----+---------+--------------+------+
- * |msgic num 2| cmd 1| id 4| flags 1 | data length 4| body n
- * +-----------+------+-----+---------+--------------+------+
+ * magic(2)+cmd(1)+version(1)+flags(1)+msgId(4)+length(4)+body(n)
  */
 public class Packet {
     public byte command;
-    public int version;
+    public byte version;
     public byte flags;
     public int msgId;
-    public int msgType;
     public byte[] body;
+
+    public int getBodyLength() {
+        return body == null ? 0 : body.length;
+    }
 }

mpush-api/src/main/java/com/shinemo/mpush/api/protocol/PacketDecoder.java

@@ -0,0 +1,71 @@
+package com.shinemo.mpush.api.protocol;
+
+import com.shinemo.mpush.api.Constants;
+import com.shinemo.mpush.api.exception.DecodeException;
+import com.shinemo.mpush.api.protocol.Packet;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.DecoderException;
+
+import java.util.List;
+
+/**
+ * Created by ohun on 2015/12/19.
+ * magic(2)+length(4)+cmd(1)+version(1)+flags(1)+msgId(4)+body(n)
+ */
+public class PacketDecoder extends ByteToMessageDecoder {
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+        decodeFrames(in, out);
+    }
+
+    private void decodeFrames(ByteBuf in, List<Object> out) throws Exception {
+        try {
+            while (in.readableBytes() >= Constants.HEADER_LEN) {
+                //1.记录当前读取位置位置.如果读取到非完整的frame,要恢复到该位置,便于下次读取
+                in.markReaderIndex();
+                out.add(decodeFrame(in));
+            }
+        } catch (DecodeException e) {
+            //2.读取到不完整的frame,恢复到最近一次正常读取的位置,便于下次读取
+            in.resetReaderIndex();
+        }
+    }
+
+    private Packet decodeFrame(ByteBuf in) throws Exception {
+        int bufferSize = in.readableBytes();
+        if (in.readByte() != Constants.MAGIC_NUM1 || in.readByte() != Constants.MAGIC_NUM2) {
+            throw new RuntimeException("ERROR MAGIC_NUM");
+        }
+        int bodyLength = in.readInt();
+        if (bufferSize < (bodyLength + Constants.HEADER_LEN)) {
+            throw new DecodeException("invalid frame");
+        }
+        return readPacket(in, bodyLength);
+    }
+
+    private Packet readPacket(ByteBuf in, int bodyLength) {
+        byte command = in.readByte();
+        byte version = in.readByte();
+        byte flags = in.readByte();
+        int msgId = in.readInt();
+        byte[] body = null;
+        if (bodyLength > 0) {
+            if (bodyLength > Constants.MAX_PACKET_SIZE) {
+                throw new RuntimeException("ERROR PACKET_SIZE：" + bodyLength);
+            }
+            body = new byte[bodyLength];
+            in.readBytes(body);
+        }
+        Packet packet = new Packet();
+        packet.command = command;
+        packet.version = version;
+        packet.flags = flags;
+        packet.msgId = msgId;
+        packet.body = body;
+        return packet;
+    }
+
+}

mpush-api/src/main/java/com/shinemo/mpush/api/protocol/PacketEncoder.java

@@ -0,0 +1,29 @@
+package com.shinemo.mpush.api.protocol;
+
+import com.shinemo.mpush.api.Constants;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+
+/**
+ * Created by ohun on 2015/12/19.
+ */
+@ChannelHandler.Sharable
+public class PacketEncoder extends MessageToByteEncoder<Packet> {
+    public static final PacketEncoder INSTANCE = new PacketEncoder();
+
+    @Override
+    protected void encode(ChannelHandlerContext ctx, Packet packet, ByteBuf out) throws Exception {
+        out.writeByte(Constants.MAGIC_NUM1);
+        out.writeByte(Constants.MAGIC_NUM2);
+        out.writeInt(packet.getBodyLength());
+        out.writeByte(packet.command);
+        out.writeByte(packet.flags);
+        out.writeByte(packet.version);
+        out.writeInt(packet.msgId);
+        if (packet.getBodyLength() > 0) {
+            out.writeBytes(packet.body);
+        }
+    }
+}

mpush-connection/src/main/java/com/shinemo/mpush/connection/ConnectionServer.java

@@ -1,5 +1,7 @@
 package com.shinemo.mpush.connection;
 
+import com.shinemo.mpush.api.protocol.PacketDecoder;
+import com.shinemo.mpush.api.protocol.PacketEncoder;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
@@ -12,18 +14,21 @@
 /**
  * Created by ohun on 2015/12/22.
  */
-public class ConnectionServer implements Runnable {
-    private int port;
-
-    public void start() {
+public class ConnectionServer {
+    private final int port;
+    private EventLoopGroup bossGroup;
+    private EventLoopGroup workerGroup;
 
+    public ConnectionServer(int port) {
+        this.port = port;
     }
 
     public void stop() {
-
+        if (workerGroup != null) workerGroup.shutdownGracefully();
+        if (bossGroup != null) bossGroup.shutdownGracefully();
     }
 
-    public void run() {
+    public void start() {
         /***
          * NioEventLoopGroup 是用来处理I/O操作的多线程事件循环器，
          * Netty提供了许多不同的EventLoopGroup的实现用来处理不同传输协议。
@@ -70,8 +75,8 @@ public void run() {
             b.childHandler(new ChannelInitializer<SocketChannel>() { // (4)
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
-                    ch.pipeline().addLast(new PacketDecode());
-                    ch.pipeline().addLast(new PacketEncode());
+                    ch.pipeline().addLast(new PacketDecoder());
+                    ch.pipeline().addLast(PacketEncoder.INSTANCE);
                     ch.pipeline().addLast(new ConnectionHandler());
                 }
             });
@@ -106,8 +111,7 @@ public void initChannel(SocketChannel ch) throws Exception {
             /***
              * 优雅关闭
              */
-            workerGroup.shutdownGracefully();
-            bossGroup.shutdownGracefully();
+            stop();
         }
     }
 }

mpush-core/src/main/java/com/shinemo/mpush/core/ConnectionImpl.java

@@ -13,6 +13,7 @@ public class ConnectionImpl implements Connection {
     private ConnectionInfo info;
     private Channel channel;
     private int status = 0;
+    private long lastHeartbeatTime = 0;
 
     public void init(Channel channel) {
         this.channel = channel;
@@ -28,6 +29,16 @@ public void send(Packet packet) {
 
     }
 
+    @Override
+    public boolean isClosed() {
+        return false;
+    }
+
+    @Override
+    public boolean isOpen() {
+        return false;
+    }
+
     public ChannelFuture close() {
         return null;
     }