开发创建XMPP“发布订阅”扩展（xmpp pubsub extend）

发布订阅（PubSub）是一个功能强大的XMPP协议扩展。用户订阅一个项目（在xmpp中叫做node），得到通知时，也即当事项节点更新时。xmpp服务器通知用户（通过message格式）。

节点类型：

Leaf node: 叶子节点，包含了发布项.

Collection node: 可以看做集合节点，它下面包含叶子.

注意：不能订阅整个Collection node，只能订阅Leaf node

<iq id="11" to="wangxin@im/PC" from="pubsub.im" type="result">
<query xmlns="http://jabber.org/protocol/disco#info">
<identity category="pubsub" name="Publish-Subscribe service" type="service"/>
<feature var="http://jabber.org/protocol/pubsub"/>
<feature var="http://jabber.org/protocol/pubsub#collections"/>
<feature var="http://jabber.org/protocol/pubsub#config-node"/>
<feature var="http://jabber.org/protocol/pubsub#create-and-configure"/>
<feature var="http://jabber.org/protocol/pubsub#create-nodes"/>
<feature var="http://jabber.org/protocol/pubsub#delete-nodes"/>
<feature var="http://jabber.org/protocol/pubsub#get-pending"/>
<feature var="http://jabber.org/protocol/pubsub#instant-nodes"/>
<feature var="http://jabber.org/protocol/pubsub#item-ids"/>
<feature var="http://jabber.org/protocol/pubsub#meta-data"/>
<feature var="http://jabber.org/protocol/pubsub#modify-affiliations"/>
<feature var="http://jabber.org/protocol/pubsub#manage-subscriptions"/>
<feature var="http://jabber.org/protocol/pubsub#multi-subscribe"/>
<feature var="http://jabber.org/protocol/pubsub#outcast-affiliation"/>
<feature var="http://jabber.org/protocol/pubsub#persistent-items"/>
<feature var="http://jabber.org/protocol/pubsub#presence-notifications"/>
<feature var="http://jabber.org/protocol/pubsub#publish"/>
<feature var="http://jabber.org/protocol/pubsub#publisher-affiliation"/>
<feature var="http://jabber.org/protocol/pubsub#purge-nodes"/>
<feature var="http://jabber.org/protocol/pubsub#retract-items"/>
<feature var="http://jabber.org/protocol/pubsub#retrieve-affiliations"/>
<feature var="http://jabber.org/protocol/pubsub#retrieve-default"/>
<feature var="http://jabber.org/protocol/pubsub#retrieve-items"/>
<feature var="http://jabber.org/protocol/pubsub#retrieve-subscriptions"/>
<feature var="http://jabber.org/protocol/pubsub#subscribe"/>
<feature var="http://jabber.org/protocol/pubsub#subscription-options"/>
<feature var="http://jabber.org/protocol/pubsub#default_access_model_open"/>
<feature var="http://jabber.org/protocol/disco#info"/>
</query>
</iq>

View Code

1.3 查询发布订阅中的某一个持久化的叶子节点

<iq type='get'
from='wangxin@im/PC'
to='pubsub.im'
id='info1'>
<query xmlns='http://jabber.org/protocol/disco#info'
node='NodeID_003'/>
</iq>

返回

<iq id="info1" to="wangxin@im/PC" from="pubsub.im" type="result">
<query xmlns="http://jabber.org/protocol/disco#info" node="NodeID_003">
<identity category="pubsub" name="null" type="leaf"/>
<feature var="http://jabber.org/protocol/pubsub"/>
<feature var="http://jabber.org/protocol/disco#info"/>
<x xmlns="jabber:x:data" type="result">
<field var="FORM_TYPE" type="hidden">
<value>http://jabber.org/protocol/pubsub#meta-data</value>
</field>
<field label="节点的简化名" var="pubsub#title" type="text-single">
<value/>
</field>
<field label="节点的描述" var="pubsub#description" type="text-single">
<value/>
</field>
<field label="Whether the node is a leaf (default) or a collection" var="pubsub#node_type" type="text-single">
<value>leaf</value>
</field>
<field label="The collection with which a node is affiliated." var="pubsub#collection" type="text-single"/>
<field label="是否允许订阅" var="pubsub#subscribe" type="boolean">
<value>1</value>
</field>
<field label="强制设置新的订阅" var="pubsub#subscription_required" type="boolean">
<value>0</value>
</field>
<field label="用事件通知投送有效载荷" var="pubsub#deliver_payloads" type="boolean">
<value>1</value>
</field>
<field label="当节点配置改变时通知订阅者" var="pubsub#notify_config" type="boolean">
<value>1</value>
</field>
<field label="当节点被删除时通知订阅者" var="pubsub#notify_delete" type="boolean">
<value>1</value>
</field>
<field label="当节点的项目被删除时通知订阅者" var="pubsub#notify_retract" type="boolean">
<value>1</value>
</field>
<field label="仅投送通知给有效的用户" var="pubsub#presence_based_delivery" type="boolean">
<value>0</value>
</field>
<field label="指定有效的数据类型给此节点" var="pubsub#type" type="text-single">
<value/>
</field>
<field label="XSLT信息体" var="pubsub#body_xslt" type="text-single">
<value/>
</field>
<field label="XSLT有效载荷" var="pubsub#dataform_xslt" type="text-single">
<value/>
</field>
<field label="指定谁可以订阅和查看项目" var="pubsub#access_model" type="list-single">
<value>open</value>
<option>
<value>authorize</value>
</option>
<option>
<value>open</value>
</option>
<option>
<value>presence</value>
</option>
<option>
<value>roster</value>
</option>
<option>
<value>whitelist</value>
</option>
</field>
<field label="指定发布者模型" var="pubsub#publish_model" type="list-single">
<value>publishers</value>
<option>
<value>publishers</value>
</option>
<option>
<value>subscribers</value>
</option>
<option>
<value>open</value>
</option>
</field>
<field label="好友列表允许订阅" var="pubsub#roster_groups_allowed" type="list-multi"/>
<field label="有问题时联系相关人员" var="pubsub#contact" type="jid-multi"/>
<field label="默认的语言" var="pubsub#language" type="text-single">
<value>English</value>
</field>
<field label="节点的主人" var="pubsub#owner" type="jid-multi">
<value>test17@im</value>
</field>
<field label="节点的发布者" var="pubsub#publisher" type="jid-multi"/>
<field label="选择实体将收到的信息回复给项目" var="pubsub#itemreply" type="list-single">
<value>owner</value>
</field>
<field label="多用户对话房间的回复将被传递" var="pubsub#replyroom" type="jid-multi"/>
<field label="给用户的回复将被传递" var="pubsub#replyto" type="jid-multi"/>
<field label="发送项目给新的订阅者" var="pubsub#send_item_subscribe" type="boolean">
<value>1</value>
</field>
<field label="持续的项目被存储" var="pubsub#persist_items" type="boolean">
<value>0</value>
</field>
<field label="项目的最大数字被持续化" var="pubsub#max_items" type="text-single">
<value>-1</value>
</field>
<field label="最大的有效载荷字节大小" var="pubsub#max_payload_size" type="text-single">
<value>5120</value>
</field>
</x>
</query>
</iq>

测试：通过smackx 和spark im客户端实现发布订阅

发布者：

import java.util.Date;

import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smackx.pubsub.AccessModel;
import org.jivesoftware.smackx.pubsub.ConfigureForm;
import org.jivesoftware.smackx.pubsub.FormType;
import org.jivesoftware.smackx.pubsub.LeafNode;
import org.jivesoftware.smackx.pubsub.PayloadItem;
import org.jivesoftware.smackx.pubsub.PubSubManager;
import org.jivesoftware.smackx.pubsub.PublishModel;
import org.jivesoftware.smackx.pubsub.SimplePayload;

public class Publisher {
private static XMPPConnection connection = new XMPPConnection("im.cvte.cn");
private static String USRE_NAME = "test17";
private static String PASSWORD = "password";
private static String nodeId = "NodeID_003";

static{
try {
connection.connect();
connection.login(USRE_NAME,PASSWORD);
} catch (Exception e) {
e.printStackTrace();
}
}

public static void main(String[] args)throws Exception{

try{
PubSubManager manager = new PubSubManager(connection,"pubsub.im");

LeafNode myNode = null;
try {
myNode = manager.getNode(nodeId);  //创建叶子节点
} catch (Exception e) {
e.printStackTrace();
}
if(myNode == null){
myNode = manager.createNode(nodeId);
}
String id1 = "1001";

SimplePayload payload1 = new SimplePayload("message","pubsub:cvtalk","<message xmlns='pubsub:cvtalk'><body>"+ id1+":消息发布:"+ new Date().toString()+"</body></message>" );

//设置叶子节点参数，目前失灵
ConfigureForm f = new ConfigureForm(FormType.submit);
//配置参数
f.setPersistentItems(true);  //是否持久化
f.setDeliverPayloads(true);
f.setAccessModel(AccessModel.open);
f.setPublishModel(PublishModel.publishers);
//f.setSubscribe(true);
//通过设置创建叶子
//myNode =(LeafNode)manager.createNode(nodeId, f);

PayloadItem<SimplePayload> item1 = new PayloadItem<SimplePayload>(id1, payload1);
//不带itemID的SimplePayload，同样是OK的
//PayloadItem<SimplePayload> item1 = new PayloadItem<SimplePayload>(payload1);
myNode.publish(item1);
System.out.println("-----publish item1-----------");

}
catch(Exception E)
{E.printStackTrace();}

}
}

订阅者，这里的代码请写到spark的LoginDialog的login()方法 ：

private boolean login() {
.......
connection.login(.......
.......

PubSubManager manager = new PubSubManager(connection,"pubsub.im");
Node eventNode = manager.getNode("NodeID_003");
eventNode.addItemEventListener(new ItemEventListener<PayloadItem>() {
public void handlePublishedItems(ItemPublishEvent evt) {
System.out.println("收到订阅的载荷数量=" + evt.getItems().size());
for (Object obj : evt.getItems()) {
PayloadItem<SimplePayload> item = (PayloadItem<SimplePayload>) obj;
System.out.println("订阅项目=" + item.getPayload().toString());
}
}
});
eventNode.subscribe(connection.getUser());

......

订阅到达的消息

<message id="NodeID_003__wangxin@im__0K463" to="wangxin@im/PC" from="pubsub.im">
<thread>n4Ch63</thread>
<event xmlns="http://jabber.org/protocol/pubsub#event">
<items node="NodeID_003">
<item id="1001">
<message xmlns="pubsub:cvtalk">
<body>1001:消息发布:Tue Dec 08 15:36:59 CST 2015</body>
</message>
</item>
</items>
</event>
<headers xmlns="http://jabber.org/protocol/shim">
<header name="pubsub#subid">GP00jOONb9Lg2PRr0K0T01xunpquPmVC2q7QhjYg</header>
</headers>
</message>

smack中的pubsub的其他操作

获取节点配置

public ConfigureForm getDefaultConfiguration()
throws XMPPException
{
// Errors will cause exceptions in getReply, so it only returns
// on success.
PubSub reply = (PubSub)sendPubsubPacket(Type.GET, new NodeExtension(PubSubElementType.DEFAULT), PubSubElementType.DEFAULT.getNamespace());
return NodeUtils.getFormFromPacket(reply, PubSubElementType.DEFAULT);
}

删除节点

public void deleteNode(String nodeId)
throws XMPPException
{
sendPubsubPacket(Type.SET, new NodeExtension(PubSubElementType.DELETE, nodeId), PubSubElementType.DELETE.getNamespace());
nodeMap.remove(nodeId);
}

监听器

一共有3个监听：

ItemDeleteListener

ItemEventListener

NodeConfigListener

其中 ItemEventListener使用的是泛型参数，类型是 org.jivesoftware.smackx.pubsub.Item

public interface ItemEventListener <T extends Item>
{
/**
* Called whenever an item is published to the node the listener
* is registered with.
*
* @param items The publishing details.
*/
void handlePublishedItems(ItemPublishEvent<T> items);
}

另外，Personal Event Publishing (XEP-163) 也是基于发布订阅，xmpp包体结构很类似，发布的代码：

PEPManager pepManager = new PEPManager(smackConnection);
pepManager.addPEPListener(new PEPListener() {
public void eventReceived(String inFrom, PEPEvent inEvent) {
LOGGER.debug("Event received: " + inEvent);
}
});

PEPProvider pepProvider = new PEPProvider();
pepProvider.registerPEPParserExtension("http://jabber.org/protocol/tune", new TuneProvider());
ProviderManager.getInstance().addExtensionProvider("event", "http://jabber.org/protocol/pubsub#event", pepProvider);

Tune tune = new Tune("jeff", "1", "CD", "My Title", "My Track");
pepManager.publish(tune);

接收的监听：

public interface PEPListener {

/**
* Called when PEP events are received as part of a presence subscribe or message filter.
*
* @param from the user that sent the entries.
* @param event the event contained in the message.
*/
public void eventReceived(String from, PEPEvent event);

}

最后一个问题，在openfire中叶子节点上的新项目持久化到哪里了？

PubSubPersistenceManager类中writePendingItems负责持久化到数据库

private static void writePendingItems(Connection con, LinkedListNode<RetryWrapper> addItem, boolean batch) throws SQLException

但每次发布却看不到数据库中的记录，可以在下面代码找到答案，原来都提交内存了

writePendingItems(Connection con, LinkedList<RetryWrapper> addList, LinkedList<PublishedItem> delList) 将数据库中的记录删除了

/**
* Flush the cache(s) of items to be persisted (itemsToAdd) and deleted (itemsToDelete).
* @param sendToCluster If true, delegate to cluster members, otherwise local only
*/
public static void flushPendingItems(boolean sendToCluster)
{
// forward to other cluster members and wait for response
if (sendToCluster) {
CacheFactory.doSynchronousClusterTask(new FlushTask(), false);
}

if (itemsToAdd.getFirst() == null && itemsToDelete.getFirst() == null) {
return;     // nothing to do for this cluster member
}

Connection con = null;
boolean rollback = false;
LinkedList<RetryWrapper> addList = null;
LinkedList<PublishedItem> delList = null;

// Swap pending items so we can parse and save the contents from this point in time
// while not blocking new entries from being cached.
synchronized(itemsPending)
{
addList = itemsToAdd;
delList = itemsToDelete;

itemsToAdd = new LinkedList<RetryWrapper>();
itemsToDelete = new LinkedList<PublishedItem>();

// Ensure pending items are available via the item read cache;
// this allows the item(s) to be fetched by other request threads
// while being written to the DB from this thread
int copied = 0;
for (String key : itemsPending.keySet()) {
if (!itemCache.containsKey(key)) {
itemCache.put(key, (((RetryWrapper)itemsPending.get(key).object)).get());
copied++;
}
}
if (log.isDebugEnabled() && copied > 0) {
log.debug("Added " + copied + " pending items to published item cache");
}
itemsPending.clear();
}

// Note that we now make multiple attempts to write cached items to the DB:
//   1) insert all pending items in a single batch
//   2) if the batch insert fails, retry by inserting each item separately
//   3) if a given item cannot be written, return it to the pending write cache
// By default step 3 will be tried once per item, but this can be configured
// (or disabled) using the "xmpp.pubsub.item.retry" property. In the event of
// a transaction rollback, items that could not be written to the database
// will be returned to the pending item write cache.
try {
con = DbConnectionManager.getTransactionConnection();
writePendingItems(con, addList, delList);
} catch (SQLException se) {
log.error("Failed to flush pending items; initiating rollback", se);
// return new items to the write cache
LinkedListNode<RetryWrapper> node = addList.getLast();
while (node != null) {
savePublishedItem(node.object);
node.remove();
node = addList.getLast();
}
rollback = true;
} finally {
DbConnectionManager.closeTransactionConnection(con, rollback);
}
}

参考网页： http://xmpp.org/extensions/xep-0060.html http://xmpp.org/extensions/xep-0163.html https://community.igniterealtime.org/thread/38433 http://www.igniterealtime.org/support/articles/pubsub.jsp http://blog.csdn.net/u011163195/article/details/17683741