Initial import of PEP code, with PubSubEngine refactoring.

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

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

@@ -25,6 +25,7 @@ import org.jivesoftware.openfire.container.PluginManager;
 import org.jivesoftware.openfire.disco.IQDiscoInfoHandler;
 import org.jivesoftware.openfire.disco.IQDiscoItemsHandler;
 import org.jivesoftware.openfire.disco.ServerFeaturesProvider;
+import org.jivesoftware.openfire.disco.ServerIdentitiesProvider;
 import org.jivesoftware.openfire.disco.ServerItemsProvider;
 import org.jivesoftware.openfire.filetransfer.DefaultFileTransferManager;
 import org.jivesoftware.openfire.filetransfer.FileTransferManager;
@@ -37,6 +38,8 @@ import org.jivesoftware.openfire.net.MulticastDNSService;
 import org.jivesoftware.openfire.net.SSLConfig;
 import org.jivesoftware.openfire.net.ServerTrafficCounter;
 import org.jivesoftware.openfire.pubsub.PubSubModule;
+import org.jivesoftware.openfire.pep.IQPEPHandler;
+import org.jivesoftware.openfire.pep.IQPEPOwnerHandler;
 import org.jivesoftware.openfire.roster.RosterManager;
 import org.jivesoftware.openfire.session.RemoteSessionLocator;
 import org.jivesoftware.openfire.spi.*;
@@ -494,9 +497,9 @@ public class XMPPServer {
         loadModule(IQLastActivityHandler.class.getName());
         loadModule(PresenceSubscribeHandler.class.getName());
         loadModule(PresenceUpdateHandler.class.getName());
-        loadModule(IQDiscoInfoHandler.class.getName());
-        loadModule(IQDiscoItemsHandler.class.getName());
         loadModule(IQOfflineMessagesHandler.class.getName());
+        loadModule(IQPEPHandler.class.getName());
+        loadModule(IQPEPOwnerHandler.class.getName());
         loadModule(MultiUserChatServerImpl.class.getName());
         loadModule(MulticastDNSService.class.getName());
         loadModule(IQSharedGroupHandler.class.getName());
@@ -507,6 +510,8 @@ public class XMPPServer {
         loadModule(MediaProxyService.class.getName());
         loadModule(STUNService.class.getName());
         loadModule(PubSubModule.class.getName());
+        loadModule(IQDiscoInfoHandler.class.getName());
+        loadModule(IQDiscoItemsHandler.class.getName());
         loadModule(UpdateManager.class.getName());
         loadModule(InternalComponentManager.class.getName());
         // Load this module always last since we don't want to start listening for clients
@@ -1024,6 +1029,17 @@ public class XMPPServer {
     public IQAuthHandler getIQAuthHandler() {
         return (IQAuthHandler) modules.get(IQAuthHandler.class);
     }
+    
+    /**
+     * Returns the <code>IQPEPHandler</code> registered with this server. The
+     * <code>IQPEPHandler</code> was registered with the server as a module while starting up
+     * the server.
+     *
+     * @return the <code>IQPEPHandler</code> registered with this server.
+     */
+    public IQPEPHandler getIQPEPHandler() {
+        return (IQPEPHandler) modules.get(IQPEPHandler.class);
+    }
 
     /**
      * Returns the <code>PluginManager</code> instance registered with this server.
@@ -1195,6 +1211,21 @@ public class XMPPServer {
         }
         return answer;
     }
+ 
+    /**
+     * Returns a list with all the modules that provide "discoverable" identities.
+     *
+     * @return a list with all the modules that provide "discoverable" identities.
+     */
+    public List<ServerIdentitiesProvider> getServerIdentitiesProviders() {
+        List<ServerIdentitiesProvider> answer = new ArrayList<ServerIdentitiesProvider>();
+        for (Module module : modules.values()) {
+            if (module instanceof ServerIdentitiesProvider) {
+                answer.add((ServerIdentitiesProvider) module);
+            }
+        }
+        return answer;
+    }
 
     /**
      * Returns a list with all the modules that provide "discoverable" items associated with

pep/src/java/org/jivesoftware/openfire/disco/IQDiscoInfoHandler.java

@@ -62,6 +62,7 @@ public class IQDiscoInfoHandler extends IQHandler implements ClusterEventListene
     private Map<String, DiscoInfoProvider> entities = new HashMap<String, DiscoInfoProvider>();
     private Set<String> localServerFeatures = new CopyOnWriteArraySet<String>();
     private Cache<String, Set<NodeID>> serverFeatures;
+    private List<Element> serverIdentities = new ArrayList<Element>();
     private Map<String, DiscoInfoProvider> serverNodeProviders = new ConcurrentHashMap<String, DiscoInfoProvider>();
     private IQHandlerInfo info;
 
@@ -285,6 +286,20 @@ public class IQDiscoInfoHandler extends IQHandler implements ClusterEventListene
         for (ServerFeaturesProvider provider : server.getServerFeaturesProviders()) {
             addServerFeaturesProvider(provider);
         }
+        // Track the implementors of ServerIdentitiesProvider so that we can collect the identities
+        // for protocols supported by the server
+        for (ServerIdentitiesProvider provider : server.getServerIdentitiesProviders()) {
+            for (Iterator<Element> it = provider.getIdentities(); it.hasNext();) {
+                serverIdentities.add(it.next());
+            }
+        }
+        // 
+        if (server.getIQPEPHandler() != null) {
+            Element userIdentity = DocumentHelper.createElement("identity");
+            userIdentity.addAttribute("category", "pubsub");
+            userIdentity.addAttribute("type", "pep");
+            registeredUserIdentities.add(0, userIdentity);
+        }
         setProvider(server.getServerInfo().getName(), getServerInfoProvider());
         // Listen to cluster events
         ClusterManager.addListener(this);
@@ -379,6 +394,11 @@ public class IQDiscoInfoHandler extends IQHandler implements ClusterEventListene
                             identity.addAttribute("type", "im");
 
                             identities.add(identity);
+                            
+                            // Include identities from modules that implement ServerIdentitiesProvider
+                            for (Element identityElement : serverIdentities) {
+                                identities.add(identityElement);
+                            }
                         }
                     }
                     return identities.iterator();

pep/src/java/org/jivesoftware/openfire/disco/ServerIdentitiesProvider.java

+/**
+ * $RCSfile: $
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 Jive Software. All rights reserved.
+ *
+ * This software is published under the terms of the GNU Public License (GPL),
+ * a copy of which is included in this distribution.
+ */
+
+package org.jivesoftware.openfire.disco;
+
+import java.util.Iterator;
+
+import org.dom4j.Element;
+
+/**
+ * <p>
+ * A <code>ServerIdentityProvider</code> is responsible for providing the identities
+ * of protocols supported by the SERVER. An example of a server identity is that
+ * for PEP (XEP-0163): <identity category="pubsub" type="pep" />
+ * <p/>
+ * 
+ * <p>
+ * When the server starts up, IQDiscoInfoHandler will request to all the services that implement
+ * the ServerIdentitiesProvider interface for their identities. Whenever a disco request is received
+ * IQDiscoInfoHandler will add to the provided information all the collected identities. Therefore, a
+ * service must implement this interface in order to offer/publish its identities as part of the
+ * server identities.
+ * </p>
+ *
+ * @author Armando Jagucki
+ */
+public interface ServerIdentitiesProvider {
+
+    /**
+     * Returns an Iterator (of Element) with the supported identities by the server. The identities to
+     * include are the identities of protocols supported by the SERVER. The idea is that
+     * different modules may provide their identities that will ultimately be included in the list
+     * of server identities.
+     *
+     * @return an Iterator (of Element) with identities of protocols supported by the server.
+     */
+    public abstract Iterator<Element> getIdentities();
+}

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

+/**
+ * $RCSfile: $
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 Jive Software. All rights reserved.
+ *
+ * This software is published under the terms of the GNU Public License (GPL),
+ * a copy of which is included in this distribution.
+ */
+
+package org.jivesoftware.openfire.pep;
+
+import org.dom4j.DocumentHelper;
+import org.dom4j.Element;
+import org.jivesoftware.openfire.IQHandlerInfo;
+import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.openfire.auth.UnauthorizedException;
+import org.jivesoftware.openfire.disco.ServerFeaturesProvider;
+import org.jivesoftware.openfire.disco.ServerIdentitiesProvider;
+import org.jivesoftware.openfire.handler.IQHandler;
+import org.jivesoftware.openfire.pubsub.LeafNode;
+import org.jivesoftware.openfire.pubsub.PubSubEngine;
+import org.jivesoftware.util.Log;
+import org.xmpp.packet.IQ;
+import org.xmpp.packet.JID;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * <p>
+ * An IQHandler used to implement XEP-0163: "Personal Eventing via Pubsub."
+ * </p>
+ * 
+ * <p>
+ * For each user on the server there is an associated PEPService interacting
+ * with a PubSubEngine for managing the user's PEP nodes.
+ * </p>
+ * 
+ * <p>
+ * An IQHandler can only handle one namespace in its IQHandlerInfo. However, PEP
+ * related packets are seen having a variety of different namespaces. Thus,
+ * classes like IQPEPOwnerHandler are used to forward packets having these other
+ * namespaces to IQPEPHandler.handleIQ().
+ * <p>
+ * 
+ * <p>
+ * This handler is used for the following namespaces:
+ * <ul>
+ * <li><i>http://jabber.org/protocol/pubsub</i></li>
+ * <li><i>http://jabber.org/protocol/pubsub#owner</i></li>
+ * </ul>
+ * </p>
+ * 
+ * @author Armando Jagucki
+ * 
+ */
+public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
+        ServerFeaturesProvider {
+
+    // Map of PEP services. Table, Key: bare JID (String); Value: PEPService
+    private Map<String, PEPService> pepServices;
+
+    private IQHandlerInfo info;
+
+    private PubSubEngine pubSubEngine = null;
+
+    public IQPEPHandler() {
+        super("Personal Eventing Handler");
+        pepServices = new ConcurrentHashMap<String, PEPService>();
+        info = new IQHandlerInfo("pubsub", "http://jabber.org/protocol/pubsub");
+    }
+
+    @Override
+    public void initialize(XMPPServer server) {
+        super.initialize(server);
+        
+        pubSubEngine = new PubSubEngine(server.getPacketRouter());
+
+    }
+
+    @Override
+    public IQHandlerInfo getInfo() {
+        return info;
+    }
+
+    @Override
+    public IQ handleIQ(IQ packet) throws UnauthorizedException {
+        // TODO: Much to be done here...
+
+        if (packet.getTo() == null) {
+            String jidFrom = packet.getFrom().toBareJID();
+            
+            PEPService pepService = pepServices.get(jidFrom);
+
+            // If no service exists yet for jidFrom, create one.
+            if (pepService == null) {
+                pepService = new PEPService(XMPPServer.getInstance(), jidFrom);
+                pepServices.put(jidFrom, pepService);
+                pubSubEngine.start(pepService);     // Keep DB synced
+                Log.debug("PEP: " + jidFrom + " had a PEPService created");
+            }
+
+            // If publishing a node, and the node doesn't exist, create it.
+            if (packet.getType() == IQ.Type.set) {
+                Element childElement = packet.getChildElement();
+                Element publishElement = childElement.element("publish");
+                if (publishElement != null) {
+                    String nodeID = publishElement.attributeValue("node");
+                    
+                    if (pepService.getNode(nodeID) == null) {
+                        // Create the node
+                        JID creator = new JID(jidFrom);
+                        LeafNode newNode = new LeafNode(pepService, null, nodeID, creator);
+                        newNode.addOwner(creator);
+                        newNode.saveToDB();
+                        Log.debug("PEP: Created node ('" + nodeID + "') for " + jidFrom);
+                    }
+                }                        
+            }
+            
+            // Process with PubSub as usual.
+            if (pubSubEngine.process(pepService, packet)) {
+                Log.debug("PEP: The pubSubEngine processed a packet for " + jidFrom + "'s pepService.");
+            }
+            else {
+                Log.debug("PEP: The pubSubEngine did not process a packet for " + jidFrom + "'s pepService.");
+            }
+
+        }
+        else {
+            // TODO: Ensure the packet gets handled elsewhere.
+        }
+
+        return null; // Error flows are handled in pubSubEngine.process(...)
+    }
+
+    public void start() {
+        super.start();
+        for (PEPService service : pepServices.values()) {
+            pubSubEngine.start(service);
+        }
+    }
+
+    public void stop() {
+        super.stop();
+        for (PEPService service : pepServices.values()) {
+            pubSubEngine.shutdown(service);
+        }
+    }
+
+    /**
+     * Implements ServerIdentitiesProvider, adding the PEP identity to the
+     * server's disco#info result.
+     */
+    public Iterator<Element> getIdentities() {
+        ArrayList<Element> identities = new ArrayList<Element>();
+        Element identity = DocumentHelper.createElement("identity");
+        identity.addAttribute("category", "pubsub");
+        identity.addAttribute("type", "pep");
+        identities.add(identity);
+        return identities.iterator();
+    }
+
+    /**
+     * Implements ServerFeaturesProvider to include all supported XEP-0060 features
+     * in the server's disco#info result (as per section 4 of XEP-0163).
+     */
+    public Iterator<String> getFeatures() {
+        return XMPPServer.getInstance().getPubSubModule().getFeatures(null, null, null);
+    }
+
+}

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

+/**
+ * $RCSfile: $
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 Jive Software. All rights reserved.
+ *
+ * This software is published under the terms of the GNU Public License (GPL),
+ * a copy of which is included in this distribution.
+ */
+
+package org.jivesoftware.openfire.pep;
+
+import org.jivesoftware.openfire.IQHandlerInfo;
+import org.jivesoftware.openfire.auth.UnauthorizedException;
+import org.jivesoftware.openfire.handler.IQHandler;
+import org.jivesoftware.openfire.XMPPServer;
+import org.xmpp.packet.IQ;
+
+/**
+ * <p>
+ * An IQHandler used to implement XEP-0163: "Personal Eventing via Pubsub."
+ * </p>
+ * 
+ * <p>
+ * An IQHandler can only handle one namespace in its IQHandlerInfo. However, PEP
+ * related packets are seen having a variety of different namespaces. This
+ * handler is needed to forward IQ packets with the
+ * <i>'http://jabber.org/protocol/pubsub#owner'</i> namespace to IQPEPHandler.
+ * </p>
+ * 
+ * @author Armando Jagucki
+ * 
+ */
+public class IQPEPOwnerHandler extends IQHandler {
+
+    private IQHandlerInfo info;
+
+    public IQPEPOwnerHandler() {
+        super("Personal Eventing 'pubsub#owner' Handler");
+        info = new IQHandlerInfo("pubsub", "http://jabber.org/protocol/pubsub#owner");
+    }
+
+    @Override
+    public IQHandlerInfo getInfo() {
+        return info;
+    }
+
+    @Override
+    public IQ handleIQ(IQ packet) throws UnauthorizedException {
+        return XMPPServer.getInstance().getIQPEPHandler().handleIQ(packet);
+    }
+}

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

@@ -51,7 +51,7 @@ public class CollectionNode extends Node {
      */
     private int maxLeafNodes = -1;
 
-    CollectionNode(PubSubService service, CollectionNode parentNode, String nodeID, JID creator) {
+    public CollectionNode(PubSubService service, CollectionNode parentNode, String nodeID, JID creator) {
         super(service, parentNode, nodeID, creator);
         // Configure node with default values (get them from the pubsub service)
         DefaultNodeConfiguration defaultConfiguration = service.getDefaultNodeConfiguration(false);

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

@@ -61,7 +61,7 @@ public class LeafNode extends Node {
 
     // TODO Add checking of max payload size. Return <not-acceptable> plus a application specific error condition of <payload-too-big/>.
 
-    LeafNode(PubSubService service, CollectionNode parentNode, String nodeID, JID creator) {
+    public LeafNode(PubSubService service, CollectionNode parentNode, String nodeID, JID creator) {
         super(service, parentNode, nodeID, creator);
         // Configure node with default values (get them from the pubsub service)
         DefaultNodeConfiguration defaultConfiguration = service.getDefaultNodeConfiguration(true);

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

@@ -26,11 +26,11 @@ import java.util.List;
  *
  * @author Matt Tucker
  */
-class PendingSubscriptionsCommand extends AdHocCommand {
+public class PendingSubscriptionsCommand extends AdHocCommand {
 
     private PubSubService service;
 
-    PendingSubscriptionsCommand(PubSubService service) {
+    public PendingSubscriptionsCommand(PubSubService service) {
         this.service = service;
     }
 

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

@@ -19,6 +19,7 @@ import org.jivesoftware.openfire.RoutingTable;
 import org.jivesoftware.openfire.XMPPServer;
 import org.jivesoftware.openfire.cluster.ClusterEventListener;
 import org.jivesoftware.openfire.cluster.ClusterManager;
+import org.jivesoftware.openfire.commands.AdHocCommandManager;
 import org.jivesoftware.openfire.component.InternalComponentManager;
 import org.jivesoftware.openfire.container.BasicModule;
 import org.jivesoftware.openfire.disco.DiscoInfoProvider;
@@ -39,6 +40,7 @@ import org.xmpp.packet.*;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * Module that implements JEP-60: Publish-Subscribe. By default node collections and
@@ -63,6 +65,47 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
      * Nodes managed by this manager, table: key nodeID (String); value Node
      */
     private Map<String, Node> nodes = new ConcurrentHashMap<String, Node>();
+    
+    /**
+     * Keep a registry of the presence's show value of users that subscribed to a node of
+     * the pubsub service and for which the node only delivers notifications for online users
+     * or node subscriptions deliver events based on the user presence show value. Offline
+     * users will not have an entry in the map. Note: Key-> bare JID and Value-> Map whose key
+     * is full JID of connected resource and value is show value of the last received presence.
+     */
+    private Map<String, Map<String, String>> barePresences =
+            new ConcurrentHashMap<String, Map<String, String>>();
+    
+    /**
+     * Queue that holds the items that need to be added to the database.
+     */
+    private Queue<PublishedItem> itemsToAdd = new LinkedBlockingQueue<PublishedItem>();
+    /**
+     * Queue that holds the items that need to be deleted from the database.
+     */
+    private Queue<PublishedItem> itemsToDelete = new LinkedBlockingQueue<PublishedItem>();
+    
+    /**
+     * Manager that keeps the list of ad-hoc commands and processing command requests.
+     */
+    private AdHocCommandManager manager;
+    
+    /**
+     * The time to elapse between each execution of the maintenance process. Default
+     * is 2 minutes.
+     */
+    private int items_task_timeout = 2 * 60 * 1000;
+
+    /**
+     * Task that saves or deletes published items from the database.
+     */
+    private PublishedItemTask publishedItemTask;
+
+    /**
+     * Timer to save published items to the database or remove deleted or old items.
+     */
+    private Timer timer = new Timer("PubSub maintenance");
+
     /**
      * Returns the permission policy for creating nodes. A true value means that not anyone can
      * create a node, only the JIDs listed in <code>allowedToCreate</code> are allowed to create
@@ -117,6 +160,15 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
 
     public PubSubModule() {
         super("Publish Subscribe Service");
+        
+        // Initialize the ad-hoc commands manager to use for this pubsub service
+        manager = new AdHocCommandManager();
+        manager.addCommand(new PendingSubscriptionsCommand(this));
+        
+        // Save or delete published items from the database every 2 minutes starting in
+        // 2 minutes (default values)
+        publishedItemTask = new PublishedItemTask(this);
+        timer.schedule(publishedItemTask, items_task_timeout, items_task_timeout);
     }
 
     public void process(Packet packet) {
@@ -127,15 +179,15 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
         try {
             // Check if the packet is a disco request or a packet with namespace iq:register
             if (packet instanceof IQ) {
-                if (!engine.process((IQ) packet)) {
+                if (!engine.process(this, (IQ) packet)) {
                     process((IQ) packet);
                 }
             }
             else if (packet instanceof Presence) {
-                engine.process((Presence) packet);
+                engine.process(this, (Presence) packet);
             }
             else {
-                engine.process((Message) packet);
+                engine.process(this, (Message) packet);
             }
         }
         catch (Exception e) {
@@ -216,23 +268,23 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
     }
 
     public Collection<String> getShowPresences(JID subscriber) {
-        return engine.getShowPresences(subscriber);
+        return PubSubEngine.getShowPresences(this, subscriber);
     }
 
     public void presenceSubscriptionNotRequired(Node node, JID user) {
-        engine.presenceSubscriptionNotRequired(node, user);
+        PubSubEngine.presenceSubscriptionNotRequired(this, node, user);
     }
 
     public void presenceSubscriptionRequired(Node node, JID user) {
-        engine.presenceSubscriptionRequired(node, user);
+        PubSubEngine.presenceSubscriptionRequired(this, node, user);
     }
 
     public void queueItemToAdd(PublishedItem newItem) {
-        engine.queueItemToAdd(newItem);
+        PubSubEngine.queueItemToAdd(this, newItem);
     }
 
     public void queueItemToRemove(PublishedItem removedItem) {
-        engine.queueItemToRemove(removedItem);
+        PubSubEngine.queueItemToRemove(this, removedItem);
     }
 
     public String getServiceName() {
@@ -334,7 +386,7 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
         routingTable = server.getRoutingTable();
         router = server.getPacketRouter();
 
-        engine = new PubSubEngine(this, server.getPacketRouter());
+        engine = new PubSubEngine(server.getPacketRouter());
 
         // Load default configuration for leaf nodes
         leafDefaultConfiguration = PubSubPersistenceManager.loadDefaultConfiguration(this, true);
@@ -406,7 +458,7 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
         // Add the route to this service
         routingTable.addComponentRoute(getAddress(), this);
         // Start the pubsub engine
-        engine.start();
+        engine.start(this);
         ArrayList<String> params = new ArrayList<String>();
         params.clear();
         params.add(getServiceDomain());
@@ -419,7 +471,7 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
         routingTable.removeComponentRoute(getAddress());
         // Stop the pubsub engine. This will gives us the chance to
         // save queued items to the database.
-        engine.shutdown();
+        engine.shutdown(this);
     }
 
     private void enableService(boolean enabled) {
@@ -750,4 +802,40 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
         }
         return buf.toString();
     }
+
+    public Map<String, Map<String, String>> getBarePresences() {
+        return barePresences;
+    }
+
+    public Queue<PublishedItem> getItemsToAdd() {
+        return itemsToAdd;
+    }
+
+    public Queue<PublishedItem> getItemsToDelete() {
+        return itemsToDelete;
+    }
+
+    public AdHocCommandManager getManager() {
+        return manager;
+    }
+
+    public PublishedItemTask getPublishedItemTask() {
+        return publishedItemTask;
+    }
+
+    public void setPublishedItemTask(PublishedItemTask task) {
+        publishedItemTask = task;
+    }
+
+    public Timer getTimer() {
+        return timer;
+    }
+    
+    public int getItemsTaskTimeout() {
+        return items_task_timeout;
+    }
+
+    public void setItemsTaskTimeout(int timeout) {
+        items_task_timeout = timeout;
+    }
 }

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

@@ -11,11 +11,15 @@
 
 package org.jivesoftware.openfire.pubsub;
 
+import org.jivesoftware.openfire.commands.AdHocCommandManager;
 import org.xmpp.packet.JID;
 import org.xmpp.packet.Message;
 import org.xmpp.packet.Packet;
 
 import java.util.Collection;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Timer;
 
 /**
  * A PubSubService is responsible for keeping the hosted nodes by the service, the default
@@ -50,6 +54,18 @@ public interface PubSubService {
      */
     String getServiceID();
 
+    /**
+     * Returns a registry of the presence's show value of users that subscribed to a node of
+     * the pubsub service and for which the node only delivers notifications for online users
+     * or node subscriptions deliver events based on the user presence show value. Offline
+     * users will not have an entry in the map. Note: Key-> bare JID and Value-> Map whose key
+     * is full JID of connected resource and value is show value of the last received presence.
+     * 
+     * @return a registry of the presence's show value of users that subscribed to a node
+     *         of the pubsub service.
+     */
+    Map<String, Map<String, String>> getBarePresences();
+
     /**
      * Returns true if the pubsub service allows the specified user to create nodes.
      *
@@ -222,6 +238,41 @@ public interface PubSubService {
      */
     void queueItemToAdd(PublishedItem newItem);
 
+    /**
+     * Gets the queue that holds the items that need to be added to the database.
+     * 
+     * @return the queue that holds the items that need to be added to the database.
+     */
+    Queue<PublishedItem> getItemsToAdd();
+
+    /**
+     * Gets the queue that holds the items that need to be deleted from the database.
+     * 
+     * @return the queue that holds the items that need to be deleted from the database.
+     */
+    Queue<PublishedItem> getItemsToDelete();
+
+    /**
+     * Returns the ad-hoc commands manager used for this service.
+     * 
+     * @return the ad-hoc commands manager used for this service.
+     */
+    AdHocCommandManager getManager();
+
+    /**
+     * Returns the published item task used for this service.
+     * 
+     * @return the published item task used for this service.
+     */
+    PublishedItemTask getPublishedItemTask();
+
+    /**
+     * Sets the published item task used for this service.
+     * 
+     * @param task the PublishedItemTask to set for this service.
+     */
+    void setPublishedItemTask(PublishedItemTask task);
+
     /**
      * Adds the item to the queue of items to remove from the database. The queue is going
      * to be processed by another thread.
@@ -229,4 +280,26 @@ public interface PubSubService {
      * @param removedItem the item to remove from the database.
      */
     void queueItemToRemove(PublishedItem removedItem);
+
+    /**
+     * Returns the timer used for the maintenance process of this service.
+     * 
+     * @return the timer used for the maintenance process of this service.
+     */
+    Timer getTimer();
+
+    /**
+     * Returns the timeout value for the published items maintenance task.
+     * 
+     * @return the timeout value for the published items maintenance task.
+     */
+    int getItemsTaskTimeout();
+
+    /**
+     * Sets the timeout value for the published items maintenance task.
+     *
+     * @param timeout the timeout value for the published items maintenance task.
+     */
+    void setItemsTaskTimeout(int timeout);
+
 }

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

+/**
+ * $RCSfile: $
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 Jive Software. All rights reserved.
+ *
+ * This software is published under the terms of the GNU Public License (GPL),
+ * a copy of which is included in this distribution.
+ */
+
+package org.jivesoftware.openfire.pubsub;
+
+import org.jivesoftware.util.LocaleUtils;
+import org.jivesoftware.util.Log;
+
+import java.util.Queue;
+import java.util.TimerTask;
+
+/**
+ * A timed maintenance task that updates the database by adding and/or
+ * removing <code>PublishedItem</code>s in regular intervals.
+ * 
+ * @author Matt Tucker
+ */
+public class PublishedItemTask extends TimerTask {
+    /**
+     * Queue that holds the items that need to be added to the database.
+     */
+    private Queue<PublishedItem> itemsToAdd = null;
+
+    /**
+     * Queue that holds the items that need to be deleted from the database.
+     */
+    private Queue<PublishedItem> itemsToDelete = null;
+
+    /**
+     * The service to perform the published item tasks on.
+     */
+    private PubSubService service = null;
+
+    /**
+     * The number of items to save on each run of the maintenance process.
+     */
+    private int items_batch_size = 50;
+
+    public PublishedItemTask(PubSubService service) {
+        this.service = service;
+        this.itemsToAdd = service.getItemsToAdd();
+        this.itemsToDelete = service.getItemsToDelete();
+    }
+
+    public void run() {
+        try {
+            PublishedItem entry;
+            boolean success;
+            // Delete from the database items contained in the itemsToDelete queue
+            for (int index = 0; index <= items_batch_size && !itemsToDelete.isEmpty(); index++) {
+                entry = itemsToDelete.poll();
+                if (entry != null) {
+                    success = PubSubPersistenceManager.removePublishedItem(service, entry);
+                    if (!success) {
+                        itemsToDelete.add(entry);
+                    }
+                }
+            }
+            // Save to the database items contained in the itemsToAdd queue
+            for (int index = 0; index <= items_batch_size && !itemsToAdd.isEmpty(); index++) {
+                entry = itemsToAdd.poll();
+                if (entry != null) {
+                    success = PubSubPersistenceManager.createPublishedItem(service, entry);
+                    if (!success) {
+                        itemsToAdd.add(entry);
+                    }
+                }
+            }
+        } catch (Throwable e) {
+            Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
+        }
+    }
+}