/***
    * 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: ConsumerManager.java,v 1.33 2003/09/25 11:24:16 tanderson Exp $
   *
   * Date         Author  Changes * 3/1/2001     jima    Created
   */
  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.Vector;
  
  import javax.jms.DeliveryMode;
  import javax.jms.InvalidSelectorException;
  import javax.jms.JMSException;
  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.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.scheduler.Scheduler;
  import org.exolab.jms.server.JmsServerSession;
  
  
  /***
   * The consumer manager is responsible for creating and managing the
   * lifecycle of Consumer. The consumer manager maintains a list of all
   * active consumers.
   * <p>
   * The consumer manager, in an attempt to make better utilization of
   * resource will also act as a proxy for all inactive durable subscribers
   * So it will act as a DestinationCacheEventListener and process the
   * message on behalf of the inactive consumer.
   *
   * @version     $Revision: 1.33 $ $Date: 2003/09/25 11:24:16 $
   * @author      <a href="mailto:jima@exoffice.com">Jim Alateras</a>
   */
  public class ConsumerManager
      implements DestinationCacheEventListener, GarbageCollectable {
  
      /***
       * Maintains a cache of all active endpoints.
       */
      private HashMap _endpoints = new HashMap();
  
     /***
      * Maintains a list of all unique consumers, durable and non-durable.
      * Each entry has an associated {@link #ConsumerEntry} record. All
      * durable subscribers are maintained in memory until they are removed
      * from the system entirely. All non-durable subscribers are maintained
      * in memory until their endpoint is removed.
      */
     private HashMap _consumerCache = new HashMap();
 
     /***
      * Maintais a mapping between destinations and consumers. A destination
      * can have more than one consumer and a consumer can also be registered
      * to more than one destination
      */
     private HashMap _destToConsumerMap = new HashMap();
 
     /***
      * Maintains a list of wildcard subscriptions using subscription name
      * and the JmsTopic.
      */
     private HashMap _wildcardConsumers = new HashMap();
 
     /***
      * Cache a copy of the scheduler instance
      */
     private Scheduler _scheduler = null;
 
     /***
      * The singleton instance of the consumer manager
      */
     private static ConsumerManager _instance = null;
 
     /***
      * The logger
      */
     private static final Log _log = LogFactory.getLog(ConsumerManager.class);
 
 
     /***
      * Create the singleton instance of the consumer manager
      *
      * @return the singleton instance
      * @throws ServiceException if the service cannot be initialised
      */
     public static ConsumerManager createInstance() throws ServiceException {
         _instance = new ConsumerManager();
         return _instance;
     }
 
     /***
      * Return the singleton instance of the ConsumerManager
      *
      * @return ConsumerManager
      */
     public static ConsumerManager instance() {
         return _instance;
     }
 
     /***
      * Construct the <code>ConsumerManager</code>
      *
      * @throws ServiceException - if it fails to initialise
      */
     private ConsumerManager() throws ServiceException {
         // register with the GarbageCollectionService
         GarbageCollectionService.instance().register(this);
 
         init();
     }
 
     /***
      * This method creates an actual durable consumer for the specified and
      * caches it. It does not create and endpoint. To create the endpoint the
      * client should call createDurableConsumerEndpoint.
      *
      * @param topic - the topic destination
      * @param name - the consumer name
      * @exception JMSException - if it cannot be created
      */
     public synchronized void createDurableConsumer(JmsTopic topic, String name)
         throws JMSException {
 
         PersistenceAdapter adapter = DatabaseService.getAdapter();
 
         Connection connection = null;
         try {
 
             connection = DatabaseService.getConnection();
 
             // ensure that we are trying to create a durable consumer to an
             // administered destination.
             if (!adapter.checkDestination(connection, topic.getName())) {
                 throw new JMSException(
                     "Cannot create durable consumer, name=" + name +
                     ", for non-administered topic=" + topic.getName());
             }
 
             if (!adapter.durableConsumerExists(connection, name)) {
                 adapter.addDurableConsumer(connection, topic.getName(), name);
             }
 
             connection.commit();
             // cache the consumer locally
             addToConsumerCache(name, topic, true);
         } catch (JMSException exception) {
             throw exception;
         } catch (Exception exception) { // PersistenceException, SQLException
             SQLHelper.rollback(connection);
             String msg = "Failed to create durable consumer, name=" + name
                 + ", for topic=" + topic.getName();
             _log.error(msg, exception);
             throw new JMSException(msg + ": " + exception.getMessage());
         } finally {
             SQLHelper.close(connection);
         }
     }
 
 
     /***
      * This method will remove the durable consumer from the database and
      * from transient memory only if it exists and is inactive. If there
      * is an active endpoint then it cannot be deleted and an exception
      * will be raise.
      * <p>
      * If the durable consumer does not exist then an exception is also
      * raised.
      *
      * @param name - the consumer name
      * @exception JMSException - if it cannot be removed
      */
     public synchronized void removeDurableConsumer(String name)
         throws JMSException {
         if (_log.isDebugEnabled()) {
             _log.debug("removeDurableConsumer(name=" + name + ")");
         }
 
         // check to see that the durable consumer exists
         if (!durableConsumerExists(name)) {
             throw new JMSException("Durable consumer " + name +
                 " is not defined.");
         }
         if (isDurableConsumerActive(name)) {
             throw new JMSException(
                 "Cannot remove durable consumer=" + name
                 + ": consumer is active");
         }
 
         // remove it from the persistent store.
         Connection connection = null;
         try {
             connection = DatabaseService.getConnection();
 
             DatabaseService.getAdapter().removeDurableConsumer(connection,
                 name);
             // if it has been successfully removed from persistent store then
             // clear up the transient references.
             ConsumerEndpoint endpoint = getConsumerEndpoint(name);
             if (endpoint != null) {
                 deleteConsumerEndpoint(endpoint);
             }
             removeFromConsumerCache(name);
             connection.commit();
         } catch (Exception exception) { // PersistenceException, SQLException
             SQLHelper.rollback(connection);
             String msg = "Failed to remove durable consumer, name=" + name;
             _log.error(msg, exception);
             throw new JMSException(msg + ":" + exception.getMessage());
         } finally {
             SQLHelper.close(connection);
         }
     }
 
     /***
      * This method will remove all the durable consumers from the database and
      * from transient memory whether they are active or not.
      * <p>
      * If we have problems removing the durable consumers then throw the
      * JMSException.
      *
      * @param topic the topic to remove consumers for
      * @throw JMSException if the consumers cannot be removed
      */
     public synchronized void removeDurableConsumers(JmsDestination topic)
         throws JMSException {
 
         Vector consumers = (Vector) _destToConsumerMap.get(topic);
         if (consumers != null) {
             Enumeration entries = consumers.elements();
             while (entries.hasMoreElements()) {
                 ConsumerEntry entry = (ConsumerEntry) entries.nextElement();
                 if (entry._durable) {
                     // remove the actual durable consumer from transient and
                     // secondary memory.
                     removeDurableConsumer(entry._name);
                 }
             }
         }
 
         // remove all consumers for the specified destination
         removeFromConsumerCache(topic);
     }
 
     /***
      * Create a transient consumer for the specified destination. The client
      * can optionally specify a selector to filter messages.
      * <p>
      * The clientId parameter is used to indirectly reference the remote client
      * which is uniquely identifiable within a session and is used during
      * asynchronous message delivery.
      *
      * @param clientId - indirect reference to the remote client
      * @param destination - consumer for this destination
      * @param selector - the consumer's selector if specified.
      * @return transient consumer
      */
     public synchronized ConsumerEndpoint createConsumerEndpoint(
         JmsServerSession session, long clientId, JmsDestination destination,
         String selector)
         throws JMSException, InvalidSelectorException {
 
         if (_log.isDebugEnabled()) {
             _log.debug("createConsumerEndpoint(session=[sessionId="
                 + session.getSessionId() + "], clientId=" + clientId
                 + ", destination=" + destination
                 + ", selector=" + selector + ")");
         }
 
         ConsumerEndpoint endpoint = null;
         DestinationManager destmgr = DestinationManager.instance();
 
         // before we create the destination we need to check the
         // characteristics of the destination. If the destination
         // is an administered destination then it must be already
         // defined. If the destination is a temporary destination
         // then we may need to add it to the cache.
         if (destination.getPersistent()) {
             if (!destmgr.destinationExists(destination)) {
                 throw new JMSException("Cannot create consumer endpoint: "
                     + "destination=" + destination
                     + " does not exist");
             }
         } else {
             if (!destmgr.destinationExists(destination)) {
                 destmgr.createDestination(destination);
             }
         }
 
         // determine what type of consumer endpoint to create based on
         // the destination it subscribes too.
         if (destination instanceof JmsTopic) {
             JmsTopic topic = (JmsTopic) destination;
             endpoint = new TopicConsumerEndpoint(session, clientId, topic,
                 selector, _scheduler);
         } else if (destination instanceof JmsQueue) {
             // this seems like a good opportunity to clean up any unreferenced
             // endpoints attached to this queue.
             cleanUnreferencedEndpoints(destination);
 
             // now we can create the endpoint
             endpoint = new QueueConsumerEndpoint(session, clientId,
                 (JmsQueue) destination, selector, _scheduler);
         }
 
         if (endpoint != null) {
             // add it to the list on managed consumers
             String id = endpoint.getPersistentId();
             _endpoints.put(id, endpoint);
             addToConsumerCache(id, destination, false);
         }
 
         return endpoint;
     }
 
     /***
      * Create a durable consumer with the specified well-known name.
      *
      * @param session the owning session
      * @param topic consumer for this topic
      * @param name the unique subscriber name
      * @param clientId the remote client identity
      * @param selector the message selector. May be <code>null</code>
      * @return the durable consumer endpoint
      * @throws JMSException if a durable consumer is already active with
      * the same <code>name</code>, or the <code>topic</code> doesn't exist,
      * or <code>selector</code> is an invalid selector
      */
     public synchronized DurableConsumerEndpoint createDurableConsumerEndpoint(
         JmsServerSession session, JmsTopic topic, String name,
         long clientId, String selector)
         throws JMSException {
 
         if (_log.isDebugEnabled()) {
             _log.debug("createDurableConsumerEndpoint(session=[sessionId="
                 + session.getSessionId() + "], topic=" + topic
                 + ", name=" + name + ", clientId=" + clientId
                 + ", selector=" + selector);
         }
 
         // check that the durable subscriber is not already registered. If it
         // is registered then check to see whether the client is still active.
         DurableConsumerEndpoint endpoint =
             (DurableConsumerEndpoint) _endpoints.get(name);
         if (endpoint != null) {
             if (endpoint.getSession().isClientEndpointActive()) {
                 throw new JMSException(name + " is already registered");
             } else {
                 // client endpoint must have been lost
                 if (_log.isDebugEnabled()) {
                     _log.debug("Closing session for inactive durable " +
                         "consumer [name=" + name + "]");
                 }
                 endpoint.getSession().close();
             }
         }
 
         // check that the destination actually exists, if the topic
         // is not a wildcard
         if (!topic.isWildCard() &&
             !DestinationManager.instance().destinationExists(topic)) {
             throw new JMSException("Cannot create a durable consumer for "
                                    + topic);
         }
 
         // if we get this far then we need to create the durable consumer
         endpoint = new DurableConsumerEndpoint(session, clientId, topic, name,
             selector, _scheduler);
         _endpoints.put(endpoint.getPersistentId(), endpoint);
 
         return endpoint;
     }
 
     /***
      * Check whether there are active durable consumers for the specified
      * destination.
      *
      * @param topic the destination to check
      * @param <code>true</code> if there is at least one active consumer
      */
     public synchronized boolean hasActiveDurableConsumers(
         JmsDestination topic) {
 
         boolean result = false;
         Vector consumers = (Vector) _destToConsumerMap.get(topic);
         if (consumers != null) {
             Enumeration iterator = consumers.elements();
             while (iterator.hasMoreElements()) {
                 ConsumerEntry entry = (ConsumerEntry) iterator.nextElement();
                 if (entry._durable) {
                     result = true;
                     break;
                 }
             }
         }
 
         return result;
     }
 
     /***
      * Create a browser for the specified destination and the selector. A
      * browser is responsible for passing all messages back to the client
      * that reside on the queue
      *
      * @param session the owning session
      * @param clientId the remote client identity, which is session scoped
      * @param queue the queue destination cache
      * @param selector optional filter
      * @return the queue browser endpoint
      */
     public synchronized ConsumerEndpoint createQueueBrowserEndpoint(
         JmsServerSession session, long clientId, JmsQueue queue,
         String selector)
         throws JMSException {
 
         ConsumerEndpoint consumer = null;
 
         if (queue != null) {
             consumer = new QueueBrowserEndpoint(session, clientId,
                 queue, selector, _scheduler);
         } else {
             throw new JMSException("Cannot create a browser for a null queue");
         }
 
         String id = consumer.getPersistentId();
         _endpoints.put(id, consumer);
         addToConsumerCache(id, queue, false);
 
         return consumer;
     }
 
     /***
      * Destroy the endpoint associated with the specified durable consumer
      *
      * @param name - name of the durable consumer
      * @exception JMSException - if itt cannot complete the request
      */
     public synchronized void deleteDurableConsumerEndpoint(String name)
         throws JMSException {
 
         if (_log.isDebugEnabled()) {
             _log.debug("deleteDurableConsumerEndpoint(name=" + name + ")");
         }
 
         ConsumerEntry entry = (ConsumerEntry) _consumerCache.get(name);
         if (entry != null) {
             if (entry._durable) {
                 deleteConsumerEndpoint((ConsumerEndpoint) _endpoints.get(name));
             } else {
                 throw new JMSException(name + " is not a durable subscriber");
             }
         } else {
             // ignore since the consumer is not active
             if (_log.isDebugEnabled()) {
                 _log.debug("deleteDurableConsumerEndpoint(name=" + name
                     + "): failed to locate consumer");
             }
         }
     }
 
     /***
      * Destroy the specified consumer
      *
      * @param consumer the consumer to destroy
      */
     public synchronized void deleteConsumerEndpoint(
         ConsumerEndpoint consumer) {
 
         if (_log.isDebugEnabled()) {
             _log.debug("deleteConsumerEndpoint(consumer=[clientId="
                 + consumer.getClientId() + ", destination="
                 + consumer.getDestination() + ", session=[sessionId="
                 + consumer.getSession().getSessionId() + "]])");
         }
 
         String id = consumer.getPersistentId();
 
         // if the consumer is currently active then delete it
         ConsumerEndpoint existing = (ConsumerEndpoint) _endpoints.get(id);
         if (existing != null) {
             // unregister itself from all the caches
             consumer.unregister();
 
             // remove it from the list of active endpoints
             // As a fix for bug 759752, only remove the consumer if it
             // matches the existing one
             if (consumer.getId().equals(existing.getId())) {
                 _endpoints.remove(id);
             } else {
                 if (_log.isDebugEnabled()) {
                     _log.debug("Existing endpoint doesn't match that to " +
                         "be deleted - retaining");
                 }
             }
 
             // close the endpoint
             consumer.close();
 
             // remove it from the consumer cache if and only if it is a
             // non-durable subscriber
             if (!(consumer instanceof DurableConsumerEndpoint)) {
                 try {
                     removeFromConsumerCache(id);
                 } catch (JMSException exception) {
                     _log.debug("Failed to remove " + id + " from the cache",
                         exception);
                 }
             }
         }
     }
 
     /***
      * Return the consumer with the specified identity
      *
      * @param id - identity of the consumer
      * @return Consumer - associated consumer object or null
      */
     public ConsumerEndpoint getConsumerEndpoint(String id) {
         return (ConsumerEndpoint) _endpoints.get(id);
     }
 
     /***
      * Return a list of {@link ConsumerEndpoint} objects, both transient and
      * durable that are currently active.
      *
      * @return Iterator - iterator of {@link ConsumerEndpoint} objects
      */
     public Iterator consumerEndpoints() {
         return _endpoints.values().iterator();
     }
 
     /***
      * Return a list of consumer names id's currently active in the
      * consumer manager.
      *
      * @return Iterator - iterator of {#link String} ids
      */
     public Iterator consumerIds() {
         return _endpoints.keySet().iterator();
     }
 
     /***
      * Check whether a consumer, with the specified identity actually
      * exists.
      *
      * @param id - identity of the consumer
      * @return boolean - true if one exists
      */
     public boolean exists(String id) {
         return (getConsumerEndpoint(id) != null);
     }
 
     /***
      * Check whether there is an active consumer for the specified
      * destination
      *
      * @param destination - the destination to check
      * @return boolean - true if it exists
      */
     public boolean hasActiveConsumers(JmsDestination destination)
         throws JMSException {
         boolean result = false;
 
         Object[] endpoints = _endpoints.values().toArray();
         for (int index = 0; index < endpoints.length; index++) {
             ConsumerEndpoint endpoint = (ConsumerEndpoint) endpoints[index];
             JmsDestination endpoint_dest = endpoint.getDestination();
 
             if ((destination instanceof JmsTopic) &&
                 (endpoint_dest instanceof JmsTopic) &&
                 (((JmsTopic) endpoint_dest).isWildCard())) {
                 if (((JmsTopic) endpoint_dest).match((JmsTopic) destination)) {
                     result = true;
                     break;
                 }
             } else {
                 if (endpoint_dest.equals(destination)) {
                     result = true;
                     break;
                 }
             }
         }
 
         return result;
     }
 
     /***
      * Check whether a particular durable consumer is active
      *
      * @param name - the consumer name
      * @return boolean - true if active
      */
     public boolean isDurableConsumerActive(String name) {
         return (_endpoints.get(name) != null);
     }
 
     /***
      * Return the destination assoicated with the specified durable
      * consumer.
      *
      * @param name - consumer name
      * @return JmsDestination - the destination is it registered under or null
      */
     public JmsDestination getDestinationForConsumerName(String name) {
         ConsumerEntry entry = (ConsumerEntry) _consumerCache.get(name);
         return (entry != null) ? entry._destination : null;
     }
 
     /***
      * Check if the specified durable consumer exists
      *
      * @param name - the name of the durable consumer
      * @return boolean - true if successful
      */
     public boolean durableConsumerExists(String name) {
         boolean result = false;
         ConsumerEntry entry = (ConsumerEntry) _consumerCache.get(name);
         if ((entry != null) &&
             (entry._durable)) {
             result = true;
         }
 
         return result;
     }
 
     /***
      * This method will check that the name-destination pair actually
      * are valid and exist as a DurableConsumer  entity
      *
      * @param topic - the name of the topic
      * @param name - the name of the durable consumer
      * @return boolean - true if valid and false otherwise
      */
     public boolean validSubscription(String topic, String name) {
 
         boolean result = false;
         ConsumerEntry entry = (ConsumerEntry) _consumerCache.get(name);
 
         if ((entry != null) &&
             (entry._destination != null) &&
             (entry._destination.getName().equals(topic))) {
             result = true;
         }
 
         return result;
     }
 
     // implementation of DestinationCacheEventListener.messageAdded
     synchronized public boolean messageAdded(MessageImpl message) {
         return false;
     }
 
     // implementation of DestinationCacheEventListener.messageRemoved
     synchronized public boolean messageRemoved(MessageImpl message) {
         return false;
     }
 
     // implementation of DestinationCacheEventListener.persistentMessageAdded
     synchronized public boolean persistentMessageAdded(Connection connection,
                                                        MessageImpl message)
         throws PersistenceException {
         try {
             JmsDestination dest = (JmsDestination) message.getJMSDestination();
 
             // check to ensure that the message is not of type queue. If it
             // is then we don't want to process it here. All the processing
             // for non-resident queues are done at the cache level
             if (dest instanceof JmsQueue) {
                 return false;
             }
 
             // for each durable consumer we need to add a persistent handle
             Vector names = getDurableConsumersForDest((JmsTopic) dest);
             while (names.size() > 0) {
                 String name = (String) names.remove(0);
                 MessageHandleFactory.createPersistentHandle(connection, dest,
                     name, message);
             }
         } catch (JMSException exception) {
             // rethrow as a PersistenceException...but does it make sense
             throw new PersistenceException(
                 "Failed to create persistent handle", exception);
         }
 
         // return false, even though we processed it to force the
         // message manager not to keep the message in memory.
         return false;
     }
 
     // implementation of DestinationCacheEventListener.persistentMessageRemoved
     synchronized public boolean persistentMessageRemoved(Connection connection,
                                                          MessageImpl message)
         throws PersistenceException {
         try {
 
             JmsDestination dest = (JmsDestination) message.getJMSDestination();
 
             // check to ensure that the message is not of type queue. If it
             // is then we don't want to process it here.
             if (dest instanceof JmsQueue) {
                 return false;
             }
 
             Vector consumers = (Vector) _destToConsumerMap.get(dest);
             if (consumers != null) {
                 // now search through the list of inactive durable consumers
                 // and remove it from their cache
                 Enumeration entries = consumers.elements();
                 while (entries.hasMoreElements()) {
                     ConsumerEntry entry = (ConsumerEntry) entries.nextElement();
                     if ((entry._durable) &&
                         (!_endpoints.containsKey(entry._name))) {
                         MessageHandleFactory.destroyPersistentHandle(connection,
                             dest, entry._name, message);
                     }
                 }
             }
         } catch (PersistenceException exception) {
             throw exception;
         } catch (Exception exception) {
             // rethrow as a PersistenceException...but does it make sense
             throw new PersistenceException(
                 "Exception in ConsumerManager.persistentMessageRemoved " +
                 exception.toString());
         }
 
         return true;
     }
 
     /***
      * Destroy this manager. This is brutal and final
      */
     public synchronized void destroy() {
 
         // clean up all the destinations
         Object[] endpoints = _endpoints.values().toArray();
         for (int index = 0; index < endpoints.length; index++) {
             deleteConsumerEndpoint((ConsumerEndpoint) endpoints[index]);
         }
         _endpoints.clear();
 
         // remove cache data structures
         _consumerCache.clear();
         _consumerCache = null;
         _destToConsumerMap.clear();
         _destToConsumerMap = null;
         _wildcardConsumers.clear();
         _wildcardConsumers = null;
 
         // reset the singleton
         _instance = null;
     }
 
     /***
      * Return a list of durable subscribers for the specified destination
      *
      * @return Vector - a vector of strings, which denote the name
      */
     public synchronized Vector getDurableConsumersForDest(JmsTopic dest) {
         Vector names = new Vector();
 
         Vector consumers = (Vector) _destToConsumerMap.get(dest);
         if (consumers != null) {
             Enumeration entries = consumers.elements();
             while (entries.hasMoreElements()) {
                 ConsumerEntry entry = (ConsumerEntry) entries.nextElement();
                 if (entry._durable) {
                     names.add(entry._name);
                 }
             }
         }
 
         // if the destination is a topic and part is also a wildcard then
         // check the wildcardConsumers for additional consumers
         Iterator wildconsumers = _wildcardConsumers.keySet().iterator();
         while (wildconsumers.hasNext()) {
             ConsumerEntry entry = (ConsumerEntry) wildconsumers.next();
             JmsDestination adest = entry._destination;
             if ((entry._durable) &&
                 (adest instanceof JmsTopic) &&
                 (((JmsTopic) adest).match((JmsTopic) dest))) {
                 names.add(entry._name);
             }
         }
 
         return names;
     }
 
     /***
      * Clean all the unreferenced endpoints for a specified destination.
      * Unreference endpoints usually occur when the client abnormally
      * terminates leaving some dangling endpoints on the server.
      *
      * @param dest- the destination to query
      */
     public void cleanUnreferencedEndpoints(JmsDestination dest) {
         Object[] endpoints = _endpoints.values().toArray();
 
         for (int index = 0; index < endpoints.length; index++) {
             ConsumerEndpoint endpoint = (ConsumerEndpoint) endpoints[index];
             if (dest.equals(endpoint.getDestination())) {
                 if (!endpoint.getSession().isClientEndpointActive()) {
                     try {
                         endpoint.getSession().close();
                     } catch (Exception ignore) {
                         // ignore this exception
                     }
                 }
             }
         }
     }
 
     /***
      * Return a list of {@link ConsumerEndpoint} objects attached to the
      * specified destination
      *
      * @param dest the destination to query
      * @return list of endpoints
      */
     public synchronized LinkedList getEndpointsForDest(JmsDestination dest) {
         LinkedList endpoints = new LinkedList();
         Iterator iter = _endpoints.values().iterator();
 
         while (iter.hasNext()) {
             ConsumerEndpoint endpoint = (ConsumerEndpoint) iter.next();
             if (dest.equals(endpoint.getDestination())) {
                 endpoints.add(endpoint);
             }
         }
 
         return endpoints;
     }
 
     // implement of GarbageCollectable.collectGarbage
     public synchronized void collectGarbage(boolean aggressive) {
         if (aggressive) {
             Object[] endpoints = _endpoints.values().toArray();
             int count = endpoints.length;
 
             for (int index = 0; index < count; index++) {
                 ((ConsumerEndpoint) endpoints[index]).collectGarbage(aggressive);
             }
         }
     }
 
     /***
      * Add the specified consumer to the cache.
      *
      * @param name - the name of the consumer
      * @param dest - the destination it is subscribed to. It can be a wildcard
      * @param durable - indicates whether it is a durable subscription
      */
     synchronized void addToConsumerCache(String name, JmsDestination dest,
                                          boolean durable)
         throws JMSException {
 
         if (_log.isDebugEnabled()) {
             _log.debug("addToConsumerCache(name=" + name + ", dest=" + dest
                 + ", durable=" + durable + ")");
         }
 
         if (!_consumerCache.containsKey(name)) {
             ConsumerEntry entry = new ConsumerEntry(name, dest, durable);
             _consumerCache.put(name, entry);
 
             // if the specified destination is a JmsTopic and also a wildcard
             // then we need to add it to all matching desitnations
             if ((dest instanceof JmsTopic) &&
                 (((JmsTopic) dest).isWildCard())) {
                 // store wild card consumers in a separate array.
                 _wildcardConsumers.put(new ConsumerEntry(name, dest, durable),
                     dest);
             } else {
                 // we also need to add the reverse mapping
                 Vector consumers = (Vector) _destToConsumerMap.get(dest);
                 if (consumers == null) {
                     consumers = new Vector();
                     _destToConsumerMap.put(dest, consumers);
                 }
 
                 // add the mapping
                 consumers.add(entry);
             }
         }
     }
 
     /***
      * Remove the specified consumer from the cache and make all necessary
      * adjustments
      *
      * @param name - name of consumer to remove
      */
     synchronized void removeFromConsumerCache(String name)
         throws JMSException {
 
         if (_log.isDebugEnabled()) {
             _log.debug("removeFromConsumerCache(name=" + name + ")");
         }
 
         if (_consumerCache.containsKey(name)) {
             ConsumerEntry entry = (ConsumerEntry) _consumerCache.remove(name);
             JmsDestination dest = entry._destination;
 
 
             if ((dest instanceof JmsTopic) &&
                 (((JmsTopic) dest).isWildCard())) {
                 // remove it from the wildcard cache.
                 _wildcardConsumers.remove(name);
             } else {
                 // remove it from the specified destination
                 Vector consumers = (Vector) _destToConsumerMap.get(dest);
                 if (consumers != null) {
                     consumers.remove(entry);
 
                     // if consumers is of size 0 then remove it
                     if (consumers.size() == 0) {
                         _destToConsumerMap.remove(dest);
                     }
                 }
             }
         } else {
             if (_log.isDebugEnabled()) {
                 _log.debug("removeFromConsumerCache(name=" + name +
                     "): consumer not found");
             }
         }
     }
 
     /***
      * Remove all the consumers for the specified destination from the
      * cache.
      *
      * @param destination - destination to remove.
      */
     synchronized void removeFromConsumerCache(JmsDestination destination) {
         if (_destToConsumerMap.containsKey(destination)) {
             _destToConsumerMap.remove(destination);
 
             // i am not too sure whether we need more house keeping
         }
     }
 
     /***
      * Return the number of active consumer endpoints
      *
      * @return int - number of active consumer endpoints
      */
     int getConsumerEndpointCount() {
         return _endpoints.size();
     }
 
     /***
      * Initialises the consumer manager
      *
      * @throws ServiceException if the manager can't be initialised
      */
     private void init() throws ServiceException {
         _scheduler = Scheduler.instance();
 
         Connection connection = null;
         try {
             connection = DatabaseService.getConnection();
 
             PersistenceAdapter adapter = DatabaseService.getAdapter();
             connection.commit();
 
             // return a list of JmsDestination objects.
             HashMap map = adapter.getAllDurableConsumers(connection);
             Iterator iter = map.keySet().iterator();
 
             // Create an endpoint for each durable consumer
             while (iter.hasNext()) {
                 // for each destination, create the destination cache
                 String consumer = (String) iter.next();
                 String deststr = (String) map.get(consumer);
 
                 JmsDestination dest =
                     DestinationManager.instance().destinationFromString(
                         deststr);
                 if (dest == null) {
                     // this maybe a wildcard subscription
                     dest = new JmsTopic(deststr);
                     if (!((JmsTopic) dest).isWildCard()) {
                         dest = null;
                     }
                 }
 
                 if (consumer != null && dest != null &&
                     dest instanceof JmsTopic) {
                     // cache the consumer-destination mapping in memory.
                     addToConsumerCache(consumer, dest, true);
                 } else {
                     // what should we do about this stage
                     _log.error(
                         "Failure in ConsumerManager.init : " + consumer +
                         ":" + dest);
                 }
             }
         } catch (ServiceException exception) {
             SQLHelper.rollback(connection);
             throw exception;
         } catch (Exception exception) {
             SQLHelper.rollback(connection);
             throw new ServiceException("Failed to initialise ConsumerManager",
                 exception);
         } finally {
             SQLHelper.close(connection);
         }
     }
 
     /***
      * This private static class is used to maintain consumer information
      */
     private static class ConsumerEntry {
 
         /***
          * The name of the consumer. This name is either the durable
          * subscriber name or a uniquely generated name.
          */
         String _name = null;
 
         /***
          * Indicated whether this entry is for a durable subscriber
          */
         boolean _durable = false;
 
         /***
          * The destination that the consumer is actually subscribed too
          */
         JmsDestination _destination = null;
 
         /***
          * Construct an instance of this class using the specified
          * name and durable subscriber indicator
          *
          * @param name - the name of the consumer
          * @param destination - the destination consumer is subscribed too
          * @param durable - indicates whether it is a durable subscription
          */
         ConsumerEntry(String name, JmsDestination destination,
                       boolean durable) {
             _name = name;
             _destination = destination;
             _durable = durable;
         }
 
         // override Object.equals
         public boolean equals(Object obj) {
 
             boolean result = false;
             if ((obj != null) &&
                 (obj instanceof ConsumerEntry) &&
                 (((ConsumerEntry) obj)._name.equals(_name))) {
                 result = true;
             }
 
             return result;
         }
 
     } //-- ConsumerEntry
 
 } //-- ConsumerManager