/********************************************************************
 * Copyright © 2004 VHA. All rights reserved
 ********************************************************************/
// Package
package gov.va.med.fw.service.jms;

// Java classes
import java.util.Iterator;
import java.util.List;
import java.util.Properties;

import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageFormatException;
import javax.jms.Session;
import javax.jms.TopicConnection;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;

import org.springframework.jms.support.JmsUtils;

import gov.va.med.fw.service.ServiceConfigurationException;
import gov.va.med.fw.service.ServiceException;

/**
 * Provides a concrete implementation of a MessageProducer that must be injected with
 * with a JMSDescriptor object.  A JMSDescriptor encapsulates specific information about 
 * a JMS Destination, Connection, and Factory in a J2EE application server.
 * 
 * <p>Treats the "providerURLs" of the TopicDescriptor as an order list of priority-based
 * targets.  This means it will attempt to send the JMS Message to the first element.  If it can
 * not connect, only then will it try the second element.  The end result is publishing the JMS
 * Message to a singular location, not all elements in the providerURLs.
 * </p>    
 * 
 * @author Venky Kullampalle
 * @version 1.0
 */
public class TopicMessageProducerService extends AbstractMessageProducerService implements TopicMessageProducer {

   /**
	 * An instance of serialVersionUID
	 */
	private static final long serialVersionUID = 2808696406833332275L;
	/**
    * A service descriptor for JMS-related properties
    */
   private TopicDescriptor descriptor = null;
   
   /** Creates a new instance of JMSService */
   public TopicMessageProducerService() {
      super();
   }

   /**
    * @return Returns the descriptor.
    */
   public TopicDescriptor getDescriptor() {
      return descriptor;
   }
   
   /**
    * @param descriptor The descriptor to set.
    */
   public void setDescriptor(TopicDescriptor descriptor) {
      this.descriptor = descriptor;
   }
   
   /** Checks required properties to ensure the required properties are
    * properly set in this bean instance.
    * 
    * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
    */
   public void afterPropertiesSet() throws Exception {
      super.afterPropertiesSet();
      if( descriptor == null ) {
         throw new ServiceConfigurationException( "Required JMS descriptor property needs to be configured");
      }
   }

   /** Sends a generic object and the specific set of string properties in a JMS Message. The
    * specific JMS Message type is determined based on a data object's class type.  If a class
    * type is either a HL7 Message or a string, a JMS TextMessage type is used.  If a class type
    * is a Map, a JMS MapMessage type is used.  If a class type is an array of bytes, a JMS
    * BytesMessage type is used.  Otherwise, an ObjectMessage is used to wrap a message data.
    *  
    * @param data An object to send to a message queue
    * @param properties A properties to send with a message
    * @throws ServiceException Thrown if failed to send a message
    */
   public void publish( Object object, Properties properties ) throws ServiceException {
       publish( object, 
              		  properties, 
               	  Message.DEFAULT_DELIVERY_MODE, 
               	  Message.DEFAULT_PRIORITY,
               	  Message.DEFAULT_TIME_TO_LIVE );
   }
   
   /** Sends a generic object and the specific set of string properties in a JMS Message. The
    * specific JMS Message type is determined based on a data object's class type.  If a class
    * type is either a HL7 Message or a string, a JMS TextMessage type is used.  If a class type
    * is a Map, a JMS MapMessage type is used.  If a class type is an array of bytes, a JMS
    * BytesMessage type is used.  Otherwise, an ObjectMessage is used to wrap a message data.
    *  
    * @param data An object to send to a message queue
    * @param properties A properties to send with a message
    * @param deliveryMode A delivery mode to send a message whether it is persistent or not
    * @param priority A property to send a message
    * @param timeToLive A total time for a message to live in a queue
    * @throws ServiceException Thrown if failed to send a message
    */
   public void publish( Object object, 
           						  Properties properties, 
           						  int deliverymode,
           						  int priority,
           						  long timeToLive ) throws ServiceException {
       
       if( object == null || properties == null || properties.isEmpty() ) {
           throw new ServiceConfigurationException( "Invalid message and properties");
       }
        
       TopicConnection connection = null;
       TopicSession session = null;
       boolean transactedJMSSession = ! super.isSupportsDistributedTransactions();
       int sessionAckMode = Session.AUTO_ACKNOWLEDGE;

       List providerURLs = descriptor.getProviderURLs();
       Iterator itr = providerURLs.iterator();
       String providerURL = null;
       while(itr.hasNext()) {
    	   providerURL = (String) itr.next();
    	   if(logger.isDebugEnabled()) {
    		   logger.debug("Connecting to JMS server at: " + providerURL);
    	   }
    	   
    	   try {
    	       // Create a queue connection
    	       connection = descriptor.createTopicConnection(providerURL);	
    	       // Create a session to send a message in a transaction
    	       session = descriptor.createTopicSession(connection, transactedJMSSession, sessionAckMode);
    		   
    		   break;
    	   }catch(ServiceException e) {
    		   logger.error("Unable to establish JMS connectivity to " + providerURL + " due to exception: ", e);
    		   
    		   JmsUtils.closeSession(session);
    		   session = null; // conservative
    		   JmsUtils.closeConnection(connection);
    		   connection = null; // conservative
    		   
    		   if(itr.hasNext()) {
    			   logger.error("...Attempting fallback to lower priority providerURL");
    		   } else {
    			   throw e;
    		   } 
    	   }
       }
       
       // Create a sender to send a message
       TopicPublisher publisher = null;
       try {
    	   if (session != null) {
	           publisher = session.createPublisher( descriptor.getTopic(providerURL) );
	
	           // Explicitly set a delivery mode here even though a JMS provider
	           // can initialize a queue to persist a message automatically.
	           publisher.setDeliveryMode( deliverymode );
	           publisher.setPriority( priority );
	           publisher.setTimeToLive( timeToLive );
	           publisher.setDisableMessageTimestamp(false);
	           publisher.setDisableMessageID(false);
	          
	           // Create a JMS message based on an input object's class type
	           // The rule here is as follow:
	           // a. bytes[] = BytesMessage
	           // b. String = TextMessage
	           // c. gov.va.med.fw.hl7.Message = TextMessage
	           // d. Anything else = ObjectMessage
	           Message jmsMessage = this.createJMSMessageByType( session, object, properties );
	           
	           // Send and acknowledge a message
	           publisher.publish( jmsMessage );
    	   }
       }
       catch (MessageFormatException e) {
           
           // Faild to send a message due to an incorrect message format
           throw new JMSServiceException("Invalid meessage format.", e);
       }
       catch (InvalidDestinationException e) {
           
           // Faild to send a message due to an invalid meessage queue
           throw new JMSServiceException("Invalid message queue.", e);
       }
       catch (JMSException e) {
           // Faild to send a message due to internal errors
           throw new JMSServiceException("Failed to send a meessage.", e);
       }
       catch (UnsupportedOperationException e) {

           // Faild to send a message due to internal errors
           throw new JMSServiceException("Missing a valid queue.", e);
       }
       finally {
    	   JmsUtils.closeMessageProducer(publisher);
    	   publisher = null; // conservative
    	   
		   close(session); // do this so that we can "commit" if transacted
		   session = null; // conservative
		   
		   JmsUtils.closeConnection(connection);
		   connection = null; // conservative
       }       
   }
}