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

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

阅读量：7029 次

发布时间：2019-06-28

本文共 12492 字，大约阅读时间需要 41 分钟。

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

节点类型：

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

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

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

访问和发布模式 Access and Publisher Models

Open: 任何人都能订阅

Authorize: 订阅请求必须由所有者批准，只有认证的用户可以订阅项目。

Whitelist: 白名单里的用户可以订阅.

Presence: 只有能收到发布者也即Owner的即席状态的用户才能收到订阅.

Roster: 只有在用户花名册或花名册组内的用户可以收到订阅事项提醒

在openfire里，Whitelist的配置如下：

发布者模式:

Open: anyone may publish items to the node.（权限最大）

Publishers: owners and publishers are allowed to publish items to the node.

Subscribers: owners, publishers and subscribers are allowed to publish items to the node.

发布订阅的过程，发布者发布到叶子节点，订阅者收到消息提醒

XMPP中的订阅流程

1、首先，需要确认你的服务器支持pubsub特性

1.1 查询XMPP服务的所有服务

View Code

1.2 查询某一项XMPP子域，如pubsub

View Code

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


     
      
       
       
       
       
        
         
          http://jabber.org/protocol/pubsub#meta-data
         
        
        
         
        
        
         
        
        
         
          leaf
         
        
        
        
         
          1
         
        
        
         
          0
         
        
        
         
          1
         
        
        
         
          1
         
        
        
         
          1
         
        
        
         
          1
         
        
        
         
          0
         
        
        
         
        
        
         
        
        
         
        
        
         
          open
         
         
         
         
         
         
        
        
         
          publishers
         
         
         
         
        
        
        
        
         
          English
         
        
        
         
          test17@im
         
        
        
        
         
          owner
         
        
        
        
        
         
          1
         
        
        
         
          0
         
        
        
         
          -1
         
        
        
         
          5120

测试：通过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","
     
      "+ id1+":消息发布:"+ new Date().toString()+"
     " );              //设置叶子节点参数，目前失灵            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
     
       item1 = new PayloadItem
      
       (id1, payload1);            //不带itemID的SimplePayload，同样是OK的            //PayloadItem
       
         item1 = new PayloadItem
        
         (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
      
       () {                          public void handlePublishedItems(ItemPublishEvent evt) {                              System.out.println("收到订阅的载荷数量=" + evt.getItems().size());                              for (Object obj : evt.getItems()) {                                  PayloadItem
       
         item = (PayloadItem
        
         ) obj;                                  System.out.println("订阅项目=" + item.getPayload().toString());                              }                          }                      });                      eventNode.subscribe(connection.getUser());

......

订阅到达的消息


       
      
       n4Ch63
        
          
             
                
                   1001:消息发布:Tue Dec 08 15:36:59 CST 2015        
               
            
         
        
          
       
        GP00jOONb9Lg2PRr0K0T01xunpquPmVC2q7QhjYg

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 
     
       {    /**     * Called whenever an item is published to the node the listener     * is registered with.     *      * @param items The publishing details.     */    void handlePublishedItems(ItemPublishEvent
      
        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
     
       addList = null;        LinkedList
      
        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
       
        ();            itemsToDelete = new LinkedList
        
         ();                        // 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
         
           node = addList.getLast();            while (node != null) {                savePublishedItem(node.object);                node.remove();                node = addList.getLast();            }            rollback = true;        } finally {            DbConnectionManager.closeTransactionConnection(con, rollback);        }    }