源码详解openfire保存消息记录_修改服务端方式

时间：2021-06-11 08:35:01

实现openfire消息记录通常有两种方式，修改服务端和添加消息记录插件。

今天，简单的说明一下修改服务端方式实现消息记录保存功能。

实现思路

修改前：

默认的，openfire只提供保存离线记录至ofOffline表中。当发送一条消息时，判断用户是否在线，若为true，不保存消息；若为fasle，保存消息至离线消息表中。

修改后：

仿照保存离线消息，用户每发送一条消息，将消息存放在ofHistory表中，ofHistory表结构同ofOffline

实现步骤：

1.修改初始数据库文件，路径src/database/openfire_sqlserver.sql

添加ofHistory表

[sql] view plaincopyprint?

CREATE TABLE ofHistory (

username NVARCHAR(64) NOT NULL,

messageID INTEGER NOT NULL,

creationDate NVARCHAR(64) NOT NULL,

messageSize INTEGER NOT NULL,

stanza TEXT NOT NULL,

CONSTRAINT ofHistory_pk PRIMARY KEY (username, messageID)

);

CREATE TABLE ofOffline (

username NVARCHAR(64) NOT NULL,

messageID INTEGER NOT NULL,

creationDate CHAR(15) NOT NULL,

messageSize INTEGER NOT NULL,

stanza NTEXT NOT NULL,

CONSTRAINT ofOffline_pk PRIMARY KEY (username, messageID)

);

注：其他数据库修改方式雷同

2.添加保存消息方法

MessageRouter类中110行

[java] view plaincopyprint?

try {

// Deliver stanza to requested route

routingTable.routePacket(recipientJID, packet, false);

//保存消息记录dml@2013.4.15

OfflineMessageStore oms = new OfflineMessageStore();

oms.addMessage_toHistory(packet);

         }

         catch (Exception e) {

            log.error("Failed to route packet: " + packet.toXML(), e);

             routingFailed(recipientJID, packet);

         }

3.修改OfflineMessageStore类，添加保存消息记录方法

[java] view plaincopyprint?

/**

  • $RCSfile$
  • $Revision: 2911 $
  • $Date: 2005-10-03 12:35:52 -0300 (Mon, 03 Oct 2005) $
  • Copyright (C) 2004-2008 Jive Software. All rights reserved.
  • Licensed under the Apache License, Version 2.0 (the "License");
  • you may not use this file except in compliance with the License.
  • You may obtain a copy of the License at
  • http://www.apache.org/licenses/LICENSE-2.0
  • Unless required by applicable law or agreed to in writing, software
  • distributed under the License is distributed on an "AS IS" BASIS,
  • WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  • See the License for the specific language governing permissions and
  • limitations under the License.

    */

package org.jivesoftware.openfire;

import java.io.StringReader;

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.text.SimpleDateFormat;

import java.util.ArrayList;

import java.util.Collection;

import java.util.Date;

import java.util.List;

import java.util.Map;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.LinkedBlockingQueue;

import java.util.regex.Matcher;

import java.util.regex.Pattern;

import org.dom4j.DocumentException;

import org.dom4j.Element;

import org.dom4j.io.SAXReader;

import org.jivesoftware.database.DbConnectionManager;

import org.jivesoftware.database.SequenceManager;

import org.jivesoftware.openfire.container.BasicModule;

import org.jivesoftware.openfire.event.UserEventDispatcher;

import org.jivesoftware.openfire.event.UserEventListener;

import org.jivesoftware.openfire.user.User;

import org.jivesoftware.openfire.user.UserManager;

import org.jivesoftware.util.JiveConstants;

import org.jivesoftware.util.LocaleUtils;

import org.jivesoftware.util.StringUtils;

import org.jivesoftware.util.XMPPDateTimeFormat;

import org.jivesoftware.util.cache.Cache;

import org.jivesoftware.util.cache.CacheFactory;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.xmpp.packet.JID;

import org.xmpp.packet.Message;

/**

  • Represents the user's offline message storage. A message store holds messages

  • that were sent to the user while they were unavailable. The user can retrieve

  • their messages by setting their presence to "available". The messages will

  • then be delivered normally. Offline message storage is optional, in which

  • case a null implementation is returned that always throws

  • UnauthorizedException when adding messages to the store.

  • @author Iain Shigeoka

    */

    public class OfflineMessageStore extends BasicModule implements

    UserEventListener {

    private static final Logger Log = LoggerFactory

    .getLogger(OfflineMessageStore.class);

    // 保存消息记录 dml@2013.4.16

    private static final String INSERT_HISTORY = "INSERT INTO ofHistory (username, messageID, creationDate, messageSize, stanza) "

    + "VALUES (?, ?, ?, ?, ?)";

    private static final String INSERT_OFFLINE = "INSERT INTO ofOffline (username, messageID, creationDate, messageSize, stanza) "

    + "VALUES (?, ?, ?, ?, ?)";

    private static final String LOAD_OFFLINE = "SELECT stanza, creationDate FROM ofOffline WHERE username=?";

    private static final String LOAD_OFFLINE_MESSAGE = "SELECT stanza FROM ofOffline WHERE username=? AND creationDate=?";

    private static final String SELECT_SIZE_OFFLINE = "SELECT SUM(messageSize) FROM ofOffline WHERE username=?";

    private static final String SELECT_SIZE_ALL_OFFLINE = "SELECT SUM(messageSize) FROM ofOffline";

    private static final String DELETE_OFFLINE = "DELETE FROM ofOffline WHERE username=?";

    private static final String DELETE_OFFLINE_MESSAGE = "DELETE FROM ofOffline WHERE username=? AND creationDate=?";

    private static final int POOL_SIZE = 10;

    private Cache<String, Integer> sizeCache;

    /**

    • Pattern to use for detecting invalid XML characters. Invalid XML
    • characters will be removed from the stored offline messages.

      */

      private Pattern pattern = Pattern.compile("&\#[\d]+;");

    /**

    • Returns the instance of OfflineMessageStore being used by the
    • XMPPServer.
    • @return the instance of OfflineMessageStore being used by the
    •     XMPPServer.

    */

    public static OfflineMessageStore getInstance() {

    return XMPPServer.getInstance().getOfflineMessageStore();

    }

    /**

    • Pool of SAX Readers. SAXReader is not thread safe so we need to have a
    • pool of readers.

      */

      private BlockingQueue xmlReaders = new LinkedBlockingQueue(

      POOL_SIZE);

    /**

    • Constructs a new offline message store.

      */

      public OfflineMessageStore() {

      super("Offline Message Store");

      sizeCache = CacheFactory.createCache("Offline Message Size");

      }

    /**

    • Adds a message to this message store. Messages will be stored and made
    • available for later delivery.
    • @param message
    •        the message to store.

    */

    public void addMessage(Message message) {

    if (message == null) {

    return;

    }

    // ignore empty bodied message (typically chat-state notifications).

    if (message.getBody() == null || message.getBody().length() == 0) {

    // allow empty pubsub messages (OF-191)

    if (message.getChildElement("event",

    "http://jabber.org/protocol/pubsub#event") == null) {

    return;

    }

    }

    JID recipient = message.getTo();

    String username = recipient.getNode();

    // If the username is null (such as when an anonymous user), don't

    // store.

    if (username == null

    || !UserManager.getInstance().isRegisteredUser(recipient)) {

    return;

    } else if (!XMPPServer.getInstance().getServerInfo().getXMPPDomain()

    .equals(recipient.getDomain())) {

    // Do not store messages sent to users of remote servers

    return;

    }

     long messageID = SequenceManager.nextID(JiveConstants.OFFLINE);  

 // Get the message in XML format.
    
 String msgXML = message.getElement().asXML();  

 Connection con = null;
    
 PreparedStatement pstmt = null;
    
 try {
    
     con = DbConnectionManager.getConnection();
    
     pstmt = con.prepareStatement(INSERT_OFFLINE);
    
     pstmt.setString(1, username);
    
     pstmt.setLong(2, messageID);
    
     pstmt.setString(3, StringUtils.dateToMillis(new java.util.Date()));
    
     // SimpleDateFormat df = new
    
     // SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
     // pstmt.setString(3, df.format(new Date()).toString());   

     pstmt.setInt(4, msgXML.length());
    
     pstmt.setString(5, msgXML);
    
     pstmt.executeUpdate();
    
 }  

 catch (Exception e) {
    
     Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
    
 } finally {
    
     DbConnectionManager.closeConnection(pstmt, con);
    
 }  

 // Update the cached size if it exists.
    
 if (sizeCache.containsKey(username)) {
    
     int size = sizeCache.get(username);
    
     size += msgXML.length();
    
     sizeCache.put(username, size);
    
 }

    }

    /**

    • 保存消息记录

    • @author dml

    • @param message

      */

      public void addMessage_toHistory(Message message) {

      if (message == null) {

      return;

      }

      // ignore empty bodied message (typically chat-state notifications).

      if (message.getBody() == null || message.getBody().length() == 0) {

      // allow empty pubsub messages (OF-191)

      if (message.getChildElement("event",

      "http://jabber.org/protocol/pubsub#event") == null) {

      return;

      }

      }

      JID recipient = message.getTo();

      String username = recipient.getNode();

      // If the username is null (such as when an anonymous user), don't

      // store.

      if (username == null

      || !UserManager.getInstance().isRegisteredUser(recipient)) {

      return;

      } else if (!XMPPServer.getInstance().getServerInfo().getXMPPDomain()

      .equals(recipient.getDomain())) {

      // Do not store messages sent to users of remote servers

      return;

      }

      long messageID = SequenceManager.nextID(JiveConstants.OFFLINE);

      // Get the message in XML format.

      String msgXML = message.getElement().asXML();

      Connection con = null;

      PreparedStatement pstmt = null;

      try {

      con = DbConnectionManager.getConnection();

      pstmt = con.prepareStatement(INSERT_HISTORY);

      pstmt.setString(1, username);

      pstmt.setLong(2, messageID);

      // pstmt.setString(3, StringUtils.dateToMillis(new

      // java.util.Date()));

      SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

      pstmt.setString(3, df.format(new Date()).toString());

       pstmt.setInt(4, msgXML.length());
      
 pstmt.setString(5, msgXML);
      
 pstmt.executeUpdate();

      }

      catch (Exception e) {

      Log.error(LocaleUtils.getLocalizedString("admin.error"), e);

      } finally {

      DbConnectionManager.closeConnection(pstmt, con);

      }

      // Update the cached size if it exists.

      if (sizeCache.containsKey(username)) {

      int size = sizeCache.get(username);

      size += msgXML.length();

      sizeCache.put(username, size);

      }

      }

    /**

    • Returns a Collection of all messages in the store for a user. Messages

    • may be deleted after being selected from the database depending on the

    • delete param.

    • @param username

    •        the username of the user who's messages you'd like to receive.

    • @param delete

    •        true if the offline messages should be deleted.

    • @return An iterator of packets containing all offline messages.

      */

      public Collection getMessages(String username,

      boolean delete) {

      List messages = new ArrayList();

      SAXReader xmlReader = null;

      Connection con = null;

      PreparedStatement pstmt = null;

      ResultSet rs = null;

      try {

      // Get a sax reader from the pool

      xmlReader = xmlReaders.take();

      con = DbConnectionManager.getConnection();

      pstmt = con.prepareStatement(LOAD_OFFLINE);

      pstmt.setString(1, username);

      rs = pstmt.executeQuery();

      while (rs.next()) {

      String msgXML = rs.getString(1);

      // 解析时间eg.Tue Apr 16 15:32:39 CST 2013

      Date creationDate = new Date(Long.parseLong(rs.getString(2)

      .trim()));

      OfflineMessage message;

      try {

      message = new OfflineMessage(creationDate, xmlReader.read(

      new StringReader(msgXML)).getRootElement());

      } catch (DocumentException e) {

      // Try again after removing invalid XML chars (e.g. )

      Matcher matcher = pattern.matcher(msgXML);

      if (matcher.find()) {

      msgXML = matcher.replaceAll("");

      }

      message = new OfflineMessage(creationDate, xmlReader.read(

      new StringReader(msgXML)).getRootElement());

      }

           // Add a delayed delivery (XEP-0203) element to the message.
      
     Element delay = message.addChildElement("delay",
      
             "urn:xmpp:delay");
      
     delay.addAttribute("from", XMPPServer.getInstance()
      
             .getServerInfo().getXMPPDomain());
      
     delay.addAttribute("stamp",
      
             XMPPDateTimeFormat.format(creationDate));
      
     // Add a legacy delayed delivery (XEP-0091) element to the
      
     // message. XEP is obsolete and support should be dropped in
      
     // future.
      
     delay = message.addChildElement("x", "jabber:x:delay");
      
     delay.addAttribute("from", XMPPServer.getInstance()
      
             .getServerInfo().getXMPPDomain());
      
     delay.addAttribute("stamp",
      
             XMPPDateTimeFormat.formatOld(creationDate));
      
     messages.add(message);
      
 }
      
 // Check if the offline messages loaded should be deleted, and that
      
 // there are
      
 // messages to delete.
      
 if (delete && !messages.isEmpty()) {
      
     PreparedStatement pstmt2 = null;
      
     try {
      
         pstmt2 = con.prepareStatement(DELETE_OFFLINE);
      
         pstmt2.setString(1, username);
      
         pstmt2.executeUpdate();
      
         removeUsernameFromSizeCache(username);
      
     } catch (Exception e) {
      
         Log.error("Error deleting offline messages of username: "
      
                 + username, e);
      
     } finally {
      
         DbConnectionManager.closeStatement(pstmt2);
      
     }
      
 }

      } catch (Exception e) {

      Log.error("Error retrieving offline messages of username: "

      + username, e);

      } finally {

      DbConnectionManager.closeConnection(rs, pstmt, con);

      // Return the sax reader to the pool

      if (xmlReader != null) {

      xmlReaders.add(xmlReader);

      }

      }

      return messages;

      }

    /**

    • Returns the offline message of the specified user with the given creation
    • date. The returned message will NOT be deleted from the database.
    • @param username
    •        the username of the user who's message you'd like to receive.
    • @param creationDate
    •        the date when the offline message was stored in the database.
    • @return the offline message of the specified user with the given creation
    •     stamp.

    */

    public OfflineMessage getMessage(String username, Date creationDate) {

    OfflineMessage message = null;

    Connection con = null;

    PreparedStatement pstmt = null;

    ResultSet rs = null;

    SAXReader xmlReader = null;

    try {

    // Get a sax reader from the pool

    xmlReader = xmlReaders.take();

    con = DbConnectionManager.getConnection();

    pstmt = con.prepareStatement(LOAD_OFFLINE_MESSAGE);

    pstmt.setString(1, username);

    pstmt.setString(2, StringUtils.dateToMillis(creationDate));

    rs = pstmt.executeQuery();

    while (rs.next()) {

    String msgXML = rs.getString(1);

    message = new OfflineMessage(creationDate, xmlReader.read(

    new StringReader(msgXML)).getRootElement());

    // Add a delayed delivery (XEP-0203) element to the message.

    Element delay = message.addChildElement("delay",

    "urn:xmpp:delay");

    delay.addAttribute("from", XMPPServer.getInstance()

    .getServerInfo().getXMPPDomain());

    delay.addAttribute("stamp",

    XMPPDateTimeFormat.format(creationDate));

    // Add a legacy delayed delivery (XEP-0091) element to the

    // message. XEP is obsolete and support should be dropped in

    // future.

    delay = message.addChildElement("x", "jabber❌delay");

    delay.addAttribute("from", XMPPServer.getInstance()

    .getServerInfo().getXMPPDomain());

    delay.addAttribute("stamp",

    XMPPDateTimeFormat.formatOld(creationDate));

    }

    } catch (Exception e) {

    Log.error("Error retrieving offline messages of username: "

    + username + " creationDate: " + creationDate, e);

    } finally {

    // Return the sax reader to the pool

    if (xmlReader != null) {

    xmlReaders.add(xmlReader);

    }

    DbConnectionManager.closeConnection(rs, pstmt, con);

    }

    return message;

    }

    /**

    • Deletes all offline messages in the store for a user.
    • @param username
    •        the username of the user who's messages are going to be
    •        deleted.

    */

    public void deleteMessages(String username) {

    Connection con = null;

    PreparedStatement pstmt = null;

    try {

    con = DbConnectionManager.getConnection();

    pstmt = con.prepareStatement(DELETE_OFFLINE);

    pstmt.setString(1, username);

    pstmt.executeUpdate();

         removeUsernameFromSizeCache(username);
    
 } catch (Exception e) {
    
     Log.error("Error deleting offline messages of username: "
    
             + username, e);
    
 } finally {
    
     DbConnectionManager.closeConnection(pstmt, con);
    
 }

    }

    private void removeUsernameFromSizeCache(String username) {

    // Update the cached size if it exists.

    if (sizeCache.containsKey(username)) {

    sizeCache.remove(username);

    }

    }

    /**

    • Deletes the specified offline message in the store for a user. The way to
    • identify the message to delete is based on the creationDate and username.
    • @param username
    •        the username of the user who's message is going to be deleted.
    • @param creationDate
    •        the date when the offline message was stored in the database.

    */

    public void deleteMessage(String username, Date creationDate) {

    Connection con = null;

    PreparedStatement pstmt = null;

    try {

    con = DbConnectionManager.getConnection();

    pstmt = con.prepareStatement(DELETE_OFFLINE_MESSAGE);

    pstmt.setString(1, username);

    pstmt.setString(2, StringUtils.dateToMillis(creationDate));

    pstmt.executeUpdate();

         // Force a refresh for next call to getSize(username),
    
     // it's easier than loading the message to be deleted just
    
     // to update the cache.
    
     removeUsernameFromSizeCache(username);
    
 } catch (Exception e) {
    
     Log.error("Error deleting offline messages of username: "
    
             + username + " creationDate: " + creationDate, e);
    
 } finally {
    
     DbConnectionManager.closeConnection(pstmt, con);
    
 }

    }

    /**

    • Returns the approximate size (in bytes) of the XML messages stored for a
    • particular user.
    • @param username
    •        the username of the user.
    • @return the approximate size of stored messages (in bytes).

      */

      public int getSize(String username) {

      // See if the size is cached.

      if (sizeCache.containsKey(username)) {

      return sizeCache.get(username);

      }

      int size = 0;

      Connection con = null;

      PreparedStatement pstmt = null;

      ResultSet rs = null;

      try {

      con = DbConnectionManager.getConnection();

      pstmt = con.prepareStatement(SELECT_SIZE_OFFLINE);

      pstmt.setString(1, username);

      rs = pstmt.executeQuery();

      if (rs.next()) {

      size = rs.getInt(1);

      }

      // Add the value to cache.

      sizeCache.put(username, size);

      } catch (Exception e) {

      Log.error(LocaleUtils.getLocalizedString("admin.error"), e);

      } finally {

      DbConnectionManager.closeConnection(rs, pstmt, con);

      }

      return size;

      }

    /**

    • Returns the approximate size (in bytes) of the XML messages stored for
    • all users.
    • @return the approximate size of all stored messages (in bytes).

      */

      public int getSize() {

      int size = 0;

      Connection con = null;

      PreparedStatement pstmt = null;

      ResultSet rs = null;

      try {

      con = DbConnectionManager.getConnection();

      pstmt = con.prepareStatement(SELECT_SIZE_ALL_OFFLINE);

      rs = pstmt.executeQuery();

      if (rs.next()) {

      size = rs.getInt(1);

      }

      } catch (Exception e) {

      Log.error(LocaleUtils.getLocalizedString("admin.error"), e);

      } finally {

      DbConnectionManager.closeConnection(rs, pstmt, con);

      }

      return size;

      }

    public void userCreated(User user, Map params) {

    // Do nothing

    }

    public void userDeleting(User user, Map params) {

    // Delete all offline messages of the user

    deleteMessages(user.getUsername());

    }

    public void userModified(User user, Map params) {

    // Do nothing

    }

    @Override

    public void start() throws IllegalStateException {

    super.start();

    // Initialize the pool of sax readers

    for (int i = 0; i < POOL_SIZE; i++) {

    SAXReader xmlReader = new SAXReader();

    xmlReader.setEncoding("UTF-8");

    xmlReaders.add(xmlReader);

    }

    // Add this module as a user event listener so we can delete

    // all offline messages when a user is deleted

    UserEventDispatcher.addListener(this);

    }

    @Override

    public void stop() {

    super.stop();

    // Clean up the pool of sax readers

    xmlReaders.clear();

    // Remove this module as a user event listener

    UserEventDispatcher.removeListener(this);

    }

    }

相关文章