import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
protected boolean canExecute() {
  if(ClusterManager.isClusteringEnabled()) {
    LOGGER.debug("canExecute : clustering is enabled");
    try {
      if(lock != null) {
        boolean result = lock.tryLock();
        LOGGER.trace("canExecute : tryLockResult={}", result);
        return result;
      } else {
        LOGGER.debug("canExecute : lock has not been set, cannot execute task");
        return false;
      }
    } catch (Exception e) {
      LOGGER.error("canExecute : error trying to acquire distributed lock", e);
      return false;
    }
  } else {
    LOGGER.debug("canExecute : clustering is disabled return true");
    return true;
  }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
/**
 * Executes the requested task considering that this JVM may still be joining the cluster.
 * This means that events regarding rooms that were not loaded yet will be stored for later
 * processing. Once the JVM is done joining the cluster queued tasks will be processed.
 *
 * @param runnable the task to execute.
 */
protected void execute(Runnable runnable) {
    // Check if we are joining a cluster
    boolean clusterStarting = ClusterManager.isClusteringStarting();
    try {
        // Check that the room exists
        getRoom();
        // Room was found so now execute the task
        runnable.run();
    }
    catch (IllegalArgumentException e) {
        // Room not found so check if we are still joining the cluster
        if (clusterStarting) {
            // Queue task in case the cluster
            QueuedTasksManager.getInstance().addTask(this);
        }
        else {
            // Task failed since room was not found
            Log.error(e.getMessage(), e);
        }
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
/**
 * Called when manager starts up, to initialize things.
 */
@Override
public void start() {
    super.start();

    loadServices();

    for (MultiUserChatService service : mucServices.values()) {
        registerMultiUserChatService(service);
    }

    // Add statistics
    addTotalRoomStats();
    addTotalOccupantsStats();
    addTotalConnectedUsers();
    addNumberIncomingMessages();
    addNumberOutgoingMessages();

    ClusterManager.addListener(this);
    UserEventDispatcher.addListener(this);
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
/**
 * Called when manager is stopped, to clean things up.
 */
@Override
public void stop() {
    super.stop();

    ClusterManager.removeListener(this);
    UserEventDispatcher.removeListener(this);

    // Remove the statistics.
    StatisticsManager.getInstance().removeStatistic(roomsStatKey);
    StatisticsManager.getInstance().removeStatistic(occupantsStatKey);
    StatisticsManager.getInstance().removeStatistic(usersStatKey);
    StatisticsManager.getInstance().removeStatistic(incomingStatKey);
    StatisticsManager.getInstance().removeStatistic(outgoingStatKey);

    for (MultiUserChatService service : mucServices.values()) {
        unregisterMultiUserChatService(service.getServiceName());
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
/**
 * Add a new session to be managed. The session has been authenticated and resource
 * binding has been done.
 *
 * @param session the session that was authenticated.
 */
public void addSession(LocalClientSession session) {
    // Add session to the routing table (routing table will know session is not available yet)
    routingTable.addClientRoute(session.getAddress(), session);
    // Remove the pre-Authenticated session but remember to use the temporary ID as the key
    localSessionManager.getPreAuthenticatedSessions().remove(session.getStreamID().toString());
    SessionEventDispatcher.EventType event = session.getAuthToken().isAnonymous() ?
            SessionEventDispatcher.EventType.anonymous_session_created :
            SessionEventDispatcher.EventType.session_created;
    // Fire session created event.
    SessionEventDispatcher.dispatchEvent(session, event);
    if (ClusterManager.isClusteringStarted()) {
        // Track information about the session and share it with other cluster nodes
        sessionInfoCache.put(session.getAddress().toString(), new ClientSessionInfo(session));
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
@Override
public void initialize(XMPPServer server) {
    super.initialize(server);
    localServer = server;
    rosterManager = server.getRosterManager();
    presenceManager = server.getPresenceManager();
    deliverer = server.getPacketDeliverer();
    messageStore = server.getOfflineMessageStore();
    sessionManager = server.getSessionManager();
    userManager = server.getUserManager();
    routingTable = server.getRoutingTable();
    directedPresencesCache = CacheFactory.createCache(PRESENCE_CACHE_NAME);
    // TODO Add as route listener (to remove direct presences info for removed routes). Mainly for c2s sessions which is uncommon.
    // Listen to cluster events
    ClusterManager.addListener(this);
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
@Override
public void leftCluster(byte[] nodeID) {
    if (ClusterManager.isSeniorClusterMember()) {
        NodeID leftNode = NodeID.getInstance(nodeID);
        for (Map.Entry<String, ClusteredServerItem> entry : serverItems.entrySet()) {
            String jid = entry.getKey();
            Lock lock = CacheFactory.getLock(jid, serverItems);
            try {
                lock.lock();
                ClusteredServerItem item = entry.getValue();
                if (item.nodes.remove(leftNode)) {
                    // Update the cache with latest info
                    if (item.nodes.isEmpty()) {
                        serverItems.remove(jid);
                    }
                    else {
                        serverItems.put(jid, item);
                    }
                }
            }
            finally {
                lock.unlock();
            }
        }
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
@Override
public void leftCluster(byte[] nodeID) {
    if (ClusterManager.isSeniorClusterMember()) {
        NodeID leftNode = NodeID.getInstance(nodeID);
        // Remove server features added by node that is gone
        for (Map.Entry<String, Set<NodeID>> entry : serverFeatures.entrySet()) {
            String namespace = entry.getKey();
            Lock lock = CacheFactory.getLock(namespace, serverFeatures);
            try {
                lock.lock();
                Set<NodeID> nodeIDs = entry.getValue();
                if (nodeIDs.remove(leftNode)) {
                    if (nodeIDs.isEmpty()) {
                        serverFeatures.remove(namespace);
                    }
                    else {
                        serverFeatures.put(namespace, nodeIDs);
                    }
                }
            }
            finally {
                lock.unlock();
            }
        }
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
public double[] getCurrentValue(String key) {
    if (ClusterManager.isSeniorClusterMember()) {
        return new double[] { engine.getDefinition(key)[0].getLastSample() };
    }
    else {
        try {
            if (RrdSqlBackend.exists(key)) {
                RrdDb db = new RrdDb(key, true);
                return new double[] { db.getLastDatasourceValues()[0] };
            }
        } catch (Exception e) {
            Log.error("Error retrieving last sample value for: " + key, e);
        }
        return new double[] { 0 };
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
public void occupantLeft(JID roomJID, JID user) {
    // Process this event in the senior cluster member or local JVM when not in a cluster
    if (ClusterManager.isSeniorClusterMember()) {
        conversationManager.leftGroupConversation(roomJID, user, new Date());
        // If there are no more occupants then consider the group conversarion over
        MUCRoom mucRoom = XMPPServer.getInstance().getMultiUserChatManager().getMultiUserChatService(roomJID).getChatRoom(roomJID.getNode());
        if (mucRoom != null &&  mucRoom.getOccupantsCount() == 0) {
            conversationManager.roomConversationEnded(roomJID, new Date());
        }
    }
    else {
        ConversationEventsQueue eventsQueue = conversationManager.getConversationEventsQueue();
        eventsQueue.addGroupChatEvent(conversationManager.getRoomConversationKey(roomJID),
                ConversationEvent.occupantLeft(roomJID, user, new Date()));
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
/**
 * Returns the set of active conversations.
 *
 * @return the active conversations.
 */
public Collection<Conversation> getConversations() {
    if (ClusterManager.isSeniorClusterMember()) {
        List<Conversation> conversationList = new ArrayList<Conversation>(conversations.values());
        // Sort the conversations by creation date.
        Collections.sort(conversationList, new Comparator<Conversation>() {
            public int compare(Conversation c1, Conversation c2) {
                return c1.getStartDate().compareTo(c2.getStartDate());
            }
        });
        return conversationList;
    } else {
        // Get this info from the senior cluster member when running in a cluster
        return (Collection<Conversation>) CacheFactory.doSynchronousClusterTask(new GetConversationsTask(), ClusterManager
                .getSeniorClusterMember().toByteArray());
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
public void nicknameChanged(JID roomJID, JID user, String oldNickname, String newNickname) {
    // Process this event in the senior cluster member or local JVM when not in a cluster
    if (ClusterManager.isSeniorClusterMember()) {
        occupantLeft(roomJID, user);
        // Sleep 1 millisecond so that there is a delay between logging out and logging in
        try {
            Thread.sleep(1);
        } catch (InterruptedException e) {
            // Ignore
        }
        occupantJoined(roomJID, user, newNickname);
    }
    else {
        ConversationEventsQueue eventsQueue = conversationManager.getConversationEventsQueue();
        eventsQueue.addGroupChatEvent(conversationManager.getRoomConversationKey(roomJID),
                ConversationEvent.nicknameChanged(roomJID, user, newNickname, new Date()));
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
private void initializeClustering() {
    logger.info("All plugins have initialized; initializing clustering");
    // Check if another cluster is installed and stop loading this plugin if found
    File pluginDir = new File(JiveGlobals.getHomeDirectory(), "plugins");
    File[] jars = pluginDir.listFiles(new FileFilter() {
        public boolean accept(File pathname) {
            String fileName = pathname.getName().toLowerCase();
            return (fileName.equalsIgnoreCase("enterprise.jar") || 
                    fileName.equalsIgnoreCase("coherence.jar"));
        }
    });
    if (jars.length > 0) {
        // Do not load this plugin if a conflicting implementation exists
        logger.warn("Conflicting clustering plugins found; remove Coherence and/or Enterprise jar files");
        throw new IllegalStateException("Clustering plugin configuration conflict (Coherence)");
    }
    ClusterManager.startup();
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
/**
    * Called when manager starts up, to initialize things.
    */
   @Override
public void start() {
       super.start();

       loadServices();

       for (MultiUserChatService service : mucServices.values()) {
           registerMultiUserChatService(service);
       }

       // Add statistics
       addTotalRoomStats();
       addTotalOccupantsStats();
       addTotalConnectedUsers();
       addNumberIncomingMessages();
       addNumberOutgoingMessages();

       ClusterManager.addListener(this);
       UserEventDispatcher.addListener(this);
   }
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
/**
    * Called when manager is stopped, to clean things up.
    */
   @Override
public void stop() {
       super.stop();

       ClusterManager.removeListener(this);
       UserEventDispatcher.removeListener(this);

       // Remove the statistics.
       StatisticsManager.getInstance().removeStatistic(roomsStatKey);
       StatisticsManager.getInstance().removeStatistic(occupantsStatKey);
       StatisticsManager.getInstance().removeStatistic(usersStatKey);
       StatisticsManager.getInstance().removeStatistic(incomingStatKey);
       StatisticsManager.getInstance().removeStatistic(outgoingStatKey);

       for (MultiUserChatService service : mucServices.values()) {
           unregisterMultiUserChatService(service.getServiceName());
       }
   }
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
/**
 * Add a new session to be managed. The session has been authenticated and resource
 * binding has been done.
 *
 * @param session the session that was authenticated.
 */
public void addSession(LocalClientSession session) {
    // Remove the pre-Authenticated session but remember to use the temporary ID as the key
    localSessionManager.getPreAuthenticatedSessions().remove(session.getStreamID().toString());
    // Add session to the routing table (routing table will know session is not available yet)
    routingTable.addClientRoute(session.getAddress(), session);
    SessionEventDispatcher.EventType event = session.getAuthToken().isAnonymous() ?
            SessionEventDispatcher.EventType.anonymous_session_created :
            SessionEventDispatcher.EventType.session_created;
    // Fire session created event.
    SessionEventDispatcher.dispatchEvent(session, event);
    if (ClusterManager.isClusteringStarted()) {
        // Track information about the session and share it with other cluster nodes
        sessionInfoCache.put(session.getAddress().toString(), new ClientSessionInfo(session));
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
@Override
public void initialize(XMPPServer server) {
       super.initialize(server);
       localServer = server;
       rosterManager = server.getRosterManager();
       presenceManager = server.getPresenceManager();
       deliverer = server.getPacketDeliverer();
       messageStore = server.getOfflineMessageStore();
       sessionManager = server.getSessionManager();
       userManager = server.getUserManager();
       routingTable = server.getRoutingTable();
       directedPresencesCache = CacheFactory.createCache(PRESENCE_CACHE_NAME);
       // TODO Add as route listener (to remove direct presences info for removed routes). Mainly for c2s sessions which is uncommon.
       // Listen to cluster events
       ClusterManager.addListener(this);
   }
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
public void leftCluster(byte[] nodeID) {
    if (ClusterManager.isSeniorClusterMember()) {
        NodeID leftNode = NodeID.getInstance(nodeID);
        for (Map.Entry<String, ClusteredServerItem> entry : serverItems.entrySet()) {
            String jid = entry.getKey();
            Lock lock = CacheFactory.getLock(jid, serverItems);
            try {
                lock.lock();
                ClusteredServerItem item = entry.getValue();
                if (item.nodes.remove(leftNode)) {
                    // Update the cache with latest info
                    if (item.nodes.isEmpty()) {
                        serverItems.remove(jid);
                    }
                    else {
                        serverItems.put(jid, item);
                    }
                }
            }
            finally {
                lock.unlock();
            }
        }
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
public void leftCluster(byte[] nodeID) {
    if (ClusterManager.isSeniorClusterMember()) {
        NodeID leftNode = NodeID.getInstance(nodeID);
        // Remove server features added by node that is gone
        for (Map.Entry<String, Set<NodeID>> entry : serverFeatures.entrySet()) {
            String namespace = entry.getKey();
            Lock lock = CacheFactory.getLock(namespace, serverFeatures);
            try {
                lock.lock();
                Set<NodeID> nodeIDs = entry.getValue();
                if (nodeIDs.remove(leftNode)) {
                    if (nodeIDs.isEmpty()) {
                        serverFeatures.remove(namespace);
                    }
                    else {
                        serverFeatures.put(namespace, nodeIDs);
                    }
                }
            }
            finally {
                lock.unlock();
            }
        }
    }
}
 

import org.jivesoftware.openfire.cluster.ClusterManager; //导入依赖的package包/类
/**
 * Returns the set of active conversations.
 *
 * @return the active conversations.
 */
public Collection<Conversation> getConversations() {
    if (ClusterManager.isSeniorClusterMember()) {
        List<Conversation> conversationList = new ArrayList<Conversation>(conversations.values());
        // Sort the conversations by creation date.
        Collections.sort(conversationList, new Comparator<Conversation>() {
            public int compare(Conversation c1, Conversation c2) {
                return c1.getStartDate().compareTo(c2.getStartDate());
            }
        });
        return conversationList;
    }
    else {
        // Get this info from the senior cluster member when running in a cluster
        return (Collection<Conversation>) CacheFactory.doSynchronousClusterTask(new GetConversationsTask(),
                ClusterManager.getSeniorClusterMember().toByteArray());
    }
}