欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识！
您现在的位置是： 首页

(二)androidpn-server tomcat版源码解析之--push消息处理 博客分类： 项目框架 androdipnjava推送

程序员文章站 2024-02-17 08:18:46
...

在 (一)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在内存中删除该会话。

 

 

 

 

 原创文章，转载请声名出处  http://spjich.iteye.com/blog/2226149
相关标签： androdipn java 推送

上一篇： Java泛型 博客分类： JavaSE Java 

下一篇： Java concurrent 之 AQS 博客分类： j2ee java

推荐阅读