destinationcache.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: DestinationCache.java,v 1.34 2003/12/29 13:09:15 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.Iterator;
import java.util.LinkedList;
import java.util.List;

import javax.transaction.TransactionManager;

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

import org.exolab.jms.Identifiable;
import org.exolab.jms.client.JmsDestination;
import org.exolab.jms.gc.GarbageCollectable;
import org.exolab.jms.lease.LeaseEventListenerIfc;
import org.exolab.jms.message.MessageHandle;
import org.exolab.jms.message.MessageImpl;
import org.exolab.jms.persistence.DatabaseService;
import org.exolab.jms.persistence.PersistenceException;
import org.exolab.jms.persistence.SQLHelper;
import org.exolab.jms.util.UUID;


/**
 * A DestinationCache is used to cache messages for a particular destination. 
 * <p>
 * It implements {@link MessageManagerEventListener} in order to be notified
 * of messages being added to and removed from 
 * {@link MessageMgr}
 * <p>
 * A {@link ConsumerEndpoint} registers with a {@link DestinationCache} to
 * receive messages for a particular destination.
 * <p>
 * In all instances this class doesn't deal with <code>Message</code> objects
 * directly, but instead uses their corresponding {@link MessageHandle},
 * which is far more lightweight.
 * <p>
 * A two level cache is used to facilitate quick seeding of registered
 * consumers and quick per-consumer-acknowledgment strategy. The two level
 * cache includes this, the DestinationCache, at the highest level and then
 * {@link ConsumerManager} at the lowest level.
 * <p>
 * In addition to registering {@link ConsumerEndpoint} objects the cache also
 * supports {@link DestinationCacheEventListener}s. A listener will be 
 * notified when messages are added to the cache but do not actually consume 
 * messages. A cache can have one or more registered listeners. This feature is
 * predominately used by browsers or iterators of destinations.
 * <p>
 * Clients can also become lifecycle listeners for this object to get notified
 * during initialization and shutdwon.
 * <p>
 * This cache is ordered on priority.
 *
 * @version     $Revision: 1.34 $ $Date: 2003/12/29 13:09:15 $
 * @author      <a href="mailto:[e-mail]">Jim Alateras</a>
 */
public abstract class DestinationCache
    implements MessageManagerEventListener, Identifiable,
    LeaseEventListenerIfc, GarbageCollectable {

    /**
     * The identity of this object
     */
    private String _id;

    /**
     * Create a message cache for this destination
     */
    private MessageCache _cache = new MessageCache();

    /**
     * Maintain the maximum size of this cache. When the cache reaches this
     * size then messages will be lost. We should only drop transient messages
     *  and maintain persistent messages. Remove transient messages from the
     * top of the cache and replace them with persistent messages.
     */
    private int _maximumSize = Integer.MAX_VALUE;

    /**
     * This is the list of consumers that have subscribed to this cache. It
     * hosts both durable and transient subscribers
     */
    protected List _consumers =
        Collections.synchronizedList(new LinkedList());

    /**
     * The message lease helper is used to manage leases for messages
     * cached by the destination
     */
    protected MessageLeaseHelper _leaseHelper = null;

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

    /**
     * Construct a message cache for a particular destination.
     */
    DestinationCache() {
        _id = UUID.next();
    }

    /**
     * Register this destination with the message manager. If it cannot
     * initialise then throw FailedToInitializeException
     */
    void init() throws FailedToInitializeException {
        // moved to after lease helper creation
        //MessageMgr.instance().addEventListener(getDestination(), this);

        try {
            _leaseHelper = new MessageLeaseHelper(this);
        } catch (PersistenceException exception) {
            String msg = "Failed to initialise destination cache";
            _log.error(msg, exception);
            throw new FailedToInitializeException(
                msg + ":" + exception.getMessage());
        }

        MessageMgr.instance().addEventListener(getDestination(), this);
    }

    /**
     * Register this destination with the message manager and create the
     * lease helper. The {@link MessageLeaseHelper} is initialized using the
     * specified {@link Connection}.
     * <p>
     * If there are problems with the initialization then throw
     * FailedToInitializeException
     *
     * @param connection - the connection to use
     * @throws FailedToInitializeException
     */
    void init(Connection connection)
        throws FailedToInitializeException {
        MessageMgr.instance().addEventListener(getDestination(), this);

        try {
            _leaseHelper = new MessageLeaseHelper(connection, this);
        } catch (Exception exception) {
            // rethrow
            throw new FailedToInitializeException("Error initialising " +
                exception.toString());
        }
    }

    /**
     * Set the maximum size of the cache. If there are more than this number
     * of messages in the cache the {@link CacheEvictionPolicy} is enforced
     * to remove messages.
     *
     * @param size - maximum number of messages a destination can hold
     */
    public void setMaximumSize(int size) {
        _maximumSize = size;
    }

    /**
     * Return the cache's maximum size
     *
     * @return int - size of cache
     */
    public int getMaximumSize() {
        return _maximumSize;
    }

    /**
     * Return a reference to the underlying destination
     *
     * @return JmsDestination
     */
    abstract public JmsDestination getDestination();

    /**
     * Return the string form of the destination
     *
     * @return  String
     */
    public String getDestinationByName() {
        return getDestination().getName();
    }

    /**
     * Set the {@link CacheEvictionPolicy} for this object. This determines
     * how messages are removed when the cache's upper limit is reached.
     *
     * @param policy the eviction policy
     */
    public void setCacheEvictionPolicy(CacheEvictionPolicy policy) {
        // not implemented
    }

    /**
     * Register a consumer with this cache. Part of the registration process
     * will be to seed the {@link ConsumerEndpoint} with an initial list of
     * messages that it needs to consume and then feed messages to it through
     * the specified listener object.
     * <p>
     * Messages are subsequently passed down to the consumer's through the
     * listener, as they enter the DestinationCache.
     *
     * @param consumer - message consumer for this destination
     * @return boolean - true if registered and false otherwise
     */
    public boolean registerConsumer(ConsumerEndpoint consumer) {

        boolean result = false;

        // check to see that the consumer is actually one for this
        // destination
        if (consumer.getDestination().equals(getDestination())) {
            if (!_consumers.contains(consumer)) {
                _consumers.add(consumer);
                consumer.setMaximumSize(this.getMaximumSize());
                result = true;
            }
        }

        return result;
    }

    /**
     * Remove the consumer for the list of registered consumers. If the
     * consumer does not exist then the call fails silently.
     *
     * @param consumer - consumer to remove.
     */
    public void unregisterConsumer(ConsumerEndpoint consumer) {
        if (_consumers.contains(consumer)) {
            _consumers.remove(consumer);
        } else {
        }
    }

    /**
     * Return an enmeration of all consumers attached to this cache.
     *
     * @return Iterator - list of registered consumers