/***
    * 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: TopicDestinationCache.java,v 1.22 2003/10/21 14:41:24 tanderson Exp $
   *
   * Date         Author  Changes
   * 3/1/2001     jima    Created
   */
  package org.exolab.jms.messagemgr;
  
  import java.sql.Connection;
  import java.util.Iterator;
  import java.util.List;
  
  import javax.jms.DeliveryMode;
  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.JmsTopic;
  import org.exolab.jms.message.MessageHandle;
  import org.exolab.jms.message.MessageImpl;
  import org.exolab.jms.persistence.PersistenceException;
  
  
  /***
   * A DestinationCache for Topics. This cache extends DestinationCache but does
   * not actually hold a reference to the messages. Instead it forwards them on
   * to registered consumers.
   * <p>
   * We may need to build the cache for clients that fail over to the new server
   * so we should maintain a cache of at least persistent messages handles. This
   * is something that needs to be considered. As for non-persistent messages well
   * that is the penalty for using them. If you want to ensure that you get every
   * message, even during failover then you best publsih them using persistent
   * delivery mode.
   *
   * @version     $Revision: 1.22 $ $Date: 2003/10/21 14:41:24 $
   * @author      <a href="mailto:jima@exoffice.com">Jim Alateras</a>
   **/
  public class TopicDestinationCache
      extends DestinationCache {
  
      /***
       * Underlying destination
       */
      private JmsTopic _topic = null;
  
      /***
       * The logger
       */
      private static final Log _log =
          LogFactory.getLog(TopicDestinationCache.class);
  
  
      /***
       * Construct a message cache for a particular destination. This object
      * does not cache any messages.
      *
      * @param destination - the destination that owns this cache
      * @throws FailedToInitializeException
      */
     TopicDestinationCache(JmsTopic destination)
         throws FailedToInitializeException {
         super();
 
         // don't need the local messages cache for the initial release, will need
         // it for clustering
         _topic = destination;
 
         // call init on the base class
         try {
             init();
         } catch (FailedToInitializeException exception) {
             // rethrow
             throw exception;
         } catch (Exception exception) {
             throw new FailedToInitializeException(
                 "Failed to construct TopicDestinationCache " +
                 exception.toString());
         }
     }
 
     /***
      * Construct a message cache for a particular destination using the specified
      * {@link Connection} to complete any database access.
      *
      * @param connection - the connection to use.
      * @param destination - the destination that owns this cache
      * @throws FailedToInitializeException
      */
     TopicDestinationCache(Connection connection, JmsTopic destination)
         throws FailedToInitializeException {
         super();
 
         // don't need the local messages cache for the initial release, will need
         // it for clustering
         _topic = destination;
 
         // call init on the base class
         try {
             init(connection);
         } catch (FailedToInitializeException exception) {
             // rethrow
             throw exception;
         } catch (Exception exception) {
             throw new FailedToInitializeException(
                 "Failed to construct TopicDestinationCache " +
                 exception.toString());
         }
     }
 
     // implementation of DestinationCache.getDestination
     public JmsDestination getDestination() {
         return _topic;
     }
 
     // implementation of MessageMgr.messageAdded
     public boolean messageAdded(JmsDestination destination, MessageImpl message) {
         boolean processed = false;
 
         if ((destination != null) &&
             (message != null)) {
             // check that it is not already present before adding it.
             if (destination.equals(_topic)) {
                 processed = notifyOnAddMessage(message);
                 // create a lease iff one is required and the message has actually
                 // been accepted by at least one endpoint
                 if (processed) {
                     checkMessageExpiry(message);
                 }
             }
         }
 
         return processed;
     }
 
     // implementation of MessageMgr.messageRemoved
     public void messageRemoved(JmsDestination destination, MessageImpl message) {
         if ((destination != null) &&
             (message != null)) {
             // call remove regardless whether it exists
             if (destination.equals(_topic)) {
                 notifyOnRemoveMessage(message);
             }
         }
     }
 
     // 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(_topic)) {
                 processed = notifyOnAddPersistentMessage(connection, message);
 
                 // create a lease iff one is required and the message has actually
                 // been accepted by at least one endpoint
                 if (processed) {
                     checkMessageExpiry(message);
                 }
             }
         }
 
         return processed;
     }
 
     // implementation of MessageMgr.persistentMessageRemoved
     public void persistentMessageRemoved(Connection connection,
                                          JmsDestination destination, MessageImpl message)
         throws PersistenceException {
         if ((destination != null) &&
             (message != null)) {
             // call remove regardless whether it exists
             if (destination.equals(_topic)) {
                 notifyOnRemovePersistentMessage(connection, message);
             }
         }
     }
 
     // implementation of DestinationCache.notifyOnAddMessage
     boolean notifyOnAddMessage(MessageImpl message) {
         boolean processed = true;
 
         // process for all the active consumers.
         Object[] iter = getConsumersByArray();
         for (int index = 0; index < iter.length; index++) {
             DestinationCacheEventListener listener =
                 (DestinationCacheEventListener) iter[index];
             processed |= listener.messageAdded(message);
         }
 
         return processed;
     }
 
     // implementation of DestinationCache.notifyOnRemoveMessage
     void notifyOnRemoveMessage(MessageImpl message) {
         Object[] iter = getConsumersByArray();
         for (int index = 0; index < iter.length; index++) {
             if (((DestinationCacheEventListener)
                 iter[index]).messageRemoved(message)) {
             }
         }
     }
 
     // implementation of DestinationCache.notifyOnAddPersistentMessage
     boolean notifyOnAddPersistentMessage(Connection connection,
                                          MessageImpl message)
         throws PersistenceException {
         boolean processed = true;
 
         // This is not done in MessageMgr
         // first let's create a persistent message handle for all registered
         // durable consumers
         //try {
         //    processed |= ConsumerManager.instance().persistentMessageAdded(
         //        connection, message);
         //} catch (Exception exception) {
         //    throw new PersistenceException("Error in notifyOnAddPersistentMessage " +
         //                                   exception);
         //}
 
         // now send the message to all active durable and non-durable consumers
         Object[] iter = getConsumersByArray();
         for (int index = 0; index < iter.length; index++) {
             DestinationCacheEventListener listener =
                 (DestinationCacheEventListener) iter[index];
             processed |= listener.persistentMessageAdded(connection, message);
         }
 
         return processed;
     }
 
     // implementation of DestinationCache.notifyOnRemovePersistentMessage
     void notifyOnRemovePersistentMessage(Connection connection,
                                          MessageImpl message)
         throws PersistenceException {
 
         Object[] iter = getConsumersByArray();
         for (int index = 0; index < iter.length; index++) {
             DestinationCacheEventListener listener =
                 (DestinationCacheEventListener) iter[index];
             // if the listener is of type {@link DurableConsumerEndpoint} then
             // indicate that a persistentMessage has been removed. If the listener
             // is a {@link TopicConsumerEndpoint} then indicatw that non-
             // persistent message has been removed.
             if (listener instanceof DurableConsumerEndpoint) {
                 listener.persistentMessageRemoved(connection, message);
             } else if (listener instanceof TopicConsumerEndpoint) {
                 listener.messageRemoved(message);
             }
         }
 
         // since it is a persistent message we need to send it
         // to inactive durable consumers subscribing to the destination
         try {
             ConsumerManager.instance().persistentMessageRemoved(connection,
                 message);
         } catch (Exception exception) {
             _log.error("Error in notifyOnRemovePersistentMessage",
                 exception);
         }
     }
 
     // override implementation of DestinationCache.registerConsumer
     public boolean registerConsumer(ConsumerEndpoint consumer) {
 
         boolean result = false;
 
         // check to see that the consumer can actually subscribe to
         // this destination
         JmsTopic cdest = (JmsTopic) consumer.getDestination();
         JmsTopic ddest = (JmsTopic) getDestination();
 
         if (cdest.match(ddest)) {
             if (!_consumers.contains(consumer)) {
                 _consumers.add(consumer);
                 consumer.setMaximumSize(this.getMaximumSize());
                 result = true;
             }
         }
 
         return result;
     }
 
     // implementation of DestinationCache.hasActiveConsumers
     boolean hasActiveConsumers() {
         return (_consumers.size() > 0);
     }
 
     // implementation of DestinationCache.getMessageCount
     public int getMessageCount() {
         return 0;
     }
 
     /***
      * Determines if this cache can be destroyed.
      * A <code>TopicDestinationCache</code> can be destroyed if there
      * are no active consumers.
      *
      * @return <code>true</code> if the cache can be destroyed, otherwise
      * <code>false</code>
      */
     public boolean canDestroy() {
         return !hasActiveConsumers();
     }
 
     // override Object.toString
     public String toString() {
         return _topic.toString();
     }
 
     // override Object.hashCode
     public int hashCode() {
         return _topic.hashCode();
     }
 
     /***
      * Resolve an expired message through its handle
      *
      * @param handle the expired message's handle
      * @return the expired message. May be null.
      */
     protected MessageImpl resolveExpiredMessage(MessageHandle handle) {
         MessageImpl message = null;
 
         if (handle.getConsumerName() != null) {
             message = super.resolveExpiredMessage(handle);
         } else {
             // can't resolve the message through the handle 
             // Need to find the first consumer which has it.
             // @todo - design flaw!
             Iterator iterator = _consumers.iterator();
             while (iterator.hasNext()) {
                 ConsumerEndpoint endpoint = (ConsumerEndpoint) iterator.next();
                 message = endpoint.getMessage(handle);
                 if (message != null) {
                     break;
                 }
             }
         }
         return message;
     }
 
 }