Contact Subscription: Implemented auto-subscriptions to PEP services via presence subscriptions.

git-svn-id: http://svn.igniterealtime.org/svn/repos/openfire/branches@8812 b35dd754-fafc-0310-a699-88a17e54d16e

pep/src/java/org/jivesoftware/openfire/handler/PresenceSubscribeHandler.java

@@ -16,6 +16,7 @@ import org.jivesoftware.openfire.container.BasicModule;
 import org.jivesoftware.openfire.roster.Roster;
 import org.jivesoftware.openfire.roster.RosterItem;
 import org.jivesoftware.openfire.roster.RosterManager;
+import org.jivesoftware.openfire.user.PresenceEventDispatcher;
 import org.jivesoftware.openfire.user.UserAlreadyExistsException;
 import org.jivesoftware.openfire.user.UserManager;
 import org.jivesoftware.openfire.user.UserNotFoundException;
@@ -168,6 +169,7 @@ public class PresenceSubscribeHandler extends BasicModule implements ChannelHand
                         JID prober = localServer.isLocal(recipientJID) ?
                                 new JID(recipientJID.toBareJID()) : recipientJID;
                         presenceManager.probePresence(prober, senderJID);
+                        PresenceEventDispatcher.subscribedToPresence(prober, senderJID);
                     }
                 }
 

pep/src/java/org/jivesoftware/openfire/pep/IQPEPHandler.java

@@ -13,6 +13,7 @@ package org.jivesoftware.openfire.pep;
 
 import org.dom4j.DocumentHelper;
 import org.dom4j.Element;
+import org.dom4j.QName;
 import org.jivesoftware.database.DbConnectionManager;
 import org.jivesoftware.openfire.IQHandlerInfo;
 import org.jivesoftware.openfire.XMPPServer;
@@ -27,11 +28,20 @@ import org.jivesoftware.openfire.pubsub.LeafNode;
 import org.jivesoftware.openfire.pubsub.Node;
 import org.jivesoftware.openfire.pubsub.PubSubEngine;
 import org.jivesoftware.openfire.pubsub.models.AccessModel;
+import org.jivesoftware.openfire.roster.Roster;
+import org.jivesoftware.openfire.roster.RosterItem;
+import org.jivesoftware.openfire.session.ClientSession;
+import org.jivesoftware.openfire.user.PresenceEventDispatcher;
+import org.jivesoftware.openfire.user.PresenceEventListener;
 import org.jivesoftware.openfire.user.UserManager;
+import org.jivesoftware.openfire.user.UserNotFoundException;
 import org.jivesoftware.util.Log;
+import org.xmpp.forms.DataForm;
+import org.xmpp.forms.FormField;
 import org.xmpp.packet.IQ;
 import org.xmpp.packet.JID;
 import org.xmpp.packet.PacketError;
+import org.xmpp.packet.Presence;
 
 import java.sql.Connection;
 import java.sql.PreparedStatement;
@@ -70,8 +80,8 @@ import java.util.concurrent.ConcurrentHashMap;
  * @author Armando Jagucki
  * 
  */
-public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
-        ServerFeaturesProvider, UserIdentitiesProvider, UserItemsProvider {
+public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider, ServerFeaturesProvider,
+        UserIdentitiesProvider, UserItemsProvider, PresenceEventListener {
 
     // Map of PEP services. Table, Key: bare JID (String); Value: PEPService
     private Map<String, PEPService> pepServices;
@@ -80,8 +90,7 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
 
     private PubSubEngine pubSubEngine = null;
 
-    private static final String GET_PEP_SERVICES =
-        "SELECT DISTINCT serviceID FROM pubsubNode";
+    private static final String GET_PEP_SERVICES = "SELECT DISTINCT serviceID FROM pubsubNode";
 
     public IQPEPHandler() {
         super("Personal Eventing Handler");
@@ -93,6 +102,9 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
     public void initialize(XMPPServer server) {
         super.initialize(server);
 
+        // Listen to presence events to manage PEP auto-subscriptions.
+        PresenceEventDispatcher.addListener(this);
+
         pubSubEngine = new PubSubEngine(server.getPacketRouter());
 
         // Restore previous PEP services for which nodes exist in the database.
@@ -104,7 +116,7 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
             pstmt = con.prepareStatement(GET_PEP_SERVICES);
             ResultSet rs = pstmt.executeQuery();
             // Restore old PEPServices
-            while(rs.next()) {
+            while (rs.next()) {
                 String serviceID = rs.getString(1);
 
                 // Create a new PEPService if serviceID looks like a bare JID.
@@ -123,10 +135,20 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
             Log.error(sqle);
         }
         finally {
-            try { if (pstmt != null) pstmt.close(); }
-            catch (Exception e) { Log.error(e); }
-            try { if (con != null) con.close(); }
-            catch (Exception e) { Log.error(e); }
+            try {
+                if (pstmt != null)
+                    pstmt.close();
+            }
+            catch (Exception e) {
+                Log.error(e);
+            }
+            try {
+                if (con != null)
+                    con.close();
+            }
+            catch (Exception e) {
+                Log.error(e);
+            }
         }
     }
 
@@ -163,6 +185,23 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
                 if (Log.isDebugEnabled()) {
                     Log.debug("PEP: " + jidFrom + " had a PEPService created");
                 }
+                
+                // Those who already have presence subscriptions to jidFrom
+                // will now automatically be subscribed to this new PEPService.
+                Roster roster;
+                try {
+                    roster = XMPPServer.getInstance().getRosterManager().getRoster(packet.getFrom().getNode());
+                    for (RosterItem item : roster.getRosterItems()) {
+                        if (item.getSubStatus() == RosterItem.SUB_BOTH || item.getSubStatus() == RosterItem.SUB_FROM) {
+                            subscribedToPresence(item.getJid(), packet.getFrom());
+                        }
+                    }
+                }
+                catch (UserNotFoundException e) {
+                    // Do nothing
+                }
+
+                
             }
 
             // If publishing a node, and the node doesn't exist, create it.
@@ -175,7 +214,7 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
                     if (pepService.getNode(nodeID) == null) {
                         // Create the node
                         JID creator = new JID(jidFrom);
-                        LeafNode newNode = new LeafNode(pepService, null, nodeID, creator);
+                        LeafNode newNode = new LeafNode(pepService, pepService.getRootCollectionNode(), nodeID, creator);
                         newNode.addOwner(creator);
                         newNode.saveToDB();
                         if (Log.isDebugEnabled()) {
@@ -200,17 +239,33 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
         }
         else {
             // TODO: Handle packets such as these.
+            if (packet.getType() == IQ.Type.result) {
                 if (Log.isDebugEnabled()) {
-                Log.debug("PEP: getTo() wasn't null.");
+                    Log.debug("PEP: Ignored result packet (probably from an auto-subscribe).");
+                }
+                return null;
             }
 
-            // Is getTo() online? If not, consider what error to use.
+            String jidTo = packet.getTo().toBareJID();
+
+            PEPService pepService = pepServices.get(jidTo);
+
+            if (pepService != null) {
+                if (pubSubEngine.process(pepService, packet)) {
+                    if (Log.isDebugEnabled()) {
+                        Log.debug("PEP: The pubSubEngine processed a packet for " + jidTo + "'s pepService.");
+                    }
+                }
+                else {
+                    if (Log.isDebugEnabled()) {
+                        Log.debug("PEP: The pubSubEngine did not process a packet for " + jidTo + "'s pepService.");
+                    }
+                }
+            }
+            else {
+                // TODO: Handle other packets here...
+            }
 
-            // FIXME: Remove this chunk of error code after such packets are handled.
-            IQ reply = IQ.createResultIQ(packet);
-            reply.setChildElement(packet.getChildElement().createCopy());
-            reply.setError(PacketError.Condition.service_unavailable);
-            return reply;
         }
 
         // Other error flows are handled in pubSubEngine.process(...)
@@ -269,7 +324,7 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
             defaultItem.addAttribute("jid", recipientJID);
 
             for (Node node : pepService.getNodes()) {
-                // Do not include the root node.
+                // Do not include the root node as an item element.
                 if (node == rootNode) {
                     continue;
                 }
@@ -286,4 +341,95 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
         return items.iterator();
     }
 
+    public void availableSession(ClientSession session, Presence presence) {
+        // TODO Auto-generated method stub
+
+    }
+
+    public void presenceChanged(ClientSession session, Presence presence) {
+        // TODO Auto-generated method stub
+
+    }
+
+    public void presencePriorityChanged(ClientSession session, Presence presence) {
+        // TODO Auto-generated method stub
+
+    }
+
+    public void subscribedToPresence(JID subscriberJID, JID authorizerJID) {
+        // If authorizerJID has a PEP service, auto generate and process a pubsub subscription packet
+        // that is equivalent to: (where 'from' is subscriberJID and 'to' is authorizerJID)
+        //
+        //        <iq type='set'
+        //            from='nurse@capulet.com/chamber'
+        //            to='juliet@capulet.com
+        //            id='collsub'>
+        //          <pubsub xmlns='http://jabber.org/protocol/pubsub'>
+        //            <subscribe jid='nurse@capulet.com'/>
+        //            <options>
+        //              <x xmlns='jabber:x:data'>
+        //                <field var='FORM_TYPE' type='hidden'>
+        //                  <value>http://jabber.org/protocol/pubsub#subscribe_options</value>
+        //                </field>
+        //                <field var='pubsub#subscription_type'>
+        //                  <value>items</value>
+        //                </field>
+        //                <field var='pubsub#subscription_depth'>
+        //                  <value>all</value>
+        //                </field>
+        //              </x>
+        //           </options>
+        //         </pubsub>
+        //        </iq>
+        PEPService pepService = pepServices.get(authorizerJID.toBareJID());
+        if (pepService != null) {
+            IQ subscriptionPacket = new IQ(IQ.Type.set);
+            subscriptionPacket.setFrom(subscriberJID);
+            subscriptionPacket.setTo(authorizerJID.toBareJID());
+
+            Element pubsubElement = subscriptionPacket.setChildElement("pubsub", "http://jabber.org/protocol/pubsub");
+
+            Element subscribeElement = pubsubElement.addElement("subscribe");
+            subscribeElement.addAttribute("jid", subscriberJID.toBareJID());
+
+            Element optionsElement = pubsubElement.addElement("options");
+            Element xElement = optionsElement.addElement(QName.get("x", "jabber:x:data"));
+
+            DataForm dataForm = new DataForm(xElement);
+
+            FormField formField = dataForm.addField();
+            formField.setVariable("FORM_TYPE");
+            formField.setType(FormField.Type.hidden);
+            formField.addValue("http://jabber.org/protocol/pubsub#subscribe_options");
+
+            formField = dataForm.addField();
+            formField.setVariable("pubsub#subscription_type");
+            formField.addValue("items");
+
+            formField = dataForm.addField();
+            formField.setVariable("pubsub#subscription_depth");
+            formField.addValue("all");
+
+            if (Log.isDebugEnabled()) {
+                Log.debug("PEP: Generated auto-subscribe packet:" + subscriptionPacket.toString());
+            }
+
+            if (pubSubEngine.process(pepService, subscriptionPacket)) {
+                if (Log.isDebugEnabled()) {
+                    Log.debug("PEP: Sent auto-subscribe packet to " + authorizerJID.toBareJID() + "'s pepService.");
+                }
+            }
+            else {
+                if (Log.isDebugEnabled()) {
+                    Log.debug("PEP: The pubSubEngine failed sending the auto-subscribe packet.");
+                }
+            }
+        }
+    }
+
+    public void unavailableSession(ClientSession session, Presence presence) {
+        // TODO Auto-generated method stub
+
+    }
+
 }

pep/src/java/org/jivesoftware/openfire/pep/PEPService.java

@@ -270,7 +270,7 @@ public class PEPService implements PubSubService {
     }
 
     public boolean isCollectionNodesSupported() {
-        return false;
+        return true;
     }
 
     public boolean isInstantNodeSupported() {

pep/src/java/org/jivesoftware/openfire/pubsub/NodeAffiliate.java

@@ -88,6 +88,13 @@ public class NodeAffiliate {
                 Element items = event.addElement("items");
                 items.addAttribute("node", getNode().getNodeID());
                 for (PublishedItem publishedItem : itemsBySubs.get(nodeSubscriptions)) {
+                    // FIXME: This was added for compatibility with PEP supporting clients like Psi.
+                    //        May not be the best of solutions.
+                    //
+                    // If the node ID looks like a JID, replace it with the published item's node ID.
+                    if (getNode().getNodeID().indexOf("@") >= 0) {
+                        items.addAttribute("node", publishedItem.getNode().getNodeID());                        
+                    }
                     // Add item information to the event notification
                     Element item = items.addElement("item");
                     if (leafNode.isItemRequired()) {

pep/src/java/org/jivesoftware/openfire/pubsub/PubSubPersistenceManager.java

@@ -35,14 +35,22 @@ import java.util.concurrent.LinkedBlockingQueue;
  */
 public class PubSubPersistenceManager {
 
-    private static final String LOAD_NODES =
+    private static final String LOAD_NON_LEAF_NODES =
             "SELECT nodeID, leaf, creationDate, modificationDate, parent, deliverPayloads, " +
             "maxPayloadSize, persistItems, maxItems, notifyConfigChanges, notifyDelete, " +
             "notifyRetract, presenceBased, sendItemSubscribe, publisherModel, " +
             "subscriptionEnabled, configSubscription, accessModel, payloadType, " +
             "bodyXSLT, dataformXSLT, creator, description, language, name, " +
             "replyPolicy, associationPolicy, maxLeafNodes FROM pubsubNode " +
-            "WHERE serviceID=? ORDER BY nodeID";
+            "WHERE serviceID=? AND leaf=0 ORDER BY nodeID";
+    private static final String LOAD_LEAF_NODES =
+            "SELECT nodeID, leaf, creationDate, modificationDate, parent, deliverPayloads, " +
+            "maxPayloadSize, persistItems, maxItems, notifyConfigChanges, notifyDelete, " +
+            "notifyRetract, presenceBased, sendItemSubscribe, publisherModel, " +
+            "subscriptionEnabled, configSubscription, accessModel, payloadType, " +
+            "bodyXSLT, dataformXSLT, creator, description, language, name, " +
+            "replyPolicy, associationPolicy, maxLeafNodes FROM pubsubNode " +
+            "WHERE serviceID=? AND leaf=1 ORDER BY nodeID";
     private static final String UPDATE_NODE =
             "UPDATE pubsubNode SET modificationDate=?, parent=?, deliverPayloads=?, " +
             "maxPayloadSize=?, persistItems=?, maxItems=?, " +
@@ -446,11 +454,22 @@ public class PubSubPersistenceManager {
         Map<String, Node> nodes = new HashMap<String, Node>();
         try {
             con = DbConnectionManager.getConnection();
-            // Get all nodes at once (with 1 query)
-            pstmt = con.prepareStatement(LOAD_NODES);
+            // Get all non-leaf nodes (to ensure parent nodes are loaded before their children)
+            pstmt = con.prepareStatement(LOAD_NON_LEAF_NODES);
             pstmt.setString(1, service.getServiceID());
             ResultSet rs = pstmt.executeQuery();
-            // Rebuild all loaded nodes
+            // Rebuild loaded non-leaf nodes
+            while(rs.next()) {
+                loadNode(service, nodes, rs);
+            }
+            rs.close();
+            pstmt.close();
+
+            // Get all leaf nodes (remaining unloaded nodes)
+            pstmt = con.prepareStatement(LOAD_LEAF_NODES);
+            pstmt.setString(1, service.getServiceID());
+            rs = pstmt.executeQuery();
+            // Rebuild loaded leaf nodes
             while(rs.next()) {
                 loadNode(service, nodes, rs);
             }

pep/src/java/org/jivesoftware/openfire/user/PresenceEventDispatcher.java

@@ -12,6 +12,7 @@
 package org.jivesoftware.openfire.user;
 
 import org.jivesoftware.openfire.session.ClientSession;
+import org.xmpp.packet.JID;
 import org.xmpp.packet.Presence;
 
 import java.util.List;
@@ -123,4 +124,19 @@ public class PresenceEventDispatcher {
             }
         }
     }
+
+    /**
+     * Notification message indicating that a user has successfully subscribed
+     * to the presence of another user.
+     * 
+     * @param subscriberJID the user that initiated the subscription.
+     * @param authorizerJID the user that authorized the subscription.
+     */
+    public static void subscribedToPresence(JID subscriberJID, JID authorizerJID) {
+        if (!listeners.isEmpty()) {
+            for (PresenceEventListener listener : listeners) {
+                listener.subscribedToPresence(subscriberJID, authorizerJID);
+            }
+        }
+    }
 }

pep/src/java/org/jivesoftware/openfire/user/PresenceEventListener.java

@@ -12,6 +12,7 @@
 package org.jivesoftware.openfire.user;
 
 import org.jivesoftware.openfire.session.ClientSession;
+import org.xmpp.packet.JID;
 import org.xmpp.packet.Presence;
 
 /**
@@ -64,4 +65,13 @@ public interface PresenceEventListener {
      * @param presence the received available presence with the new information.
      */
     public void presenceChanged(ClientSession session, Presence presence);
+
+    /**
+     * Notification message indicating that a user has successfully subscribed
+     * to the presence of another user.
+     * 
+     * @param subscriberJID the user that initiated the subscription.
+     * @param authorizerJID the user that authorized the subscription.
+     */
+    public void subscribedToPresence(JID subscriberJID, JID authorizerJID);
 }