/***
    * 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 2000-2003 (C) Exoffice Technologies Inc. All Rights Reserved.
   */
  package org.exolab.jms.messagemgr;
  
  import java.sql.Connection;
  import java.sql.SQLException;
  import java.util.Date;
  import java.util.HashMap;
  import java.util.Iterator;
  
  import javax.jms.JMSException;
  import javax.jms.Message;
  import javax.transaction.TransactionManager;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import org.exolab.core.service.BasicService;
  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.DatabaseConfiguration;
  import org.exolab.jms.config.MessageManagerConfiguration;
  import org.exolab.jms.config.Subscriber;
  import org.exolab.jms.message.MessageHandle;
  import org.exolab.jms.message.MessageImpl;
  import org.exolab.jms.persistence.DatabaseService;
  import org.exolab.jms.persistence.PersistenceException;
  
  
  /***
   * This is the active message handling component within the JMS server.
   * Messages are passed in and added to the appropriate dispatchers for delivery
   * to the clients.
   *
   * @version     $Revision: 1.93 $ $Date: 2003/08/17 01:32:24 $
   * @author      <a href="mailto:mourikis@intalio.com">Jim Mourikis</a>
   * @author      <a href="mailto:tma@netspace.net.au">Tim Anderson</a>
   */
  public class MessageMgr extends BasicService {
  
      /***
       * The service name of the message manager
       */
      private final static String MM_SERVICE_NAME = "MessageManager";
  
      /***
       * Caches the singleton instance of the message manager.
       */
      private static MessageMgr _instance;
  
      /***
       * used to synchronise the singleton construction
       */
     private static final Object _block = new Object();
 
     /***
      * Maintain a list of registered MessageManagerEventListener objects, that
      * get notified when certain events occur in the MessageManager
      */
     private transient HashMap _listeners = new HashMap(1023);
 
     /***
      * The sequence number generator is used to differentiate messages arriving
      * on the same millisecond
      */
     private long _sequenceNumberGenerator = 0;
 
     /***
      * This is the maximum size the cache can reach before we are forced to
      * run garbage collection. Garbage collection will also execute in the
      * background periodically to remove processed messages from the cache.
      */
     private int _maximumSize = 2500;
 
     /***
      * Tracks the number of messages processed
      */
     private long _messagesProcessed;
 
     /***
      * The logger
      */
     private static final Log _log = LogFactory.getLog(MessageMgr.class);
 
 
     /***
      * Create and return an instance of the singleton. If the singleton already
      * exists then simply return it. If there is a problem creating the
      * singleton then throw an exception
      *
      * @return MessageMgr - the singleton instance
      * @throws MessageMgrException
      */
     public static MessageMgr createInstance() throws MessageMgrException {
         if (_instance == null) {
             synchronized (_block) {
                 if (_instance == null) {
                     _instance = new MessageMgr();
                 }
             }
         }
         return _instance;
     }
 
     /***
      * Return an instance to the MessageMgr singleton. This method assumes that
      * the singleton has already been created with a call to
      * {@link #createInstance}
      *
      * @return MessageMgr
      */
     public static MessageMgr instance() {
         return _instance;
     }
 
     /***
      * The constructor will register itself with the garbage collection
      * service.
      *
      * @throws MessageMgrException - if there are problems during construction
      */
     private MessageMgr() throws MessageMgrException {
         super(MM_SERVICE_NAME);
     }
 
     // ovverride BasicService.start
     public void start() throws ServiceException {
         try {
             DestinationManager.createInstance();
             ConsumerManager.createInstance();
         } catch (ServiceException exception) {
             throw exception;
         } catch (Exception exception) {
             String msg = "Failed to start MessageMgr";
             _log.error(msg, exception);
             throw new ServiceException(msg + ":" + exception);
         }
     }
 
     // implement BasicService.run
     public void run() {
         // do nothing
     }
 
     // override BasicService.stop
     public synchronized void stop() throws ServiceException {
         try {
             // destroy the consumer manager.
             ConsumerManager.instance().destroy();
 
             // destroy the destination manager.
             DestinationManager.instance().destroy();
 
             // clear state
             _listeners.clear();
         } catch (Exception error) {
             error.printStackTrace();
             throw new ServiceException("Failed to stop MessageMgr : " +
                 error.toString());
         }
 
         // clear the static reference
         synchronized (_block) {
             _instance = null;
 
         }
     }
 
     /***
      * Create the specified destination. The destination is a container
      * for messages and consumers. Consumers listen for messages posted on a
      * particular destination.
      * <p>
      * This can be called multiple times without any side effects. If the
      * destination is null then it throws a JMSException
      *
      * @param destination - create this destination
      * @throws JMSException - if the params is null
      */
     public void addDestination(JmsDestination destination)
         throws JMSException {
 
         // check the methods preconditions
         if (destination == null) {
             throw new JMSException("Call to addDestination with null object");
         }
 
         DestinationManager.instance().createDestinationCache(destination);
     }
 
     /***
      * Remove this destination and all attached consumers. If the destination
      * is null then throw an exception.
      *
      * @param destination - the destination to remove
      * @throws JMSException
      */
     public void removeDestination(JmsDestination destination)
         throws JMSException {
 
         // check the method's preconditions
         if (destination == null) {
             throw new JMSException("Call to removeDestination with null object");
         }
 
         DestinationManager.instance().destroyDestinationCache(destination);
     }
 
     /***
      * Return true if the specified destination exists.
      *
      * @param destination - destination to check
      * @return boolean - true if a it exists
      */
     public boolean exists(JmsDestination destination) {
         return DestinationManager.instance().hasDestinationCache(destination);
     }
 
     /***
      * Add a message to the message manager for the specified destination.
      * If the message or the destination are null then throw a JMSException
      * <p>
      * If the destination, specified in the message, does not exist then
      * create it.
      * destinations
      *
      * @param       message             the message to add
      * @throws      JMSException        if the message cannot be added
      */
     public void add(MessageImpl message) throws JMSException {
         if (message != null) {
             // if the message is persistent then process it accordingly,
             // otherwise use the non-persistent quality of service
             if (message.isPersistent()) {
                 addPersistentMessage(message);
             } else {
                 addNonPersistentMessage(message);
             }
         } else {
             _log.error("Cannot process a null message.");
         }
     }
 
     /***
      * Add a message to the message manager for the specified destination.
      * Note that this method is called exclusively by the 
      * {@link ResourceManager} and should not be used for any other purpose.
      *
      * @param connection - this is the database connection that is used
      * @param message - the message to add
      * @throws JMSException - thrown if  there is a problem processing msg
      */
     void add(Connection connection, MessageImpl message) throws JMSException {
         if (message != null) {
             // if the message is persistent then process it accordingly,
             // otherwise use the non-persistent quality of service
             if (message.isPersistent()) {
                 addPersistentMessage(connection, message);
             } else {
                 addNonPersistentMessage(message);
             }
         } else {
             _log.error("Cannot process a null message.");
         }
     }
 
     /***
      * This method is used to process non-persistent messages.
      *
      * @param message - the message to process
      * @throws JMSException - if the message cannot be processed
      */
     protected void addNonPersistentMessage(MessageImpl message)
         throws JMSException {
         // mark the message as accepted and attach a sequence number
         message.setAcceptedTime((new Date()).getTime());
         message.setSequenceNumber(++_sequenceNumberGenerator);
         message.setReadOnly(true);
 
         // Use the message to retrieve the corresponding destination object.
         JmsDestination destination =
             (JmsDestination) message.getJMSDestination();
         if (destination != null) {
             // notify all registered listeners that a new message has arrived
             // for the specified destination.
             notifyOnAddMessage(destination, message);
             _messagesProcessed++;
         } else {
             _log.error("Can't locate destination for message");
         }
     }
 
     /***
      * This method is used to process persistent messages.
      *
      * @param message - the message to process
      * @throws JMSException - if the message cannot be processed
      */
     protected void addPersistentMessage(MessageImpl message)
         throws JMSException {
 
         // mark the message as accepted and attach a sequence number
         message.setAcceptedTime((new Date()).getTime());
         message.setSequenceNumber(++_sequenceNumberGenerator);
         message.setReadOnly(true);
 
         JmsDestination destination =
             (JmsDestination) message.getJMSDestination();
         if (destination == null) {
             throw new JMSException(
                 "Can't process message - JMSDestination is null");
         }
 
         Connection connection = null;
         TransactionManager tm = null;
 
         // do all persistent work in this block
         try {
             // get a database connection
             connection = DatabaseService.getConnection();
 
             // add the message to the database
             DatabaseService.getAdapter().addMessage(connection, message);
 
             if (destination instanceof JmsTopic) {
                 // let the consumer manager handle this
                 ConsumerManager.instance().persistentMessageAdded(
                     connection, message);
             } else {
                 // if this message is for a queue then simply create a
                 // persistent handle for the destination
                 MessageHandleFactory.createPersistentHandle(
                     connection, destination, null, message);
             }
 
             // commit the persistent message and handle
             connection.commit();
 
             // Notify all listeners that a persistent message has arrived
             notifyOnAddPersistentMessage(null, destination, message);
             _messagesProcessed++;
         } catch (Exception exception) {
             if (connection != null) {
                 try {
                     connection.rollback();
                 } catch (SQLException ignore) {
                     // no-op
                 }
             }
             _log.error("Failed to make message persistent", exception);
             throw new JMSException(
                 "Failed to make message persistent: " +
                 exception.toString());
         } finally {
             if (connection != null) {
                 try {
                     connection.close();
                 } catch (Exception ignore) {
                     // no-op
                 }
             }
         }
     }
 
     /***
      * This method is used to process persistent messages published through
      * the resource manager.
      *
      * @param connection - the database connection to use.
      * @param message - the message to process
      * @throws JMSException - if the message cannot be processed
      */
     protected void addPersistentMessage(Connection connection,
                                         MessageImpl message)
         throws JMSException {
 
         // Use the message to retrieve the corresponding destination object.
         // This method will create the object if one does not already exist.
         JmsDestination destination = (JmsDestination) message.getJMSDestination();
         if (destination != null) {
             try {
                 // notify all listeensers that a persistent message has arrived
                 notifyOnAddPersistentMessage(connection, destination, message);
                 _messagesProcessed++;
             } catch (PersistenceException exception) {
                 throw new JMSException("Failed in addPersistentMessage : " +
                     exception.toString());
             } catch (Exception exception) {
                 throw new JMSException("Failed in addPersistentMessage : " +
                     exception.toString());
             }
         } else {
             // shouldn't really get here, since the message should have been
             // checked and prepared before passed to this routine.
             _log.error("Can't locate destination for message");
         }
     }
 
     /***
      * Return the message given the specified message handle.
      * This will delegate to the appropriate {@link DestinationCache} or
      * {@link ConsumerManager}
      *
      * @param handle - the handle
      * @return MessageImpl - the associated message or null
      */
     MessageImpl getMessage(MessageHandle handle) {
         // precondition; ensure that the handle is not null
         if (handle == null) {
             return null;
         }
 
         MessageImpl message = null;
         if (handle.getDestination() instanceof JmsTopic) {
             // is for a topic so check the consumer endpoint
             // cache
             TopicConsumerEndpoint endpoint = (TopicConsumerEndpoint)
                 ConsumerManager.instance().getConsumerEndpoint(
                     handle.getConsumerName());
             if (endpoint != null) {
                 message = endpoint.getMessage(handle);
             }
         } else {
             // must be for a queue so check the destination cache
             DestinationCache cache =
                 DestinationManager.instance().getDestinationCache(
                     handle.getDestination());
             if (cache != null) {
                 message = cache.getMessage(handle);
             }
         }
 
         return message;
     }
 
     /***
      * This method prepares the message without actually passing it through the
      * system. It is used by the {@link ResourceManager} to process incoming
      * messages.
      * <p>
      * If there are any issues with the message the method will throw an
      * exception
      *
      * @param message - the message
      * @throws JMSException - if the message is invalid or cannot be prep'ed
      */
     void checkAndPrepareMessage(MessageImpl message)
         throws JMSException {
         if (message != null) {
             // mark the message as accepted and attach a sequence number
             message.setAcceptedTime((new Date()).getTime());
             message.setSequenceNumber(++_sequenceNumberGenerator);
             message.setReadOnly(true);
 
             if (message.getJMSDestination() == null) {
                 throw new JMSException("Null destination specified in message");
             }
         } else {
             throw new JMSException("checkAndPrepareMessage failed for null message");
         }
     }
 
     /***
      * Returns true if there are any messages for the specified consumer
      *
      * @param consumer - the consumer to check
      * @return boolean - true if messages are queued
      * @throws JMSException - if the consumer can't be checked
      */
     public boolean hasMessages(ConsumerEndpoint consumer) throws JMSException {
         if (consumer == null) {
             throw new JMSException(
                 "Can't call hasMessages with null consumer");
         }
         return (consumer.getMessageCount() > 0);
     }
 
     /***
      * Returns a list of active destinations
      *
      * @return List a list of JmsDestination objects
      */
     public Iterator getDestinations() {
         return DestinationManager.instance().destinations();
     }
 
     /***
      * Returns an iterator of active consumers registered to a given
      * destination
      *
      * @return Iterator - iterator of {@link ConsumerEndpoint} objects.
      * @throws JMSException
      */
     public Iterator getConsumers(JmsDestination destination)
         throws JMSException {
         //check to see that the destination is not null
         if (destination == null) {
             throw new JMSException("destination is null in getConsumer");
         }
 
         DestinationCache dest =
             DestinationManager.instance().getDestinationCache(destination);
 
         return (dest == null) ? null : dest.getConsumers();
     }
 
     /***
      * Resolves a destination given its name
      *
      * @param       name                the name of the destination
      * @return      JmsDestination      if an active destination exists for
      *                                  the given name, else it returns
      *                                  <tt>null</tt>
      */
     public JmsDestination resolve(String name) {
         return DestinationManager.instance().destinationFromString(name);
     }
 
     /***
      * Resolves a consumer given its destination and an identity. Should look
      * removing t from here.
      *
      * @param       destination         the destination
      * @param       name                the name of the consumer
      * @return      ConsumerIfc         if an active consumer exists for
      *                                  the given name, else it returns
      *                                  <tt>null</tt>
      */
     public ConsumerEndpoint resolveConsumer(JmsDestination destination,
                                             String id) {
         return ConsumerManager.instance().getConsumerEndpoint(id);
     }
 
     /***
      * Stop/start a consumer. When stopped, the consumer will not receive
      * messages until the consumer is re-started.
      * This is invoked when the underlying connection is stopped or started
      *
      * @param       consumer            the consumer to stop/start
      * @param       stop                when <tt>true</tt> stop the consumer
      *                                  else start it.
      */
     public void setStopped(ConsumerEndpoint consumer, boolean stop)
         throws JMSException {
         // need to implement this for the consumer
     }
 
     /***
      * Add a message listener for a specific destination to be informed
      * when messages, for the destination are added or removed from the
      * queue. More than one listener can be registered per desitnation
      * and the same listener can be registered for multiple destinations.
      * <p>
      * If a listener is already registered for a particuler destination
      * then it fails silently.
      *
      * @param destination - what messgaes to listen for
      * @param listener - a JmsMessageListener instance
      */
     public void addEventListener(JmsDestination destination,
                                  MessageManagerEventListener listener) {
 
         if ((destination != null) &&
             (listener != null)) {
             synchronized (_listeners) {
                 if (!_listeners.containsKey(destination)) {
                     _listeners.put(destination, listener);
                 }
             }
         }
     }
 
     /***
      * Remove the listener for the specified destination. If one is not
      * registered then ignore it.
      *
      * @param destination - destination that it listens for
      * @param listener - listener for that destination.
      */
     public void removeEventListener(JmsDestination destination,
                                     MessageManagerEventListener listener) {
         if ((destination != null) &&
             (listener != null)) {
             synchronized (_listeners) {
                 if (_listeners.containsKey(destination)) {
                     _listeners.remove(destination);
                 }
             }
         }
     }
 
     /***
      * Notify the listeners, registered for the destination that a message has
      * been added to the message manager.
      * <p>
      * All errors are propagated as JMSException exceptions
      *
      * @param destination - destination for which message exits
      * @param message - message that was added
      * @return boolean - true if the message was processed
      * @throws JMSException - for any processing error
      */
     boolean notifyOnAddMessage(JmsDestination destination,
                                MessageImpl message) throws JMSException {
         boolean result = false;
         MessageManagerEventListener listener =
             (MessageManagerEventListener) _listeners.get(destination);
 
         if (listener != null) {
             // if there is a registered destination cache then let the cache
             // process it.
             result = listener.messageAdded(destination, message);
         } else {
             // let the {@link DestinationManager handle the message
             result = DestinationManager.instance().messageAdded(destination,
                 message);
         }
 
         return result;
     }
 
     /***
      * Notify the listeners, registered for the destination that a message has
      * been removed from the message manager. There maybe several reason why
      * this has happened (i.e the message has expired, message has been
      * purged, message has been consumed etc).
      *
      * @param destination - destination for which message exits
      * @param message - message that was removed
      * @throws JMSException for any processing error
      */
     void notifyOnRemoveMessage(JmsDestination destination,
                                MessageImpl message) throws JMSException {
 
         MessageManagerEventListener listener =
             (MessageManagerEventListener) _listeners.get(destination);
 
         if (listener != null) {
             // send the notification to the active listener
             listener.messageRemoved(destination, message);
         } else {
             // there is not active listener, send it to the Destination
             // Manager
             DestinationManager.instance().messageRemoved(destination, message);
         }
     }
 
     /***
      * Notify the listeners, registered for the destination that a persistent
      * message has been added to the message manager.
      *
      * @param connection - the database connection to use.
      * @param destination - destination for which message exits
      * @param message - message that was added
      * @return boolean - true if the message was processed
      * @throws JMSException - is a processing error occured
      * @throws PersistenceException - if a persistence error occured
      */
     boolean notifyOnAddPersistentMessage(Connection connection,
                                          JmsDestination destination,
                                          MessageImpl message)
         throws JMSException, PersistenceException {
 
         boolean result = false;
         MessageManagerEventListener listener =
             (MessageManagerEventListener) _listeners.get(destination);
 
         if (listener != null) {
             // if there is a registered destination cache then let the cache
             // process it.
             result = listener.persistentMessageAdded(connection,
                 destination, message);
         } else {
             // let the {@link DestinationManager} handle the message
             result = DestinationManager.instance().persistentMessageAdded(
                 connection, destination, message);
         }
 
         return result;
     }
 
     /***
      * Notify the listeners, registered for the destination that a persistent
      * message has been removed from the message manager. There maybe several
      * reason why this has happened (i.e the message has expired, message has
      * been purged, message has been consumed etc).
      *
      * @param connection - the database connection to use
      * @param destination - destination for which message exits
      * @param message - message that was removed
      * @throws JMSException - for any processing problem
      * @throws PersistenceException - for any persistence related problem
      */
     void notifyOnRemovePersistentMessage(Connection connection,
                                          JmsDestination destination,
                                          MessageImpl message)
         throws JMSException, PersistenceException {
 
         MessageManagerEventListener listener =
             (MessageManagerEventListener) _listeners.get(destination);
 
         if (listener != null) {
             // send the notification to the active listener
             listener.persistentMessageRemoved(connection, destination,
                 message);
         } else {
             // there is not active listener, send it to the Destination
             // Manager
             DestinationManager.instance().persistentMessageRemoved(connection,
                 destination, message);
         }
     }
 
     /***
      * Return the maximum size of the cache
      *
      * @return int - maximum size of cache
      */
     public int getMaximumSize() {
         return _maximumSize;
     }
 
     /***
      * Notify the destruction of a handle.
      * <p>
      * If the handle has been destroyed then we need to do the following
      * 1. if the handle is for a queue then we can remove the message
      *    from the cache
      * 2. if the handle is for a topic then we need to see whether we can
      *    garbage collect it
      *
      * @param handle a TransientMessageHandle
      */
     public void handleDestroyed(MessageHandle handle) {
 
         // precondition: handle != null
         if (handle == null) {
             return;
         }
 
         if (handle.getDestination() instanceof JmsTopic) {
             TopicConsumerEndpoint endpoint = (TopicConsumerEndpoint)
                 ConsumerManager.instance().getConsumerEndpoint(
                     handle.getConsumerName());
 
             if (endpoint != null) {
                 endpoint.deleteMessage(handle);
             }
         } else {
             DestinationCache cache =
                 DestinationManager.instance().getDestinationCache(
                     handle.getDestination());
 
             if (cache != null) {
                 cache.deleteMessage(handle);
             }
         }
     }
 }