在?(一)androidpn-server tomcat版源码解析之--项目启动这篇中，已经描述了整个推送服务器的启动过程，并且把握到了消息的入口即XmppIoHandler这个类，今天我将继续往下分析下面的核心代码，主要分为3大块，链接创建，消息的发送，链接关闭。

先贴一段XmppIoHandler的部分代码

/**     * Invoked from an I/O processor thread when a new connection has been created.     */    public void sessionCreated(IoSession session) throws Exception {        log.debug("sessionCreated()...");        session.getConfig().setBothIdleTime(IDLE_TIME);;    }    /**     * Invoked when a connection has been opened.     */    public void sessionOpened(IoSession session) throws Exception {        log.debug("sessionOpened()...");        log.debug("remoteAddress=" + session.getRemoteAddress());        // Create a new XML parser        XMLLightweightParser parser = new XMLLightweightParser("UTF-8");        session.setAttribute(XML_PARSER, parser);        // Create a new connection        Connection connection = new Connection(session);        session.setAttribute(CONNECTION, connection);        session.setAttribute(STANZA_HANDLER, new StanzaHandler(serverName,                connection));    }    /**     * Invoked when a connection is closed.     */    public void sessionClosed(IoSession session) throws Exception {        log.debug("sessionClosed()...");        Connection connection = (Connection) session.getAttribute(CONNECTION);        connection.close();    }    /**     * Invoked with the related IdleStatus when a connection becomes idle.     */    public void sessionIdle(IoSession session, IdleStatus status)            throws Exception {        log.debug("sessionIdle()...");        Connection connection = (Connection) session.getAttribute(CONNECTION);        if (log.isDebugEnabled()) {            log.debug("Closing connection that has been idle: " + connection);        }        connection.close();    }    /**     * Invoked when any exception is thrown.     */    public void exceptionCaught(IoSession session, Throwable cause)            throws Exception {        log.debug("exceptionCaught()...");        log.error(cause);    }    /**     * Invoked when a message is received.     */    public void messageReceived(IoSession session, Object message)            throws Exception {        log.debug("messageReceived()...");        log.debug("RCVD: " + message);        // Get the stanza handler        StanzaHandler handler = (StanzaHandler) session                .getAttribute(STANZA_HANDLER);        // Get the XMPP packet parser        int hashCode = Thread.currentThread().hashCode();        XMPPPacketReader parser = parsers.get(hashCode);        if (parser == null) {            parser = new XMPPPacketReader();            parser.setXPPFactory(factory);            parsers.put(hashCode, parser);        }        // The stanza handler processes the message        try {            handler.process((String) message, parser);        } catch (Exception e) {            log.error(                    "Closing connection due to error while processing message: "                            + message, e);            Connection connection = (Connection) session                    .getAttribute(CONNECTION);            connection.close();        }    }    /**     * Invoked when a message written by IoSession.write(Object) is sent out.     */    public void messageSent(IoSession session, Object message) throws Exception {        log.debug("messageSent()...");    }

• ?链接创建

androidpn-server的消息处理步奏为：?sessionCreated->sessionOpened->Router->handle

对应的处理类为：XmppIoHandler- >StanzaHandler->PacketRouterIQAuthHandler.java->IQHandler.java

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? IQRegisterHandler.java

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? IQRosterHandler.java

?

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? PresenceUpdateHandler.java

androidpn是以xmpp为协议的，要看懂以下的处理步奏也就必须补充点xmpp的一个相关知识

xmpp的三种消息类型:

message：message是一种基本推送消息方法，它不要求响应。

presence：presence用来表明用户的状态，如：online、away、dnd(请勿打扰)等

iq：一种请求／响应机制，从一个实体从发送请求，另外一个实体接受请求，并进行响应。

XmppIoHandler.java-> ?sessionCreated：

public void sessionCreated(IoSession session) throws Exception {        log.debug("sessionCreated()...");        session.getConfig().setBothIdleTime(IDLE_TIME);;    }

?client的请求第一次到达时候为session设置了一个idletime

?

XmppIoHandler.java->?sessionOpened

public void sessionOpened(IoSession session) throws Exception {        log.debug("sessionOpened()...");        log.debug("remoteAddress=" + session.getRemoteAddress());        // Create a new XML parser        XMLLightweightParser parser = new XMLLightweightParser("UTF-8");        session.setAttribute(XML_PARSER, parser);        // Create a new connection        Connection connection = new Connection(session);        session.setAttribute(CONNECTION, connection);        session.setAttribute(STANZA_HANDLER, new StanzaHandler(serverName,                connection));    }

?初始化了xml解析器，并且创建了StanzaHandler实例，StanzaHandler这个类时后续处理的关键

StanzaHandler.java->?StanzaHandler

    /**     * Constructor.     *      * @param serverName the server name     * @param connection the connection     */    public StanzaHandler(String serverName, Connection connection) {        this.serverName = serverName;        this.connection = connection;        this.router = new PacketRouter();//该构造中初始化了三个router实现,分别是 MessageRouter,PresenceRouter,IQRouter,分别处理xmpp的三种消息类型        notificationService = ServiceLocator.getNotificationService();        notificationManager = new NotificationManager();    }

?XmppIoHandler.java->?messageReceived

    public void messageReceived(IoSession session, Object message)            throws Exception {        log.debug("messageReceived()...");        log.debug("RCVD: " + message);        // Get the stanza handler        StanzaHandler handler = (StanzaHandler) session                .getAttribute(STANZA_HANDLER);        // Get the XMPP packet parser        int hashCode = Thread.currentThread().hashCode();        XMPPPacketReader parser = parsers.get(hashCode);        if (parser == null) {            parser = new XMPPPacketReader();            parser.setXPPFactory(factory);            parsers.put(hashCode, parser);        }        // The stanza handler processes the message        try {            handler.process((String) message, parser);//###这个方法中，程序会根据xmpp的xml头来判断消息类型，并且传递到对应的Router处理类        } catch (Exception e) {            log.error(                    "Closing connection due to error while processing message: "                            + message, e);            Connection connection = (Connection) session                    .getAttribute(CONNECTION);            connection.close();        }    }

?因为当前做的需求是消息推送，而MessageRouter实用在IM这种对话场合，所以MessageRouter并没有实现处理PresenceRouter和IQRouter都有对应的功能实现。

IQRoute.java ->?IQRouter

?

public IQRouter() {        sessionManager = SessionManager.getInstance();        iqHandlers.add(new IQAuthHandler());        iqHandlers.add(new IQRegisterHandler());        iqHandlers.add(new IQRosterHandler());        notificationService = ServiceLocator.getNotificationService();    }

?构造中又实现了3和handler，用户建立好链接后又会对用户进行IQAuthHandler（鉴权），IQRegisterHandler（解析用户携带参数，往数据库中插入用户信息，将链接保存在sessionManager中），IQRosterHandler（没进行什么操作）

?

?

?

• 消息的发送

以广播消息为例

SessionManager.java

/**     * Returns a list that contains all authenticated client sessions.     *      * @return a list that contains all client sessions     */    public Collection<ClientSession> getSessions() {        return clientSessions.values();    }

?获得链接池中所有链接

Session.java

?

    /**     * Delivers the packet to the associated connection.     *      * @param packet the packet to deliver     */    public void deliver(Packet packet) {        if (connection != null && !connection.isClosed()) {            connection.deliver(packet);        }    }

?Connection.java

/**     * Delivers the packet to this connection (without checking the recipient).     *      * @param packet the packet to deliver     */    public void deliver(Packet packet) {        log.debug("SENT: " + packet.toXML());        if (!isClosed()) {            IoBuffer buffer = IoBuffer.allocate(4096);            buffer.setAutoExpand(true);            boolean errorDelivering = false;            try {                XMLWriter xmlSerializer = new XMLWriter(new IoBufferWriter(                        buffer, (CharsetEncoder) encoder.get()),                        new OutputFormat());                xmlSerializer.write(packet.getElement());                xmlSerializer.flush();                buffer.flip();                ioSession.write(buffer);//###通过mina的isSession对象传递            } catch (Exception e) {                log.debug("Connection: Error delivering packet" + "\n"                        + this.toString(), e);                errorDelivering = true;            }            if (errorDelivering) {                close();            } else {                session.incrementServerPacketCount();            }        }    }

?封装层次为isSession->Connection->Session

?

?

?

• 链接关闭

XmppIoHandler.java

/**     * Invoked when a connection is closed.     */    public void sessionClosed(IoSession session) throws Exception {        log.debug("sessionClosed()...");        Connection connection = (Connection) session.getAttribute(CONNECTION);        connection.close();    }

?Connection.java

/**     * Closes the session including associated socket connection,     * notifing all listeners that the channel is shutting down.     */    public void close() {        boolean closedSuccessfully = false;        synchronized (this) {            if (!isClosed()) {                try {                    deliverRawText("</stream:stream>", false);                } catch (Exception e) {                    // Ignore                }                if (session != null) {                    session.setStatus(Session.STATUS_CLOSED);                }                ioSession.close(false);                closed = true;                closedSuccessfully = true;            }        }        if (closedSuccessfully) {            notifyCloseListeners();        }    }

?此处使用了观察者模式,首先传递了Xmpp的结束xml“</stream:stream>”，再关闭了mina最低层的ioSession，再通知观察者，也就是androidpn封装的ClientSession在内存中删除该会话。

?

?

?

?