OF-205 Initial refactoring for clustering support.

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

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

@@ -157,6 +157,11 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
         pepServiceManager = new PEPServiceManager();
     }
 
+	public PEPServiceManager getServiceManager()
+	{
+		return pepServiceManager;
+	}
+
     /*
 	 * (non-Javadoc)
 	 * 
@@ -441,7 +446,7 @@ public class IQPEPHandler extends IQHandler implements ServerIdentitiesProvider,
      *
      * @return the knownRemotePresences map
      */
-    public Map<String, Set<JID>> getKnownRemotePresenes() {
+    public Map<String, Set<JID>> getKnownRemotePresences() {
         return knownRemotePresences;
     }
 

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

@@ -348,7 +348,8 @@ public class PEPService implements PubSubService, Cacheable {
         else {
             // Since recipientJID is not local, try to get presence info from cached known remote
             // presences.
-            Map<String, Set<JID>> knownRemotePresences = XMPPServer.getInstance().getIQPEPHandler().getKnownRemotePresenes();
+			Map<String, Set<JID>> knownRemotePresences = XMPPServer.getInstance().getIQPEPHandler()
+					.getKnownRemotePresences();
 
             Set<JID> remotePresenceSet = knownRemotePresences.get(getAddress().toBareJID());
             if (remotePresenceSet != null) {

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

@@ -21,19 +21,15 @@
 package org.jivesoftware.openfire.pubsub;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.Date;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.dom4j.Element;
 import org.jivesoftware.util.LocaleUtils;
-import org.jivesoftware.util.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.xmpp.forms.DataForm;
@@ -76,11 +72,9 @@ public class LeafNode extends Node {
      */
     private boolean sendItemSubscribe;
     /**
-     * List of items that were published to the node and that are still active. If the node is
-     * not configured to persist items then the last published item will be kept. The list is
-     * sorted cronologically.
+     * The last item published to this node.  In a cluster this may have occurred on a different cluster node.
      */
-    volatile private PublishedItem lastPublished;
+    private PublishedItem lastPublished;
 
     // TODO Add checking of max payload size. Return <not-acceptable> plus a application specific error condition of <payload-too-big/>.
 
@@ -173,8 +167,10 @@ public class LeafNode extends Node {
 	protected void deletingNode() {
     }
 
-    void setLastPublishedItem(PublishedItem item) {
-    	lastPublished = item;
+	public synchronized void setLastPublishedItem(PublishedItem item)
+	{
+		if ((lastPublished == null) || (item != null) && item.getCreationDate().after(lastPublished.getCreationDate()))
+			lastPublished = item;
     }
 
     public int getMaxPayloadSize() {
@@ -360,7 +356,7 @@ public class LeafNode extends Node {
     }
 
     @Override
-	public PublishedItem getLastPublishedItem() {
+	public synchronized PublishedItem getLastPublishedItem() {
     	return lastPublished;
     }
 

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

@@ -20,21 +20,28 @@
 
 package org.jivesoftware.openfire.pubsub;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
 import org.dom4j.Element;
+import org.jivesoftware.openfire.pubsub.cluster.NewSubscriptionTask;
 import org.jivesoftware.openfire.pubsub.models.AccessModel;
 import org.jivesoftware.openfire.pubsub.models.PublisherModel;
 import org.jivesoftware.util.LocaleUtils;
 import org.jivesoftware.util.StringUtils;
+import org.jivesoftware.util.cache.CacheFactory;
 import org.xmpp.forms.DataForm;
 import org.xmpp.forms.FormField;
 import org.xmpp.packet.IQ;
 import org.xmpp.packet.JID;
 import org.xmpp.packet.Message;
 
-import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArrayList;
-
 /**
  * A virtual location to which information can be published and from which event
  * notifications and/or payloads can be received (in other pubsub systems, this may
@@ -1734,7 +1741,8 @@ public abstract class Node {
         affiliates.add(affiliate);
     }
 
-    void addSubscription(NodeSubscription subscription) {
+	public void addSubscription(NodeSubscription subscription)
+	{
         subscriptionsByID.put(subscription.getID(), subscription);
         subscriptionsByJID.put(subscription.getJID().toString(), subscription);
     }
@@ -2017,8 +2025,7 @@ public abstract class Node {
         // Generate a subscription ID (override even if one was sent by the client)
         String id = StringUtils.randomString(40);
         // Create new subscription
-        NodeSubscription subscription =
-                new NodeSubscription(service, this, owner, subscriber, subState, id);
+        NodeSubscription subscription = new NodeSubscription(this, owner, subscriber, subState, id);
         // Configure the subscription with the specified configuration (if any)
         if (options != null) {
             subscription.configure(options);
@@ -2041,6 +2048,10 @@ public abstract class Node {
         if (subscription.isAuthorizationPending()) {
             subscription.sendAuthorizationRequest();
         }
+        
+		// Synchronous so the task can flush all other cluster nodes before
+		// the last item is retrieved.
+		CacheFactory.doSynchronousClusterTask(new NewSubscriptionTask(subscription), false);
 
         // Send last published item (if node is leaf node and subscription status is ok)
         if (isSendItemSubscribe() && subscription.isActive()) {

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

@@ -69,10 +69,7 @@ public class NodeSubscription {
 
     private static final SimpleDateFormat dateFormat;
     private static final FastDateFormat fastDateFormat;
-    /**
-     * Reference to the publish and subscribe service.
-     */
-    private PubSubService service;
+
     /**
      * The node to which this subscription is interested in.
      */
@@ -160,8 +157,8 @@ public class NodeSubscription {
      * @param state the state of the subscription with the node.
      * @param id the id the uniquely identifies this subscriptin within the node.
      */
-    NodeSubscription(PubSubService service, Node node, JID owner, JID jid, State state, String id) {
-        this.service = service;
+	public NodeSubscription(Node node, JID owner, JID jid, State state, String id)
+	{
         this.node = node;
         this.jid = jid;
         this.owner = owner;
@@ -405,7 +402,7 @@ public class NodeSubscription {
         configure(options);
         if (originalIQ != null) {
             // Return success response
-            service.send(IQ.createResultIQ(originalIQ));
+			node.getService().send(IQ.createResultIQ(originalIQ));
         }
 
         if (wasUnconfigured) {
@@ -519,10 +516,10 @@ public class NodeSubscription {
         // Check if the service needs to subscribe or unsubscribe from the owner presence
         if (!node.isPresenceBasedDelivery() && wasUsingPresence != !presenceStates.isEmpty()) {
             if (presenceStates.isEmpty()) {
-                service.presenceSubscriptionNotRequired(node, owner);
+				node.getService().presenceSubscriptionNotRequired(node, owner);
             }
             else {
-                service.presenceSubscriptionRequired(node, owner);
+				node.getService().presenceSubscriptionRequired(node, owner);
             }
         }
     }
@@ -719,7 +716,7 @@ public class NodeSubscription {
         }
         // Check if delivery is subject to presence-based policy
         if (!getPresenceStates().isEmpty()) {
-            Collection<String> shows = service.getShowPresences(jid);
+			Collection<String> shows = node.getService().getShowPresences(jid);
             if (shows.isEmpty() || Collections.disjoint(getPresenceStates(), shows)) {
                 return false;
             }
@@ -727,7 +724,8 @@ public class NodeSubscription {
         // Check if node is only sending events when user is online
         if (node.isPresenceBasedDelivery()) {
             // Check that user is online
-            if (service.getShowPresences(jid).isEmpty()) {
+			if (node.getService().getShowPresences(jid).isEmpty())
+			{
                 return false;
             }
         }
@@ -795,7 +793,7 @@ public class NodeSubscription {
             subscribeOptions.addElement("required");
         }
         // Send the result
-        service.send(result);
+		node.getService().send(result);
     }
 
     /**
@@ -834,7 +832,7 @@ public class NodeSubscription {
         notification.getElement().addElement("delay", "urn:xmpp:delay")
                 .addAttribute("stamp", fastDateFormat.format(publishedItem.getCreationDate()));
         // Send the event notification to the subscriber
-        service.sendNotification(node, notification, jid);
+		node.getService().sendNotification(node, notification, jid);
     }
 
     /**
@@ -846,7 +844,7 @@ public class NodeSubscription {
      * @return true if the specified user is allowed to modify or cancel the subscription.
      */
     boolean canModify(JID user) {
-        return user.equals(getJID()) || user.equals(getOwner()) || service.isServiceAdmin(user);
+		return user.equals(getJID()) || user.equals(getOwner()) || node.getService().isServiceAdmin(user);
     }
 
     /**
@@ -899,9 +897,9 @@ public class NodeSubscription {
         Message authRequest = new Message();
         authRequest.addExtension(node.getAuthRequestForm(this));
         authRequest.setTo(owner);
-        authRequest.setFrom(service.getAddress());
+		authRequest.setFrom(node.getService().getAddress());
         // Send authentication request to node owners
-        service.send(authRequest);
+		node.getService().send(authRequest);
     }
 
     /**
@@ -912,7 +910,7 @@ public class NodeSubscription {
         Message authRequest = new Message();
         authRequest.addExtension(node.getAuthRequestForm(this));
         // Send authentication request to node owners
-        service.broadcast(node, authRequest, node.getOwners());
+		node.getService().broadcast(node, authRequest, node.getOwners());
     }
 
     /**

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

@@ -39,9 +39,11 @@ import org.jivesoftware.openfire.RoutingTable;
 import org.jivesoftware.openfire.XMPPServer;
 import org.jivesoftware.openfire.XMPPServerListener;
 import org.jivesoftware.openfire.component.InternalComponentManager;
+import org.jivesoftware.openfire.pubsub.cluster.RefreshNodeTask;
 import org.jivesoftware.openfire.pubsub.models.AccessModel;
 import org.jivesoftware.openfire.user.UserManager;
 import org.jivesoftware.util.StringUtils;
+import org.jivesoftware.util.cache.CacheFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.xmpp.forms.DataForm;
@@ -1210,6 +1212,8 @@ public class PubSubEngine {
                     else {
                         newNode.saveToDB();
                     }
+
+					CacheFactory.doClusterTask(new RefreshNodeTask(newNode));
                 }
                 else {
                     conflict = true;
@@ -1302,6 +1306,8 @@ public class PubSubEngine {
                 // Update node configuration with the provided data form
                 // (and update the backend store)
                 node.configure(completedForm);
+
+				CacheFactory.doClusterTask(new RefreshNodeTask(node));
                 // Return that node configuration was successful
                 router.route(IQ.createResultIQ(iq));
             }

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

@@ -25,11 +25,8 @@ import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Queue;
-import java.util.Timer;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.LinkedBlockingQueue;
 
 import org.dom4j.DocumentHelper;
 import org.dom4j.Element;
@@ -166,10 +163,6 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
     }
 
     public void process(Packet packet) {
-        // TODO Remove this method when moving PubSub as a component and removing module code
-        // The MUC service will receive all the packets whose domain matches the domain of the MUC
-        // service. This means that, for instance, a disco request should be responded by the
-        // service itself instead of relying on the server to handle the request.
         try {
             // Check if the packet is a disco request or a packet with namespace iq:register
             if (packet instanceof IQ) {
@@ -510,27 +503,32 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
         return serviceEnabled;
     }
 
+	@Override
     public void joinedCluster() {
         // Disable the service until we know that we are the senior cluster member
-        enableService(false);
+		// enableService(false);
     }
 
+	@Override
     public void joinedCluster(byte[] nodeID) {
         // Do nothing
     }
 
+	@Override
     public void leftCluster() {
         // Offer the service when not running in a cluster
-        enableService(true);
+		// enableService(true);
     }
 
+	@Override
     public void leftCluster(byte[] nodeID) {
         // Do nothing
     }
 
+	@Override
     public void markedAsSeniorClusterMember() {
         // Offer the service since we are the senior cluster member
-        enableService(true);
+		// enableService(true);
     }
 
     public Iterator<DiscoServerItem> getItems() {

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

@@ -61,31 +61,28 @@ public class PubSubPersistenceManager {
     private static final Logger log = LoggerFactory.getLogger(PubSubPersistenceManager.class);
     
     private static final String PURGE_FOR_SIZE = 
-	    "DELETE ofPubsubItem FROM ofPubsubItem LEFT JOIN " +
-	    "(SELECT id FROM ofPubsubItem WHERE serviceID=? AND nodeID=? " +
-	    "ORDER BY creationDate DESC LIMIT ?) AS noDelete " +
- "ON ofPubsubItem.id = noDelete.id WHERE noDelete.id IS NULL AND "
-	    + "ofPubsubItem.serviceID = ? AND nodeID = ?";
+    		"DELETE ofPubsubItem FROM ofPubsubItem LEFT JOIN " +
+			"(SELECT id FROM ofPubsubItem WHERE serviceID=? AND nodeID=? " +
+			"ORDER BY creationDate DESC LIMIT ?) AS noDelete " +
+			"ON ofPubsubItem.id = noDelete.id WHERE noDelete.id IS NULL AND " +
+			"ofPubsubItem.serviceID = ? AND nodeID = ?";
 	
 	private static final String PURGE_FOR_SIZE_HSQLDB = "DELETE FROM ofPubsubItem WHERE serviceID=? AND nodeID=? AND id NOT IN "
 			+ "(SELECT id FROM ofPubsubItem WHERE serviceID=? AND nodeID=? ORDER BY creationDate DESC LIMIT ?)";
 
-    private static final String LOAD_NON_LEAF_NODES =
+	private static final String LOAD_NODE_BASE =
             "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 ofPubsubNode " +
-            "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 ofPubsubNode " +
-            "WHERE serviceID=? AND leaf=1 ORDER BY nodeID";
+ "WHERE serviceID=?";
+
+	private static final String LOAD_NODES = LOAD_NODE_BASE + " AND leaf=? ORDER BY nodeID";
+
+	private static final String LOAD_NODE = LOAD_NODE_BASE + " AND nodeID=?";
+
     private static final String UPDATE_NODE =
             "UPDATE ofPubsubNode SET modificationDate=?, parent=?, deliverPayloads=?, " +
             "maxPayloadSize=?, persistItems=?, maxItems=?, " +
@@ -107,7 +104,8 @@ public class PubSubPersistenceManager {
 
     private static final String LOAD_NODES_JIDS =
             "SELECT nodeID, jid, associationType FROM ofPubsubNodeJIDs WHERE serviceID=?";
-    private static final String ADD_NODE_JIDS =
+	private static final String LOAD_NODE_JIDS = "SELECT nodeID, jid, associationType FROM ofPubsubNodeJIDs WHERE serviceID=? AND nodeID=?";
+	private static final String ADD_NODE_JIDS =
             "INSERT INTO ofPubsubNodeJIDs (serviceID, nodeID, jid, associationType) " +
             "VALUES (?,?,?,?)";
     private static final String DELETE_NODE_JIDS =
@@ -115,6 +113,7 @@ public class PubSubPersistenceManager {
 
     private static final String LOAD_NODES_GROUPS =
             "SELECT nodeID, rosterGroup FROM ofPubsubNodeGroups WHERE serviceID=?";
+	private static final String LOAD_NODE_GROUPS = "SELECT nodeID, rosterGroup FROM ofPubsubNodeGroups WHERE serviceID=? AND nodeID=?";
     private static final String ADD_NODE_GROUPS =
             "INSERT INTO ofPubsubNodeGroups (serviceID, nodeID, rosterGroup) " +
             "VALUES (?,?,?)";
@@ -124,6 +123,7 @@ public class PubSubPersistenceManager {
     private static final String LOAD_AFFILIATIONS =
             "SELECT nodeID,jid,affiliation FROM ofPubsubAffiliation WHERE serviceID=? " +
             "ORDER BY nodeID";
+	private static final String LOAD_NODE_AFFILIATIONS = "SELECT nodeID,jid,affiliation FROM ofPubsubAffiliation WHERE serviceID=? AND nodeID=?";
     private static final String ADD_AFFILIATION =
             "INSERT INTO ofPubsubAffiliation (serviceID,nodeID,jid,affiliation) VALUES (?,?,?,?)";
     private static final String UPDATE_AFFILIATION =
@@ -133,10 +133,13 @@ public class PubSubPersistenceManager {
     private static final String DELETE_AFFILIATIONS =
             "DELETE FROM ofPubsubAffiliation WHERE serviceID=? AND nodeID=?";
 
-    private static final String LOAD_SUBSCRIPTIONS =
-            "SELECT nodeID, id, jid, owner, state, deliver, digest, digest_frequency, " +
-            "expire, includeBody, showValues, subscriptionType, subscriptionDepth, " +
-            "keyword FROM ofPubsubSubscription WHERE serviceID=? ORDER BY nodeID";
+	private static final String LOAD_SUBSCRIPTIONS_BASE = "SELECT nodeID, id, jid, owner, state, deliver, digest, digest_frequency, "
+			+ "expire, includeBody, showValues, subscriptionType, subscriptionDepth, "
+			+ "keyword FROM ofPubsubSubscription WHERE serviceID=? ";
+	private static final String LOAD_NODE_SUBSCRIPTION = LOAD_SUBSCRIPTIONS_BASE + "AND nodeID=? AND id=?";
+	private static final String LOAD_NODE_SUBSCRIPTIONS = LOAD_SUBSCRIPTIONS_BASE + "AND nodeID=?";
+	private static final String LOAD_SUBSCRIPTIONS = LOAD_SUBSCRIPTIONS_BASE + "ORDER BY nodeID";
+
     private static final String ADD_SUBSCRIPTION =
             "INSERT INTO ofPubsubSubscription (serviceID, nodeID, id, jid, owner, state, " +
             "deliver, digest, digest_frequency, expire, includeBody, showValues, " +
@@ -544,8 +547,9 @@ public class PubSubPersistenceManager {
         try {
             con = DbConnectionManager.getConnection();
             // Get all non-leaf nodes (to ensure parent nodes are loaded before their children)
-            pstmt = con.prepareStatement(LOAD_NON_LEAF_NODES);
+			pstmt = con.prepareStatement(LOAD_NODES);
             pstmt.setString(1, service.getServiceID());
+			pstmt.setInt(2, 0);
             rs = pstmt.executeQuery();
             // Rebuild loaded non-leaf nodes
             while(rs.next()) {
@@ -554,8 +558,9 @@ public class PubSubPersistenceManager {
             DbConnectionManager.fastcloseStmt(rs, pstmt);
 
             // Get all leaf nodes (remaining unloaded nodes)
-            pstmt = con.prepareStatement(LOAD_LEAF_NODES);
+			pstmt = con.prepareStatement(LOAD_NODES);
             pstmt.setString(1, service.getServiceID());
+			pstmt.setInt(2, 1);
             rs = pstmt.executeQuery();
             // Rebuild loaded leaf nodes
             while(rs.next()) {
@@ -620,6 +625,105 @@ public class PubSubPersistenceManager {
         }
     }
 
+	/**
+	 * Loads all nodes from the database and adds them to the PubSub service.
+	 * 
+	 * @param service
+	 *            the pubsub service that is hosting the nodes.
+	 */
+	public static void loadNode(PubSubService service, String nodeId)
+	{
+		Connection con = null;
+		PreparedStatement pstmt = null;
+		ResultSet rs = null;
+		Map<String, Node> nodes = new HashMap<String, Node>();
+		try
+		{
+			con = DbConnectionManager.getConnection();
+			// Get all non-leaf nodes (to ensure parent nodes are loaded before
+			// their children)
+			pstmt = con.prepareStatement(LOAD_NODE);
+			pstmt.setString(1, service.getServiceID());
+			pstmt.setString(2, nodeId);
+			rs = pstmt.executeQuery();
+
+			// Rebuild loaded non-leaf nodes
+			if (rs.next())
+			{
+				loadNode(service, nodes, rs);
+			}
+			DbConnectionManager.fastcloseStmt(rs, pstmt);
+
+			// Get JIDs associated with all nodes
+			pstmt = con.prepareStatement(LOAD_NODE_JIDS);
+			pstmt.setString(1, service.getServiceID());
+			pstmt.setString(2, nodeId);
+			rs = pstmt.executeQuery();
+			// Add to each node the associated JIDs
+			while (rs.next())
+			{
+				loadAssociatedJIDs(nodes, rs);
+			}
+			DbConnectionManager.fastcloseStmt(rs, pstmt);
+
+			// Get roster groups associated with all nodes
+			pstmt = con.prepareStatement(LOAD_NODE_GROUPS);
+			pstmt.setString(1, service.getServiceID());
+			pstmt.setString(2, nodeId);
+			rs = pstmt.executeQuery();
+			// Add to each node the associated Groups
+			while (rs.next())
+			{
+				loadAssociatedGroups(nodes, rs);
+			}
+			DbConnectionManager.fastcloseStmt(rs, pstmt);
+
+			// Get affiliations of all nodes
+			pstmt = con.prepareStatement(LOAD_NODE_AFFILIATIONS);
+			pstmt.setString(1, service.getServiceID());
+			pstmt.setString(2, nodeId);
+			rs = pstmt.executeQuery();
+			// Add to each node the corresponding affiliates
+			while (rs.next())
+			{
+				loadAffiliations(nodes, rs);
+			}
+			DbConnectionManager.fastcloseStmt(rs, pstmt);
+
+			// Get subscriptions to all nodes
+			pstmt = con.prepareStatement(LOAD_NODE_SUBSCRIPTIONS);
+			pstmt.setString(1, service.getServiceID());
+			pstmt.setString(2, nodeId);
+			rs = pstmt.executeQuery();
+			// Add to each node the corresponding subscriptions
+			while (rs.next())
+			{
+				loadSubscriptions(service, nodes, rs);
+			}
+			DbConnectionManager.fastcloseStmt(rs, pstmt);
+		}
+		catch (SQLException sqle)
+		{
+			log.error(sqle.getMessage(), sqle);
+		}
+		finally
+		{
+			DbConnectionManager.closeConnection(rs, pstmt, con);
+		}
+
+		for (Node node : nodes.values())
+		{
+			// Set now that the node is persistent in the database. Note: We
+			// need to
+			// set this now since otherwise the node's affiliations will be
+			// saved to the database
+			// "again" while adding them to the node!
+			node.setSavedToDB(true);
+			// Add the node to the service
+			service.addNode(node);
+		}
+	}
+
     private static void loadNode(PubSubService service, Map<String, Node> loadedNodes, ResultSet rs) {
         Node node;
         try {
@@ -632,9 +736,14 @@ public class PubSubPersistenceManager {
                 // Check if the parent has already been loaded
                 parentNode = (CollectionNode) loadedNodes.get(parent);
                 if (parentNode == null) {
-                    // Parent is not in memory so try to load it
-                    log.warn("Node not loaded due to missing parent. NodeID: " + nodeID);
-                    return;
+					parentNode = (CollectionNode) service.getNode(parent);
+
+					if (parentNode == null)
+					{
+						// Parent is not in memory so try to load it
+						log.warn("Node not loaded due to missing parent. NodeID: " + nodeID);
+						return;
+					}
                 }
             }
 
@@ -746,6 +855,41 @@ public class PubSubPersistenceManager {
         }
     }
 
+	public static void loadSubscription(PubSubService service, Node node, String subId)
+	{
+		Connection con = null;
+		PreparedStatement pstmt = null;
+		ResultSet rs = null;
+		Map<String, Node> nodes = new HashMap<String, Node>();
+		nodes.put(node.getNodeID(), node);
+
+		try
+		{
+			con = DbConnectionManager.getConnection();
+
+			// Get subscriptions to all nodes
+			pstmt = con.prepareStatement(LOAD_NODE_SUBSCRIPTION);
+			pstmt.setString(1, service.getServiceID());
+			pstmt.setString(2, node.getNodeID());
+			pstmt.setString(3, subId);
+			rs = pstmt.executeQuery();
+
+			// Add to each node the corresponding subscription
+			if (rs.next())
+			{
+				loadSubscriptions(service, nodes, rs);
+			}
+		}
+		catch (SQLException sqle)
+		{
+			log.error(sqle.getMessage(), sqle);
+		}
+		finally
+		{
+			DbConnectionManager.closeConnection(rs, pstmt, con);
+		}
+	}
+
     private static void loadSubscriptions(PubSubService service, Map<String, Node> nodes, ResultSet rs) {
         try {
             String nodeID = decodeNodeID(rs.getString(1));
@@ -763,8 +907,7 @@ public class PubSubPersistenceManager {
                 return;
             }
             NodeSubscription.State state = NodeSubscription.State.valueOf(rs.getString(5));
-            NodeSubscription subscription =
-                    new NodeSubscription(service, node, owner, subscriber, state, subID);
+			NodeSubscription subscription = new NodeSubscription(node, owner, subscriber, state, subID);
             subscription.setShouldDeliverNotifications(rs.getInt(6) == 1);
             subscription.setUsingDigest(rs.getInt(7) == 1);
             subscription.setDigestFrequency(rs.getInt(8));
@@ -993,7 +1136,7 @@ public class PubSubPersistenceManager {
     /**
      * Flush the cache of items to be persisted and deleted.
      */
-    public static void flushItems()
+	public static void flushItems()
     {
     	log.debug("Flushing items to database");