public final class PacketReceiver {

    private static final Logger logger = LoggerFactory.getLogger(PacketReceiver.class);
    private static List<Filter> filterList = new ArrayList<>();
    private static final Map<Command, PacketHandler> handlers = new HashMap<>();

    public void addHandler(PacketHandler handler) {
        logger.debug("register : {}", handler.cmd().name());
        handlers.put(handler.cmd(), handler);
    }

    public void addFilter(Filter filter) {
        logger.debug("filter : {}", filter.getClass());
        filterList.add(filter);
    }

    public void onReceive(Packet packet, Connection connection) {
        Command command = Command.toCMD(packet.getCmd());
        PacketHandler handler = handlers.get(command);
        logger.debug("new packet : {}", command);
        if (handler == null) return;
        for (Filter filter : filterList) {
            if (filter.exec(packet, connection).equals(FilterResult.END)) return;
        }
        try {
            handler.handle(packet, connection);
        } catch (Throwable e) {
            e.printStackTrace();
        }
    }
} public enum Command {
    HEARTBEAT(1),
    HANDSHAKE_REQ(10),
    HANDSHAKE_RESP(11),
    ERROR(2),
    LOGIN(3),
    LOGOUT(4),
    BIND(5),
    UNBIND(6),
    PUSH(7),
    TEXT_MESSAGE(100),
    UNKNOWN(-1);

    Command(int cmd) {
        this.cmd = (byte) cmd;
    }

    public final byte cmd;

    public static Command toCMD(byte b) {
        for (Command command : values()) {
            if (command.cmd == b) {
                return command;
            }
        }
        return UNKNOWN;
    }

} public class AuthFilter implements Filter {

    private Logger logger = LoggerFactory.getLogger(AuthFilter.class);

    public Set<Command> commands = new HashSet<>();

    public void addCmd(Command command) {
        commands.add(command);
    }

    @Autowired
    private MessagePack messagePack;
    @Autowired
    private SessionManager sessionManager;

    @Override
    public FilterResult exec(Packet packet, Connection connection) {
        Command command = Command.toCMD(packet.getCmd());
        logger.debug("command : {}", command);
        if (commands.contains(command)) {
            String sessionId = packet.getSessionId();
            if (sessionId == null) {
                logger.info("sessionId is null");
                sendOffline(packet,connection);
                return FilterResult.END;
            } else {
                logger.info("sessionId is [{}]", sessionId);
                Session session = sessionManager.get(sessionId);
                if (session == null) {
                    logger.info("session is Invalid : {}", sessionId);
                    sendOffline(packet,connection);
                    return FilterResult.END;
                }
                logger.debug("session is {}", session);
            }
        }
        return FilterResult.NEXT;
    }

    public void sendOffline(Packet packet, Connection connection){
        Packet re = new Packet(Command.ERROR);
        re.setSessionId(packet.getSessionId());
        Error error = new Error();
        error.setCode(ErrorCode.OFFLINE, "用户不在线");
        try {
            re.setBody(messagePack.write(error));
        } catch (IOException e) {
            e.printStackTrace();
        }
        connection.send(re);
    }

} public class HandshakeHandler extends AbstractHandler<HandshakeReq> {

    @Autowired
    private SessionManager sessionManager;
    @Autowired
    private MessagePack messagePack;

    @Override
    public Command cmd() {
        return Command.HANDSHAKE_REQ;
    }

    @Override
    public Class<HandshakeReq> getType() {
        return HandshakeReq.class;
    }

    @Override
    public void handle(HandshakeReq message, Connection connection) {
        logger.debug("接收到新的握手请求:{}", message.deviceId);
        PushSession session = new PushSession();
        session.setSessionId(SessionUtil.get());
        session.setDeviceId(message.deviceId);
        session.setOsName(message.osName);
        session.setClientVersion(message.clientVersion);
        session.setOsVersion(message.osVersion);
        sessionManager.on(session);
        Packet packet = new Packet(Command.HANDSHAKE_RESP);
        HandshakeResp handshakeResp = new HandshakeResp();
        handshakeResp.sessionId = session.getId();
        try {
            packet.setBody(messagePack.write(handshakeResp));
        } catch (IOException e) {
            e.printStackTrace();
        }
        connection.send(packet);
    }
} public final class HeartBeatHandler extends PacketHandler {

    @Override
    public Command cmd() {
        return Command.HEARTBEAT;
    }

    @Override
    public void handle(Packet packet, Connection connection) {
        logger.debug("heart : {}", connection.getChc().channel());
        connection.getChc().channel().writeAndFlush(packet);
    }

} public final class TextMessageHandler extends AbstractHandler<TextMessage> {

    @Override
    public Class<TextMessage> getType() {
        return TextMessage.class;
    }

    @Override
    public void handle(TextMessage message, Connection connection) {
        logger.debug(message.getContent());
    }

    @Override
    public Command cmd() {
        return Command.TEXT_MESSAGE;
    }
} @Sharable
public class ServerHandler extends ChannelInboundHandlerAdapter {
    private Logger logger = LoggerFactory.getLogger(ServerHandler.class);
    public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    private PacketReceiver packetReceiver;

    public ServerHandler(PacketReceiver packetReceiver) {
        this.packetReceiver = packetReceiver;
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) { // (2)
        Channel incoming = ctx.channel();
        logger.debug("handlerAdded : {}", incoming.remoteAddress());
        channels.add(ctx.channel());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) { // (3)
        Channel incoming = ctx.channel();
        logger.debug("handlerRemoved : {}", incoming.remoteAddress());
        channels.remove(ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // (4)
        Packet packet = (Packet) msg;
        Connection connection = new Connection();
        connection.setChc(ctx);
        Channel channel = ctx.channel();
        packetReceiver.onReceive(packet, connection);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception { // (5)
        Channel incoming = ctx.channel();
        logger.debug("channelActive : {}", incoming.remoteAddress());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception { // (6)
        Channel incoming = ctx.channel();
        logger.debug("channelInactive : {}", incoming.remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 当出现异常就关闭连接
        Channel incoming = ctx.channel();
        logger.error("exceptionCaught : " + incoming.remoteAddress(), cause);
        ctx.close();
    }
} public class MgsServer implements DisposableBean, Runnable {

    private static final Logger logger = LoggerFactory.getLogger(MgsServer.class);

    /**
     * 用于分配处理业务线程的线程组个数
     */
    protected static final int BIZGROUPSIZE = Runtime.getRuntime().availableProcessors() * 2; // 默认
    /**
     * 业务出现线程大小
     */
    protected static final int BIZTHREADSIZE = 4;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private Channel channel;
    private final int port;
    private String status = "stop";
    private String id;

    public MgsServer(int port, ServerHandler serverHandler, MsgPackDecode msgPackDecode, MsgPackEncode msgPackEncode) {
        super();
        this.id = UUID.randomUUID().toString();
        this.port = port;
        this.serverHandler = serverHandler;
        this.msgPackDecode = msgPackDecode;
        this.msgPackEncode = msgPackEncode;
    }

    private final ServerHandler serverHandler;
    private final MsgPackDecode msgPackDecode;
    private final MsgPackEncode msgPackEncode;

    @Override
    public void run() {
        ServerBootstrap b = new ServerBootstrap();// 引导辅助程序

        bossGroup = new NioEventLoopGroup(BIZGROUPSIZE);
        workerGroup = new NioEventLoopGroup(BIZTHREADSIZE);
        try {
            b.group(bossGroup, workerGroup);
            b.channel(NioServerSocketChannel.class);// 设置nio类型的channel
            b.childHandler(new ChannelInitializer<SocketChannel>() {// 有连接到达时会创建一个channel
                protected void initChannel(SocketChannel ch) throws Exception {
                    logger.debug("客户端:{} 初始化", ch.remoteAddress());
                    // pipeline管理channel中的Handler，在channel队列中添加一个handler来处理业务
                    ch.pipeline().addLast("frameDecoder",
                            new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                    ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));
                    ch.pipeline().addLast("decoder", msgPackDecode);
                    ch.pipeline().addLast("encoder", msgPackEncode);
                    ch.pipeline().addLast(serverHandler);
                }
            });
            b.option(ChannelOption.SO_BACKLOG, 128);
            b.childOption(ChannelOption.SO_KEEPALIVE, true);
            logger.info("server start : {}", port);
            status = "run";
            ChannelFuture f = b.bind(port).sync();// 配置完成，开始绑定server，通过调用sync同步方法阻塞直到绑定成功
            channel = f.channel();
            f.channel().closeFuture().sync();// 应用程序会一直等待，直到channel关闭
        } catch (Exception e) {
            status = "error";
            e.printStackTrace();
        }

    }

    public void stop() {
        status = "stop";
        logger.info("destroy server resources");
        if (null == channel) {
            logger.error("server channel is null");
        }
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        channel.closeFuture().syncUninterruptibly();
        channel = null;
    }

    @Override
    public void destroy() throws Exception {
        stop();
    }

    public String getId() {
        return id;
    }

    public int getPort() {
        return port;
    }

} @Sharable
public class MsgPackDecode extends MessageToMessageDecoder<ByteBuf> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
        final int length = msg.readableBytes();
        final byte[] array = new byte[length];
        msg.getBytes(msg.readerIndex(), array, 0, length);
        out.add(new MessagePack().read(array, Packet.class));
    }
} @Sharable
public class MsgPackEncode extends MessageToByteEncoder<Packet> {

    @Override
    protected void encode(ChannelHandlerContext ctx, Packet packet, ByteBuf out) throws Exception {
        out.writeBytes(new MessagePack().write(packet));
    }

}
import cn.ctodb.push.core.PacketReceiver;
import cn.ctodb.push.dto.Command;
import cn.ctodb.push.server.filter.AuthFilter;
import cn.ctodb.push.server.handler.HandshakeHandler;
import cn.ctodb.push.server.handler.HeartBeatHandler;
import cn.ctodb.push.server.handler.TextMessageHandler;
import org.msgpack.MessagePack;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.web.client.RestTemplateBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import cn.ctodb.push.server.ServerHandler;
import cn.ctodb.push.server.service.MgsServer;
import cn.ctodb.push.utils.MsgPackDecode;
import cn.ctodb.push.utils.MsgPackEncode;
import org.springframework.web.client.RestTemplate;
import java.io.IOException;
import java.net.ServerSocket;
package cn.ctodb.push.server.conf;




/**
 * All rights Reserved, Designed By www.ctodb.cn
 *
 * @version V1.0
 * @author: lichaohn@163.com
 * @Copyright: 2018 www.ctodb.cn Inc. All rights reserved.
 */
@Configuration
@EnableConfigurationProperties(ApplicationProperties.class)
public class ServerConfiguration {
    private Logger logger = LoggerFactory.getLogger(ServerConfiguration.class);

    @Autowired
    private ApplicationProperties serverProperties;

    @Autowired
    private RestTemplateBuilder builder;

    @Bean
    public RestTemplate restTemplate() {
        return builder.build();
    }

    @Bean
    public MgsServer mgsServer() throws IOException {
        if (serverProperties.getServer().getPort() < 0) {
            ServerSocket serverSocket = new ServerSocket(0); //读取空闲的可用端口
            int port = serverSocket.getLocalPort();
            serverProperties.getServer().setPort(port);
            logger.info("使用随机端口号:" + port);
        }
        return new MgsServer(serverProperties.getServer().getPort(), serverHandler(), msgPackDecode(), msgPackEncode());
    }

    @Bean
    public ServerHandler serverHandler() {
        return new ServerHandler(packetReceiver());
    }

    @Bean
    public MsgPackEncode msgPackEncode() {
        return new MsgPackEncode();
    }

    @Bean
    public MessagePack messagePack() {
        return new MessagePack();
    }

    @Bean
    public MsgPackDecode msgPackDecode() {