topicdestinationcache.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: TopicDestinationCache.java,v 1.22 2003/10/21 14:41:24 tanderson Exp $
 *
 * Date         Author  Changes
 * 3/1/2001     jima    Created
 */
package org.exolab.jms.messagemgr;

import java.sql.Connection;
import java.util.Iterator;
import java.util.List;

import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.transaction.TransactionManager;

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

import org.exolab.jms.client.JmsDestination;
import org.exolab.jms.client.JmsTopic;
import org.exolab.jms.message.MessageHandle;
import org.exolab.jms.message.MessageImpl;
import org.exolab.jms.persistence.PersistenceException;


/**
 * A DestinationCache for Topics. This cache extends DestinationCache but does
 * not actually hold a reference to the messages. Instead it forwards them on
 * to registered consumers.
 * <p>
 * We may need to build the cache for clients that fail over to the new server
 * so we should maintain a cache of at least persistent messages handles. This
 * is something that needs to be considered. As for non-persistent messages well
 * that is the penalty for using them. If you want to ensure that you get every
 * message, even during failover then you best publsih them using persistent
 * delivery mode.
 *
 * @version     $Revision: 1.22 $ $Date: 2003/10/21 14:41:24 $
 * @author      <a href="mailto:[e-mail]">Jim Alateras</a>
 **/
public class TopicDestinationCache
    extends DestinationCache {

    /**
     * Underlying destination
     */
    private JmsTopic _topic = null;

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


    /**
     * Construct a message cache for a particular destination. This object
     * does not cache any messages.
     *
     * @param destination - the destination that owns this cache
     * @throws FailedToInitializeException
     */
    TopicDestinationCache(JmsTopic destination)
        throws FailedToInitializeException {
        super();

        // don't need the local messages cache for the initial release, will need
        // it for clustering
        _topic = destination;

        // call init on the base class
        try {
            init();
        } catch (FailedToInitializeException exception) {
            // rethrow
            throw exception;
        } catch (Exception exception) {
            throw new FailedToInitializeException(
                "Failed to construct TopicDestinationCache " +
                exception.toString());
        }
    }

    /**
     * Construct a message cache for a particular destination using the specified
     * {@link Connection} to complete any database access.
     *
     * @param connection - the connection to use.
     * @param destination - the destination that owns this cache
     * @throws FailedToInitializeException
     */
    TopicDestinationCache(Connection connection, JmsTopic destination)
        throws FailedToInitializeException {
        super();

        // don't need the local messages cache for the initial release, will need
        // it for clustering
        _topic = destination;

        // call init on the base class
        try {
            init(connection);
        } catch (FailedToInitializeException exception) {
            // rethrow
            throw exception;
        } catch (Exception exception) {
            throw new FailedToInitializeException(
                "Failed to construct TopicDestinationCache " +
                exception.toString());
        }
    }

    // implementation of DestinationCache.getDestination
    public JmsDestination getDestination() {
        return _topic;
    }

    // implementation of MessageMgr.messageAdded
    public boolean messageAdded(JmsDestination destination, MessageImpl message) {
        boolean processed = false;

        if ((destination != null) &&
            (message != null)) {
            // check that it is not already present before adding it.
            if (destination.equals(_topic)) {
                processed = notifyOnAddMessage(message);
                // create a lease iff one is required and the message has actually
                // been accepted by at least one endpoint
                if (processed) {
                    checkMessageExpiry(message);
                }
            }
        }

        return processed;
    }

    // implementation of MessageMgr.messageRemoved
    public void messageRemoved(JmsDestination destination, MessageImpl message) {
        if ((destination != null) &&
            (message != null)) {
            // call remove regardless whether it exists
            if (destination.equals(_topic)) {
                notifyOnRemoveMessage(message);
            }
        }
    }

    // implementation of MessageMgr.persistentMessageAdded
    public boolean persistentMessageAdded(Connection connection,
                                          JmsDestination destination, MessageImpl message)
        throws PersistenceException {
        boolean processed = false;

        if ((destination != null) &&
            (message != null)) {
            // check that it is not already present before adding it.
            if (destination.equals(_topic)) {
                processed = notifyOnAddPersistentMessage(connection, message);

                // create a lease iff one is required and the message has actually
                // been accepted by at least one endpoint
                if (processed) {
                    checkMessageExpiry(message);
                }
            }
        }

        return processed;
    }

    // implementation of MessageMgr.persistentMessageRemoved
    public void persistentMessageRemoved(Connection connection,
                                         JmsDestination destination, MessageImpl message)
        throws PersistenceException {
        if ((destination != null) &&
            (message != null)) {
            // call remove regardless whether it exists
            if (destination.equals(_topic)) {
                notifyOnRemovePersistentMessage(connection, message);
            }
        }
    }

    // implementation of DestinationCache.notifyOnAddMessage
    boolean notifyOnAddMessage(MessageImpl message) {
        boolean processed = true;

        // process for all the active consumers.
        Object[] iter = getConsumersByArray();
        for (int index = 0; index < iter.length; index++) {
            DestinationCacheEventListener listener =
                (DestinationCacheEventListener) iter[index];
            processed |= listener.messageAdded(message);
        }

        return processed;
    }

    // implementation of DestinationCache.notifyOnRemoveMessage
    void notifyOnRemoveMessage(MessageImpl message) {
        Object[] iter = getConsumersByArray();
        for (int index = 0; index < iter.length; index++) {
            if (((DestinationCacheEventListener)
                iter[index]).messageRemoved(message)) {
            }
        }
    }

    // implementation of DestinationCache.notifyOnAddPersistentMessage
    boolean notifyOnAddPersistentMessage(Connection connection,
                                         MessageImpl message)
        throws PersistenceException {
        boolean processed = true;

        // This is not done in MessageMgr
        // first let's create a persistent message handle for all registered
        // durable consumers
        //try {
        //    processed |= ConsumerManager.instance().persistentMessageAdded(
        //        connection, message);
        //} catch (Exception exception) {
        //    throw new PersistenceException("Error in notifyOnAddPersistentMessage " +
        //                                   exception);
        //}

        // now send the message to all active durable and non-durable consumers
        Object[] iter = getConsumersByArray();
        for (int index = 0; index < iter.length; index++) {
            DestinationCacheEventListener listener =
                (DestinationCacheEventListener) iter[index];
            processed |= listener.persistentMessageAdded(connection, message);
        }

        return processed;
    }

    // implementation of DestinationCache.notifyOnRemovePersistentMessage
    void notifyOnRemovePersistentMessage(Connection connection,
                                         MessageImpl message)
        throws PersistenceException {

        Object[] iter = getConsumersByArray();
        for (int index = 0; index < iter.length; index++) {
            DestinationCacheEventListener listener =
                (DestinationCacheEventListener) iter[index];
            // if the listener is of type {@link DurableConsumerEndpoint} then
            // indicate that a persistentMessage has been removed. If the listener
            // is a {@link TopicConsumerEndpoint} then indicatw that non-
            // persistent message has been removed.
            if (listener instanceof DurableConsumerEndpoint) {
                listener.persistentMessageRemoved(connection, message);
            } else if (listener instanceof TopicConsumerEndpoint) {
                listener.messageRemoved(message);
            }
        }

        // since it is a persistent message we need to send it
        // to inactive durable consumers subscribing to the destination
        try {
            ConsumerManager.instance().persistentMessageRemoved(connection,
                message);
        } catch (Exception exception) {
            _log.error("Error in notifyOnRemovePersistentMessage",
                exception);
        }
    }

    // override implementation of DestinationCache.registerConsumer
    public boolean registerConsumer(ConsumerEndpoint consumer) {

        boolean result = false;

        // check to see that the consumer can actually subscribe to
        // this destination
        JmsTopic cdest = (JmsTopic) consumer.getDestination();
        JmsTopic ddest = (JmsTopic) getDestination();

        if (cdest.match(ddest)) {
            if (!_consumers.contains(consumer)) {
                _consumers.add(consumer);
                consumer.setMaximumSize(this.getMaximumSize());
                result = true;
            }
        }

        return result;
    }

    // implementation of DestinationCache.hasActiveConsumers
    boolean hasActiveConsumers() {
        return (_consumers.size() > 0);
    }

    // implementation of DestinationCache.getMessageCount
    public int getMessageCount() {
        return 0;
    }

    /**
     * Determines if this cache can be destroyed.
     * A <code>TopicDestinationCache</code> can be destroyed if there
     * are no active consumers.
     *
     * @return <code>true</code> if the cache can be destroyed, otherwise
     * <code>false</code>
     */
    public boolean canDestroy() {
        return !hasActiveConsumers();
    }

    // override Object.toString
    public String toString() {
        return _topic.toString();
    }

    // override Object.hashCode
    public int hashCode() {
        return _topic.hashCode();
    }

    /**
     * Resolve an expired message through its handle
     *
     * @param handle the expired message's handle
     * @return the expired message. May be null.
     */
    protected MessageImpl resolveExpiredMessage(MessageHandle handle) {
        MessageImpl message = null;

        if (handle.getConsumerName() != null) {
            message = super.resolveExpiredMessage(handle);
        } else {
            // can't resolve the message through the handle 
            // Need to find the first consumer which has it.
            // @todo - design flaw!
            Iterator iterator = _consumers.iterator();
            while (iterator.hasNext()) {
                ConsumerEndpoint endpoint = (ConsumerEndpoint) iterator.next();
                message = endpoint.getMessage(handle);
                if (message != null) {
                    break;
                }
            }
        }
        return message;
    }

}