基于XMPP协议的aSmack源码分析

在研究如何实现Pushing功能期间,收集了很多关于Pushing的资料,其中有一个androidnp开源项目用的人比较多,但是由于长时间没有什么人去维护,听说bug的几率挺多的,为了以后自己的产品稳定些,所以就打算自己研究一下asmack的源码,自己做一个插件,androidnp移动端的源码中包含了一个叫做asmack的jar。

Reader和Writer

在asmack中有两个非常重要的对象PacketReader和PacketWriter,那么从类名上看Packet + (Reader/Wirter),而TCP/IP传输的数据,叫做Packet(包),asmack使用的是XMPP协议,XMPP简单讲就是使用TCP/IP协议 + XML流协议的组合。所以这个了对象的作用从字面上看应该是,写包与读包,作用为从服务端读写数据。

PacketWriter中一定含有一个Writer对象,这个Writer是一个输出流,同样的PacketReader对象中有一个Reader,而这个Reader是一个输入流,Writer和Reader对象就是一个简单的读写器,他们是从socket对象中获取出来后,经过装饰变成现在这个样子。

1 reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8"));
2 writer = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream(), "UTF-8"));

没有什么神奇的地方,主要看PacketWriter/Reader,这两个对象分别把对应的Writer和Reader引用到自己的内部进行操作,下面就先看一个PacketWriter。

    /**
     * Creates a new packet writer with the specified connection.
     *
     * @param connection the connection.
     */
    protected PacketWriter(XMPPConnection connection) {
        this.queue = new ArrayBlockingQueue<Packet>(500, true);
        this.connection = connection;
        init();
    }

还有就是PacketWriter初始化的时候将XMPPConnection对象传了进来,因为在init方法中使用到了XMPPConnection对象的writer成员,我想说的是,为什么不直接传递writer成员?而是将整个对象XMPPConnection传了过来?其实这就是设计模式的好处,我们如果每次都传递的是自己的成员,那么如果后期有改动,实现一个新的XMPPConnection与PacketWriter关联,那么老的代码维护起来是很巨大的,如果这里XMPPConnection和他的同事类PacketWriter都有相对应的接口,(XMPPConnection的接口是Connection)那就更完美了,而这里用到的模式应该是中介者,不是绝对意义的中介者,由于形成中介者的条件比较高,所以实际开发中多是变形使用。PacketWriter对象在XMPPConnection中的connect方法中被初始化,它的最大作用是在其自身的内部创建了两个消息循环,其中一个用30s的heartbeats向服务器发送空白字符,保持长连接。而第二个循环则时刻从队列中主动取消息并发往服务器,而向外部提供的sendPacket方法则是向queue中添加消息,前面提到的循环机制都是在线程中工作,而消息的队列用的是ArrayBlockingQueue,这个无边界阻塞队列可以存放任何对象,这里存放的是Packet对象。

 1 public void sendPacket(Packet packet) {
 2         if (!done) {
 3             try {
 4                 queue.put(packet);
 5             }
 6             catch (InterruptedException ie) {
 7                 ie.printStackTrace();
 8                 return;
 9             }
10             synchronized (queue) {
11                 queue.notifyAll();
12             }
13         }
14     }
while (!done && (writerThread == thisThread)) {
                Packet packet = nextPacket();
                if (packet != null) {
                    synchronized (writer) {
                        writer.write(packet.toXML());
                        writer.flush();
                        // Keep track of the last time a stanza was sent to the server
                        lastActive = System.currentTimeMillis();
                    }
                }
            }

消息循环则是一个通过各种成员变量控制的while loop,第一行的nextPacket方法是向queue中获取Packet消息,并且通过weiter将包发出去,这样生产/消费的模型就搭建好了,这里需要注意的是,我删减了很多影响阅读的代码,并没有全部贴上。关于heartbeats循环其实也是一个在线程中运行的while loop,也是通过一些成员控制。wirter向服务端写了写什么?看下面的这个方法

void openStream() throws IOException {
        StringBuilder stream = new StringBuilder();
        stream.append("<stream:stream");
        stream.append(" to=\"").append(connection.getServiceName()).append("\"");
        stream.append(" xmlns=\"jabber:client\"");
        stream.append(" xmlns:stream=\"http://etherx.jabber.org/streams\"");
        stream.append(" version=\"1.0\">");
        writer.write(stream.toString());
        writer.flush();
    }

XML,没错,这也是符合XMPP协议规范的一种表现吧,至于更多XMPP协议的好处,由于本人的经验有限,就不多做点评,希望后续会对其深入了解。

下面看一个PacketReader这个类都包含了什么职责。

PacketReader

PacketReader所有的核心逻辑都在一个线程中完成的,PacketReader的工作很专注,同样的在一个while loop中 不停的解析、刷新reader对象、同时作为事件源发送解析过后的各种Packet,解析这里用的是Android独特的Pull解析,Pull解析的特点事件驱动,在这里被完全的利用了起来,随着不同的标签,PacketReader都会做出不同的处理,处理完这些数据用不同Pocket对象封装,最后,分发出去,由监听者做最后的业务处理。

readerThread = new Thread() {
    public void run() {
        parsePackets(this);
    }
};

由于解析过程的代码量过于多,我写到什么地方就分解什么地方,大家有时间最好自己看源码。

一、初始化/重置解析器

private void resetParser() {
    try {
        //用的是Pull解析
        parser = XmlPullParserFactory.newInstance().newPullParser();
        parser.setFeature(XmlPullParser.FEATURE_PROCESS_NAMESPACES, true);
        parser.setInput(connection.reader);
    }
    catch (XmlPullParserException xppe) {
        xppe.printStackTrace();
    }
}

上面这个resetParser方法还会在解析的过程中碰到不同的业务需求会不断的被调用,有用和业务逻辑比较紧密,没什么技术含量,关键是要看解析的方式和同时作为事件源发送解析过后的各种Packet,这两部分的设计,是非常的迷人的。

二、解析

do {
    if (eventType == XmlPullParser.START_TAG) {
        if (parser.getName().equals("message")) {
            processPacket(PacketParserUtils.parseMessage(parser));
    }
    else if (parser.getName().equals("iq")) {
        processPacket(PacketParserUtils.parseIQ(parser, connection));
    }
    else if (parser.getName().equals("presence")) {
        processPacket(PacketParserUtils.parsePresence(parser));
    }

PacketParserUtils是一个工具类,各个静态方法传入的还是Parser对象,内部同样的使用Pull的方式进行解析,但是由于Pull是驱动解析,不会无故的浪费资源只会加载感兴趣的内容,试想一下,如果这里用Dom解析……PacketParserUtils的这些静态解析方法返回的实例对象也不一样,从方法名可以看出有IQ、message、presence等,他们的父类为Packet,这些对象又被执行processPacket方法的时候传入

private void processPacket(Packet packet) {
    if (packet == null) {
        return;
    }

// Loop through all collectors and notify the appropriate ones.
    for (PacketCollector collector: connection.getPacketCollectors()) {
        collector.processPacket(packet);
    }

// Deliver the incoming packet to listeners.
    listenerExecutor.submit(new ListenerNotification(packet));
}

processPacket方法内部有一个循环来转调collector.processPacket(packet);方法,前提是connection.getPacketCollectors()内部有货,到目前位置都没有涉及到PacketCollector这个接口的内容,他的作用其实是一个观察者模式中的执行者的作用,也就是传说中的监听器,凡是注册了它的对象,都可以通过processPacket这个抽象方法,监听packet的变化。可是到现在任何对象都没有注册它,所以这个Loop还没有作用,因为目前我们还处在连接的步骤(还没绕出来)。

listenerExecutor.submit(new ListenerNotification(packet));其中ListenerNotification是个Runnable
/**
 * A runnable to notify all listeners of a packet.
 */
private class ListenerNotification implements Runnable {

private Packet packet;

public ListenerNotification(Packet packet) {
        this.packet = packet;
    }

public void run() {
        for (ListenerWrapper listenerWrapper : connection.recvListeners.values()) {
            listenerWrapper.notifyListener(packet);
        }
    }
}

我们上面看到listenerExecutor是一个线程池,在线程池中执行了一个凡是注册了ListenerWrapper的对象,都将接收到packet,同样的,到目前为止没有对象注册,(在RegisterTask过程中ListenerWrapper被注册)

else if (eventType == XmlPullParser.END_TAG) {
    if (parser.getName().equals("stream")) {
        // Disconnect the connection
        connection.disconnect();
    }
}

当文档读取结束是将断开连接

void cleanup() {
    connection.recvListeners.clear();
    connection.collectors.clear();
}

看到了吗,只是将监听器接口集合清空而已,并没有断开连接,或者取消消息循环

PacketReader对象的startup方法比较复杂,大体上执行了读取流,并将解析好的Packet对象发送给观察者,由观察者继续后续操作,目前观察者还没有出现,还有就是使用了线程池和令牌来操作执行线程,而且维护了一个connectionID成员,这个成员的作用还需要再看,这就不多说了。
关于Packet对象,packet对象有很多子类,上面举例了3个,其实还有很多,都是在parser时封装的
AuthMechanism\Challenge\Failure\IQ\Message\Presence\Response\Success
还有就是Pull解析的优点体现了出来,可以一个parser对象包含了很多信息,但可能没到一个时刻我们需要的信息只是一小部分,这样用Pull解析的驱动式就大大减少了冗余的过程,PacketReader对象使用了2个监听器集合对象,PacketCollector、listenerWrapper,还是那句话,还没看到观察者,所以还不知道什么情况下需要注册这两个监听。
到目前位置packetReader.startup()方法终于告一个段落了。

register过程分析

RegisterTask这个task在运行中,添加了一个监听,上面说道的PacketReader中有一个消息机制,在不停的解析服务器返回的结果,然后将解析过后的包分发给各个监听器(观察者),而register中就注册了一个监听器,比较有意思的是,监听器被注册时还加了一个过滤器,这个过滤器的目的是监听器只接收自己感兴趣的内容,这个设计真的很赞。这样就不必在数据源头PacketReader中对数据进行过滤了,只要后期扩展自己Packet和自己的过滤器,就能达到排除自己不关心的信息的功能。

Registration registration = new Registration();
    PacketFilter packetFilter = new AndFilter(new PacketIDFilter(registration.getPacketID()), new PacketTypeFilter(IQ.class));

其中Registration的类型其实一个IQ的子类,IQ是Packet的子类。
AndFilter是PacketFilter的子类,PacketFilter的种类型有很多,也可以自己扩展,AndFilter就是其中一个、PacketTypeFilter也是、PacketIDFilter也是,
其中PacketTypeFilter的构造方法传入一个IQ.class,其实就是通过这个类文件来过滤packet,这个PacketTypeFilter就是要设置关心的Packet,这里面它告诉监听器,只接收类型为IQ的Packet,这些Filter中都有一个关键方法,accept(Packet packet).这个accept方法每个Filter的实现方式都不一样,我们可可以扩展自己的Filter并且重写这个方法,最有意思的是AndFilter这个类,他的构造方法传入的是一个动态数组,类型为PacketFilter,你可以传入你需要的过滤器,将他们当成组合条件使用来过滤Packet,这个就是典型的装饰设计模式和职责链模式的组合使用。

注册监听器

 1 PacketListener packetListener = new PacketListener() {
 2     //这一部分就是监听器接收到Packet后执行的后续操作
 3     public void processPacket(Packet packet) {
 4         Log.d("RegisterTask.PacketListener", "processPacket().....");
 5         Log.d("RegisterTask.PacketListener", "packet=" + packet.toXML());
 6
 7         if (packet instanceof IQ) {
 8             IQ response = (IQ) packet;
 9             if (response.getType() == IQ.Type.ERROR) {
10                 if (!response.getError().toString().contains("409")) {
11                     Log.e(LOGTAG,
12                             "Unknown error while registering XMPP account! "
13                                     + response.getError()
14                                             .getCondition());
15                 }
16             } else if (response.getType() == IQ.Type.RESULT) {
17                 xmppManager.setUsername(newUsername);
18                 xmppManager.setPassword(newPassword);
19                 Log.d(LOGTAG, "username=" + newUsername);
20                 Log.d(LOGTAG, "password=" + newPassword);
21
22                 Editor editor = sharedPrefs.edit();
23                 editor.putString(Constants.XMPP_USERNAME,
24                         newUsername);
25                 editor.putString(Constants.XMPP_PASSWORD,
26                         newPassword);
27                 editor.commit();
28                 Log
29                         .i(LOGTAG,
30                                 "Account registered successfully");
31                 //执行task
32                 xmppManager.runTask();
33             }
34         }
35     }
36 };

addPacketListener方法传入一个监听器和过滤器,看一下内部

/**
 * Registers a packet listener with this connection. A packet filter determines
 * which packets will be delivered to the listener. If the same packet listener
 * is added again with a different filter, only the new filter will be used.
 *
 * @param packetListener the packet listener to notify of new received packets.
 * @param packetFilter   the packet filter to use.
 */
public void addPacketListener(PacketListener packetListener, PacketFilter packetFilter) {
    if (packetListener == null) {
        throw new NullPointerException("Packet listener is null.");
    }
    ListenerWrapper wrapper = new ListenerWrapper(packetListener, packetFilter);
    recvListeners.put(packetListener, wrapper);
}

可以看到,监听器和过滤器被 ListenerWrapper 再次封装,后续的recvListeners这个集合将ListenerWrapper收入囊中,好整个注册过程完毕,就等待接收信息了,那么发送信息的地方在什么地方呢?分析connect过程时,上面的PacketReader中已经开始循环发送了,代码如下

listenerExecutor.submit(new ListenerNotification(packet));其中ListenerNotification是个Runnable

/**
 * A runnable to notify all listeners of a packet.
 */
private class ListenerNotification implements Runnable {

    private Packet packet;

    public ListenerNotification(Packet packet) {
        this.packet = packet;
    }

    public void run() {
        for (ListenerWrapper listenerWrapper : connection.recvListeners.values()) {
            listenerWrapper.notifyListener(packet);
        }
    }
}

而listenerWrapper的notifyListener(packet)内部,使用了传入的过滤器对Packet进行了过滤

/**
 * Notify and process the packet listener if the filter matches the packet.
 *
 * @param packet the packet which was sent or received.
 */
public void notifyListener(Packet packet) {
    if (packetFilter == null || packetFilter.accept(packet)) {
        packetListener.processPacket(packet);
    }

而具体的过滤机制还是转调了传入的过滤器本身的过滤方式accept,非常的灵活。过滤完的Packet将被发送出去

这个方法connection.sendPacket(registration);将一个Registration对象发了出去,

public void sendPacket(Packet packet) {
    if (!isConnected()) {
        throw new IllegalStateException("Not connected to server.");
    }
    if (packet == null) {
        throw new NullPointerException("Packet is null.");
    }
    packetWriter.sendPacket(packet);
}

内部转调的是 packetWriter.sendPacket(packet);以前提到过PacketWirter中有两个循环机制,其中一个就是在不停的访问队列来获取Packet,而这个sendPacket方法就是将消息写入队列中供消费者使用。

/**
 * Sends the specified packet to the server.
 *
 * @param packet the packet to send.
 */
public void sendPacket(Packet packet) {
    if (!done) {
        // Invoke interceptors for the new packet that is about to be sent. Interceptors
        // may modify the content of the packet.
        //内部执行了一个发送数据源的动作,也是为某些监听器对象服务的interceptorWrapper.notifyListener(packet);
        connection.firePacketInterceptors(packet);

        try {
            //将一个Packet对象放入到阻塞队列中,在上面的witerPacket方法中的wile循环中发送出去
            queue.put(packet);
        }
        catch (InterruptedException ie) {
            ie.printStackTrace();
            return;
        }
        synchronized (queue) {
            queue.notifyAll();
        }

        // Process packet writer listeners. Note that we‘re using the sending
        // thread so it‘s expected that listeners are fast.
        connection.firePacketSendingListeners(packet);
    }
}   

其实,注册的过程就是在注册监听,这样在有消息发出时,才可以根据业务需求对消息进行接收和处理。

http://www.cnblogs.com/rioder/archive/2013/01/23/2873176.html

时间: 2024-08-05 19:11:22

基于XMPP协议的aSmack源码分析的相关文章

Java集合基于JDK1.8的ArrayList源码分析

本篇分析ArrayList的源码,在分析之前先跟大家谈一谈数组.数组可能是我们最早接触到的数据结构之一,它是在内存中划分出一块连续的地址空间用来进行元素的存储,由于它直接操作内存,所以数组的性能要比集合类更好一些,这是使用数组的一大优势.但是我们知道数组存在致命的缺陷,就是在初始化时必须指定数组大小,并且在后续操作中不能再更改数组的大小.在实际情况中我们遇到更多的是一开始并不知道要存放多少元素,而是希望容器能够自动的扩展它自身的容量以便能够存放更多的元素.ArrayList就能够很好的满足这样的

Java -- 基于JDK1.8的ThreadLocal源码分析

1,最近在做一个需求的时候需要对外部暴露一个值得应用  ,一般来说直接写个单例,将这个成员变量的值暴露出去就ok了,但是当时突然灵机一动(现在回想是个多余的想法),想到handle源码里面有使用过ThreadLocal这个类,想了想为什么不想直接用ThreadLocal保存数据源然后使用静态方法暴露出去呢,结果发现使用ThreadLocal有时候会获取不到值,查了下原因原来同事是在子线程中调用的(捂脸哭泣),所以还是要来看一波源码,看看ThreadLocal底层实现,适用于哪些场景 2,我们现在

AFNetworking源码分析

来源:zongmumask 链接:http://www.jianshu.com/p/8eac5b1975de 简述 在iOS开发中,与直接使用苹果框架中提供的NSURLConnection或NSURLSession进行网络请求相比,使用AFNetworking会有哪些好处?当同时发起多个网络请求AFNetworking是如何实现并发的,在并发的时候,AFNetworking是如何管理线程的?苹果重构NSURLConnetion推出新的网络加载系统NSURLSession解决了什么问题或者是与NS

docker 源码分析 一(基于1.8.2版本),docker daemon启动过程;

最近在研究golang,也学习一下比较火的开源项目docker的源代码,国内比较出名的docker源码分析是孙宏亮大牛写的一系列文章,但是基于的docker版本有点老:索性自己就git 了一下最新的代码研读: docker是c/s的架构,分为docker client 和 docker daemon,client端发送命令,daemon端负责完成client发送过来的命令(如获取和存储镜像.管理容器等).两者之间可以通过TCP,HTTP和UNIX SOCKET来进行通信: docker的启动入口

Spark资源调度机制源码分析--基于spreadOutApps及非spreadOutApps两种资源调度算法

Spark资源调度机制源码分析--基于spreadOutApps及非spreadOutApps两种资源调度算法 1.spreadOutApp尽量平均分配到每个executor上: 2.非spreadOutApp尽量在使用单个executor的资源. 源码分析 org.apache.spark.deploy.master.Master 1.首先判断,master状态不是ALIVE的话,直接返回2.调度driver3. Application的调度机制(核心之核心,重中之重) 源码如下: 1 /*

AtomicInteger源码分析——基于CAS的乐观锁实现

AtomicInteger源码分析--基于CAS的乐观锁实现 1. 悲观锁与乐观锁 我们都知道,cpu是时分复用的,也就是把cpu的时间片,分配给不同的thread/process轮流执行,时间片与时间片之间,需要进行cpu切换,也就是会发生进程的切换.切换涉及到清空寄存器,缓存数据.然后重新加载新的thread所需数据.当一个线程被挂起时,加入到阻塞队列,在一定的时间或条件下,在通过notify(),notifyAll()唤醒回来.在某个资源不可用的时候,就将cpu让出,把当前等待线程切换为阻

vlc源码分析(七) 调试学习HLS协议

HTTP Live Streaming(HLS)是苹果公司提出来的流媒体传输协议.与RTP协议不同的是,HLS可以穿透某些允许HTTP协议通过的防火墙. 一.HLS播放模式 (1) 点播模式(Video on demand, VOD) 点播模式是指当前时间点可以获取到所有index文件和ts文件,二级index文件中记录了所有ts文件的地址.这种模式允许客户端访问全部内容.上面的例子中就是一个点播模式下的m3u8的结构. (2) 直播模式(Live) 直播模式是指实时生成M3u8和ts文件.它的

协议的注册与维护——ndpi源码分析

在前面的文章中,我们对ndpi中的example做了源码分析.这一次我们将尽可能深入的了解ndpi内部的结构和运作.我们将带着下面三个目的(问题)去阅读ndpi的源代码. 1.ndpi内部是怎么样注册和维护需要检测的协议呢? 2.ndpi在初始化的过程中,做了怎么样的工作? 3.ndpi在底层的实现中具体又是使用怎样的数据结构? 注:这里限于篇幅,本文章指针对使用中的初始化部分进行源码分析.主体的分析函数和具体的各个协议将在后面的文中陆续介绍.如果有不正确或者理解不到位的地方,欢迎大家一起讨论.

基于TCP网络通信的自动升级程序源码分析-客户端请求服务器上的升级信息

每次升级,客户端都会获取服务器端存放在upgradefile文件夹下的需要升级的文件和升级信息配置文件(即upgradeconfig.xml文件) 我们来看一下代码 //升级信息配置文件相对应的类 ( 升级信息配置文件是由这个类转化成的) private UpgradeConfig upgradeConfig = null; //客户端存储升级配置文件的地址 是放在客户端根目录下的 (就是把服务器 upgradefile/upgradeconfig.xml下载到客户端存放的位置) string