android smack源码分析——接收消息以及如何解析消息
在android里面用的smack包其实叫做asmack，该包提供了两种不同的连接方式：socket和httpclient。该并且提供了很多操作xmpp协议的API，也方便各种不同自定义协议的扩展。我们不需要自己重新去定义一套接收机制来扩展新的协议，只需继承然后在类里处理自己的协议就可以了。而本文今天主要说两点，一点就是消息是如何接收的，另一点就是消息是如何通知事件的。
总的思路
1.使用socket连接服务器
2.将XmlPullParser的数据源关联到socket的InputStream
3.启动线程不断循环处理消息
4.将接收到的消息解析xml处理封装好成一个Packet包
5.将包广播给所有注册事件监听的类
逐步击破
（声明在看下面的文章时，最好先理解一下smack的使用，这样才能达到深入的理解）
（谨记：上图只显示本文章解释所要用到的类和方法，减缩了一些跟本文主题无关的代码，只留一条贯穿着从建立连接到接收消息的线。）
解析这块东西打算从最初的调用开始作为入口，抽丝剥茧，逐步揭开。
1.
PacketListener packetListener = new PacketListener() {
@Override
public void processPacket(Packet packet) {
System.out
.println("Activity----processPacket" + packet.toXML());
}
};
PacketFilter packetFilter = new PacketFilter() {
@Override
public boolean accept(Packet packet) {
System.out.println("Activity----accept" + packet.toXML());
return true;
}
};
解释：创建包的监听以及包的过滤，当有消息到时就会广播到所有注册的监听，当然前提是要通过packetFilter的过滤。
2.
connection = new XMPPConnection();
XMPPConnection在这构造函数里面主要配置ip地址和端口（super(new ConnectionConfiguration("169.254.141.109", 9991));）
3.
connection.addPacketListener(packetListener, packetFilter);
connection.connect();
注册监听，开始初始化连接。
4.
public void connect() {
// Stablishes the connection, readers and writers
connectUsingConfiguration(config);
}5.
private void connectUsingConfiguration(ConnectionConfiguration config) {
String host = config.getHost();
int port = config.getPort();
try {
this.socket = new Socket(host, port);
} catch (UnknownHostException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
initConnection();
}通过之前设置的ip和端口，建立socket对象
6.
protected void initDebugger() {
Class<?> debuggerClass = null;
try {
debuggerClass = Class.forName("com.simualteSmack.ConsoleDebugger");
Constructor<?> constructor = debuggerClass.getConstructor(
Connection.class, Writer.class, Reader.class);
debugger = (SmackDebugger) constructor.newInstance(this, writer,
reader);
reader = debugger.getReader();
} catch (ClassNotFoundException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
} catch (Exception e) {
throw new IllegalArgumentException(
"Can't initialize the configured debugger!", e);
}
}private void initReaderAndWriter() {
try {
reader = new BufferedReader(new InputStreamReader(socket
.getInputStream(), "UTF-8"));
} catch (UnsupportedEncodingException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
initDebugger();
}private void initConnection() {
// Set the reader and writer instance variables
initReaderAndWriter();
packetReader = new PacketReader(this);
addPacketListener(debugger.getReaderListener(), null);
// Start the packet reader. The startup() method will block until we
// get an opening stream packet back from server.
packetReader.startup();
}从三个方法可以看出，建立reader和writer的对象关联到socket的InputStream，实例化ConsoleDebugger，该类主要是打印出接收到的消息，给reader设置了一个消息的监听。接着建立PacketReader对象，并启动。PacketReader主要负责消息的处理和通知
7.
public class PacketReader {
private ExecutorService listenerExecutor;
private boolean done;
private XMPPConnection connection;
private XmlPullParser parser;
private Thread readerThread;
protected PacketReader(final XMPPConnection connection) {
this.connection = connection;
this.init();
}
/**
* Initializes the reader in order to be used. The reader is initialized
* during the first connection and when reconnecting due to an abruptly
* disconnection.
*/
protected void init() {
done = false;
readerThread = new Thread() {
public void run() {
parsePackets(this);
}
};
readerThread.setName("Smack Packet Reader ");
readerThread.setDaemon(true);
// create an executor to deliver incoming packets to listeners.
// we will use a single thread with an unbounded queue.
listenerExecutor = Executors
.newSingleThreadExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r,
"smack listener processor");
thread.setDaemon(true);
return thread;
}
});
resetParser();
}
/**
* Starts the packet reader thread and returns once a connection to the
* server has been established. A connection will be attempted for a maximum
* of five seconds. An XMPPException will be thrown if the connection fails.
*
*/
public void startup() {
readerThread.start();
}
/**
* Shuts the packet reader down.
*/
public void shutdown() {
done = true;
// Shut down the listener executor.
listenerExecutor.shutdown();
}
private void resetParser() {
try {
parser = XmlPullParserFactory.newInstance().newPullParser();
parser.setFeature(XmlPullParser.FEATURE_PROCESS_NAMESPACES, true);
parser.setInput(connection.reader);
} catch (XmlPullParserException xppe) {
xppe.printStackTrace();
}
}
/**
* Parse top-level packets in order to process them further.
*
* @param thread
* the thread that is being used by the reader to parse incoming
* packets.
*/
private void parsePackets(Thread thread) {
try {
int eventType = parser.getEventType();
do {
if (eventType == XmlPullParser.START_TAG) {
if (parser.getName().equals("message")) {
processPacket(PacketParserUtils.parseMessage(parser));
}
System.out.println("START_TAG");
} else if (eventType == XmlPullParser.END_TAG) {
System.out.println("END_TAG");
}
eventType = parser.next();
} while (!done && eventType != XmlPullParser.END_DOCUMENT
&& thread == readerThread);
} catch (Exception e) {
e.printStackTrace();
if (!done) {
}
}
}
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));
}
/**
* 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);
}
}
}
}创建该类时就初始化线程和ExecutorService ，接着调用resetParser() 方法为parser设置输入源（这里是重点，parser的数据都是通过这里获取），调用startup启动线程，循环监听parser，如果接收到消息根据消息协议的不同将调用PacketParserUtils类里的不同方法，这里调用parseMessage（）该方法主要处理message的消息，在该方法里分析message消息并返回packet包。返回的包将调用processPacket方法，先通知所有注册了PacketCollector的监听，接着消息（listenerExecutor.submit(new ListenerNotification(packet)); ）传递给所有注册了PacketListener的监听。这样在activity开始之前注册的那个监听事件就会触发，从而完成了整个流程。
7以上.
剩下的就是一些辅助包，很简单。比如PacketCollector 这个类，它的用处主要用来处理一些需要在发送后需要等待一个答复这样的请求。
protected synchronized void processPacket(Packet packet) {
System.out.println("PacketCollector---processPacket");
if (packet == null) {
return;
}
if (packetFilter == null || packetFilter.accept(packet)) {
while (!resultQueue.offer(packet)) {
resultQueue.poll();
}
}
}public Packet nextResult(long timeout) {
long endTime = System.currentTimeMillis() + timeout;
System.out.println("nextResult");
do {
try {
return resultQueue.poll(timeout, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) { /* ignore */
}
} while (System.currentTimeMillis() < endTime);
return null;
}
该方法就是将获取到的包，先过滤然后放到队列里，最后通过nextResult来获取包，这样就完成一个请求收一个答复。
这样整个流程就完成了，最后总结一下，如图（就这么简单^0^）：
相关新闻>>
- 发表评论
-
- 最新评论 更多>>
