conversationmanager.java

openfire 服务器源码下载

/**
 * $Revision: $
 * $Date: $
 *
 * Copyright (C) 2008 Jive Software. All rights reserved.
 *
 * This software is published under the terms of the GNU Public License (GPL),
 * a copy of which is included in this distribution, or a commercial license
 * agreement with Jive.
 */

package org.jivesoftware.openfire.archive;

import org.jivesoftware.openfire.archive.cluster.GetConversationCountTask;
import org.jivesoftware.openfire.archive.cluster.GetConversationTask;
import org.jivesoftware.openfire.archive.cluster.GetConversationsTask;
import org.jivesoftware.openfire.reporting.util.TaskEngine;
import org.dom4j.Element;
import org.jivesoftware.database.DbConnectionManager;
import org.jivesoftware.openfire.XMPPServer;
import org.jivesoftware.openfire.XMPPServerInfo;
import org.jivesoftware.openfire.cluster.ClusterManager;
import org.jivesoftware.openfire.component.ComponentEventListener;
import org.jivesoftware.openfire.component.InternalComponentManager;
import org.jivesoftware.openfire.stats.Statistic;
import org.jivesoftware.openfire.stats.StatisticsManager;
import org.jivesoftware.util.*;
import org.jivesoftware.util.cache.CacheFactory;
import org.picocontainer.Startable;
import org.xmpp.component.Component;
import org.xmpp.packet.IQ;
import org.xmpp.packet.JID;
import org.xmpp.packet.Message;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * Manages all conversations in the system. Optionally, conversations (messages plus
 * meta-data) can be archived to the database. Archiving of conversation data is
 * enabled by default, but can be disabled by setting "conversation.metadataArchiving" to
 * <tt>false</tt>. Archiving of messages in a conversation is disabled by default, but
 * can be enabled by setting "conversation.messageArchiving" to <tt>true</tt>.<p>
 *
 * When running in a cluster only the senior cluster member will keep track of the active
 * conversations. Other cluster nodes will forward conversation events that occured in the
 * local node to the senior cluster member. If the senior cluster member goes down then
 * current conversations will be terminated and if users keep sending messages between them
 * then new converstions will be created. 
 *
 * @author Matt Tucker
 */
public class ConversationManager implements Startable, ComponentEventListener {

    private static final String UPDATE_CONVERSATION =
            "UPDATE ofConversation SET lastActivity=?, messageCount=? WHERE conversationID=?";
    private static final String UPDATE_PARTICIPANT =
            "UPDATE ofConParticipant SET leftDate=? WHERE conversationID=? AND bareJID=? AND jidResource=? AND joinedDate=?";
    private static final String INSERT_MESSAGE =
            "INSERT INTO ofMessageArchive(conversationID, fromJID, toJID, sentDate, body) " +
                    "VALUES (?,?,?,?,?)";
    private static final String CONVERSATION_COUNT =
            "SELECT COUNT(*) FROM ofConversation";
    private static final String MESSAGE_COUNT =
            "SELECT COUNT(*) FROM ofMessageArchive";

    private static final int DEFAULT_IDLE_TIME = 10;
    private static final int DEFAULT_MAX_TIME = 60;

    public static final String CONVERSATIONS_KEY = "conversations";

    private ConversationEventsQueue conversationEventsQueue;
    private TaskEngine taskEngine;

    private Map<String, Conversation> conversations = new ConcurrentHashMap<String, Conversation>();
    private boolean metadataArchivingEnabled;
    /**
     * Flag that indicates if messages of one-to-one chats should be archived.
     */
    private boolean messageArchivingEnabled;
    /**
     * Flag that indicates if messages of group chats (in MUC rooms) should be archived.
     */
    private boolean roomArchivingEnabled;
    /**
     * List of room names to archive. When list is empty then all rooms are archived (if
     * roomArchivingEnabled is enabled).
     */
    private Collection<String> roomsArchived;
    private long idleTime;
    private long maxTime;
    private PropertyEventListener propertyListener;

    private Queue<Conversation> conversationQueue;
    private Queue<ArchivedMessage> messageQueue;
    /**
     * Queue of participants that joined or left a conversation. This queue is processed by the
     * ArchivingTask.
     */
    private Queue<RoomParticipant> participantQueue;

    private boolean archivingRunning = false;

    private TimerTask archiveTask;
    private TimerTask cleanupTask;

    private Collection<ConversationListener> conversationListeners;

    /**
     * Keeps the address of those components that provide the gateway service.
     */
    private List<String> gateways;
    private XMPPServerInfo serverInfo;

    public ConversationManager(TaskEngine taskEngine) {
        this.taskEngine = taskEngine;
        this.gateways = new CopyOnWriteArrayList<String>();
        this.serverInfo = XMPPServer.getInstance().getServerInfo();
        this.conversationEventsQueue = new ConversationEventsQueue(this, taskEngine);
    }

    public void start() {
        metadataArchivingEnabled = JiveGlobals.getBooleanProperty("conversation.metadataArchiving", true);
        messageArchivingEnabled = JiveGlobals.getBooleanProperty("conversation.messageArchiving", false);
        if (messageArchivingEnabled && !metadataArchivingEnabled) {
            Log.warn("Metadata archiving must be enabled when message archiving is enabled. Overriding setting.");
            metadataArchivingEnabled = true;
        }
        roomArchivingEnabled = JiveGlobals.getBooleanProperty("conversation.roomArchiving", false);
        roomsArchived = StringUtils.stringToCollection(JiveGlobals.getProperty("conversation.roomsArchived", ""));
        if (roomArchivingEnabled && !metadataArchivingEnabled) {
            Log.warn("Metadata archiving must be enabled when room archiving is enabled. Overriding setting.");
            metadataArchivingEnabled = true;
        }
        idleTime = JiveGlobals.getIntProperty("conversation.idleTime", DEFAULT_IDLE_TIME) *
                JiveConstants.MINUTE;
        maxTime = JiveGlobals.getIntProperty("conversation.maxTime",
                DEFAULT_MAX_TIME) * JiveConstants.MINUTE;
        // Listen for any changes to the conversation properties.
        propertyListener = new ConversationPropertyListener();
        PropertyEventDispatcher.addListener(propertyListener);

        conversationQueue = new ConcurrentLinkedQueue<Conversation>();
        messageQueue = new ConcurrentLinkedQueue<ArchivedMessage>();
        participantQueue = new ConcurrentLinkedQueue<RoomParticipant>();

        conversationListeners = new CopyOnWriteArraySet<ConversationListener>();

        // Schedule a task to do conversation archiving.
        archiveTask = new TimerTask() {
            public void run() {
                new ArchivingTask().run();
            }
        };
        taskEngine.scheduleAtFixedRate(archiveTask, JiveConstants.MINUTE, JiveConstants.MINUTE);

        // Schedule a task to do conversation cleanup.
        cleanupTask = new TimerTask() {
            public void run() {
                for (String key : conversations.keySet()) {
                    Conversation conversation = conversations.get(key);
                    long now = System.currentTimeMillis();
                    if ((now - conversation.getLastActivity().getTime() > idleTime) ||
                            (now - conversation.getStartDate().getTime() > maxTime)) {
                        removeConversation(key, conversation, new Date(now));
                    }
                }
            }
        };
        taskEngine.scheduleAtFixedRate(cleanupTask, JiveConstants.MINUTE * 5, JiveConstants.MINUTE * 5);

        // Register a statistic.
        Statistic conversationStat = new Statistic() {

            public String getName() {
                return LocaleUtils.getLocalizedString("stat.conversation.name", "monitoring");
            }

            public Type getStatType() {
                return Type.count;
            }

            public String getDescription() {
                return LocaleUtils.getLocalizedString("stat.conversation.desc", "monitoring");
            }

            public String getUnits() {
                return LocaleUtils.getLocalizedString("stat.conversation.units", "monitoring");
            }

            public double sample() {
                return getConversationCount();
            }

            public boolean isPartialSample() {
                return false;
            }
        };
        StatisticsManager.getInstance().addStatistic(CONVERSATIONS_KEY, conversationStat);
        InternalComponentManager.getInstance().addListener(this);
    }

    public void stop() {
        archiveTask.cancel();
        archiveTask = null;
        cleanupTask.cancel();
        cleanupTask = null;

        // Remove the statistics.
        StatisticsManager.getInstance().removeStatistic(CONVERSATIONS_KEY);

        PropertyEventDispatcher.removeListener(propertyListener);
        propertyListener = null;
        conversations.clear();
        conversations = null;

        // Archive anything remaining in the queue before quitting.
        new ArchivingTask().run();

        conversationQueue.clear();
        conversationQueue = null;

        messageQueue.clear();
        messageQueue = null;

        conversationListeners.clear();
        conversationListeners = null;

        serverInfo = null;
        InternalComponentManager.getInstance().removeListener(this);
    }

    /**
     * Returns true if metadata archiving is enabled. Conversation meta-data includes
     * the participants, start date, last activity, and the count of messages sent.
     * When archiving is enabled, all meta-data is written to the database.
     *
     * @return true if metadata archiving is enabled.
     */
    public boolean isMetadataArchivingEnabled() {
        return metadataArchivingEnabled;
    }

    /**
     * Sets whether metadata archiving is enabled. Conversation meta-data includes
     * the participants, start date, last activity, and the count of messages sent.
     * When archiving is enabled, all meta-data is written to the database.
     *
     * @param enabled true if archiving should be enabled.
     */
    public void setMetadataArchivingEnabled(boolean enabled) {
        this.metadataArchivingEnabled = enabled;
        JiveGlobals.setProperty("conversation.metadataArchiving", Boolean.toString(enabled));
    }

    /**
     * Returns true if one-to-one chats or group chats messages are being archived.
     *
     * @return true if one-to-one chats or group chats messages are being archived.
     */
    public boolean isArchivingEnabled() {
        return isMessageArchivingEnabled() || isRoomArchivingEnabled();
    }

    /**
     * Returns true if message archiving is enabled for one-to-one chats. When enabled, all messages
     * in one-to-one conversations are stored in the database. Note: it's not possible for
     * meta-data archiving to be disabled when message archiving is enabled; enabling
     * message archiving automatically enables meta-data archiving.
     *
     * @return true if message archiving is enabled.
     */
    public boolean isMessageArchivingEnabled() {
        return messageArchivingEnabled;
    }

    /**
     * Sets whether message archiving is enabled. When enabled, all messages
     * in conversations are stored in the database. Note: it's not possible for
     * meta-data archiving to be disabled when message archiving is enabled; enabling
     * message archiving automatically enables meta-data archiving.
     *
     * @param enabled true if message should be enabled.
     */
    public void setMessageArchivingEnabled(boolean enabled) {
        this.messageArchivingEnabled = enabled;
        JiveGlobals.setProperty("conversation.messageArchiving", Boolean.toString(enabled));
        // Force metadata archiving enabled.
        if (enabled) {
            this.metadataArchivingEnabled = true;
        }
    }

    /**
     * Returns true if message archiving is enabled for group chats. When enabled, all messages
     * in group conversations are stored in the database unless a list of rooms was specified
     * in {@link #getRoomsArchived()} . Note: it's not possible for meta-data archiving to be
     * disabled when room archiving is enabled; enabling room archiving automatically
     * enables meta-data archiving.
     *
     * @return true if room archiving is enabled.
     */
    public boolean isRoomArchivingEnabled() {
        return roomArchivingEnabled;
    }

    /**
     * Sets whether message archiving is enabled for group chats. When enabled, all messages
     * in group conversations are stored in the database unless a list of rooms was specified
     * in {@link #getRoomsArchived()} . Note: it's not possible for meta-data archiving to be
     * disabled when room archiving is enabled; enabling room archiving automatically
     * enables meta-data archiving.
     *
     * @param enabled if room archiving is enabled.
     */
    public void setRoomArchivingEnabled(boolean enabled) {
        this.roomArchivingEnabled = enabled;
        JiveGlobals.setProperty("conversation.roomArchiving", Boolean.toString(enabled));
        // Force metadata archiving enabled.
        if (enabled) {
            this.metadataArchivingEnabled = true;
        }
    }

    /**
     * Returns list of room names whose messages will be archived. When room archiving is enabled and
     * this list is empty then messages of all local rooms will be archived. However, when name of
     * rooms are defined in this list then only messages of those rooms will be archived.
     *
     * @return list of local room names whose messages will be archived.
     */
    public Collection<String> getRoomsArchived() {
        return roomsArchived;
    }

    /**
     * Sets list of room names whose messages will be archived. When room archiving is enabled and
     * this list is empty then messages of all local rooms will be archived. However, when name of
     * rooms are defined in this list then only messages of those rooms will be archived.
     *