PEP: Added initial support for sending Last Item Published on newly-available...

PEP: Added initial support for sending Last Item Published on newly-available presence notifications to the PEP service.

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

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

@@ -16,6 +16,7 @@ import org.dom4j.Element;
 import org.dom4j.QName;
 import org.jivesoftware.database.DbConnectionManager;
 import org.jivesoftware.openfire.IQHandlerInfo;
+import org.jivesoftware.openfire.PresenceManager;
 import org.jivesoftware.openfire.XMPPServer;
 import org.jivesoftware.openfire.auth.UnauthorizedException;
 import org.jivesoftware.openfire.disco.ServerFeaturesProvider;
@@ -31,6 +32,7 @@ import org.jivesoftware.openfire.pubsub.LeafNode;
 import org.jivesoftware.openfire.pubsub.Node;
 import org.jivesoftware.openfire.pubsub.NodeSubscription;
 import org.jivesoftware.openfire.pubsub.PubSubEngine;
+import org.jivesoftware.openfire.pubsub.PublishedItem;
 import org.jivesoftware.openfire.pubsub.models.AccessModel;
 import org.jivesoftware.openfire.roster.Roster;
 import org.jivesoftware.openfire.roster.RosterItem;
@@ -427,7 +429,63 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
     }
 
     public void availableSession(ClientSession session, Presence presence) {
+        // FIXME: this method is never called for remote sessions...
+        /**
+         * When a PEP service receives initial presence information from a subscriber's
+         * resource with a non-negative priority and including XEP-0115 information that
+         * indicates an interest in the data format, it MUST generate a notification
+         * containing the last published item for that node and send it to the newly-available
+         * resource.
+         * 
+         * As an exception to the foregoing MUST rules, a PEP service MUST NOT send notifications
+         * to a subscriber if the user has blocked the subscriber from receiving all or any kinds
+         * of stanza (presence, message, IQ, or any combination thereof) using communiations blocking
+         * as specified in XMPP IM.
+         */
+
+
+        JID newlyAvailableJID = presence.getFrom();
+        
+        PresenceManager presenceManager = XMPPServer.getInstance().getPresenceManager();
+
+        for (PEPService pepService : pepServices.values()) {
+            try {
+                if (presenceManager.canProbePresence(newlyAvailableJID, pepService.getAddress().getNode())) {
+                    // Retrieve last published item.
+                    CollectionNode rootNode = pepService.getRootCollectionNode();
+                    PublishedItem lastPublishedItem = null;
+
+                    for (Node leafNode : rootNode.getNodes()) {
+                        PublishedItem leafLastPublishedItem = leafNode.getLastPublishedItem();
+                        if (leafLastPublishedItem == null) {
+                            continue;
+                        }
+
+                        if (lastPublishedItem == null) {
+                            lastPublishedItem = leafLastPublishedItem;
+                            continue;
+                        }
+                        
+                        if (leafLastPublishedItem.getCreationDate().compareTo(lastPublishedItem.getCreationDate()) > 0) {
+                            lastPublishedItem = leafLastPublishedItem;
+                        }
+                    }
+                    
+                    // Send last published item to the newly-available resource.
+                    NodeSubscription subscription = rootNode.getSubscription(newlyAvailableJID);
+                    if (subscription == null) {
+                        subscription = rootNode.getSubscription(new JID(newlyAvailableJID.toBareJID()));
+                    }
+
+                    if (subscription != null) {
+                        pepService.sendLastPublishedItem(subscription, lastPublishedItem);
+                    }
+                }
+            }
+            catch (UserNotFoundException e) {
                 // Do nothing
+            }
+        }
 
     }
 

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

@@ -17,7 +17,9 @@ import org.dom4j.QName;
 import org.jivesoftware.openfire.commands.AdHocCommandManager;
 import org.jivesoftware.openfire.pubsub.CollectionNode;
 import org.jivesoftware.openfire.pubsub.DefaultNodeConfiguration;
+import org.jivesoftware.openfire.pubsub.LeafNode;
 import org.jivesoftware.openfire.pubsub.Node;
+import org.jivesoftware.openfire.pubsub.NodeSubscription;
 import org.jivesoftware.openfire.pubsub.PendingSubscriptionsCommand;
 import org.jivesoftware.openfire.pubsub.PubSubEngine;
 import org.jivesoftware.openfire.pubsub.PubSubPersistenceManager;
@@ -33,6 +35,8 @@ import org.jivesoftware.openfire.user.UserNotFoundException;
 import org.jivesoftware.openfire.PacketRouter;
 import org.jivesoftware.openfire.SessionManager;
 import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.util.FastDateFormat;
+import org.jivesoftware.util.LocaleUtils;
 import org.jivesoftware.util.StringUtils;
 import org.xmpp.packet.JID;
 import org.xmpp.packet.Message;
@@ -44,6 +48,7 @@ import java.util.Collection;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Queue;
+import java.util.TimeZone;
 import java.util.Timer;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -130,7 +135,7 @@ public class PEPService implements PubSubService {
      * The time to elapse between each execution of the maintenance process.
      * Default is 2 minutes.
      */
-    public int items_task_timeout = 2 * 60 * 1000;
+    private int items_task_timeout = 2 * 60 * 1000;
 
     /**
      * Task that saves or deletes published items from the database.
@@ -143,6 +148,12 @@ public class PEPService implements PubSubService {
      */
     private Timer timer = new Timer("PEP service maintenance");
     
+    private static final FastDateFormat fastDateFormat;
+    
+    static {
+        fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", TimeZone.getTimeZone("UTC"));
+    }
+
     /**
      * Constructs a PEPService.
      * 
@@ -456,4 +467,44 @@ public class PEPService implements PubSubService {
         items_task_timeout = timeout;
     }
 
+    /**
+     * Sends an event notification for the last published item to the subscription's subscriber. If
+     * the subscription has not yet been authorized or is pending to be configured then
+     * no notification is going to be sent.<p>
+     *
+     * Depending on the subscription configuration the event notification may or may not have
+     * a payload, may not be sent if a keyword (i.e. filter) was defined and it was not matched.
+     *
+     * @param subscription the subscription the published item is being sent for.
+     * @param publishedItem the last item that was published to the node.
+     */
+    public void sendLastPublishedItem(NodeSubscription subscription, PublishedItem publishedItem) {
+        // Check if the published item can be sent to the subscriber
+        if (!subscription.canSendPublicationEvent(publishedItem.getNode(), publishedItem)) {
+            return;
+        }
+        // Send event notification to the subscriber
+        Message notification = new Message();
+        Element event = notification.getElement()
+                .addElement("event", "http://jabber.org/protocol/pubsub#event");
+        Element items = event.addElement("items");
+        items.addAttribute("node", publishedItem.getNode().getNodeID());
+        Element item = items.addElement("item");
+        if (((LeafNode) publishedItem.getNode()).isItemRequired()) {
+            item.addAttribute("id", publishedItem.getID());
+        }
+        if (publishedItem.getNode().isPayloadDelivered() && publishedItem.getPayload() != null) {
+            item.add(publishedItem.getPayload().createCopy());
+        }
+        // Add a message body (if required)
+        if (subscription.isIncludingBody()) {
+            notification.setBody(LocaleUtils.getLocalizedString("pubsub.notification.message.body"));
+        }
+        // Include date when published item was created
+        notification.getElement().addElement("x", "jabber:x:delay")
+                .addAttribute("stamp", fastDateFormat.format(publishedItem.getCreationDate()));
+        // Send the event notification to the subscriber
+        this.sendNotification(subscription.getNode(), notification, subscription.getJID());
+    }
+
 }

src/java/org/jivesoftware/openfire/pubsub/Node.java

@@ -1735,7 +1735,7 @@ public abstract class Node {
      * @throws IllegalStateException If this message was used when the node supports multiple
      *         subscriptions.
      */
-    NodeSubscription getSubscription(JID subscriberJID) {
+    public NodeSubscription getSubscription(JID subscriberJID) {
         // Check that node does not support multiple subscriptions
         if (isMultipleSubscriptionsEnabled()) {
             throw new IllegalStateException("Multiple subscriptions is enabled so subscriptions " +

src/java/org/jivesoftware/openfire/pubsub/NodeSubscription.java

@@ -129,7 +129,7 @@ public class NodeSubscription {
         dateFormat = new SimpleDateFormat("yyyy-MM-DD'T'HH:mm:ss.SSS'Z'");
         dateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
         fastDateFormat = FastDateFormat
-                .getInstance("yyyy-MM-DD'T'HH:mm:ss.SSS'Z'", TimeZone.getTimeZone("UTC"));
+                .getInstance("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", TimeZone.getTimeZone("UTC"));
     }
 
     /**
@@ -613,7 +613,7 @@ public class NodeSubscription {
      * @return true if an event notification can be sent to the subscriber for the specified
      *         published item.
      */
-    boolean canSendPublicationEvent(LeafNode leafNode, PublishedItem publishedItem) {
+    public boolean canSendPublicationEvent(LeafNode leafNode, PublishedItem publishedItem) {
         if (!canSendEvents()) {
             return false;
         }