messagecache.java

实现了Jms的服务器源码,支持多种适配器,DB,FTP,支持多种数据库

/**
 * Redistribution and use of this software and associated documentation
 * ("Software"), with or without modification, are permitted provided
 * that the following conditions are met:
 *
 * 1. Redistributions of source code must retain copyright
 *    statements and notices.  Redistributions must also contain a
 *    copy of this document.
 *
 * 2. Redistributions in binary form must reproduce the
 *    above copyright notice, this list of conditions and the
 *    following disclaimer in the documentation and/or other
 *    materials provided with the distribution.
 *
 * 3. The name "Exolab" must not be used to endorse or promote
 *    products derived from this Software without prior written
 *    permission of Exoffice Technologies.  For written permission,
 *    please contact [e-mail].
 *
 * 4. Products derived from this Software may not be called "Exolab"
 *    nor may "Exolab" appear in their names without prior written
 *    permission of Exoffice Technologies. Exolab is a registered
 *    trademark of Exoffice Technologies.
 *
 * 5. Due credit should be given to the Exolab Project
 *    (http://www.exolab.org/).
 *
 * THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
 * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
 * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
 * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
 * EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
 * OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * Copyright 2001-2003 (C) Exoffice Technologies Inc. All Rights Reserved.
 *
 * $Id: MessageCache.java,v 1.9 2003/12/29 13:09:25 tanderson Exp $
 *
 * Date         Author  Changes
 * 3/1/2001     jima    Created
 */
package org.exolab.jms.messagemgr;

import java.sql.Connection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import javax.jms.JMSException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.exolab.jms.message.MessageHandle;
import org.exolab.jms.message.MessageId;
import org.exolab.jms.message.MessageImpl;
import org.exolab.jms.persistence.DatabaseService;
import org.exolab.jms.persistence.PersistenceAdapter;
import org.exolab.jms.persistence.PersistenceException;


/**
 * The MessageCache is responsible for managing the collection of persistent
 * messages for a particular {@link DestinationCache} or {@link Endpoint}.
 * <p>
 * Persistent messages and transient messages are stored in different caches
 * since they behave slightly different. Persistent messages can be passivated
 * to disk when memory is low.
 * <p>
 * We leave garbage collection of messages to the Java Garbage Collection
 * mechanism.
 *
 * @version     $Revision: 1.9 $ $Date: 2003/12/29 13:09:25 $
 * @author      <a href="mailto:[e-mail]">Jim Alateras</a>
 **/
class MessageCache {

    /**
     * Maintain the pool of transient messages.
     */
    private transient Map _transient =
        Collections.synchronizedMap(new HashMap(1023));

    /**
     * Maintain the pool of persistent messages.
     */
    private transient Map _persistent =
        Collections.synchronizedMap(new HashMap(1023));

    /**
     * Maintains a list of {@link MessageHandles}, both transient and
     * persistent. The messages are sorted based on priority
     */
    private MessageQueue _handles =
        new MessageQueue(new MessageHandleComparator());

    /**
     * The logger
     */
    private static final Log _log = LogFactory.getLog(MessageCache.class);


    /**
     * Default constructor
     */
    MessageCache() {
    }

    /**
     * Add the following message to the cache
     *
     * @param id - the message identity
     * @param message - the message to add
     */
    void addMessage(MessageId id, MessageImpl message) {
        if (message.isPersistent()) {
            addPersistentMessage(message);
        } else {
            addNonPersistentMessage(message);
        }
    }

    /**
     * Add the following handle and associated message to the cache
     *
     * @param handle - the message handle to add
     * @param message - the message to add
     */
    void addMessage(MessageHandle handle, MessageImpl message) {
        addMessage(handle.getMessageId(), message);
        addHandle(handle);
    }

    /**
     * Add the following message handle to the cache
     *
     * @param handle - the message handle to add
     */
    void addHandle(MessageHandle handle) {
        _handles.add(handle);
    }

    /**
     * Remove the message with the following id from the cache and return
     * it to the client.
     *
     * @param id - the id of the message to remove
     * @param MessageImpl - the removed message or null
     */
    MessageImpl removeMessage(MessageId id) {
        MessageImpl message = removePersistentMessage(id);
        if (message == null) {
            message = removeNonPersistentMessage(id);
        }

        return message;
    }

    /**
     * Remove the specified handle from the cache
     *
     * @param handle - the handle to remove
     * @return boolean - true if successful
     */
    boolean removeHandle(MessageHandle handle) {
        return _handles.remove(handle);
    }

    /**
     * Remove and return the first message handle in the cache
     *
     * @return MessageHandle - the first handle or null if cache is empty
     */
    final MessageHandle removeFirstHandle() {
        return (MessageHandle) _handles.removeFirst();
    }

    /**
     * Return an array of all handles in the cache
     *
     * @return Object[]
     */
    Object[] getHandleArray() {
        return _handles.toArray();
    }

    /**
     * Check whether the specified handle is cache
     *
     * @param handle - the handle to check
     * @return boolean - true if it is
     */
    final boolean containsHandle(MessageHandle handle) {
        return _handles.contains(handle);
    }

    /**
     * Return the message corresponding to the specified handle
     *
     * @param handle - the handle to check
     * @param MessageImpl - the associated message or null
     */
    MessageImpl getMessage(final MessageHandle handle) {
        if (handle == null) {
            return null;
        }

        MessageImpl message = null;
        MessageId id = handle.getMessageId();

        if (handle.isPersistent()) {
            message = (MessageImpl) _persistent.get(id);

            // if the message is not cached then try and retrieve it from the
            // database and cache it.
            if (message == null) {
                // fault in at least the next message from the database
                PersistentMessageHandle phandle = null;
                if (handle instanceof PersistentMessageHandle) {
                    phandle = (PersistentMessageHandle) handle;
                } else {
                    // this is a transient handle holding a reference to
                    // a persistent message. Create a persistent handle
                    // from it
                    try {
                        phandle = (PersistentMessageHandle)
                            MessageHandleFactory.createPersistentHandle(
                                (TransientMessageHandle) handle);
                    } catch (Exception exception) {
                        _log.error("Failed to retrieve message", exception);
                    }
                }

                // if we have a phandle then retrieve it from the database
                if (phandle != null) {
                    PersistenceAdapter adapter = DatabaseService.getAdapter();
                    Connection connection = null;
                    try {
                        connection = DatabaseService.getConnection();
                        message = adapter.getMessage(
                            connection, phandle.getMessageId().getId());
                        connection.commit();
                    } catch (PersistenceException exception) {
                        if (connection != null) {
                            try {
                                connection.rollback();
                            } catch (Exception nested) {
                                // ignore
                            }
                        }
                        _log.error("Failed to retrieve message", exception);
                    } catch (Exception exception) {
                        _log.error("Failed to retrieve message", exception);
                    } finally {
                        if (connection != null) {
                            try {
                                connection.close();
                            } catch (Exception nested) {
                                // ignore
                            }
                        }
                    }

                    // add the message to the persistent cache once it has been
                    // retrieved from the datastore
                    if (message != null) {
                        synchronized (_persistent) {
                            _persistent.put(id, message);
                        }
                    }
                }
            }
        } else {
            message = (MessageImpl) _transient.get(id);
        }

        if ((message != null) &&
            (!message.getReadOnly())) {
            // mark the message as read-only. If this fails, log the
            // error and return null.
            try {
                message.setReadOnly(true);
            } catch (JMSException exception) {
                _log.error(exception);
                message = null;
            }
        }

        return message;
    }

    /**
     * Clear the persistent and non-persistent message cache
     */
    void clear() {
        _transient.clear();
        _persistent.clear();
        _handles.clear();
    }

    /**
     * Clear only the persistent messages in the cache
     *
     */
    void clearPersistentMessages() {
        _persistent.clear();
    }

    /**
     * Clear only the transient messages in the cache
     *
     */
    void clearTransientMessages() {
        _transient.clear();
    }

    /**
     * Add a non-persistent message to the cache
     *
     * @param message - the message to add
     */
    private void addNonPersistentMessage(MessageImpl message) {
        synchronized (_transient) {
            _transient.put(message.getMessageId(), message);
        }
    }

    /**
     * Remove the non persistnet message with the specified identity from the
     * transient cache.
     *
     * @param id - the id of the message to remove
     * @return MessageImpl - the remove message or null
     */
    private MessageImpl removeNonPersistentMessage(MessageId id) {
        MessageImpl message = null;
        synchronized (_transient) {
            message = (MessageImpl) _transient.remove(id);
        }

        return message;
    }

    /**
     * Remove the  persistnet message with the specified identity from the
     * persistent cache.
     *
     * @param id - the id of the message to remove
     * @return MessageImpl - the remove message or null
     */
    private MessageImpl removePersistentMessage(MessageId id) {
        MessageImpl message = null;
        synchronized (_persistent) {
            message = (MessageImpl) _persistent.remove(id);
        }

        return message;
    }

    /**
     * Add persistent message to the persistent cache
     *
     * @param message - the message to add
     */
    private void addPersistentMessage(MessageImpl message) {
        synchronized (_persistent) {
            _persistent.put(message.getMessageId(), message);
        }
    }

    /**
     * Return the number of messages in the transient cache
     *
     * @return int
     */
    public int getTransientCount() {
        return _transient.size();
    }

    /**
     * Return the number of messages in the persistent cache
     *
     * @return int
     */
    public int getPersistentCount() {
        return _persistent.size();
    }

    /**
     * Return the number of handles in the cache
     *
     * @return int
     */
    public int getHandleCount() {
        return _handles.size();
    }
}