/***
    * 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 info@exolab.org.
   *
   * 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: DestinationManager.java,v 1.36 2003/08/16 06:37:43 tanderson Exp $
   *
   * Date         Author  Changes
   * 3/1/2001     jima    Created
   */
  package org.exolab.jms.messagemgr;
  
  import java.sql.Connection;
  import java.sql.SQLException;
  import java.util.Collections;
  import java.util.Enumeration;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.LinkedList;
  import java.util.Map;
  import java.util.Vector;
  
  import javax.jms.DeliveryMode;
  import javax.jms.JMSException;
  import javax.naming.Context;
  import javax.naming.InitialContext;
  import javax.naming.NamingException;
  import javax.transaction.TransactionManager;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import org.exolab.core.service.ServiceException;
  import org.exolab.jms.client.JmsDestination;
  import org.exolab.jms.client.JmsQueue;
  import org.exolab.jms.client.JmsTopic;
  import org.exolab.jms.config.AdministeredDestinations;
  import org.exolab.jms.config.AdministeredQueue;
  import org.exolab.jms.config.AdministeredTopic;
  import org.exolab.jms.config.Configuration;
  import org.exolab.jms.config.ConfigurationManager;
  import org.exolab.jms.config.MessageManagerConfiguration;
  import org.exolab.jms.config.Subscriber;
  import org.exolab.jms.gc.GarbageCollectable;
  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.persistence.SQLHelper;
  import org.exolab.jms.server.NamingHelper;
  
  
  // @todo - need to restructure dirs
  
  
  /***
   * 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}
   *
  * @version     $Revision: 1.36 $ $Date: 2003/08/16 06:37:43 $
  * @author      <a href="mailto:jima@exoffice.com">Jim Alateras</a>
  */
 public class DestinationManager
     implements MessageManagerEventListener, GarbageCollectable {
 
     /***
      * This structure maintains a list of active caches.
      */
     private Map _caches = Collections.synchronizedMap(new HashMap());
 
     /***
      * A list of administered and non-administered destinations are maintained
      * in this data structure.
      */
     private HashMap _destinationCache = new HashMap();
 
     /***
      * The maximum no. of messages that each destination cache can hold
      */
     private final int _maxCacheSize;
 
     /***
      * Maintains a list of wildcard destinations, which can either be durable
      * or transient
      */
     private LinkedList _wildcardDestinations = new LinkedList();
 
     /***
      * Maintains a linked list of DestinationEventListener objects. These
      * listeners will be informated when destination are added or destroyed
      */
     private LinkedList _listeners = new LinkedList();
 
     /***
      * Manage the singleton instance of the DestinationManager
      */
     private static volatile DestinationManager _instance = null;
 
     /***
      * Caches the root context for all jndi binding.
      */
     private InitialContext _context = null;
 
     /***
      * The logger
      */
     private static final Log _log =
         LogFactory.getLog(DestinationManager.class);
 
     /***
      * Construct a new <code>DestinationManager</code>
      *
      * @throws ServiceException if the service cannot be initialised
      */
     private DestinationManager() throws ServiceException {
         MessageManagerConfiguration config =
             ConfigurationManager.getConfig().getMessageManagerConfiguration();
         _maxCacheSize = config.getDestinationCacheSize();
 
         init();
 
         // register with the GarbageCollectionService
         GarbageCollectionService.instance().register(this);
     }
 
     /***
      * Create the singleton instance of the destination manager
      *
      * @return the singleton instance
      * @throws ServiceException if the service cannot be initialised
      */
     public static DestinationManager createInstance() throws ServiceException {
         _instance = new DestinationManager();
         return _instance;
     }
 
     /***
      * Return the singleton destination manager
      *
      * @return the singleton instance, or <code>null</code> if it hasn't
      * been initialised
      */
     public static DestinationManager instance() {
         return _instance;
     }
 
     /***
      * Create a destination of the specified type. If the destination already
      * exists then simply return it. If it doesn't exist then create it.
      *
      * @param dest - the destination to create
      * @return DestinationCache - the created destination cache
      */
     public synchronized DestinationCache createDestinationCache(
         JmsDestination dest) {
         return createDestinationCache(null, dest);
     }
 
     /***
      * Create a destination of the specified type. If the destination already
      * exists then simply return it. If it doesn't exist then create it. If
      * a connection is supplied then create the cache using the connection
      * context, otherwise create it without a context.
      *
      * @param connection - the connection to use.
      * @param dest - the destination to create
      * @return DestinationCache - the created destination cache
      */
     public synchronized DestinationCache createDestinationCache(
         Connection connection, JmsDestination dest) {
 
         // check to see if it exists first
         DestinationCache cache = (DestinationCache) _caches.get(dest);
 
         if (cache == null) {
             // create a destination based on its type
             try {
                 if (dest instanceof JmsTopic) {
                     cache = (connection != null) ?
                         new TopicDestinationCache(connection, (JmsTopic) dest) :
                         new TopicDestinationCache((JmsTopic) dest);
                 } else if (dest instanceof JmsQueue) {
                     cache = (connection != null) ?
                         new QueueDestinationCache(connection, (JmsQueue) dest) :
                         new QueueDestinationCache((JmsQueue) dest);
                 }
                 // set the configured size of each destination cache
                 cache.setMaximumSize(_maxCacheSize);
 
                 // notify the listeners that a new destination has been added
                 // to the destination manager
                 notifyDestinationAdded(dest, cache);
                 //cache it first
                 _caches.put(dest, cache);
             } catch (Exception exception) {
                 _log.error("Failed to createDestinationCache", exception);
             }
         }
 
         return cache;
     }
 
     /***
      * Delete the specfied destination
      *
      * @param cahe - the destination to destroy
      */
     public synchronized void destroyDestinationCache(DestinationCache cache) {
         destroyDestinationCache(cache.getDestination());
     }
 
     /***
      * Delete the specfied destination
      *
      * @deprecated use destroyDestination(JmsDestination) instead
      * @param name - destination name
      */
     public synchronized void destroyDestinationCache(String name) {
 
         // currently we are doing a linear search through the
         // destinations until we find the corresponding {@link JmsDestination}
         // We want to discourage the use of this function.
         Iterator iter = _caches.keySet().iterator();
         while (iter.hasNext()) {
             JmsDestination dest = (JmsDestination) iter.next();
             if (dest.getName().equals(name)) {
                 destroyDestinationCache(dest);
                 break;
             }
         }
     }
 
     /***
      * Delete the specfied destination
      *
      * @param dest - the destination to destroy
      */
     public synchronized void destroyDestinationCache(JmsDestination dest) {
         DestinationCache cache = (DestinationCache) _caches.remove(dest);
         if (cache != null) {
             cache.destroy();
 
             // notify the listeners that a destination has been removed from
             // the destination manager
             notifyDestinationRemoved(dest, cache);
         }
     }
 
     /***
      * Return the JmsDestination corresponding to the specified destination
      * name
      *
      * @param name - destination name
      * @return JmsDestination - the corresponding object or null
      */
     public synchronized JmsDestination destinationFromString(String name) {
         return (JmsDestination) _destinationCache.get(name);
     }
 
     /***
      * Register the specified DestinationEventListener. If the listener is
      * already registered then do not re-register it again.
      *
      * @param listener - listener to add
      */
     void addDestinationEventListener(DestinationEventListener listener) {
         synchronized (_listeners) {
             if (!_listeners.contains(listener)) {
                 _listeners.add(listener);
             }
 
         }
     }
 
     /***
      * Remove the specified DestinationEventListener from the list
      *
      * @param listener - listener to remove
      */
     void removeDestinationEventListener(DestinationEventListener listener) {
         synchronized (_listeners) {
             _listeners.remove(listener);
         }
     }
 
     /***
      * Return the destination cache associated with the dest object
      *
      * @param dest - the destination
      * @return DestinationCache - associated destination object or null
      */
     public DestinationCache getDestinationCache(JmsDestination dest) {
         return (DestinationCache) _caches.get(dest);
     }
 
     /***
      * Return the destination object associated with destination
      *
      * @param dest - the name of the destination
      * @return DestinationCache - associated destination object or null
      */
     public DestinationCache getDestinationCache(String dest) {
         return getDestinationCache(destinationFromString(dest));
     }
 
     /***
      * Return the {@link DestinationCache} for this message.
      *
      * @param message - the message to query
      * @return DestinationCache - the corresponding cahce or null
      */
     public DestinationCache getDestinationCache(MessageImpl message) {
         DestinationCache cache = null;
 
         if (message != null) {
             try {
                 cache = getDestinationCache(
                     (JmsDestination) message.getJMSDestination());
             } catch (JMSException exception) {
                 // ignore it, probably means thant no destination has been
                 // assigned.
             }
         }
 
         return cache;
     }
 
     /***
      * Check if the specified destination exists.
      *
      * @param dest - destination to check
      * @return boolean - true if it exists
      */
     public boolean hasDestinationCache(JmsDestination dest) {
         return (_caches.containsKey(dest));
     }
 
     /***
      * Create a non-administered destination and cache it. It does not
      * check to see whether or not it is an administered destination
      * this  must be done by the caller
      *
      * @param destination  - the destination to create
      */
     public synchronized void createDestination(JmsDestination destination) {
         addToDestinationCache(destination);
     }
 
     /***
      * Create an administered destination using the specified destination.
      * It will create the destination in the database and register it with
      * the jndi context.
      *
      * @param dest - the destination
      * @return boolean - true if successful
      */
     public synchronized boolean createAdministeredDestination(
         JmsDestination dest) throws JMSException {
         if (_log.isDebugEnabled()) {
             _log.debug("createAdministeredDestination(dest=" + dest + ")");
         }
 
         boolean success = true;
         boolean queue = (dest instanceof JmsQueue) ? true : false;
         PersistenceAdapter adapter = DatabaseService.getAdapter();
 
         // check that the destination does not exist. If it exists then return
         // false. If it doesn't exists the create it and bind it to the jndi
         // context
 
         Connection connection = null;
         try {
 
             connection = DatabaseService.getConnection();
 
             if (!adapter.checkDestination(connection, dest.getName())) {
                 adapter.addDestination(connection, dest.getName(), queue);
 
                 // destination was created in persistent store, now create it
                 // in transient memory and also bind it in the jndi context
                 addToDestinationCache(dest);
                 try {
                     dest.setPersistent(true);
                     ContextHelper.rebind(getContext(), dest.getName(), dest);
                 } catch (NamingException exception) {
                     String msg = "Failed to add destination " + dest.getName()
                         + " to JNDI context";
                     _log.error(msg, exception);
                     throw new JMSException(msg + ": " +
                         exception.getMessage());
                 }
             } else {
                 success = false;
             }
             connection.commit();
         } catch (JMSException exception) {
             SQLHelper.rollback(connection);
             throw exception;
         } catch (Exception exception) { // PersistenceException, SQLException
             SQLHelper.rollback(connection);
             String msg = "Failed to create administered destination"
                 + dest.getName();
             _log.error(msg, exception);
             throw new JMSException(msg + ": " + exception.getMessage());
         } finally {
             SQLHelper.close(connection);
         }
 
         return success;
     }
 
     /***
      * Remove the corresponding administered destination from persistent
      * store, from transient memory and from the jndi context. This will
      * also remove all durable consumers for this topic.
      *
      * @param dest - the destination to remove
      * @return boolean - true if successful
      */
     public synchronized void deleteAdministeredDestination(JmsDestination dest)
         throws JMSException {
 
         if (_log.isDebugEnabled()) {
             _log.debug("deleteAdministeredDestination(dest=" + dest + ")");
         }
 
         boolean success = false;
         boolean queue = (dest instanceof JmsQueue) ? true : false;
         ConsumerManager consumerMgr = ConsumerManager.instance();
 
         // If we are dealing with a topic then first check that there are
         // no active durable consumers for the destination
         if (!queue) {
             if (consumerMgr.hasActiveDurableConsumers(dest)) {
                 throw new JMSException(
                     "Cannot delete the administered destination " + dest
                     + " since there are active durable consumers.");
             }
             // no active consumers. Remove all durable consumers to this
             // destination
             consumerMgr.removeDurableConsumers(dest);
         }
 
         // make sure there are not active endpoints, but first clear
         // unreferenced endpoints.
         consumerMgr.cleanUnreferencedEndpoints(dest);
         int active = consumerMgr.getEndpointsForDest(dest).size();
         if (active > 0) {
             throw new JMSException(
                 "Cannot delete the administered destination" + dest
                 + " since there are " + active + " active endpoints.");
         }
 
         // unbind it from the jndi context so that now it is unavailable
         // to other consumers
         try {
             getContext().unbind(dest.getName());
         } catch (NamingException error) {
             _log.error("Failed to remove destination " + dest.getName()
                 + " from JNDI", error);
         }
 
         // 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
         Connection connection = null;
         try {
             connection = DatabaseService.getConnection();
 
             DatabaseService.getAdapter().removeDestination(connection,
                 dest.getName());
             destroyDestinationCache(dest);
             removeFromDestinationCache(dest);
             connection.commit();
         } catch (PersistenceException exception) {
             SQLHelper.rollback(connection);
             String msg = "Failed to remove destination " + dest.getName();
             _log.error(msg, exception);
             throw new JMSException(msg + ":" + exception.getMessage());
         } catch (SQLException exception) {
             SQLHelper.rollback(connection);
             String msg = "Failed to remove destination " + dest.getName();
             _log.error(msg, exception);
             throw new JMSException(msg + ":" + exception.getMessage());
         } finally {
             SQLHelper.close(connection);
         }
     }
 
     /***
      * Return a list of destination names currently supported by the destination
      * manager. This includes all types of destinations.
      *
      * @return Iterator - iterator for {@link JmsDestination} objects
      */
     public Iterator destinationNames() {
         return _destinationCache.values().iterator();
     }
 
     /***
      * Return a list of {@link DestinationCache} objects that are currently
      * active and in memory. This will return a list of all destination
      * types (temporary. transient, administered}.
      *
      * @return Iterator - set of DestinationCache objects
      */
     public Iterator destinations() {
         return _caches.values().iterator();
     }
 
     /***
      * This method will create the administered destinations specified in
      * the configuration. A topic may also have zero or more preconfigured
      * durable sunbscribers. An equivalent entity for queues does not
      * exist.
      */
     public void registerConfiguredAdministeredDestinations() {
         AdministeredDestinations destinations =
             ConfigurationManager.getConfig().getAdministeredDestinations();
         if (destinations != null) {
 
             // first process the topics
             int count = destinations.getAdministeredTopicCount();
             for (int index = 0; index < count; index++) {
                 AdministeredTopic topic = destinations.getAdministeredTopic(index);
 
                 // define a persistent topic destination and then use the
                 // message manager administrator to add it
                 JmsTopic destination = new JmsTopic(topic.getName());
                 destination.setPersistent(true);
                 try {
 
                     createAdministeredDestination(destination);
 
                     // register the subscribers for each topic.
                     int scount = topic.getSubscriberCount();
                     ConsumerManager mgr = ConsumerManager.instance();
                     for (int sindex = 0; sindex < scount; sindex++) {
                         Subscriber subscriber = topic.getSubscriber(sindex);
 
                         // create the durable consumer only if one does
                         // not already exist
                         if (!mgr.exists(subscriber.getName())) {
                             mgr.createDurableConsumer(destination,
                                 subscriber.getName());
                         }
                     }
                 } catch (JMSException exception) {
                     _log.error("Failed to register persistent topic "
                         + topic.getName(), exception);
                 }
             }
 
             // next process the queue destinations. QueueDestinations do not
             // have any associated durable subscribers
             count = destinations.getAdministeredQueueCount();
             for (int index = 0; index < count; index++) {
                 AdministeredQueue queue = destinations.getAdministeredQueue(index);
 
                 // define a persistent topic destination and then use the
                 // message manager administrator to add it
                 JmsQueue destination = new JmsQueue(queue.getName());
                 destination.setPersistent(true);
                 try {
                     createAdministeredDestination(destination);
                 } catch (JMSException exception) {
                     _log.error("Failed to register persistent queue "
                         + queue.getName(), exception);
                 }
             }
         }
     }
 
     // implementation of MessageManagerEventListener.messageAdded
     public synchronized boolean messageAdded(JmsDestination destination, MessageImpl message) {
         boolean result = false;
         try {
             if (destination instanceof JmsTopic) {
                 // check to see whether there are active consumers for the
                 // specified destination. If there are then we need to
                 // create a destination cache and pass the message to it.
                 if (ConsumerManager.instance().hasActiveConsumers(destination)) {
                     if (!destinationExists(destination)) {
                         createDestination(destination);
                     }
                     DestinationCache cache = createDestinationCache(destination);
                     result = cache.messageAdded(destination, message);
                 }
             } else {
                 // assume that the destination is a queue. since the message
                 // is non-persistent then we need to create the cache and pass the
                 // message to it.
                 if (!destinationExists(destination)) {
                     createDestination(destination);
                 }
                 DestinationCache cache = createDestinationCache(destination);
                 result = cache.messageAdded(destination, message);
             }
         } catch (Exception exception) {
             _log.error("Exception in DestinationManager.messageAdded",
                 exception);
         }
 
         return result;
     }
 
     // implementation of MessageManagerEventListener.messageRemoved
     public void messageRemoved(JmsDestination destination, MessageImpl message) {
         // removing a non-persistent messages, when the associated destination
         // is not active is a noop
     }
 
     // implementation of MessageManagerEventListener.persistentMessageAdded
     public synchronized boolean persistentMessageAdded(Connection connection,
                                                        JmsDestination destination, MessageImpl message)
         throws PersistenceException {
 
         boolean result = false;
 
         try {
             if (destination instanceof JmsTopic) {
                 // the cache for this destination is inactive. Determine, if
                 // there are any active wildcard consumers for this destination
                 // If there are then create the destination cache and let it
                 // handle the message. Otherwise send it to the ConsumerManager
                 // for processing
                 ConsumerManager manager = ConsumerManager.instance();
                 if (manager.hasActiveConsumers(destination)) {
                     // create the destincation cache and let it process the
                     // message
                     DestinationCache cache = createDestinationCache(connection,
                         destination);
                     result = cache.persistentMessageAdded(connection,
                         destination, message);
                 } else {
                     // This is now handled by the MessageMgr when the message
                     // enters the system
                     // let the consumer manager handle this
                     // result = ConsumerManager.instance().persistentMessageAdded(
                     //    connection, message);
                 }
             } else {
                 // This is now handled by the MessageMgr when the message
                 // enters the system
                 // assume that the destination is a queue. Since the message is
                 // persistent then we do not need to activate the cache, simply
                 // create a persistent handle and be done with it.
                 // MessageHandleFactory.createPersistentHandle(connection,
                 //    destination, null, message);
             }
         } catch (Exception exception) {
             // rethrow as a PersistenceException
             exception.printStackTrace();
             throw new PersistenceException(
                 "Exception in DestinationManager.messageAdded " +
                 exception.toString());
         }
 
         return result;
     }
 
     // implementation of MessageManagerEventListener.persistentMessageRemoved
     public void persistentMessageRemoved(Connection connection,
                                          JmsDestination destination, MessageImpl message)
         throws PersistenceException {
         try {
             if (destination instanceof JmsTopic) {
                 // this is a persistent message so we need to retrieve the
                 // set of durable subscribers for this.
                 Vector names =
                     ConsumerManager.instance().getDurableConsumersForDest(
                         (JmsTopic) destination);
 
                 // for each durable consumer we need to destory that handle
                 while (names.size() > 0) {
                     String name = (String) names.remove(0);
                     MessageHandleFactory.destroyPersistentHandle(connection,
                         destination, name, message);
                 }
             } else {
                 // assume it is a queue and destroy the handle.
                 MessageHandleFactory.destroyPersistentHandle(connection,
                     destination, null, message);
             }
         } catch (PersistenceException exception) {
             // catch and rethrow
             throw exception;
         } catch (Exception exception) {
             throw new PersistenceException(
                 "Exception in DestinationManager.messageRemoved " +
                 exception.toString());
         }
     }
 
     // implement of GarbageCollectable.collectGarbage
     public synchronized void collectGarbage(boolean aggressive) {
         // before continuing we should change the priority of the thread
         // to the lowest value.
         int gc_caches = 0;
         int gc_destinations = 0;
 
         Object[] caches = _caches.values().toArray();
         for (int index = 0; index < caches.length; index++) {
             DestinationCache cache = (DestinationCache) caches[index];
             if (cache.canDestroy()) {
                 if (_log.isDebugEnabled()) {
                     _log.debug("Garbage collecting destination cache="
                         + cache);
                 }
                 destroyDestinationCache(cache);
                 gc_caches++;
             } else {
                 // the cache is active, so issue a garbage collection
                 // request on it
                 cache.collectGarbage(aggressive);
             }
         }
 
         // get rid of non-administered destinations, without
         // associated caches.
         Iterator destinations = _destinationCache.values().iterator();
         Vector to_delete = new Vector();
         while (destinations.hasNext()) {
             JmsDestination dest = (JmsDestination) destinations.next();
             if (!(dest.getPersistent()) &&
                 (!_caches.containsKey(dest))) {
                 to_delete.add(dest);
                 gc_destinations++;
             }
         }
 
         // now delete the actual destinations
         Enumeration todel = to_delete.elements();
         while (todel.hasMoreElements()) {
             _destinationCache.remove(((JmsDestination) todel.nextElement()).getName());
         }
 
         // log the information
         _log.info("DMGC Collected " + gc_caches + " caches, " + _caches.size()
             + " remaining.");
         _log.info("DMGC Collected " + gc_destinations + " destinations, "
             + _destinationCache.size() + " remaining.");
     }
 
     /***
      * Return a HashMap of all destinations that match the specified topic
      * If the topic represents a wildcard then it may match none, one or more
      * destinations. The results are returns as a mapping of destination and
      * the corresponding cache
      * <p>
      * The topic maybe a straight topic name or a wildcard
      *
      * @param topic - topic to query
      * @return HashMap
      */
     synchronized HashMap getTopicDestinationCaches(JmsTopic topic) {
         HashMap result = new HashMap();
 
         Iterator iter = _caches.keySet().iterator();
         while (iter.hasNext()) {
             JmsDestination dest = (JmsDestination) iter.next();
             if ((dest instanceof JmsTopic) &&
                 (topic.match((JmsTopic) dest))) {
                 result.put(dest, _caches.get(dest));
             }
         }
 
         return result;
     }
 
     /***
      * Destroy this manager. This is brutal and final
      */
     public synchronized void destroy() {
         // clean up all the destinations
         Object[] dests = _caches.keySet().toArray();
         for (int index = 0; index < dests.length; index++) {
             destroyDestinationCache((JmsDestination) dests[index]);
         }
 
         _caches.clear();
         _caches = null;
 
         _destinationCache.clear();
         _destinationCache = null;
 
         // remove all the listeners
         _listeners.clear();
         _listeners = null;
 
         _context = null;
 
         // reset the singleton
         _instance = null;
     }
 
     /***
      * Test whether the specified destination is an administered destination. It
      * assumes that the destination exsits.
      *
      * @param name - the name of the destination
      * @return boolean - true if it is and false otherwise
      */
     public boolean isAdministeredDestination(String name) {
         boolean result = false;
         JmsDestination dest = (JmsDestination) _destinationCache.get(name);
         if ((dest != null) &&
             (dest.getPersistent())) {
             result = true;
         }
 
         return result;
     }
 
     /***
      * Test whether the specified message is for an administered destination.
      * This would be the case if the destination is administered or if there
      * is an administered wildcard destination that is satisfied by the
      * message destination
      *
      * @param message - the message to check
      * @return boolean
      */
     public boolean isMessageForAdministeredDestination(MessageImpl msg) {
         boolean result = false;
         try {
             JmsDestination mdest = (JmsDestination) msg.getJMSDestination();
             JmsDestination dest = (JmsDestination) _destinationCache.get(mdest.getName());
 
             if (dest != null) {
                 if (dest.getPersistent()) {
                     result = true;
                 } else if (mdest instanceof JmsTopic) {
                     // check if any of the wildcards are administered
                     Object[] dests = _wildcardDestinations.toArray();
                     for (int index = 0; index < dests.length; index++) {
                         JmsTopic adest = (JmsTopic) dests[index];
                         if ((adest.match((JmsTopic) mdest)) &&
                             (adest.getPersistent())) {
                             result = true;
                             break;
                         }
                     }
 
                 }
             }
         } catch (JMSException ignore) {
         }
 
         return result;
     }
 
     /***
      * Add the specified entry to the destination cache, if it doesn't
      * already exist.
      *
      * @param destination - destination to add
      */
     void addToDestinationCache(JmsDestination destination) {
         synchronized (_destinationCache) {
             if (!_destinationCache.containsKey(destination.getName())) {
                 _destinationCache.put(destination.getName(), destination);
 
                 // check whether it is a wildcard destination
                 if (((destination instanceof JmsTopic) &&
                     (((JmsTopic) destination).isWildCard()))) {
                     _wildcardDestinations.add(destination);
                 }
             }
         }
     }
 
     /***
      * Remove the specified destination from the cache
      *
      * @param destination - the destination to remove
      */
     void removeFromDestinationCache(JmsDestination destination) {
         synchronized (_destinationCache) {
             if (_destinationCache.remove(destination.getName()) != null) {
 
                 // check whether we also need to delete it from the
                 // list of wildcard subscriptions
                 if (((destination instanceof JmsTopic) &&
                     (((JmsTopic) destination).isWildCard()))) {
                     _wildcardDestinations.remove(destination);
                 }
             }
 
         }
     }
 
     /***
      * Check if the specified destination exists.
      *
      * @param destination - the destination to check
      * @return boolean - true if it exists and false otherwise
      */
     public boolean destinationExists(JmsDestination destination) {
         return _destinationCache.containsKey(destination.getName());
     }
 
     /***
      * Initialises the destination manager.
      *
      * @throws ServiceException if the service cannot be initialised
      */
     protected void init() throws ServiceException {
         Connection connection = null;
         TransactionManager tm = null;
         try {
             connection = DatabaseService.getConnection();
 
             // return a list of JmsDestination objects.
             Enumeration iter =
                 DatabaseService.getAdapter().getAllDestinations(connection);
             connection.commit();
 
             while (iter.hasMoreElements()) {
                 // add each destination to the cache and also bind
                 // it to the context
                 JmsDestination dest = (JmsDestination) iter.nextElement();
                 addToDestinationCache(dest);
                 try {
                     // for each of the administered destinations rebind it to
                     // the jndi context
                     dest.setPersistent(true);
                     ContextHelper.rebind(getContext(), dest.getName(), dest);
                 } catch (NamingException error) {
                     throw new ServiceException(
                         "Failed to add destination " + dest.getName()
                         + " to JNDI", error);
                 }
             }
         } catch (PersistenceException exception) {
             SQLHelper.rollback(connection);
             String msg = "Failed to initialise DestinationManager";
             _log.error(msg, exception);
             throw exception;
         } catch (SQLException exception) {
             SQLHelper.rollback(connection);
             String msg = "Failed to initialise DestinationManager";
             _log.error(msg, exception);
             throw new ServiceException(msg, exception);
         } finally {
             SQLHelper.close(connection);
         }
     }
 
     /***
      * Notyify the list of DestinationEventListener objects that the specified
      * destination has been added
      *
      * @param dest - the destination that was added
      * @param cache - the corresponding cache
      */
     private void notifyDestinationAdded(JmsDestination dest,
                                         DestinationCache cache) {
         synchronized (_listeners) {
             Iterator iter = _listeners.iterator();
             while (iter.hasNext()) {
                 ((DestinationEventListener) iter.next()).destinationAdded(dest, cache);
             }
         }
     }
 
     /***
      * Notyify the list of DestinationEventListener objects that the specified
      * destination has been removed
      *
      * @param dest - the destination that was removed
      * @param cache - the corresponding cache
      */
     private void notifyDestinationRemoved(JmsDestination dest,
                                           DestinationCache cache) {
         synchronized (_listeners) {
             Iterator iter = _listeners.iterator();
             while (iter.hasNext()) {
                 ((DestinationEventListener) iter.next()).destinationRemoved(dest, cache);
             }
         }
     }
 
     /***
      * Return the initial context using the JndiHelper. It assumes that the
      * configuration manager has been successfully initialized. If the context
      * has been retrieved it is cached so subsequent gets will return the
      * cached instance
      *
      * @return the root context
      */
     private static Context getContext() throws NamingException {
         return NamingHelper.getInitialContext(
             ConfigurationManager.getConfig());
     }
 
     /***
      * This static class is used to maintain information about a
      * destination
      */
     private static class DestinationEntry {
 
         /***
          * The destination that this entry pertains too
          */
         public JmsDestination _destination = null;
 
         /***
          * Indicates whether the destination is administered
          */
         public boolean _administered = false;
 
 
         /***
          * Construct an instance of the entry using the specified
          * parameters
          *
          * @param dest - the destination
          * @param administered - true if the destination is administered
          */
         DestinationEntry(JmsDestination dest, boolean administered) {
             _destination = dest;
             _administered = administered;
         }
 
         // override Object.equals
         public boolean equals(Object obj) {
 
             boolean result = false;
 
             if ((obj != null) &&
                 (obj instanceof DestinationEntry)) {
 
                 result = _destination.equals(((DestinationEntry) obj)._destination);
             }
 
             return result;
         }
     }
 }