OfflineMessageStore.java 16.6 KB
Newer Older
1 2 3 4 5
/**
 * $RCSfile$
 * $Revision: 2911 $
 * $Date: 2005-10-03 12:35:52 -0300 (Mon, 03 Oct 2005) $
 *
6
 * Copyright (C) 2004-2008 Jive Software. All rights reserved.
7 8
 *
 * This software is published under the terms of the GNU Public License (GPL),
9 10
 * a copy of which is included in this distribution, or a commercial license
 * agreement with Jive.
11 12
 */

13
package org.jivesoftware.openfire;
14

15
import org.dom4j.DocumentException;
16 17 18 19
import org.dom4j.Element;
import org.dom4j.io.SAXReader;
import org.jivesoftware.database.DbConnectionManager;
import org.jivesoftware.database.SequenceManager;
20
import org.jivesoftware.util.*;
21
import org.jivesoftware.util.cache.Cache;
22
import org.jivesoftware.util.cache.CacheFactory;
23 24 25 26 27
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;
28
import org.xmpp.packet.JID;
29 30 31 32 33 34 35 36 37
import org.xmpp.packet.Message;

import java.io.StringReader;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
38 39
import java.util.regex.Matcher;
import java.util.regex.Pattern;
40 41 42 43 44 45 46 47 48 49

/**
 * 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
 */
50
public class OfflineMessageStore extends BasicModule implements UserEventListener {
51 52

    private static final String INSERT_OFFLINE =
53
        "INSERT INTO ofOffline (username, messageID, creationDate, messageSize, stanza) " +
54 55
        "VALUES (?, ?, ?, ?, ?)";
    private static final String LOAD_OFFLINE =
56
        "SELECT stanza, creationDate FROM ofOffline WHERE username=?";
57
    private static final String LOAD_OFFLINE_MESSAGE =
58
        "SELECT stanza FROM ofOffline WHERE username=? AND creationDate=?";
59
    private static final String SELECT_SIZE_OFFLINE =
60
        "SELECT SUM(messageSize) FROM ofOffline WHERE username=?";
61
    private static final String SELECT_SIZE_ALL_OFFLINE =
62
        "SELECT SUM(messageSize) FROM ofOffline";
63
    private static final String DELETE_OFFLINE =
64
        "DELETE FROM ofOffline WHERE username=?";
65
    private static final String DELETE_OFFLINE_MESSAGE =
66
        "DELETE FROM ofOffline WHERE username=? AND creationDate=?";
67

Gaston Dombiak's avatar
Gaston Dombiak committed
68
    private Cache<String, Integer> sizeCache;
69
    private FastDateFormat dateFormat;
70 71 72 73 74
    /**
     * 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]+;");
75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94

    /**
     * Returns the instance of <tt>OfflineMessageStore</tt> being used by the XMPPServer.
     *
     * @return the instance of <tt>OfflineMessageStore</tt> 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<SAXReader> xmlReaders = new LinkedBlockingQueue<SAXReader>();

    /**
     * Constructs a new offline message store.
     */
    public OfflineMessageStore() {
        super("Offline Message Store");
95 96
        dateFormat = FastDateFormat.getInstance(JiveConstants.XMPP_DELAY_DATETIME_FORMAT,
                TimeZone.getTimeZone("UTC"));
97
        sizeCache = CacheFactory.createCache("Offline Message Size");
98 99 100 101 102 103 104 105 106 107 108 109
    }

    /**
     * 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;
        }
110 111 112 113
        if (message.getBody() == null || message.getBody().length() == 0) {
        	// ignore empty bodied message (typically chat-state notifications).
        	return;
        }
114 115
        JID recipient = message.getTo();
        String username = recipient.getNode();
116
        // If the username is null (such as when an anonymous user), don't store.
117
        if (username == null || !UserManager.getInstance().isRegisteredUser(recipient)) {
118 119
            return;
        }
120
        else
121
        if (!XMPPServer.getInstance().getServerInfo().getXMPPDomain().equals(recipient.getDomain())) {
122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147
            // 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()));
            pstmt.setInt(4, msgXML.length());
            pstmt.setString(5, msgXML);
            pstmt.executeUpdate();
        }

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

        // Update the cached size if it exists.
        if (sizeCache.containsKey(username)) {
Gaston Dombiak's avatar
Gaston Dombiak committed
153
            int size = sizeCache.get(username);
154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182
            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<OfflineMessage> getMessages(String username, boolean delete) {
        List<OfflineMessage> messages = new ArrayList<OfflineMessage>();
        Connection con = null;
        PreparedStatement pstmt = null;
        SAXReader xmlReader = null;
        try {
            // Get a sax reader from the pool
            xmlReader = xmlReaders.take();
            con = DbConnectionManager.getConnection();
            pstmt = con.prepareStatement(LOAD_OFFLINE);
            pstmt.setString(1, username);
            ResultSet rs = pstmt.executeQuery();
            while (rs.next()) {
                String msgXML = rs.getString(1);
                Date creationDate = new Date(Long.parseLong(rs.getString(2).trim()));
183 184 185 186 187 188 189 190 191 192 193 194 195
                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. &#12;)
                    Matcher matcher = pattern.matcher(msgXML);
                    if (matcher.find()) {
                        msgXML = matcher.replaceAll("");
                    }
                    message = new OfflineMessage(creationDate,
                            xmlReader.read(new StringReader(msgXML)).getRootElement());
                }
196 197
                // Add a delayed delivery (JEP-0091) element to the message.
                Element delay = message.addChildElement("x", "jabber:x:delay");
198
                delay.addAttribute("from", XMPPServer.getInstance().getServerInfo().getXMPPDomain());
199
                delay.addAttribute("stamp", dateFormat.format(creationDate));
200 201 202
                messages.add(message);
            }
            rs.close();
203 204 205
            // Check if the offline messages loaded should be deleted, and that there are
            // messages to delete.
            if (delete && !messages.isEmpty()) {
206 207 208 209 210 211 212 213 214 215
                pstmt.close();

                pstmt = con.prepareStatement(DELETE_OFFLINE);
                pstmt.setString(1, username);
                pstmt.executeUpdate();
                
                removeUsernameFromSizeCache(username);
            }
        }
        catch (Exception e) {
Gaston Dombiak's avatar
Gaston Dombiak committed
216
            Log.error("Error retrieving offline messages of username: " + username, e);
217 218 219
        }
        finally {
            // Return the sax reader to the pool
220 221 222 223
            if (xmlReader != null) {
                xmlReaders.add(xmlReader);
            }
            DbConnectionManager.closeConnection(pstmt, con);
224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239
        }
        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;
240
        ResultSet rs = null;
241 242 243 244 245 246 247 248
        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));
249
            rs = pstmt.executeQuery();
250 251
            while (rs.next()) {
                String msgXML = rs.getString(1);
252 253
                message = new OfflineMessage(creationDate,
                        xmlReader.read(new StringReader(msgXML)).getRootElement());
254 255
                // Add a delayed delivery (JEP-0091) element to the message.
                Element delay = message.addChildElement("x", "jabber:x:delay");
256
                delay.addAttribute("from", XMPPServer.getInstance().getServerInfo().getXMPPDomain());
257
                delay.addAttribute("stamp", dateFormat.format(creationDate));
258 259 260
            }
        }
        catch (Exception e) {
Gaston Dombiak's avatar
Gaston Dombiak committed
261 262
            Log.error("Error retrieving offline messages of username: " + username +
                    " creationDate: " + creationDate, e);
263 264 265
        }
        finally {
            // Return the sax reader to the pool
266 267 268 269
            if (xmlReader != null) {
                xmlReaders.add(xmlReader);
            }
            DbConnectionManager.closeConnection(rs, pstmt, con);
270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290
        }
        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) {
Gaston Dombiak's avatar
Gaston Dombiak committed
291
            Log.error("Error deleting offline messages of username: " + username, e);
292 293
        }
        finally {
294
            DbConnectionManager.closeConnection(pstmt, con);
295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321
        }
    }

    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();
            
322 323 324
            // Force a refresh for next call to getSize(username),
            // it's easier than loading the message to be deleted just
            // to update the cache.
325 326 327
            removeUsernameFromSizeCache(username);
        }
        catch (Exception e) {
Gaston Dombiak's avatar
Gaston Dombiak committed
328 329
            Log.error("Error deleting offline messages of username: " + username +
                    " creationDate: " + creationDate, e);
330 331
        }
        finally {
332
            DbConnectionManager.closeConnection(pstmt, con);
333 334 335 336 337 338 339 340 341 342 343 344 345
        }
    }

    /**
     * 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)) {
Gaston Dombiak's avatar
Gaston Dombiak committed
346
            return sizeCache.get(username);
347 348 349 350
        }
        int size = 0;
        Connection con = null;
        PreparedStatement pstmt = null;
351
        ResultSet rs = null;
352 353 354 355
        try {
            con = DbConnectionManager.getConnection();
            pstmt = con.prepareStatement(SELECT_SIZE_OFFLINE);
            pstmt.setString(1, username);
356
            rs = pstmt.executeQuery();
357 358 359 360 361 362 363 364 365 366
            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 {
367
            DbConnectionManager.closeConnection(rs, pstmt, con);
368 369 370 371 372 373 374 375 376 377 378 379 380 381
        }
        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;
382
        ResultSet rs = null;
383 384 385
        try {
            con = DbConnectionManager.getConnection();
            pstmt = con.prepareStatement(SELECT_SIZE_ALL_OFFLINE);
386
            rs = pstmt.executeQuery();
387 388 389 390 391 392 393 394
            if (rs.next()) {
                size = rs.getInt(1);
            }
        }
        catch (Exception e) {
            Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
        }
        finally {
395
            DbConnectionManager.closeConnection(rs, pstmt, con);
396 397 398 399
        }
        return size;
    }

400 401 402 403 404 405 406 407 408 409 410 411 412
    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
    }

413 414 415 416
    public void start() throws IllegalStateException {
        super.start();
        // Initialize the pool of sax readers
        for (int i=0; i<10; i++) {
417 418 419
            SAXReader xmlReader = new SAXReader();
            xmlReader.setEncoding("UTF-8");
            xmlReaders.add(xmlReader);
420
        }
421 422 423
        // Add this module as a user event listener so we can delete
        // all offline messages when a user is deleted
        UserEventDispatcher.addListener(this);
424 425 426 427 428 429
    }

    public void stop() {
        super.stop();
        // Clean up the pool of sax readers
        xmlReaders.clear();
430 431
        // Remove this module as a user event listener
        UserEventDispatcher.removeListener(this);
432 433
    }
}