/*
    * Copyright (C) The Apache Software Foundation. All rights reserved.
    *
    * This software is published under the terms of the Apache Software License
    * version 1.1, a copy of which has been included with this distribution in
    * the LICENSE file.
    * 
    * $Id: ConsumerThread.java 155459 2005-02-26 13:24:44Z dirkv $
    */
  package org.apache.commons.messagelet;
  
  import javax.jms.Destination;
  import javax.jms.JMSException;
  import javax.jms.Message;
  import javax.jms.MessageConsumer;
  import javax.jms.MessageListener;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.commons.messenger.Messenger;
  
  /** 
   * <p><code>ConsumerThread</code> is a thread which will repeatedly consume JMS messages
   * using a receive() method on Messenger and then process the message.
   * This class is a good base class when implementing some kind of transactional processing of 
   * JMS messages
   *
   * @author <a href="mailto:jstrachan@apache.org">James Strachan</a>
   * @version $Revision: 155459 $
   */
  public class ConsumerThread extends Thread {
  
      /** Logger */
      private static final Log log = LogFactory.getLog(ConsumerThread.class);
  
  
      private MessageConsumer consumer;
      private Messenger messenger;
      private Destination destination;
      private String selector;
      private MessageListener listener;
      private boolean shouldStop;
      
      public ConsumerThread() {
          setName("Consumer" + getName());
      }
  
  
      /**
       * Starts all the JMS connections and consumes JMS messages, 
       * passing them onto the MessageListener and Message Driven Objects
       */
      public void run() {
          if (log.isDebugEnabled()) {
              log.debug( "Starting consumer thread: " + getName());
          }
          try {
              startConsumer();
          }
          catch (JMSException e) {
              log.error("Failed to start consumer thread: " + e, e);
              setShouldStop(true);
          }
          
          while (! isShouldStop()) {
          	try {
  	            startTransaction();
          	}
          	catch (Exception e) {
          		log.error("Caught exception trying to start transaction. This thread will terminate. Reason: " + e, e);
          		break;
          	}
  
              try {
                  Message message = receive();
  
                  if (log.isTraceEnabled()) {
                      log.trace( "Found: " + message );
                  }
                  
                  if (message != null) {
                      processMessage(message);
                      commitTransaction();
                  }
                  else {
                      cancelTransaction();
                  }
              }
              catch (Exception e) {
                  rollbackTransaction(e);
              }
          }
          
          try {
              stopConsumer();
          }
          catch (JMSException e) {
              log.error("Failed to stop consuming messages: " + e, e);
          }
     }
     
     // Properties
     //-------------------------------------------------------------------------    
 
     /**
      * Returns the destination.
      * @return Destination
      */
     public Destination getDestination() {
         return destination;
     }
 
     /**
      * Returns the listener.
      * @return MessageListener
      */
     public MessageListener getListener() {
         return listener;
     }
 
     /**
      * Returns the messenger.
      * @return Messenger
      */
     public Messenger getMessenger() {
         return messenger;
     }
 
     /**
      * Returns the selector.
      * @return String
      */
     public String getSelector() {
         return selector;
     }
 
     /**
      * Returns the shouldStop.
      * @return boolean
      */
     public boolean isShouldStop() {
         return shouldStop;
     }
 
     /**
      * Sets the destination.
      * @param destination The destination to set
      */
     public void setDestination(Destination destination) {
         this.destination = destination;
     }
 
     /**
      * Sets the listener.
      * @param listener The listener to set
      */
     public void setListener(MessageListener listener) {
         this.listener = listener;
     }
 
     /**
      * Sets the messenger.
      * @param messenger The messenger to set
      */
     public void setMessenger(Messenger messenger) {
         this.messenger = messenger;
     }
 
     /**
      * Sets the selector.
      * @param selector The selector to set
      */
     public void setSelector(String selector) {
         this.selector = selector;
     }
 
     /**
      * Sets the shouldStop.
      * @param shouldStop The shouldStop to set
      */
     public void setShouldStop(boolean shouldStop) {
         this.shouldStop = shouldStop;
     }
 
     // Implementation methods
     //-------------------------------------------------------------------------    
     
     /**
      * Starts consuming messages        
      */    
     protected void startConsumer() throws JMSException {
         consumer = createConsumer();
     }
 
     /**
      * Stops consuming messages        
      */    
     protected void stopConsumer() throws JMSException {
         consumer.close();
     }
 
     /**
      * Factory method to create a new MessageConsumer 
      */
     protected MessageConsumer createConsumer() throws JMSException {
         String selector = getSelector();
         if (selector != null) {
             return getMessenger().createConsumer(getDestination(), selector);
         }
         else {
             return getMessenger().createConsumer(getDestination());
         }
     }
 
     /**
      * Strategy method to consume a message using a receive() kind of method.
      * @return the message or null if a message could not be found after waiting for
      * some period of time.
      */
     private Message receive() throws JMSException {
         return getConsumer().receive();
     }    
     
     /**
      * Strategy method to process a given message. 
      * By default this will just invoke the MessageListener
      */
     protected void processMessage(Message message) throws JMSException {
         MessageListener listener = getListener();
         if (listener != null) {
             listener.onMessage(message);
         }
     }
 
 
     /**
      * Strategy method to represent the code required to start
      * a transaction.
      */
     protected void startTransaction() throws Exception {
     }
 
     /**
      * Strategy method to represent the code required to commit
      * a transaction.
      */
     protected void commitTransaction() throws Exception {
     }
 
     /**
      * Strategy method to represent the code required to rollback
      * a transaction.
      */
     protected void rollbackTransaction(Exception e) {
     }
 
     /**
      * Strategy method to represent the code required to cancel
      * a transaction. 
      * This is called when a message is not received.
      */
     protected void cancelTransaction() throws Exception {
     }
 
 
     /**
      * @erturn the consumer of messages 
      */
     protected MessageConsumer getConsumer() {
         return consumer;
     }
 }