More work on clustering events.

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

src/java/org/jivesoftware/openfire/SessionManager.java

@@ -46,6 +46,10 @@ import java.util.concurrent.locks.Lock;
  */
 public class SessionManager extends BasicModule {
 
+    public static final String COMPONENT_SESSION_CACHE_NAME = "Components Sessions";
+    public static final String CM_CACHE_NAME = "Connection Managers Sessions";
+    public static final String ISS_CACHE_NAME = "Incoming Server Sessions";
+
     public static final int NEVER_KICK = -1;
 
     private XMPPServer server;
@@ -1149,9 +1153,9 @@ public class SessionManager extends BasicModule {
 
         // Initialize caches.
         countersCache = CacheFactory.createCache("Session Manager Counters");
-        componentSessionsCache = CacheFactory.createCache("Components Sessions");
-        multiplexerSessionsCache = CacheFactory.createCache("Connection Managers Sessions");
-        incomingServerSessionsCache = CacheFactory.createCache("Incoming Server Sessions");
+        componentSessionsCache = CacheFactory.createCache(COMPONENT_SESSION_CACHE_NAME);
+        multiplexerSessionsCache = CacheFactory.createCache(CM_CACHE_NAME);
+        incomingServerSessionsCache = CacheFactory.createCache(ISS_CACHE_NAME);
         hostnameSessionsCache = CacheFactory.createCache("Sessions by Hostname");
     }
 

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

@@ -131,6 +131,7 @@ public class XMPPServer {
             "org.jivesoftware.openfire.starter.ServerStarter";
     private static final String WRAPPER_CLASSNAME =
             "org.tanukisoftware.wrapper.WrapperManager";
+    private boolean shuttingDown;
 
     /**
      * Returns a singleton instance of XMPPServer.
@@ -877,6 +878,7 @@ public class XMPPServer {
      * Makes a best effort attempt to shutdown the server
      */
     private void shutdownServer() {
+        shuttingDown = true;
         // Notify server listeners that the server is about to be stopped
         for (XMPPServerListener listener : listeners) {
             listener.serverStopping();
@@ -904,6 +906,15 @@ public class XMPPServer {
         Log.info("Openfire stopped");
     }
 
+    /**
+     * Returns true if the server is being shutdown.
+     *
+     * @return true if the server is being shutdown.
+     */
+    public boolean isShuttingDown() {
+        return shuttingDown;
+    }
+
     /**
      * Returns the <code>ConnectionManager</code> registered with this server. The
      * <code>ConnectionManager</code> was registered with the server as a module while starting up

src/java/org/jivesoftware/openfire/cluster/ClusterEventListener.java

@@ -20,14 +20,32 @@ package org.jivesoftware.openfire.cluster;
 public interface ClusterEventListener {
 
     /**
-     * Notification event indication that this JVM is now part of a cluster. The
-     * {@link org.jivesoftware.openfire.XMPPServer#getNodeID()} will now return
-     * a new value.<p>
+     * Notification event indicating that this JVM is now part of a cluster. At this point the
+     * {@link org.jivesoftware.openfire.XMPPServer#getNodeID()} holds the new nodeID value.<p>
      *
      * When joining the cluster as the senior cluster member the {@link #markedAsSeniorClusterMember()}
-     * event will be sent right after this event.
+     * event will be sent right after this event.<p>
+     *
+     * At this point the CacheFactory holds clustered caches. That means that modifications
+     * to the caches will be reflected in the cluster. The clustered caches were just
+     * obtained from the cluster and no local cached data was automatically moved.<p>
+     *
+     * @param oldNodeID nodeID used by this JVM before joining the cluster.
+     */
+    void joinedCluster(byte[] oldNodeID);
+
+    /**
+     * Notification event indicating that this JVM is about to leave the cluster. This could
+     * happen when disabling clustering support, removing the enterprise plugin that provides
+     * clustering support or even shutdown the server.<p>
+     *
+     * At this point the CacheFactory is still holding clustered caches. That means that
+     * modifications to the caches will be reflected in the cluster.<p>
+     *
+     * Use {@link org.jivesoftware.openfire.XMPPServer#isShuttingDown()} to figure out if the
+     * server is being shutdown.
      */
-    void joinedCluster();
+    void leavingCluster();
 
     /**
      * Notification event indicating that this JVM is no longer part of the cluster. This could
@@ -39,7 +57,10 @@ public interface ClusterEventListener {
      * get the <tt>left cluster event</tt> and <tt>joined cluster events</tt>. That means that
      * caches will be reset and thus will need to be repopulated again with fresh data from this JVM.
      * This also includes the case where this JVM was the senior cluster member and when the islands
-     * met again then this JVM stopped being the senior member.
+     * met again then this JVM stopped being the senior member.<p>
+     *
+     * At this point the CacheFactory holds local caches. That means that modifications to
+     * the caches will only affect this JVM.
      */
     void leftCluster();
 
@@ -52,7 +73,7 @@ public interface ClusterEventListener {
      * island will have its own senior cluster member. However, when the islands meet again there
      * could only be one senior cluster member so one of the senior cluster members will stop playing
      * that role. When that happens the JVM no longer playing that role will receive the
-     * {@link #leftCluster()} and {@link #joinedCluster()} events.
+     * {@link #leftCluster()} and {@link #joinedCluster(byte[])} events.
      */
     void markedAsSeniorClusterMember();
 }

src/java/org/jivesoftware/openfire/muc/spi/MultiUserChatServerImpl.java

@@ -923,11 +923,15 @@ public class MultiUserChatServerImpl extends BasicModule implements MultiUserCha
         outMessages.addAndGet(numOccupants);
     }
 
-    public void joinedCluster() {
+    public void joinedCluster(byte[] oldNodeID) {
         // Disable the service until we know that we are the senior cluster member
         enableService(false, false);
     }
 
+    public void leavingCluster() {
+        // Do nothing
+    }
+
     public void leftCluster() {
         // Offer the service when not running in a cluster
         enableService(true, false);

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

@@ -439,11 +439,15 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
         }
     }
 
-    public void joinedCluster() {
+    public void joinedCluster(byte[] oldNodeID) {
         // Disable the service until we know that we are the senior cluster member
         enableService(false);
     }
 
+    public void leavingCluster() {
+        // Do nothing
+    }
+
     public void leftCluster() {
         // Offer the service when not running in a cluster
         enableService(true);

src/java/org/jivesoftware/openfire/spi/LocalRoutingTable.java

@@ -13,10 +13,7 @@ package org.jivesoftware.openfire.spi;
 
 import org.jivesoftware.openfire.RoutableChannelHandler;
 import org.jivesoftware.openfire.SessionManager;
-import org.jivesoftware.openfire.session.ClientSession;
-import org.jivesoftware.openfire.session.LocalSession;
-import org.jivesoftware.openfire.session.OutgoingServerSession;
-import org.jivesoftware.openfire.session.Session;
+import org.jivesoftware.openfire.session.*;
 import org.jivesoftware.util.LocaleUtils;
 import org.jivesoftware.util.Log;
 import org.jivesoftware.util.TaskEngine;
@@ -60,11 +57,41 @@ class LocalRoutingTable {
      *
      * @return the client sessions that are connected to this JVM.
      */
-    Collection<ClientSession> getClientRoutes() {
-        List<ClientSession> sessions = new ArrayList<ClientSession>();
+    Collection<LocalClientSession> getClientRoutes() {
+        List<LocalClientSession> sessions = new ArrayList<LocalClientSession>();
         for (RoutableChannelHandler route : routes.values()) {
-            if (route instanceof ClientSession) {
-                sessions.add((ClientSession) route);
+            if (route instanceof LocalClientSession) {
+                sessions.add((LocalClientSession) route);
+            }
+        }
+        return sessions;
+    }
+
+    /**
+     * Returns the outgoing server sessions that are connected to this JVM.
+     *
+     * @return the outgoing server sessions that are connected to this JVM.
+     */
+    Collection<LocalOutgoingServerSession> getServerRoutes() {
+        List<LocalOutgoingServerSession> sessions = new ArrayList<LocalOutgoingServerSession>();
+        for (RoutableChannelHandler route : routes.values()) {
+            if (route instanceof LocalOutgoingServerSession) {
+                sessions.add((LocalOutgoingServerSession) route);
+            }
+        }
+        return sessions;
+    }
+
+    /**
+     * Returns the external component sessions that are connected to this JVM.
+     *
+     * @return the external component sessions that are connected to this JVM.
+     */
+    Collection<RoutableChannelHandler> getComponentRoute() {
+        List<RoutableChannelHandler> sessions = new ArrayList<RoutableChannelHandler>();
+        for (RoutableChannelHandler route : routes.values()) {
+            if (!(route instanceof LocalOutgoingServerSession || route instanceof LocalClientSession)) {
+                sessions.add(route);
             }
         }
         return sessions;

src/java/org/jivesoftware/openfire/spi/RoutingTableImpl.java

@@ -13,7 +13,10 @@ package org.jivesoftware.openfire.spi;
 
 import org.jivesoftware.openfire.*;
 import org.jivesoftware.openfire.auth.UnauthorizedException;
+import org.jivesoftware.openfire.cluster.ClusterEventListener;
+import org.jivesoftware.openfire.cluster.ClusterManager;
 import org.jivesoftware.openfire.container.BasicModule;
+import org.jivesoftware.openfire.handler.PresenceUpdateHandler;
 import org.jivesoftware.openfire.server.OutgoingSessionPromise;
 import org.jivesoftware.openfire.session.*;
 import org.jivesoftware.util.JiveGlobals;
@@ -42,7 +45,12 @@ import java.util.concurrent.locks.Lock;
  *
  * @author Gaston Dombiak
  */
-public class RoutingTableImpl extends BasicModule implements RoutingTable {
+public class RoutingTableImpl extends BasicModule implements RoutingTable, ClusterEventListener {
+
+    public static final String C2S_CACHE_NAME = "Routing Users Cache";
+    public static final String ANONYMOUS_C2S_CACHE_NAME = "Routing AnonymousUsers Cache";
+    public static final String S2S_CACHE_NAME = "Routing Servers Cache";
+    public static final String COMPONENT_CACHE_NAME = "Routing Components Cache";
 
     /**
      * Cache (unlimited, never expire) that holds outgoing sessions to remote servers from this server.
@@ -69,7 +77,7 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable {
      * (includes anonymous).
      * Key: bare JID, Value: list of full JIDs of the user
      */
-    private Cache<String, List<String>> usersSessions;
+    private Cache<String, Collection<String>> usersSessions;
 
     private String serverName;
     private XMPPServer server;
@@ -81,10 +89,10 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable {
 
     public RoutingTableImpl() {
         super("Routing table");
-        serversCache = CacheFactory.createCache("Routing Servers Cache");
-        componentsCache = CacheFactory.createCache("Routing Components Cache");
-        usersCache = CacheFactory.createCache("Routing Users Cache");
-        anonymousUsersCache = CacheFactory.createCache("Routing AnonymousUsers Cache");
+        serversCache = CacheFactory.createCache(S2S_CACHE_NAME);
+        componentsCache = CacheFactory.createCache(COMPONENT_CACHE_NAME);
+        usersCache = CacheFactory.createCache(C2S_CACHE_NAME);
+        anonymousUsersCache = CacheFactory.createCache(ANONYMOUS_C2S_CACHE_NAME);
         usersSessions = CacheFactory.createCache("Routing User Sessions");
         localRoutingTable = new LocalRoutingTable();
     }
@@ -137,9 +145,9 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable {
                 Lock lock = LockManager.getLock(route.toBareJID());
                 try {
                     lock.lock();
-                    List<String> jids = usersSessions.get(route.toBareJID());
+                    Collection<String> jids = usersSessions.get(route.toBareJID());
                     if (jids == null) {
-                        jids = new ArrayList<String>();
+                        jids = new HashSet<String>();
                     }
                     jids.add(route.toString());
                     usersSessions.put(route.toBareJID(), jids);
@@ -540,7 +548,7 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable {
             }
             else {
                 // Address is a bare JID so return all AVAILABLE resources of user
-                List<String> sessions = usersSessions.get(route.toBareJID());
+                Collection<String> sessions = usersSessions.get(route.toBareJID());
                 if (sessions != null) {
                     // Select only available sessions
                     for (String jid : sessions) {
@@ -584,7 +592,7 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable {
                     usersSessions.remove(route.toBareJID());
                 }
                 else {
-                    List<String> jids = usersSessions.get(route.toBareJID());
+                    Collection<String> jids = usersSessions.get(route.toBareJID());
                     if (jids != null) {
                         jids.remove(route.toString());
                         if (!jids.isEmpty()) {
@@ -649,6 +657,8 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable {
         iqRouter = server.getIQRouter();
         messageRouter = server.getMessageRouter();
         presenceRouter = server.getPresenceRouter();
+        // Listen to cluster events
+        ClusterManager.addListener(this);
     }
 
     public void start() throws IllegalStateException {
@@ -660,4 +670,60 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable {
         super.stop();
         localRoutingTable.stop();
     }
+
+    public void joinedCluster(byte[] oldNodeID) {
+        // Add outgoing server sessions hosted locally to the cache (using new nodeID)
+        for (LocalOutgoingServerSession session : localRoutingTable.getServerRoutes()) {
+            addServerRoute(session.getAddress(), session);
+        }
+
+        // Add component sessions hosted locally to the cache (using new nodeID) and remove traces to old nodeID
+        for (RoutableChannelHandler route : localRoutingTable.getComponentRoute()) {
+            addComponentRoute(route.getAddress(), route);
+        }
+
+        // Add client sessions hosted locally to the cache (using new nodeID)
+        for (LocalClientSession session : localRoutingTable.getClientRoutes()) {
+            addClientRoute(session.getAddress(), session);
+        }
+
+        // Broadcast presence of local sessions to remote sessions when subscribed to presence
+        // Probe presences of remote sessions when subscribed to presence of local session
+        // Send pending subscription requests to local sessions from remote sessions
+        // Deliver offline messages sent to local sessions that were unavailable in other nodes
+        // Send available presences of local sessions to other resources of the same user
+        PresenceUpdateHandler presenceUpdateHandler = XMPPServer.getInstance().getPresenceUpdateHandler();
+        for (LocalClientSession session : localRoutingTable.getClientRoutes()) {
+            // Simulate that the local session has just became available
+            session.setInitialized(false);
+            // Simulate that current session presence has just been received
+            presenceUpdateHandler.process(session.getPresence());
+        }
+    }
+
+    public void leavingCluster() {
+        if (XMPPServer.getInstance().isShuttingDown()) {
+            // Do nothing since local sessions will be closed. Local session manager
+            // and local routing table will be correctly updated thus updating the
+            // other cluster nodes correctly
+        }
+        else {
+            // This JVM is leaving the cluster but will continue to work. That means
+            // that clients connected to this JVM will be able to keep talking.
+            // In other words, their sessions will not be closed (and not removed from
+            // the routing table or the session manager). However, other nodes should
+            // get their routing tables correctly updated.
+            // TODO Implement this. Remove local sessions from caches
+        }
+    }
+
+    public void leftCluster() {
+        if (!XMPPServer.getInstance().isShuttingDown()) {
+            // TODO Implement this. Add local sessions to caches
+        }
+    }
+
+    public void markedAsSeniorClusterMember() {
+        // Do nothing
+    }
 }

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

@@ -20,7 +20,6 @@ import org.jivesoftware.util.Log;
 
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
  * Creates Cache objects. The returned caches will either be local or clustered
@@ -44,11 +43,6 @@ public class CacheFactory {
      */
     private static Map<String, Cache> caches = new ConcurrentHashMap<String, Cache>();
 
-    /**
-     * List of registered listeners to be notified when clustering is enabled or disabled.
-     */
-    private static List<ClusteringListener> listeners = new CopyOnWriteArrayList<ClusteringListener>();
-
     private static String localCacheFactoryClass;
     private static String clusteredCacheFactoryClass;
     private static CacheFactoryStrategy cacheFactoryStrategy;
@@ -67,7 +61,7 @@ public class CacheFactory {
      * Returns an array of all caches in the system.
      * @return an array of all caches in the system.
      */
-    public static synchronized Cache[] getAllCaches() {
+    public static Cache[] getAllCaches() {
         List<Cache> values = new ArrayList<Cache>();
         for (Cache cache : caches.values()) {
             values.add(cache);
@@ -280,26 +274,6 @@ public class CacheFactory {
         }
     }
 
-    public static void addClusteringListener(ClusteringListener listener) {
-        listeners.add(listener);
-    }
-
-    public static void removeClusteringListener(ClusteringListener listener) {
-        listeners.remove(listener);
-    }
-
-    private static void fireClusteringStarted() {
-        for (ClusteringListener listener : listeners) {
-            (listener).clusteringStarted();
-        }
-    }
-
-    private static void fireClusteringStopped() {
-        for (ClusteringListener listener : listeners) {
-            (listener).clusteringStopped();
-        }
-    }
-
     public static synchronized void initialize() throws InitializationException {
         try {
             cacheFactoryStrategy = (CacheFactoryStrategy) Class
@@ -389,30 +363,16 @@ public class CacheFactory {
 
     private static void startClustering() {
         clusteringStarted = false;
-        boolean clusterStarted = false;
         try {
             cacheFactoryStrategy = (CacheFactoryStrategy) Class.forName(clusteredCacheFactoryClass, true,
                     getClusteredCacheStrategyClassLoader("enterprise"))
                     .newInstance();
-            clusterStarted = cacheFactoryStrategy.startCluster();
-            if (clusterStarted) {
-                // Loop through local caches and switch them to clustered cache.
-                for (String cacheName : caches.keySet()) {
-                    CacheWrapper wrapper = (CacheWrapper) caches.get(cacheName);
-                    wrapper.setWrappedCache(cacheFactoryStrategy.createCache(cacheName));
-                }
-
-                clusteringStarted = true;
-                // Set the ID of this cluster node
-                XMPPServer.getInstance().setNodeID(CacheFactory.getClusterMemberID());
-                // Fire event that cluster has been started
-                fireClusteringStarted();
-            }
+            clusteringStarted = cacheFactoryStrategy.startCluster();
         }
         catch (Exception e) {
             Log.error("Unable to start clustering - continuing in local mode", e);
         }
-        if (!clusterStarted) {
+        if (!clusteringStarted) {
             // Revert to local cache factory if cluster fails to start
             try {
                 cacheFactoryStrategy = (CacheFactoryStrategy) Class.forName(localCacheFactoryClass).newInstance();
@@ -428,35 +388,39 @@ public class CacheFactory {
             cacheFactoryStrategy = (CacheFactoryStrategy) Class.forName(localCacheFactoryClass)
                     .newInstance();
 
-            // Loop through clustered caches and change them to local caches.
-            for (String cacheName : caches.keySet()) {
-                CacheWrapper wrapper = (CacheWrapper) caches.get(cacheName);
-                wrapper.setWrappedCache(cacheFactoryStrategy.createCache(cacheName));
-            }
-
             clusteringStarted = false;
             // Reset the node ID
             XMPPServer.getInstance().setNodeID(null);
 
             // Stop the cluster
             clusteredFactory.stopCluster();
-            // Fire event that cluster has been stopped
-            fireClusteringStopped();
         }
         catch (Exception e) {
             Log.error("Unable to stop clustering - continuing in clustered mode", e);
         }
     }
 
-
     /**
-     * Listener interface for any object which needs to be notified when clustering starts or stops
+     * Notification message indicating that this JVM has joined a cluster.
      */
-    public static interface ClusteringListener {
-
-        public void clusteringStarted();
-
-        public void clusteringStopped();
+    public static void joinedCluster() {
+        // Loop through local caches and switch them to clustered cache (migrate content)
+        for (Cache cache : getAllCaches()) {
+            CacheWrapper cacheWrapper = ((CacheWrapper) cache);
+            Cache clusteredCache = cacheFactoryStrategy.createCache(cacheWrapper.getName());
+            cacheWrapper.setWrappedCache(clusteredCache);
+        }
     }
 
+    /**
+     * Notification message indicating that this JVM has left the cluster.
+     */
+    public static void leftCluster() {
+        // Loop through clustered caches and change them to local caches (migrate content)
+        for (Cache cache : getAllCaches()) {
+            CacheWrapper cacheWrapper = ((CacheWrapper) cache);
+            Cache standaloneCache = cacheFactoryStrategy.createCache(cacheWrapper.getName());
+            cacheWrapper.setWrappedCache(standaloneCache);
+        }
+    }
 }
\ No newline at end of file