/***
    * 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: TopicConsumerEndpoint.java,v 1.40 2003/10/21 14:41:23 tanderson Exp $
   *
   * Date         Author  Changes
   * 3/1/2001     jima    Created
   */
  package org.exolab.jms.messagemgr;
  
  import java.sql.Connection;
  import java.util.HashMap;
  import java.util.Iterator;
  
  import javax.jms.InvalidSelectorException;
  import javax.jms.JMSException;
  import javax.jms.MessageListener;
  import javax.jms.Session;
  import javax.transaction.TransactionManager;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import org.exolab.jms.JMSErrorCodes;
  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;
  import org.exolab.jms.scheduler.Scheduler;
  import org.exolab.jms.server.ClientDisconnectionException;
  import org.exolab.jms.server.JmsServerSession;
  
  
  /***
   * The topic destination cache is specific to a consumer that
   * subscribes to a topic destination. 
   *
   * @version     $Revision: 1.40 $ $Date: 2003/10/21 14:41:23 $
   * @author      <a href="mailto:jima@exoffice.com">Jim Alateras</a>
   */
  public class TopicConsumerEndpoint
      extends ConsumerEndpoint
      implements DestinationEventListener {
  
      /***
       * Maintains a list of TopicDestinationCache that this endpoint subscribes
       * too. A wildcard subscription may point to more than one.
       */
      protected HashMap _caches = new HashMap();
  
      /***
       * Cache the destination that this endpoint subscribes too
       */
      protected JmsTopic _topic = null;
  
      /***
       * The maximum number of messages that a dispatch can deliver at any one
       * time
       */
     private final int MAX_MESSAGES = 200;
 
     /***
      * The logger
      */
     private static final Log _log =
         LogFactory.getLog(TopicConsumerEndpoint.class);
 
 
     /***
      * Construct a new <code>TopicConsumerEndpoint</code>.<p/>
      * The destination and selector determine where it will be sourcing
      * its messages from, and scheduler is used to asynchronously deliver
      * messages to the consumer.
      *
      * @param session - the owning session
      * @param clientId - uniquely identifies the remote client within session
      * @param topic - destination that this object was created for
      * @param selector - the selector attached to the consumer, if any.
      * @param scheduler - used to schedule async message delivery.
      * @throws InvalidSelectorException - if the selector is invalid
      * @throws JMSException - for other general exceptions
      */
     TopicConsumerEndpoint(JmsServerSession session, long clientId,
                           JmsTopic topic, String selector, Scheduler scheduler)
         throws InvalidSelectorException, JMSException {
         super(session, clientId, selector, scheduler);
 
         // register the endpoint with the destination
         DestinationManager destmgr = DestinationManager.instance();
         if (topic != null) {
             _topic = topic;
 
             if (_topic.isWildCard()) {
                 // if the topic is a wild card then we need to retrieve a
                 // set of matching destination caches.
                 _caches = destmgr.getTopicDestinationCaches(topic);
                 // for each cache register this endpoint as a consumer of
                 // it's messages. Before doing so register as a destination
                 // event listener with the DestinationManager
                 destmgr.addDestinationEventListener(this);
                 Iterator iter = _caches.values().iterator();
                 while (iter.hasNext()) {
                     ((TopicDestinationCache) iter.next()).registerConsumer(this);
                 }
             } else {
                 // if the topic is not a wildcard then we need to get the
                 // destination cache. If one does not exist then we need to
                 // create it.
                 DestinationCache cache = destmgr.getDestinationCache(topic);
                 if (cache == null) {
                     cache = destmgr.createDestinationCache(topic);
                 }
                 _caches.put(topic, cache);
                 cache.registerConsumer(this);
             }
         }
     }
 
     /***
      * Deliver messages in the cache to the consumer
      *
      * @return <code>true</code> if the endpoint should be rescheduled
      */
     public boolean deliverMessages() {
         boolean reschedule = true;
 
         for (int index = 0; index < MAX_MESSAGES;) {
             
             // check if we should exit the loop
             if (stopDelivery()) {
                 reschedule = false;
                 break;
             }
 
             // Process the first message on the list.
             boolean processed = true;
             MessageHandle handle = removeFirstMessage();
             try {
                 if (_selector != null) {
                     MessageImpl m = handle.getMessage();
                     if ((m != null) && _selector.selects(m)) {
                         // this message has been selected by the selector
                         handle.setClientId(getClientId());
                         _listener.onMessage(handle, true);
                         index++;
                     } else {
                         // this message has not been selected
                         handle.destroy();
                     }
                 } else {
                     // send the message to the consumer
                     handle.setClientId(getClientId());
                     _listener.onMessage(handle, true);
                     index++;
                 }
             } catch (ClientDisconnectionException exception) {
                 _listener = null;
                 returnMessage(handle);
             } catch (JMSException exception) {
                 if (exception.getErrorCode().equals(
                         JMSErrorCodes.FailedToResolveHandle)) {
                     // do not return message back to the cache
                     _log.error("Dropping handle " + handle
                                + " since we cannot resolve it.");
                 } else {
                     _log.error(exception, exception);
                     returnMessage(handle);
                 }
             } catch (Exception exception) {
                 _log.error(exception, exception);
                 returnMessage(handle);
             }
         }
         return reschedule;
     }
 
     // implementation of ConsumerEndpoint.receiveMessage
     public MessageHandle receiveMessage(long wait) {
         synchronized (_waitingForMessageMonitor) {
             MessageHandle handle = receiveNoWait();
             if ((handle == null) &&
                 (wait >= 0)) {
                 // no message is currently available for this
                 // consumer. So set the flag to indicate that
                 // the consumer is waiting for a message
                 setWaitingForMessage();
             }
 
             return handle;
         }
     }
 
     /***
      * Returns the first available message
      */
     public MessageHandle receiveNoWait() {
         MessageHandle handle = null;
         while ((handle = removeFirstMessage()) != null) {
             // make sure we can still get access to the message
             MessageImpl message = handle.getMessage();
             if (message != null) {
                 if (_selector == null || _selector.selects(message)) {
                     // got a message which is applicable for the endpoint
                     handle.setClientId(getClientId());
                     break;
                 } else {
                     // this message has been filtered out so we can destroy
                     // the handle.
                     handle.destroy();
                 }
             }
             handle = null;
         }
         return handle;
     }
 
     // override ConsumerEndpoint.messageRemoved
     public boolean messageRemoved(MessageImpl message) {
 
         boolean result = false;
 
         if (message != null) {
             try {
                 MessageHandle handle =
                     MessageHandleFactory.getHandle(this, message);
 
                 // if the message in the main cache then destroy the
                 // handle
                 if (removeMessage(handle)) {
                     if (handle.getConsumerName() == null) {
                         // @todo - the follwoing is required in order for 
                         // MessageMgr to resolve this endpoint when the 
                         // handle is destroyed. However, its not clear
                         // why topics need to go through MessageMgr at all...
                         handle.setConsumerName(getPersistentId());
                     }
                     handle.destroy();
                     result = true;
                 }
 
             } catch (JMSException exception) {
                 _log.error("Failed to remove message from endpoint",
                     exception);
             }
         }
 
         return result;
     }
 
     // override ConsumerEndpoint.messageRemoved
     public boolean persistentMessageRemoved(Connection connection,
                                             MessageImpl message)
         throws PersistenceException {
 
         boolean result = false;
 
         if (message != null) {
             try {
                 PersistentMessageHandle handle = (PersistentMessageHandle)
                     MessageHandleFactory.getHandle(this, message);
 
                 // if the message in the main cache then destroy the handle
                 if (removeMessage(handle)) {
                     MessageHandleFactory.destroyPersistentHandle(connection,
                         handle);
                     result = true;
                 }
 
             } catch (JMSException exception) {
                 _log.error("Failed to remove persistent message from endpoint",
                     exception);
             }
         }
 
         return result;
     }
 
     // implementation of ConsumerEndpoint.getDestination
     public JmsDestination getDestination() {
         return _topic;
     }
 
     // override ConsumerEndpoint.unregister
     public void unregister() {
         Iterator iter = _caches.values().iterator();
         while (iter.hasNext()) {
             ((TopicDestinationCache) iter.next()).unregisterConsumer(this);
         }
         _caches.clear();
     }
 
     public void recover() {
         // no-op
     }
 
     // implementation of DestinationEventListener.destinationAdded
     public void destinationAdded(JmsDestination destination,
                                  DestinationCache cache) {
         if ((destination instanceof JmsTopic) &&
             (_topic.match((JmsTopic) destination)) &&
             (!_caches.containsKey(destination))) {
             synchronized (_caches) {
                 _caches.put(destination, cache);
             }
 
             ((TopicDestinationCache) cache).registerConsumer(this);
             _log.debug(_topic + " registered with " + destination);
         }
     }
 
     // implementation of DestinationEventListener.destinationRemoved
     public void destinationRemoved(JmsDestination destination,
                                    DestinationCache cache) {
         if ((destination instanceof JmsTopic) &&
             (_caches.containsKey(destination))) {
             synchronized (_caches) {
                 _caches.remove(destination);
             }
             _log.debug(_topic + " unregistered with " + destination);
         }
     }
 
     /***
      * Closes this endpoint
      */
     protected void doClose() {
         // unregister as a destination event listener
         DestinationManager.instance().removeDestinationEventListener(this);
         // unregister from the destination before continuing
         Iterator iter = _caches.values().iterator();
         while (iter.hasNext()) {
             ((TopicDestinationCache) iter.next()).unregisterConsumer(this);
         }
         _caches.clear();
     }
 
 } //-- TopicConsumerEndpoint