Partial rewrite of the PEP implementation. A cache is used that's regularly...

Partial rewrite of the PEP implementation. A cache is used that's regularly cleaned, which should improve concurrency and reduce the memory footprint (OF-82).

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

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

@@ -20,6 +20,16 @@
 
 package org.jivesoftware.openfire.pep;
 
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.Timer;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+
 import org.dom4j.DocumentHelper;
 import org.dom4j.Element;
 import org.dom4j.QName;
@@ -29,7 +39,17 @@ import org.jivesoftware.openfire.XMPPServer;
 import org.jivesoftware.openfire.commands.AdHocCommandManager;
 import org.jivesoftware.openfire.entitycaps.EntityCapabilities;
 import org.jivesoftware.openfire.entitycaps.EntityCapabilitiesManager;
-import org.jivesoftware.openfire.pubsub.*;
+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;
+import org.jivesoftware.openfire.pubsub.PubSubService;
+import org.jivesoftware.openfire.pubsub.PublishedItem;
+import org.jivesoftware.openfire.pubsub.PublishedItemTask;
 import org.jivesoftware.openfire.pubsub.models.AccessModel;
 import org.jivesoftware.openfire.pubsub.models.PublisherModel;
 import org.jivesoftware.openfire.roster.Roster;
@@ -44,10 +64,6 @@ import org.xmpp.packet.Message;
 import org.xmpp.packet.Packet;
 import org.xmpp.packet.PacketExtension;
 
-import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-
 /**
  * A PEPService is a {@link PubSubService} for use with XEP-0163: "Personal Eventing via
  * Pubsub" Version 1.0
@@ -165,7 +181,9 @@ public class PEPService implements PubSubService {
 
         // Save or delete published items from the database every 2 minutes
         // starting in 2 minutes (default values)
-        publishedItemTask = new PublishedItemTask(this);
+        publishedItemTask = new PublishedItemTask(this) {
+        	
+        };
         timer.schedule(publishedItemTask, items_task_timeout, items_task_timeout);
 
         // Load default configuration for leaf nodes
@@ -530,12 +548,10 @@ public class PEPService implements PubSubService {
 
     public void queueItemToAdd(PublishedItem newItem) {
         PubSubEngine.queueItemToAdd(this, newItem);
-
     }
 
     public void queueItemToRemove(PublishedItem removedItem) {
         PubSubEngine.queueItemToRemove(this, removedItem);
-
     }
 
     public Map<String, Map<String, String>> getBarePresences() {

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

+/**
+ * Copyright (C) 2004-2008 Jive Software. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jivesoftware.openfire.pep;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.concurrent.locks.Lock;
+
+import org.jivesoftware.database.DbConnectionManager;
+import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.openfire.pubsub.CollectionNode;
+import org.jivesoftware.openfire.pubsub.Node;
+import org.jivesoftware.openfire.pubsub.PubSubEngine;
+import org.jivesoftware.openfire.user.UserManager;
+import org.jivesoftware.util.cache.Cache;
+import org.jivesoftware.util.cache.CacheFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.xmpp.packet.IQ;
+import org.xmpp.packet.JID;
+
+/**
+ * Manages the creation, persistence and removal of {@link PEPService}
+ * instances.
+ * 
+ * @author Guus der Kinderen, guus.der.kinderen@gmail.com
+ * 
+ */
+public class PEPServiceManager {
+
+	public static final Logger Log = LoggerFactory
+			.getLogger(PEPServiceManager.class);
+
+	private final static String GET_PEP_SERVICE = "SELECT DISTINCT serviceID FROM ofPubsubNode WHERE serviceID=?";
+
+	/**
+	 * Cache of PEP services. Table, Key: bare JID (String); Value: PEPService
+	 */
+	private final Cache<String, PEPService> pepServices = CacheFactory
+			.createCache("PEPServiceManager");
+
+	private PubSubEngine pubSubEngine = null;
+
+	/**
+	 * Retrieves a PEP service -- attempting first from memory, then from the
+	 * database.
+	 * 
+	 * @param jid
+	 *            the bare JID of the user that owns the PEP service.
+	 * @return the requested PEP service if found or null if not found.
+	 */
+	public PEPService getPEPService(String jid) {
+		PEPService pepService = null;
+
+		final Lock lock = CacheFactory.getLock(jid, pepServices);
+		try {
+			lock.lock();
+			if (pepServices.containsKey(jid)) {
+				// lookup in cache
+				pepService = pepServices.get(jid);
+			} else {
+				// lookup in database.
+				pepService = loadPEPServiceFromDB(jid);
+				
+				// always add to the cache, even if it doesn't exist. This will
+				// prevent future database lookups.
+				pepServices.put(jid, pepService);
+			}
+		} finally {
+			lock.unlock();
+		}
+
+		return pepService;
+	}
+
+	public PEPService create(JID owner) {
+		// Return an error if the packet is from an anonymous, unregistered user
+		// or remote user
+		if (!XMPPServer.getInstance().isLocal(owner)
+				|| !UserManager.getInstance().isRegisteredUser(owner.getNode())) {
+			throw new IllegalArgumentException(
+					"Request must be initiated by a local, registered user, but is not: "
+							+ owner);
+		}
+
+		PEPService pepService = null;
+		final String bareJID = owner.toBareJID();
+		final Lock lock = CacheFactory.getLock(owner, pepServices);
+		try {
+			lock.lock();
+
+			pepService = pepServices.get(bareJID);
+			if (pepService == null) {
+				pepService = new PEPService(XMPPServer.getInstance(), bareJID);
+				pepServices.put(bareJID, pepService);
+
+				if (Log.isDebugEnabled()) {
+					Log.debug("PEPService created for : " + bareJID);
+				}
+			}
+		} finally {
+			lock.unlock();
+		}
+
+		return pepService;
+	}
+
+	/**
+	 * Loads a PEP service from the database, if it exists.
+	 * 
+	 * @param jid
+	 *            the JID of the owner of the PEP service.
+	 * @return the loaded PEP service, or null if not found.
+	 */
+	private PEPService loadPEPServiceFromDB(String jid) {
+		PEPService pepService = null;
+
+		Connection con = null;
+		PreparedStatement pstmt = null;
+		ResultSet rs = null;
+		try {
+			con = DbConnectionManager.getConnection();
+			// Get all PEP services
+			pstmt = con.prepareStatement(GET_PEP_SERVICE);
+			pstmt.setString(1, jid);
+			rs = pstmt.executeQuery();
+			// Restore old PEPServices
+			while (rs.next()) {
+				String serviceID = rs.getString(1);
+
+				// Create a new PEPService
+				pepService = new PEPService(XMPPServer.getInstance(), serviceID);
+				pepServices.put(serviceID, pepService);
+				pubSubEngine.start(pepService);
+
+				if (Log.isDebugEnabled()) {
+					Log.debug("PEP: Restored service for " + serviceID
+							+ " from the database.");
+				}
+			}
+		} catch (SQLException sqle) {
+			Log.error(sqle.getMessage(), sqle);
+		} finally {
+			DbConnectionManager.closeConnection(rs, pstmt, con);
+		}
+
+		return pepService;
+	}
+
+	/**
+	 * Deletes the {@link PEPService} belonging to the specified owner.
+	 * 
+	 * @param owner
+	 *            The JID of the owner of the service to be deleted.
+	 */
+	public void remove(JID owner) {
+		PEPService service = null;
+
+		final Lock lock = CacheFactory.getLock(owner, pepServices);
+		try {
+			lock.lock();
+			service = pepServices.remove(owner.toBareJID());
+		} finally {
+			lock.unlock();
+		}
+
+		if (service == null) {
+			return;
+		}
+
+		// Delete the user's PEP nodes from memory and the database.
+		CollectionNode rootNode = service.getRootCollectionNode();
+		for (final Node node : service.getNodes()) {
+			if (rootNode.isChildNode(node)) {
+				node.delete();
+			}
+		}
+		rootNode.delete();
+	}
+
+	public void start(PEPService pepService) {
+		pubSubEngine.start(pepService);
+	}
+
+	public void start() {
+		pubSubEngine = new PubSubEngine(XMPPServer.getInstance()
+				.getPacketRouter());
+	}
+
+	public void stop() {
+
+		for (PEPService service : pepServices.values()) {
+			pubSubEngine.shutdown(service);
+		}
+
+		pubSubEngine = null;
+	}
+
+	public void process(PEPService service, IQ iq) {
+		pubSubEngine.process(service, iq);
+	}
+	
+	public boolean hasCachedService(JID owner) {
+		return pepServices.get(owner) != null;
+	}
+	
+	// mimics Shutdown, without killing the timer.
+	public void unload(PEPService service) {	
+		pubSubEngine.shutdown(service);
+	}
+}

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

+/**
+ * $RCSfile: $
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2005-2008 Jive Software. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jivesoftware.openfire.pep;
+
+import org.jivesoftware.openfire.pubsub.PublishedItemTask;
+
+/**
+ * TimerTask that unloads services from memory, after they have been expired
+ * from cache.
+ * 
+ * @author Guus der Kinderen, guus.der.kinderen@gmail.com
+ */
+public class PublishedPEPServiceTask extends PublishedItemTask {
+
+	private final PEPServiceManager manager;
+
+	public PublishedPEPServiceTask(PEPService service, PEPServiceManager manager) {
+		super(service);
+		this.manager = manager;
+	}
+
+	@Override
+	public void run() {
+		// Somewhat of a hack to unload the PEPService after it has been removed
+		// from the cache. New scheduled packets will re-instate the service.
+		PEPService service = (PEPService) this.getService();
+		if (manager.hasCachedService(service.getAddress())) {
+			super.run();
+		} else {
+			manager.unload(service);
+		}
+	}
+}

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

@@ -1731,7 +1731,7 @@ public class PubSubEngine {
 
     public void shutdown(PubSubService service) {
         // Stop the maintenance processes
-        service.getTimer().cancel();
+    	service.getPublishedItemTask().cancel();
         // Delete from the database items contained in the itemsToDelete queue
         PublishedItem entry;
         while (!service.getItemsToDelete().isEmpty()) {
@@ -1749,6 +1749,10 @@ public class PubSubEngine {
         }
         // Stop executing ad-hoc commands
         service.getManager().stop();
+        
+        // clear all nodes for this service, to remove circular references back to the service instance.
+		service.getNodes().clear(); // FIXME: this is an ugly hack. getNodes() is documented to return an unmodifiable collection (but does not).
+
     }
 
     /*******************************************************************************

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

@@ -23,6 +23,7 @@ package org.jivesoftware.openfire.pubsub;
 import java.util.Queue;
 import java.util.TimerTask;
 
+import org.jivesoftware.openfire.pep.PEPService;
 import org.jivesoftware.util.LocaleUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -91,4 +92,8 @@ public class PublishedItemTask extends TimerTask {
             Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
         }
     }
+    
+	protected PubSubService getService() {
+		return service;
+	}
 }

src/java/org/jivesoftware/util/cache/CacheFactory.java

@@ -128,6 +128,7 @@ public class CacheFactory {
         cacheNames.put("Entity Capabilities Users", "entityCapabilitiesUsers");
         cacheNames.put("Entity Capabilities Pending Hashes", "entityCapabilitiesPendingHashes");
         cacheNames.put("Clearspace SSO Nonce", "clearspaceSSONonce");
+        cacheNames.put("PEPServiceManager", "pepServiceManager");
 
         cacheProps.put("cache.fileTransfer.size", 128 * 1024l);
         cacheProps.put("cache.fileTransfer.maxLifetime", 1000 * 60 * 10l);
@@ -199,6 +200,8 @@ public class CacheFactory {
         cacheProps.put("cache.pluginCacheInfo.maxLifetime", -1l);
         cacheProps.put("cache.clearspaceSSONonce.size", -1l);
         cacheProps.put("cache.clearspaceSSONonce.maxLifetime", JiveConstants.MINUTE * 2);
+        cacheProps.put("cache.pepServiceManager.size", 1024l * 1024 * 10);
+        cacheProps.put("cache.pepServiceManager.maxLifetime", JiveConstants.MINUTE * 30);
     }
 
     private CacheFactory() {

src/plugins/clustering/coherence-cache-config.xml

@@ -883,6 +883,25 @@ http://www.tangosol.com/UserGuide-Reference-CacheConfig.jsp
             </init-params>
         </cache-mapping>
 
+        <cache-mapping>
+            <cache-name>PEPServiceManager</cache-name>
+            <scheme-name>near-distributed</scheme-name>
+            <init-params>
+                <init-param>
+                    <param-name>back-size-high</param-name>
+                    <param-value>4000</param-value>
+                </init-param>
+                <init-param>
+                    <param-name>back-expiry</param-name>
+                    <param-value>30m</param-value>
+                </init-param>
+                <init-param>
+                    <param-name>back-size-low</param-name>
+                    <param-value>3000</param-value>
+                </init-param>
+            </init-params>
+        </cache-mapping>
+
     </caching-scheme-mapping>
 
 </cache-config>