destinationmanagerimpl.java

来自「OpenJMS是一个开源的Java Message Service API 1.」· Java 代码 · 共 788 行 · 第 1/2 页

/**
 * 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-2005 (C) Exoffice Technologies Inc. All Rights Reserved.
 *
 * $Id: DestinationManagerImpl.java,v 1.2 2005/11/12 10:49:48 tanderson Exp $
 */
package org.exolab.jms.messagemgr;

import java.sql.Connection;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.List;
import java.util.ArrayList;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;

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

import org.exolab.jms.client.JmsDestination;
import org.exolab.jms.client.JmsQueue;
import org.exolab.jms.client.JmsTopic;
import org.exolab.jms.gc.GarbageCollectionService;
import org.exolab.jms.message.MessageImpl;
import org.exolab.jms.persistence.DatabaseService;
import org.exolab.jms.persistence.PersistenceAdapter;
import org.exolab.jms.persistence.PersistenceException;
import org.exolab.jms.service.Service;
import org.exolab.jms.service.ServiceException;


/**
 * The destination manager is responsible for creating and managing the
 * lifecycle of {@link DestinationCache} objects. The destination manager is
 * also responsible for managing messages, that are received by the message
 * manager, which do not have any registered {@link DestinationCache}.
 *
 * @author <a href="mailto:[e-mail]">Jim Alateras</a>
 * @author <a href="mailto:[e-mail]">Tim Anderson</a>
 * @version $Revision: 1.2 $ $Date: 2005/11/12 10:49:48 $
 */
public class DestinationManagerImpl extends Service
        implements DestinationManager {

    /**
     * The set of persistent and non-persistent destinations, keyed on name.
     */
    private final HashMap _destinations = new HashMap();

    /**
     * The set of active DestinationCache instances, keyed on destination.
     */
    private final HashMap _caches = new HashMap();

    /**
     * Synchronization helper. Should be synchronized on whenever accessing
     * _destinations, or _caches
     */
    private final Object _lock = _destinations;

    /**
     * Maintains a linked list of DestinationEventListener objects. These
     * listeners will be informed when destinations are added or destroyed.
     */
    private LinkedList _listeners = new LinkedList();

    /**
     * The message manager.
     */
    private final MessageManager _messages;

    /**
     * The destination cache factory.
     */
    private final DestinationCacheFactory _factory;

    /**
     * The consumer manager.
     */
    private ConsumerManager _consumers;

    /**
     * The database service.
     */
    private final DatabaseService _database;

    /**
     * The garbage collection service.
     */
    private final GarbageCollectionService _collector;

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

    /**
     * Construct a new <code>DestinationManagerImpl</code>.
     *
     * @param messages  the message manager
     * @param factory   the destination cache factory
     * @param database  the database service
     * @param collector the garbage collection service
     */
    public DestinationManagerImpl(MessageManager messages,
                                  DestinationCacheFactory factory,
                                  DatabaseService database,
                                  GarbageCollectionService collector) {
        if (messages == null) {
            throw new IllegalArgumentException("Argument 'messages' is null");
        }
        if (factory == null) {
            throw new IllegalArgumentException("Argument 'factory' is null");
        }
        if (database == null) {
            throw new IllegalArgumentException("Argument 'database' is null");
        }
        if (collector == null) {
            throw new IllegalArgumentException("Argument 'collector' is null");
        }
        _messages = messages;
        _factory = factory;
        _database = database;
        _collector = collector;
    }

    /**
     * Sets the consumer manager.
     *
     * @param consumers the consumer manager
     */
    public void setConsumerManager(ConsumerManager consumers) {
        _consumers = consumers;
    }

    /**
     * Returns the cache for the supplied destination.
     * <p/>
     * If the cache doesn't exist, it will be created, and any registered {@link
     * DestinationEventListener}s will be notified.
     *
     * @param destination the destination of the cache to return
     * @return the cache associated with <code>destination</code>
     * @throws InvalidDestinationException if <code>destination</code> doesn't
     *                                     exist
     * @throws JMSException                if the cache can't be created
     */
    public DestinationCache getDestinationCache(JmsDestination destination)
            throws JMSException {
        DestinationCache result;
        boolean created = false;

        synchronized (_lock) {
            final String name = destination.getName();

            // make sure the managed destination instance is used.
            destination = getExistingDestination(name);

            result = (DestinationCache) _caches.get(destination);
            if (result == null) {
                checkWildcard(destination);
                result = _factory.createDestinationCache(destination);
                _caches.put(destination, result);
                _messages.addEventListener(destination, result);
                created = true;
            }
        }

        if (created) {
            // notify the listeners that a new cache has been added,
            // outside the sync of _lock
            notifyCacheAdded(result);
        }

        return result;
    }

    /**
     * Returns a destination given its name.
     *
     * @param name the name of the destination
     * @return the destination corresponding to <code>name</code> or
     *         <code>null</code> if none exists
     */
    public JmsDestination getDestination(String name) {
        synchronized (_lock) {
            return (JmsDestination) _destinations.get(name);
        }
    }

    /**
     * Register an event listener to be notified when destinations are created
     * and destroyed.
     *
     * @param listener the listener to add
     */
    public void addDestinationEventListener(DestinationEventListener listener) {
        synchronized (_listeners) {
            if (!_listeners.contains(listener)) {
                _listeners.add(listener);
            }
        }
    }

    /**
     * Remove an event listener.
     *
     * @param listener the listener to remove
     */
    public void removeDestinationEventListener(
            DestinationEventListener listener) {
        synchronized (_listeners) {
            _listeners.remove(listener);
        }
    }

    /**
     * Create a destination.
     * <p/>
     * Any registered {@link DestinationEventListener}s will be notified.
     *
     * @param destination the destination to create
     * @throws InvalidDestinationException if the destination already exists or
     *                                     is a wildcard destination
     * @throws JMSException                if the destination can't be created
     */
    public void createDestination(JmsDestination destination)
            throws JMSException {
        checkWildcard(destination);
        synchronized (_lock) {
            if (exists(destination.getName())) {
                throw new InvalidDestinationException(
                        "Destination already exists: " + destination.getName());
            }
            if (destination.getPersistent()) {
                createPersistentDestination(destination);
            }
            addToDestinations(destination);
        }

        notifyDestinationAdded(destination);
    }

    /**
     * Remove a destination.
     * <p/>
     * All messages and durable consumers will be removed. Any registered {@link
     * DestinationEventListener}s will be notified.
     *
     * @param destination the destination to remove
     * @throws InvalidDestinationException if the destination is invalid
     * @throws JMSException                if the destination can't be removed
     */
    public void removeDestination(JmsDestination destination)
            throws JMSException {

        if (_log.isDebugEnabled()) {
            _log.debug("removeDestination(destination=" + destination + ")");
        }

        // make sure the managed destination instance is used.
        destination = getExistingDestination(destination.getName());

        boolean queue = (destination instanceof JmsQueue) ? true : false;

        if (!queue) {
            // If its a topic, unsubscribe any inactive durable subscribers.
            // The following will fail if there are active subscribers
            _consumers.unsubscribe((JmsTopic) destination);
        }

        synchronized (_lock) {
            DestinationCache cache =
                    (DestinationCache) _caches.get(destination);
            // make sure there are no consumers
            if (cache != null && cache.hasConsumers()) {
                throw new JMSException("Cannot delete destination"
                                       + destination + " since there are "
                                       + " active consumers.");
            }
        }

        // now that we have removed all the durable consumers we can remove
        // the administered topic. First delete it from memory and then
        // from the persistent store
        try {
            _database.begin();
            Connection connection = _database.getConnection();

            _database.getAdapter().removeDestination(connection,
                                                     destination.getName());
            destroyDestinationCache(destination);
            removeFromDestinations(destination);
            _database.commit();
        } catch (Exception exception) { // JMSException, PersistenceException
            String msg = "Failed to remove destination "
                    + destination.getName();
            cleanup(msg, exception);
        }

        notifyDestinationRemoved(destination);
    }

    /**
     * Invoked when the {@link MessageManager} receives a non-persistent
     * message.
     *
     * @param destination the message's destination
     * @param message     the message
     * @throws JMSException if the listener fails to handle the message
     */
    public void messageAdded(JmsDestination destination,
                             MessageImpl message)
            throws JMSException {
        if (destination instanceof JmsTopic) {
            // check to see whether there are active consumers interested
            // in the specified destination. If there are then we need to
            // create a destination cache and pass the message to it.
            if (_consumers.hasActiveConsumers(destination)) {
                if (!exists(destination.getName())) {
                    createDestination(destination);
                }
                DestinationCache cache = getDestinationCache(destination);
                cache.messageAdded(destination, message);
            }
        } else {
            // destination is a queue. Since the message is non-persistent,
            // create the cache and pass the message to it.
            if (!exists(destination.getName())) {
                createDestination(destination);
            }
            DestinationCache cache = getDestinationCache(destination);
            cache.messageAdded(destination, message);
        }
    }

    /**
     * Invoked when the {@link MessageManager} receives a persistent message.
     *
     * @param destination the message's destination
     * @param message     the message
     * @throws JMSException         if the listener fails to handle the message
     * @throws PersistenceException if there is a persistence related problem
     */
    public void persistentMessageAdded(JmsDestination destination,
                                       MessageImpl message)
            throws JMSException, PersistenceException {
        DestinationCache cache = getDestinationCache(destination);
        cache.persistentMessageAdded(destination, message);
    }

    /**