/***
    * 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: QueueDestinationCache.java,v 1.34.2.1 2004/05/01 12:05:26 tanderson Exp $
   *
   * Date         Author  Changes
   * 3/1/2001     jima    Created
   */
  package org.exolab.jms.messagemgr;
  
  import java.sql.Connection;
  import java.util.Collections;
  import java.util.Enumeration;
  import java.util.Iterator;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Vector;
  
  import javax.jms.JMSException;
  import javax.transaction.TransactionManager;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import org.exolab.jms.client.JmsDestination;
  import org.exolab.jms.client.JmsQueue;
  import org.exolab.jms.client.JmsTemporaryDestination;
  import org.exolab.jms.message.MessageHandle;
  import org.exolab.jms.message.MessageImpl;
  import org.exolab.jms.persistence.DatabaseService;
  import org.exolab.jms.persistence.PersistenceException;
  import org.exolab.jms.selector.Selector;
  import org.exolab.jms.server.JmsServerConnectionManager;
  
  
  /***
   * A DestinationCache for Queues
   *
   * @version     $Revision: 1.34.2.1 $ $Date: 2004/05/01 12:05:26 $
   * @author      <a href="mailto:jima@exoffice.com">Jim Alateras</a>
   */
  public class QueueDestinationCache
      extends DestinationCache {
  
      /***
       * Maintains a list of queue listeners for this cache
       */
      protected List _queueListeners =
          Collections.synchronizedList(new LinkedList());
  
      /***
       * Underlying destination
       */
      private JmsQueue _queue = null;
  
      /***
       * Index of the last listener that received a message from this
       * destination. If multiple listeners are attached to this queue then
       * messages will be sent to each in a round robin fashion
       */
     private int _lastConsumerIndex = 0;
 
     /***
      * Tracks the number of messages added to the destination cache
      */
     private long _publishCount;
 
     /***
      * Tracks the number of messages consumed from the destination cache
      */
     private long _consumeCount;
 
     /***
      * The logger
      */
     private static final Log _log = LogFactory.getLog(
         QueueDestinationCache.class);
 
 
     /***
      * Construct a message cache for a queue destination. This cache will
      * receive all messages published under the specified destination.
      * <p>
      * the constructor will also attempt to load any persistent messages
      * from the database.
      * <p>
      *
      * @param destination the queue
      * @throws FailedToInitializeException
      */
     QueueDestinationCache(JmsQueue destination)
         throws FailedToInitializeException {
 
         _queue = destination;
         // call init on the base class
         init();
         if (DestinationManager.instance().isAdministeredDestination(destination.getName())) {
             // call the persistence adapter to determine if we have unacknowledged
             // messages for this queue. The persistent handles are keyed on the
             // queue name
             Connection connection = null;
             TransactionManager tm = null;
             try {
                 connection = DatabaseService.getConnection();
 
                 // initialise the cache
                 init(destination, connection);
 
                 // commit the work
                 connection.commit();
             } catch (PersistenceException exception) {
                 if (connection != null) {
                     try {
                         connection.rollback();
                     } catch (Exception nested) {
                         // ignore
                     }
                 }
                 throw new FailedToInitializeException(
                     "QueueDestinationCache init failed " + exception);
             } catch (Exception exception) {
                 // rethrow as a JMSException
                 throw new FailedToInitializeException(
                     "QueueDestinationCache init failed " + exception);
             } finally {
                 if (connection != null) {
                     try {
                         connection.close();
                     } catch (Exception nested) {
                         // ignore
                     }
                 }
             }
         }
     }
 
 
     /***
      * Construct a message cache for a queue destination using the specified
      * database connection. This cache will receive all messages published
      * under the destination.
      * <p>
      * the constructor will also attempt to load any persistent messages
      * from the database using the specified connection
      * <p>
      * If there is any problem during construction a FailedToInitializeException
      * will be raised.
      *
      * @param destination - the queue
      * @throws FialedToInitializeException
      */
     QueueDestinationCache(Connection connection, JmsQueue destination)
         throws FailedToInitializeException {
         super();
         _queue = destination;
 
         // call init on the base class
         init(connection);
         if (DestinationManager.instance().isAdministeredDestination(destination.getName())) {
             // call the persistence adapter to determine if we have unacknowledged
             // messages for this queue. The persistent handles are keyed on the
             // queue name
             try {
                 // initialise the cache
                 init(destination, connection);
             } catch (Exception exception) {
                 // rethrow as a JMSException
                 throw new FailedToInitializeException(
                     "QueueDestinationCache init failed " + exception);
             }
         }
     }
 
 
     /***
      * This common method is used to help initialise the cache. It basically
      * removes all the expired messages and then retrieves all unacked messages
      * from the database and stores them locally.
      * <p>
      * It will throw a PersistenceException if there is a database related
      * problem.
      *
      * @param destination - the queue
      * @param connection - the database connection to use
      * @throws PersistenceException
      */
     void init(JmsQueue destination, Connection connection)
         throws PersistenceException {
 
         DatabaseService.getAdapter().removeExpiredMessageHandles(
             connection, destination.getName());
         Vector handles = DatabaseService.getAdapter().getMessageHandles(
             connection, destination, destination.getName());
         if (handles != null) {
             Enumeration iter = handles.elements();
             while (iter.hasMoreElements()) {
                 addMessage((MessageHandle) iter.nextElement());
             }
         }
     }
 
     // implementation of DestinationCache.getDestination
     public JmsDestination getDestination() {
         return _queue;
     }
 
     /***
      * A Queue can also hav a queue listener, which simply gets informed of all
      * messages that arrive at this destination
      *
      * @param listener - queue listener
      */
     public void addQueueListener(QueueListener listener) {
         // add if not present
         if (!_queueListeners.contains(listener)) {
             _queueListeners.add(listener);
         }
     }
 
     /***
      * Remove the queue listener associated with this cache
      *
      * @param listener - queue listener to remove
      */
     public void removeQueueListener(QueueListener listener) {
         // add if not present
         if (_queueListeners.contains(listener)) {
             _queueListeners.remove(listener);
         }
     }
 
     // implementation of MessageMgr.messageAdded
     public boolean messageAdded(JmsDestination destination,
                                 MessageImpl message) {
         boolean processed = false;
 
         if ((destination != null) &&
             (message != null)) {
 
             // check that the message is for this queue
             if (destination.equals(_queue)) {
 
                 // create a handle for the message
                 try {
                     MessageHandle handle =
                         MessageHandleFactory.createHandle(this, message);
 
 
                     // all messages are added to this queue. Receivers will
                     // then pick messages of it as required.
                     addMessage(handle, message);
 
                     // update the publishedCount
                     _publishCount++;
 
                     // if we have any registered consumers then we need to
                     // send the message to one of them first. If none are
                     // registered then cache it.
                     QueueConsumerEndpoint endpoint =
                         getEndpointForMessage(message);
                     if (endpoint != null) {
                         endpoint.messageAdded(message);
                     }
 
                     // notify any queue listeners that a message has arrived
                     notifyQueueListeners(message);
 
                     // create a lease iff one is required
                     checkMessageExpiry(message);
 
                     // check the message as processed
                     processed = true;
                 } catch (JMSException exception) {
                     _log.error("Failed to add message", exception);
                 }
             } else {
                 // need to notify someone or something that we are
                 // dropping messages. Do we throw an exception
                 _log.error("Dropping message " + message.getMessageId()
                     + " for destination " + destination.getName());
             }
         }
 
         return processed;
     }
 
     /***
      * This method is called when the {@link MessageMgr} removes a message
      * from the cache.
      *
      * @param destination the message destination
      * @param message the message removed from cache
      */
     public void messageRemoved(JmsDestination destination,
                                MessageImpl message) {
 
         if ((destination != null) &&
             (message != null)) {
 
             try {
                 MessageHandle handle =
                     MessageHandleFactory.getHandle(this, message);
 
                 // call remove regardless whether it exists
                 if (destination.equals(_queue)) {
                     removeMessage(handle);
                     notifyOnRemoveMessage(message);
                     handle.destroy();
                 }
             } catch (JMSException exception) {
                 _log.error("Failed to remove message", exception);
             }
         }
     }
 
     // implementation of MessageMgr.persistentMessageAdded
     public boolean persistentMessageAdded(Connection connection,
                                           JmsDestination destination,
                                           MessageImpl message)
         throws PersistenceException {
 
         boolean processed = false;
 
         if ((destination != null) &&
             (message != null)) {
 
             // check that it is not already present before adding it.
             if (destination.equals(_queue)) {
 
                 // create a handle for the message
                 try {
 
                     // all messages are added to this queue. Receivers will
                     // then pick messages of it as required.
                     MessageHandle handle =
                         MessageHandleFactory.getHandle(this, message);
                     addMessage(handle, message);
 
                     // increment the number of messages received
                     _publishCount++;
 
                     // if we have any registered consumers then we need to
                     // send the message to one of them first. If none are
                     // registered then cache it.
                     QueueConsumerEndpoint endpoint =
                         getEndpointForMessage(message);
                     if (endpoint != null) {
                         endpoint.persistentMessageAdded(connection, message);
                     }
 
                     // notify any queue listeners that a message has arrived
                     notifyQueueListeners(message);
 
                     // create a lease iff one is required
                     checkMessageExpiry(message);
 
                     // check the message as processed
                     processed = true;
                 } catch (JMSException exception) {
                     _log.error("Failed to add persistent message",
                         exception);
                 }
             } else {
                 // need to notify someone or something that we are
                 // dropping messages. Do we throw an exception
             }
         }
 
         return processed;
     }
 
     // implementation of MessageMgr.persistentMessageAdded
     public synchronized void persistentMessageRemoved(
         Connection connection, JmsDestination destination,
         MessageImpl message)
         throws PersistenceException {
 
         if ((destination != null) &&
             (message != null)) {
 
             try {
                 PersistentMessageHandle handle = (PersistentMessageHandle)
                     MessageHandleFactory.getHandle(this, message);
 
                 // call remove regardless whether it exists
                 if (destination.equals(_queue)) {
                     removeMessage(handle);
                     notifyOnRemovePersistentMessage(connection, message);
                     MessageHandleFactory.destroyPersistentHandle(connection,
                         handle);
                 }
             } catch (JMSException exception) {
                 _log.error("Failed to remove persistent message", exception);
             }
         }
     }
 
     /***
      * Return the next {@link ConsumerEndpoint} that can consume the specified
      * message or null if there is none.
      *
      * @param  message - the message to consume
      * @return the consumer who should receive this message or null
      */
     private synchronized QueueConsumerEndpoint getEndpointForMessage(
         MessageImpl message) {
         QueueConsumerEndpoint selectedEndpoint = null;
 
         if (_consumers.size() > 0) {
             // roll over the consumer index if it is greater
             // than the number of registered consumers
             if ((_lastConsumerIndex + 1) > _consumers.size()) {
                 _lastConsumerIndex = 0;
             }
 
             // look over the list of consumers and return the
             // first endpoint that can process this message
             int index = _lastConsumerIndex;
             do {
                 QueueConsumerEndpoint endpoint =
                     (QueueConsumerEndpoint) _consumers.get(index);
                 Selector selector = endpoint.getSelector();
 
                 // if the endpoint has a message listener registered
                 // or the endpoint is waiting for a message and the
                 // message satisfies the selector then return it to
                 // the client.
                 if (((endpoint.hasMessageListener()) ||
                     (endpoint.isWaitingForMessage())) &&
                     ((selector == null) ||
                     (selector.selects(message)))) {
                     _lastConsumerIndex = ++index;
                     selectedEndpoint = endpoint;
                     break;
                 }
 
                 // advance to the next consumer
                 if (++index >= _consumers.size()) {
                     index = 0;
                 }
             } while (index != _lastConsumerIndex);
         }
 
         return selectedEndpoint;
     }
 
     /***
      * Return the first message of the queue or null if there are no messages
      * in the cache
      *
      * @param QueueConsumerEndpoint - the consumer who will receive the message
      * @return MessageHandle - handle to the first message
      */
     public synchronized MessageHandle getMessage(
         QueueConsumerEndpoint endpoint) {
         MessageHandle handle = null;
         // do not return a message is the endpoint is null;
         if ((endpoint != null) &&
             (getMessageCount() > 0)) {
             Selector selector = endpoint.getSelector();
             if (selector == null) {
                 // if no selector has been specified then remove and return
                 // the first message
                 handle = removeFirstMessage();
                 _consumeCount++;
             } else {
                 // for non null selector we must find the first matching
                 Object[] handles = toMessageArray();
                 for (int i = 0; i < handles.length; ++i) {
                     MessageHandle hdl = (MessageHandle) handles[i];
                     MessageImpl message = hdl.getMessage();
                     if (message != null && selector.selects(message)) {
                         handle = hdl;
                         removeMessage(hdl);
                         _consumeCount++;
                         break;
                     }
                 }
             }
         }
 
         return handle;
     }
 
     /***
      * Playback all the messages in the cache to the specified
      * {@link QueueListener}
      *
      * @param listener - the queue listener
      */
     public void playbackMessages(QueueListener listener) {
 
         Object[] messages = toMessageArray();
         if ((listener != null) &&
             (messages.length > 0)) {
             try {
                 for (int index = 0; index < messages.length; index++) {
                     listener.onMessage(((MessageHandle) messages[index]).getMessage());
                 }
             } catch (IndexOutOfBoundsException exception) {
                 // ignore the exception since the list is dynamic and may
                 // be modified while it is being processed.
             }
         }
     }
 
     /***
      * Return the specified message to top of the queue. This is called to
      * recover unsent or unacked messages
      *
      * @param message - message to return
      */
     public synchronized void returnMessage(MessageHandle handle) {
 
         // add the message to the destination cache
         addMessage(handle);
 
         // if there are registered consumers then check whether
         // any of them have registered message listeners
         if (_consumers.size() > 0) {
             // roll over the consumer index if it is greater
             // than the number of registered consumers
             if ((_lastConsumerIndex + 1) > _consumers.size()) {
                 _lastConsumerIndex = 0;
             }
 
             int index =
                 (_lastConsumerIndex >= _consumers.size()) ?
                 0 : _lastConsumerIndex;
 
             do {
 
                 QueueConsumerEndpoint endpoint =
                     (QueueConsumerEndpoint) _consumers.get(index);
 
                 // if we find an endpoint with a listener then
                 // we should reschedule it.
                 if (endpoint.hasMessageListener()) {
                     endpoint.schedule();
                     _lastConsumerIndex = ++index;
                     break;
                 }
 
                 // advance to the next consumer
                 if (++index >= _consumers.size()) {
                     index = 0;
                 }
             } while (index != _lastConsumerIndex);
         }
     }
 
     /***
      * Notify all the queue listeners, that this message has arrived. This is
      * ideal for browsers and iterators
      *
      * @param message - message to deliver
      */
     void notifyQueueListeners(MessageImpl message) {
         if (!_queueListeners.isEmpty()) {
             QueueListener[] listeners =
                 (QueueListener[]) _queueListeners.toArray(
                     new QueueListener[0]);
 
             int size = listeners.length;
             for (int index = 0; index < size; ++index) {
                 QueueListener listener = listeners[index];
                 if (listener instanceof QueueBrowserEndpoint) {
                     QueueBrowserEndpoint browser =
                         (QueueBrowserEndpoint) listener;
                     Selector selector = browser.getSelector();
 
                     // if a selector has been specified then apply the filter
                     // before sending down the message
                     if ((selector == null) ||
                         (selector.selects(message))) {
                         browser.onMessage(message);
                     }
                 } else {
                     // if there is any other type of subscriber then just
                     // send the message to it.
                     listener.onMessage(message);
                 }
             }
         }
     }
 
     // implementation of DestinationCache.notifyOnAddMessage
     boolean notifyOnAddMessage(MessageImpl message) {
         return true;
     }
 
     // implementation of DestinationCache.notifyOnRemoveMessage
     void notifyOnRemoveMessage(MessageImpl message) {
     }
 
     // implementation of DestinationCache.hasActiveConsumers
     boolean hasActiveConsumers() {
         boolean active = true;
         if (_queueListeners.isEmpty() && _consumers.isEmpty()) {
             active = false;
         }
         if (_log.isDebugEnabled()) {
             _log.debug("hasActiveConsumers()[queue=" + _queue + "]=" + active);
         }
         return active;
     }
 
     /***
      * Determines if this cache can be destroyed.
      * A <code>QueueDestinationCache</code> can be destroyed if there
      * are no active consumers and:
      * <ul>
      *   <li>the queue is persistent and there are no messages</li>
      *   <li>
      *     the queue is temporary and the corresponding connection is closed
      *   </li>
      * </ul>
      *
      * @return <code>true</code> if the cache can be destroyed, otherwise
      * <code>false</code>
      */
     public boolean canDestroy() {
         boolean destroy = false;
         if (!hasActiveConsumers()) {
             JmsDestination queue = getDestination();
             if (queue.getPersistent() && getMessageCount() == 0) {
                 destroy = true;
             } else if (queue.isTemporaryDestination()) {
                 // check if there is a corresponding connection. If
                 // not, it has been closed, and the cache can be removed
                 String connectionId =
                     ((JmsTemporaryDestination) queue).getConnectionId();
                 JmsServerConnectionManager manager =
                     JmsServerConnectionManager.instance();
                 if (manager.getConnection(connectionId) == null) {
                     destroy = true;
                 }
             }
         }
         return destroy;
     }
 
     /***
      * Destroy this object
      */
     synchronized void destroy() {
         super.destroy();
         _queueListeners.clear();
     }
 
     // override Object.toString
     public String toString() {
         return _queue.toString();
     }
 
     // override Object.hashCode
     public int hashCode() {
         return _queue.hashCode();
     }
 
 } //-- QueueDestinationCache