conversationmanager.java

openfire 服务器源码下载

     * @param roomsArchived list of local room names whose messages will be archived.
     */
    public void setRoomsArchived(Collection<String> roomsArchived) {
        this.roomsArchived = roomsArchived;
        JiveGlobals.setProperty("conversation.roomsArchived", StringUtils.collectionToString(roomsArchived));
    }

    /**
     * Returns the number of minutes a conversation can be idle before it's ended.
     *
     * @return the conversation idle time.
     */
    public int getIdleTime() {
        return (int)(idleTime / JiveConstants.MINUTE);
    }

    /**
     * Sets the number of minutes a conversation can be idle before it's ended.
     *
     * @param idleTime the max number of minutes a conversation can be idle before it's ended.
     * @throws IllegalArgumentException if idleTime is less than 1.
     */
    public void setIdleTime(int idleTime) {
        if (idleTime < 1) {
            throw new IllegalArgumentException("Idle time less than 1 is not valid: " + idleTime);
        }
        JiveGlobals.setProperty("conversation.idleTime", Integer.toString(idleTime));
        this.idleTime = idleTime * JiveConstants.MINUTE;
    }

    /**
     * Returns the maximum number of minutes a conversation can last before it's ended.
     * Any additional messages between the participants in the chat will be associated
     * with a new conversation.
     *
     * @return the maximum number of minutes a conversation can last.
     */
    public int getMaxTime() {
        return (int)(maxTime / JiveConstants.MINUTE);
    }

    /**
     * Sets the maximum number of minutes a conversation can last before it's ended.
     * Any additional messages between the participants in the chat will be associated
     * with a new conversation.
     *
     * @param maxTime the maximum number of minutes a conversation can last.
     * @throws IllegalArgumentException if maxTime is less than 1.
     */
    public void setMaxTime(int maxTime) {
        if (maxTime < 1) {
            throw new IllegalArgumentException("Max time less than 1 is not valid: " + maxTime);
        }
        JiveGlobals.setProperty("conversation.maxTime", Integer.toString(maxTime));
        this.maxTime = maxTime * JiveConstants.MINUTE;
    }

    public ConversationEventsQueue getConversationEventsQueue() {
        return conversationEventsQueue;
    }

    /**
     * Returns the count of active conversations.
     *
     * @return the count of active conversations.
     */
    public int getConversationCount() {
        if (ClusterManager.isSeniorClusterMember()) {
            return conversations.size();
        }
        return (Integer) CacheFactory.doSynchronousClusterTask(new GetConversationCountTask(),
                ClusterManager.getSeniorClusterMember().toByteArray());
    }

    /**
     * Returns a conversation by ID.
     *
     * @param conversationID the ID of the conversation.
     * @return the conversation.
     * @throws NotFoundException if the conversation could not be found.
     */
    public Conversation getConversation(long conversationID) throws NotFoundException {
        if (ClusterManager.isSeniorClusterMember()) {
            // Search through the currently active conversations.
            for (Conversation conversation : conversations.values()) {
                if (conversation.getConversationID() == conversationID) {
                    return conversation;
                }
            }
            // Otherwise, it might be an archived conversation, so attempt to load it.
            return new Conversation(this, conversationID);
        }
        else {
            // Get this info from the senior cluster member when running in a cluster
            Conversation conversation = (Conversation) CacheFactory.doSynchronousClusterTask(
                    new GetConversationTask(conversationID), ClusterManager.getSeniorClusterMember().toByteArray());
            if (conversation == null) {
                throw new NotFoundException("Conversation not found: " + conversationID);
            }
            return conversation;
        }
    }

    /**
     * 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());
        }
    }

    /**
     * Returns the total number of conversations that have been archived to the database.
     * The archived conversation may only be the meta-data, or it might include messages
     * as well if message archiving is turned on.
     *
     * @return the total number of archived conversations.
     */
    public int getArchivedConversationCount() {
        int conversationCount = 0;
        Connection con = null;
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            con = DbConnectionManager.getConnection();
            pstmt = con.prepareStatement(CONVERSATION_COUNT);
            rs = pstmt.executeQuery();
            if (rs.next()) {
                conversationCount = rs.getInt(1);
            }
        }
        catch (SQLException sqle) {
            Log.error(sqle);
        }
        finally {
            DbConnectionManager.closeConnection(rs, pstmt, con);
        }
        return conversationCount;
    }

    /**
     * Returns the total number of messages that have been archived to the database.
     *
     * @return the total number of archived messages.
     */
    public int getArchivedMessageCount() {
        int messageCount = 0;
        Connection con = null;
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            con = DbConnectionManager.getConnection();
            pstmt = con.prepareStatement(MESSAGE_COUNT);
            rs = pstmt.executeQuery();
            if (rs.next()) {
                messageCount = rs.getInt(1);
            }
        }
        catch (SQLException sqle) {
            Log.error(sqle);
        }
        finally {
            DbConnectionManager.closeConnection(rs, pstmt, con);
        }
        return messageCount;
    }

    /**
     * Adds a conversation listener, which will be notified of newly created conversations,
     * conversations ending, and updates to conversations.
     *
     * @param listener the conversation listener.
     */
    public void addConversationListener(ConversationListener listener) {
        conversationListeners.add(listener);
    }

    /**
     * Removes a conversation listener.
     *
     * @param listener the conversation listener.
     */
    public void removeConversationListener(ConversationListener listener) {
        conversationListeners.remove(listener);
    }

    /**
     * Processes an incoming message of a one-to-one chat. The message will mapped to a
     * conversation and then queued for storage if archiving is turned on.
     *
     * @param sender sender of the message.
     * @param receiver receiver of the message.
     * @param body body of the message.
     * @param date date when the message was sent.
     */
    void processMessage(JID sender, JID receiver, String body, Date date) {
        String conversationKey = getConversationKey(sender, receiver);
        synchronized (conversationKey.intern()) {
            Conversation conversation = conversations.get(conversationKey);
            // Create a new conversation if necessary.
            if (conversation == null) {
                Collection<JID> participants = new ArrayList<JID>(2);
                participants.add(sender);
                participants.add(receiver);
                XMPPServer server = XMPPServer.getInstance();
                // Check to see if this is an external conversation; i.e. one of the participants
                // is on a different server. We can use XOR since we know that both JID's can't
                // be external.
                boolean external = isExternal(server, sender) ^ isExternal(server, receiver);
                // Make sure that the user joined the conversation before a message was received
                Date start = new Date(date.getTime() - 1);
                conversation = new Conversation(this, participants, external, start);
                conversations.put(conversationKey, conversation);
                // Notify listeners of the newly created conversation.
                for (ConversationListener listener : conversationListeners) {
                    listener.conversationCreated(conversation);
                }
            }
            // Check to see if the current conversation exceeds either the max idle time
            // or max conversation time.
            else if ((date.getTime() - conversation.getLastActivity().getTime() > idleTime) ||
                    (date.getTime() - conversation.getStartDate().getTime() > maxTime)) {
                removeConversation(conversationKey, conversation, conversation.getLastActivity());

                Collection<JID> participants = new ArrayList<JID>(2);
                participants.add(sender);
                participants.add(receiver);
                XMPPServer server = XMPPServer.getInstance();
                // Check to see if this is an external conversation; i.e. one of the participants
                // is on a different server. We can use XOR since we know that both JID's can't
                // be external.
                boolean external = isExternal(server, sender) ^ isExternal(server, receiver);
                // Make sure that the user joined the conversation before a message was received
                Date start = new Date(date.getTime() - 1);
                conversation = new Conversation(this, participants, external, start);
                conversations.put(conversationKey, conversation);
                // Notify listeners of the newly created conversation.
                for (ConversationListener listener : conversationListeners) {
                    listener.conversationCreated(conversation);
                }
            }
            // Record the newly received message.
            conversation.messageReceived(sender, date);
            if (metadataArchivingEnabled) {
                conversationQueue.add(conversation);
            }
            if (messageArchivingEnabled) {
                messageQueue
                        .add(new ArchivedMessage(conversation.getConversationID(), sender, receiver, date, body, false));
            }
            // Notify listeners of the conversation update.
            for (ConversationListener listener : conversationListeners) {
                listener.conversationUpdated(conversation, date);
            }
        }
    }

    /**
     * Processes an incoming message sent to a room. The message will mapped to a conversation and then
     * queued for storage if archiving is turned on.
     *
     * @param roomJID  the JID of the room where the group conversation is taking place.
     * @param sender   the JID of the entity that sent the message.
     * @param nickname nickname of the user in the room when the message was sent.
     * @param body  the message sent to the room.
     * @param date  date when the message was sent.
     */
    void processRoomMessage(JID roomJID, JID sender, String nickname, String body, Date date) {
        String conversationKey = getRoomConversationKey(roomJID);
        synchronized (conversationKey.intern()) {
            Conversation conversation = conversations.get(conversationKey);
            // Create a new conversation if necessary.
            if (conversation == null) {
                // Make sure that the user joined the conversation before a message was received
                Date start = new Date(date.getTime() - 1);
                conversation = new Conversation(this, roomJID, false, start);
                conversations.put(conversationKey, conversation);
                // Notify listeners of the newly created conversation.
                for (ConversationListener listener : conversationListeners) {
                    listener.conversationCreated(conversation);
                }
            }
            // Check to see if the current conversation exceeds either the max idle time
            // or max conversation time.
            else if ((date.getTime() - conversation.getLastActivity().getTime() > idleTime) ||
                    (date.getTime() - conversation.getStartDate().getTime() > maxTime)) {
                removeConversation(conversationKey, conversation, conversation.getLastActivity());
                // Make sure that the user joined the conversation before a message was received
                Date start = new Date(date.getTime() - 1);
                conversation = new Conversation(this, roomJID, false, start);
                conversations.put(conversationKey, conversation);
                // Notify listeners of the newly created conversation.
                for (ConversationListener listener : conversationListeners) {
                    listener.conversationCreated(conversation);
                }
            }
            // Record the newly received message.
            conversation.messageReceived(sender, date);
            if (metadataArchivingEnabled) {
                conversationQueue.add(conversation);
            }
            if (roomArchivingEnabled && (roomsArchived.isEmpty() || roomsArchived.contains(roomJID.getNode()))) {
                JID jid = new JID(roomJID + "/" + nickname);
                messageQueue.add(new ArchivedMessage(conversation.getConversationID(), sender, jid, date, body, false));
            }
            // Notify listeners of the conversation update.
            for (ConversationListener listener : conversationListeners) {
                listener.conversationUpdated(conversation, date);
            }
        }
    }

    /**
     * Notification message indicating that a user joined a groupchat conversation. If
     * no groupchat conversation was taking place in the specified room then ignore this
     * event.<p>
     * <p/>
     * Eventually, when a new conversation will start in the room and if this user is
     * still in the room then the new conversation will detect this user and mark like
     * if the user joined the converstion from the beginning.
     *
     * @param room     the room where the user joined.
     * @param user     the user that joined the room.
     * @param nickname nickname of the user in the room.
     * @param date date when the user joined the group coversation.
     */
    void joinedGroupConversation(JID room, JID user, String nickname, Date date) {
        Conversation conversation = getRoomConversation(room);
        if (conversation != null) {
            conversation.participantJoined(user, nickname, date.getTime());
        }
    }