OF-205 Initial refactoring for clustering support.

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

src/java/org/jivesoftware/openfire/pubsub/cluster/CancelSubscriptionTask.java

+package org.jivesoftware.openfire.pubsub.cluster;
+
+import org.jivesoftware.openfire.pubsub.Node;
+import org.jivesoftware.openfire.pubsub.NodeSubscription;
+
+public class CancelSubscriptionTask extends SubscriptionTask
+{
+	public CancelSubscriptionTask()
+	{
+	}
+
+	public CancelSubscriptionTask(NodeSubscription subscription)
+	{
+		super(subscription);
+	}
+
+	@Override
+	public void run()
+	{
+		System.out.println("Running DeleteSubscriptionTask: " + toString());
+
+		Node node = getNode();
+		
+		// This will only occur if a PEP service is not loaded.  We can safely do nothing in this 
+		// case since any changes will get loaded from the db when it is loaded.
+		if (node == null)
+			return;
+		
+		// This method will make a db call, but it will simply do nothing since
+		// the record will already be deleted.
+		node.cancelSubscription(getSubscription());
+	}
+}

src/java/org/jivesoftware/openfire/pubsub/cluster/ModifySubscriptionTask.java

+package org.jivesoftware.openfire.pubsub.cluster;
+
+import org.jivesoftware.openfire.pubsub.NodeSubscription;
+import org.jivesoftware.openfire.pubsub.PubSubPersistenceManager;
+
+public class ModifySubscriptionTask extends SubscriptionTask
+{
+	public ModifySubscriptionTask()
+	{
+
+	}
+
+	public ModifySubscriptionTask(NodeSubscription subscription)
+	{
+		super(subscription);
+	}
+
+	@Override
+	public void run()
+	{
+		PubSubPersistenceManager.loadSubscription(getService(), getNode(), getSubscriptionId());
+	}
+}

src/java/org/jivesoftware/openfire/pubsub/cluster/NewSubscriptionTask.java

+package org.jivesoftware.openfire.pubsub.cluster;
+
+import org.jivesoftware.openfire.pubsub.Node;
+import org.jivesoftware.openfire.pubsub.NodeSubscription;
+import org.jivesoftware.openfire.pubsub.PubSubPersistenceManager;
+
+public class NewSubscriptionTask extends SubscriptionTask
+{
+	public NewSubscriptionTask()
+	{
+
+	}
+
+	public NewSubscriptionTask(NodeSubscription subscription)
+	{
+		super(subscription);
+	}
+
+	@Override
+	public void run()
+	{
+		System.out.println("Running NewSubscriptionTask: " + toString());
+
+		Node node = getNode();
+
+		// This will only occur if a PEP service is not loaded.  We can safely do nothing in this 
+		// case since any changes will get loaded from the db when it is loaded.
+		if (node == null)
+			return;
+
+		if (node.getAffiliate(getOwner()) == null)
+		{
+			node.addNoneAffiliation(getOwner());
+		}
+		node.addSubscription(getSubscription());
+
+		if (node.isPresenceBasedDelivery() && node.getSubscriptions(getSubscription().getOwner()).size() == 1)
+		{
+			if (getSubscription().getPresenceStates().isEmpty())
+			{
+				// Subscribe to the owner's presence since the node is only
+				// sending events to online subscribers and this is the first
+				// subscription of the user and the subscription is not
+				// filtering notifications based on presence show values.
+				getService().presenceSubscriptionRequired(getNode(), getOwner());
+			}
+		}
+		// We have to flush so the originating node can do a get last item.
+		PubSubPersistenceManager.flushItems();
+	}
+}

src/java/org/jivesoftware/openfire/pubsub/cluster/NodeChangeTask.java

+package org.jivesoftware.openfire.pubsub.cluster;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+
+import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.openfire.pubsub.Node;
+import org.jivesoftware.util.cache.ClusterTask;
+import org.jivesoftware.util.cache.ExternalizableUtil;
+
+/**
+ * Base class of clustering tasks for pubsub. It simply stores/retrieves the
+ * node.
+ * 
+ * @author Robin Collier
+ * 
+ */
+public abstract class NodeChangeTask implements ClusterTask
+{
+	private String nodeId;
+	transient private Node node;
+
+	public NodeChangeTask()
+	{
+
+	}
+
+	public NodeChangeTask(String nodeIdent)
+	{
+		nodeId = nodeIdent;
+	}
+
+	public NodeChangeTask(Node node)
+	{
+		this.node = node;
+		nodeId = node.getNodeID();
+	}
+
+	@Override
+	public void writeExternal(ObjectOutput out) throws IOException
+	{
+		ExternalizableUtil.getInstance().writeSafeUTF(out, nodeId);
+	}
+
+	@Override
+	public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
+	{
+		nodeId = ExternalizableUtil.getInstance().readSafeUTF(in);
+	}
+
+	public Node getNode()
+	{
+		if (node == null)
+			node = XMPPServer.getInstance().getPubSubModule().getNode(nodeId);
+		return node;
+	}
+
+	public String getNodeId()
+	{
+		return nodeId;
+	}
+}

src/java/org/jivesoftware/openfire/pubsub/cluster/NodeTask.java

+package org.jivesoftware.openfire.pubsub.cluster;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+
+import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.openfire.pep.PEPServiceManager;
+import org.jivesoftware.openfire.pubsub.Node;
+import org.jivesoftware.openfire.pubsub.PubSubService;
+import org.jivesoftware.util.cache.ClusterTask;
+import org.jivesoftware.util.cache.ExternalizableUtil;
+import org.xmpp.packet.JID;
+
+public abstract class NodeTask implements ClusterTask
+{
+	private static final String PUBSUB_SVC_ID = XMPPServer.getInstance().getPubSubModule().getServiceID();
+
+	protected String nodeId;
+	protected String serviceId;
+
+	protected NodeTask()
+	{
+
+	}
+
+	protected NodeTask(Node node)
+	{
+		nodeId = node.getNodeID();
+		serviceId = node.getService().getServiceID();
+	}
+
+	public String getNodeId()
+	{
+		return nodeId;
+	}
+
+	public Node getNode()
+	{
+		PubSubService svc = getService();
+
+		return svc != null ? svc.getNode(nodeId) : null;
+	}
+
+	public PubSubService getService()
+	{
+		if (PUBSUB_SVC_ID.equals(serviceId))
+			return XMPPServer.getInstance().getPubSubModule();
+		else
+		{
+			PEPServiceManager serviceMgr = XMPPServer.getInstance().getIQPEPHandler().getServiceManager();
+			return serviceMgr.hasCachedService(new JID(serviceId)) ? serviceMgr.getPEPService(serviceId) : null;
+		}
+	}
+
+	@Override
+	public void writeExternal(ObjectOutput out) throws IOException
+	{
+		ExternalizableUtil.getInstance().writeSafeUTF(out, nodeId);
+		ExternalizableUtil.getInstance().writeSafeUTF(out, serviceId);
+	}
+
+	@Override
+	public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
+	{
+		nodeId = ExternalizableUtil.getInstance().readSafeUTF(in);
+		serviceId = ExternalizableUtil.getInstance().readSafeUTF(in);
+	}
+}

src/java/org/jivesoftware/openfire/pubsub/cluster/RefreshNodeTask.java

+package org.jivesoftware.openfire.pubsub.cluster;
+
+import org.jivesoftware.openfire.pubsub.Node;
+import org.jivesoftware.openfire.pubsub.PubSubPersistenceManager;
+
+/**
+ * Forces the node to be refreshed from the database. This will load a node from
+ * the database and then add it to the service. If the node already existed it
+ * will be replaced, thereby refreshing it from persistence.
+ * 
+ * @author Robin Collier
+ * 
+ */
+public class RefreshNodeTask extends NodeTask
+{
+	public RefreshNodeTask()
+	{
+	}
+
+	public RefreshNodeTask(Node node)
+	{
+		super(node);
+	}
+
+	@Override
+	public Object getResult()
+	{
+		return null;
+	}
+
+	@Override
+	public void run()
+	{
+		System.out.println("Refreshing node task");
+		PubSubPersistenceManager.loadNode(getService(), getNodeId());
+	}
+
+}

src/java/org/jivesoftware/openfire/pubsub/cluster/SubscriptionTask.java

+package org.jivesoftware.openfire.pubsub.cluster;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+
+import org.jivesoftware.openfire.pubsub.NodeSubscription;
+import org.jivesoftware.openfire.pubsub.NodeSubscription.State;
+import org.jivesoftware.util.cache.ExternalizableUtil;
+import org.xmpp.packet.JID;
+
+public abstract class SubscriptionTask extends NodeTask
+{
+	private String subId;
+	private JID owner;
+	private JID subJid;
+	private NodeSubscription.State state;
+	transient private NodeSubscription subscription;
+
+	public SubscriptionTask()
+	{
+	}
+
+	public SubscriptionTask(NodeSubscription subscription)
+	{
+		super(subscription.getNode());
+		subId = subscription.getID();
+		state = subscription.getState();
+		owner = subscription.getOwner();
+		subJid = subscription.getJID();
+	}
+
+	public String getSubscriptionId()
+	{
+		return subId;
+	}
+
+	public JID getOwner()
+	{
+		return owner;
+	}
+
+	public JID getSubscriberJid()
+	{
+		return subJid;
+	}
+
+	public NodeSubscription.State getState()
+	{
+		return state;
+	}
+
+	public NodeSubscription getSubscription()
+	{
+		if (subscription == null)
+		{
+			subscription = new NodeSubscription(getNode(), owner, subJid, state, subId);
+		}
+		return subscription;
+	}
+
+	@Override
+	public void writeExternal(ObjectOutput out) throws IOException
+	{
+		super.writeExternal(out);
+		ExternalizableUtil.getInstance().writeSafeUTF(out, subId);
+		ExternalizableUtil.getInstance().writeSafeUTF(out, owner.toString());
+		ExternalizableUtil.getInstance().writeSafeUTF(out, subJid.toString());
+		ExternalizableUtil.getInstance().writeSerializable(out, state);
+	}
+
+	@Override
+	public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
+	{
+		super.readExternal(in);
+		subId = ExternalizableUtil.getInstance().readSafeUTF(in);
+		owner = new JID(ExternalizableUtil.getInstance().readSafeUTF(in));
+		subJid = new JID(ExternalizableUtil.getInstance().readSafeUTF(in));
+		state = (State) ExternalizableUtil.getInstance().readSerializable(in);
+	}
+
+	@Override
+	public Object getResult()
+	{
+		return null;
+	}
+
+	@Override
+	public String toString()
+	{
+		return getClass().getSimpleName() + " [(service=" + serviceId + "), (nodeId=" + nodeId + "), (owner=" + owner
+				+ "),(subscriber=" + subJid + "),(state=" + state + "),(id=" + subId + ")]";
+	}
+}