/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.uima.adapter.jms.activemq;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.ConnectionFailedException;
import org.apache.activemq.advisory.ConsumerEvent;
import org.apache.activemq.advisory.ConsumerListener;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.uima.UIMAFramework;
import org.apache.uima.aae.InputChannel;
import org.apache.uima.aae.UIMAEE_Constants;
import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
import org.apache.uima.aae.controller.AnalysisEngineController;
import org.apache.uima.aae.controller.BaseAnalysisEngineController;
import org.apache.uima.aae.controller.Endpoint;
import org.apache.uima.aae.controller.LocalCache.CasStateEntry;
import org.apache.uima.aae.delegate.Delegate;
import org.apache.uima.aae.error.AsynchAEException;
import org.apache.uima.aae.error.DelegateConnectionLostException;
import org.apache.uima.aae.error.InvalidMessageException;
import org.apache.uima.aae.error.ServiceShutdownException;
import org.apache.uima.aae.message.AsynchAEMessage;
import org.apache.uima.adapter.jms.JmsConstants;
import org.apache.uima.adapter.jms.activemq.JmsOutputChannel.BrokerConnectionEntry;
import org.apache.uima.cas.CAS;
import org.apache.uima.util.Level;


public class JmsEndpointConnection_impl implements ConsumerListener {
  private static final Class CLASS_NAME = JmsEndpointConnection_impl.class;

  // timestamp containing time when the last message was dispatched from this dispatcher to 
  // jms destination. This is updated every time a message is dispatched to a queue. 
  // At fixed intervals a cleanup thread wakes up and checks for unused dispatchers by
  // comparing value in lastDispatchTimestamp to a max allowed which by default is 5 minutes.
  protected AtomicLong lastDispatchTimestamp = new AtomicLong(0);

  protected Destination destination;

  protected Session producerSession;

  private MessageProducer producer;

  private BrokerConnectionEntry brokerDestinations;

  private String serverUri;

  private String endpoint;

  private String endpointName;

  protected Endpoint delegateEndpoint;

  private volatile boolean retryEnabled;

  private AnalysisEngineController controller = null;

  private volatile boolean connectionAborted = false;

  protected static long connectionCreationTimestamp = 0L;

  private Object semaphore = new Object();

  protected boolean isReplyEndpoint;

  private volatile boolean failed = false;

  private Object lock = new Object();

  private final String componentName;

  // Create semaphore to control creation of a JMS connection.
  // This semaphore is shared by all instances of this class
  private static Semaphore connectionSemaphore = new Semaphore(1);
  
  public JmsEndpointConnection_impl(BrokerConnectionEntry aBrokerDestinationMap,
          Endpoint anEndpoint, AnalysisEngineController aController) {
    brokerDestinations = aBrokerDestinationMap;
    if ( anEndpoint.isFreeCasEndpoint() && anEndpoint.isCasMultiplier() && anEndpoint.isReplyEndpoint()) {
      serverUri = anEndpoint.getServerURI();
    } else {
      //  If this is a reply to a client, use the same broker URL that manages this service input queue.
      //  Otherwise this is a request so use a broker specified in the endpoint object.
      serverUri = (anEndpoint.isReplyEndpoint()) ? 
              ((JmsOutputChannel) aController.getOutputChannel()).getServerURI() : anEndpoint.getServerURI();
    }
    isReplyEndpoint = anEndpoint.isReplyEndpoint();
    controller = aController;

    if ((anEndpoint.getCommand() == AsynchAEMessage.Stop || isReplyEndpoint)
            && anEndpoint.getDestination() != null
            && anEndpoint.getDestination() instanceof ActiveMQDestination) {
      endpoint = ((ActiveMQDestination) anEndpoint.getDestination()).getPhysicalName();
    } else {
      endpoint = anEndpoint.getEndpoint();
    }
    anEndpoint.remove();
    componentName = controller.getComponentName();
    delegateEndpoint = anEndpoint;
  }

  public boolean isRetryEnabled() {
    return retryEnabled;
  }

  public void setRetryEnabled(boolean retryEnabled) {
    this.retryEnabled = retryEnabled;
  }

  public boolean isOpen() {
	synchronized (lock) {
	    if (failed || producerSession == null || connectionClosedOrFailed(brokerDestinations)) {
	        return false;
	      }
	      return ((ActiveMQSession) producerSession).isRunning();
	}
  }

  protected static boolean connectionClosedOrFailed(BrokerConnectionEntry aBrokerDestinationMap) {
    Connection connection = aBrokerDestinationMap.getConnection();
    if (connection == null
            || ((ActiveMQConnection) connection).isClosed()
            || ((ActiveMQConnection) connection).isClosing()
            || ((ActiveMQConnection) connection).isTransportFailed()) {
      return true;
    }
    return false;
  }

  private void openChannel() throws AsynchAEException, ServiceShutdownException {
    openChannel(getServerUri(), componentName, endpoint, controller);
  }

  private void openChannel(String brokerUri, String aComponentName,
          String anEndpointName, AnalysisEngineController aController) throws AsynchAEException,
          ServiceShutdownException {
	  synchronized (lock) {
		    try {

		        // If replying to http request, reply to a queue managed by this service broker using tcp
		        // protocol
		        if (isReplyEndpoint && brokerUri.startsWith("http")) {
		          brokerUri = ((JmsOutputChannel) aController.getOutputChannel()).getServerURI();

		          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
		            UIMAFramework.getLogger(CLASS_NAME).logrb(
		                    Level.FINE,
		                    CLASS_NAME.getName(),
		                    "open",
		                    JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
		                    "UIMAJMS_override_connection_to_endpoint__FINE",
		                    new Object[] {  aComponentName, getEndpoint(),
		                      ((JmsOutputChannel) aController.getOutputChannel()).getServerURI() });
		          }
		        } else if ( !brokerUri.startsWith("http") && !brokerUri.startsWith("failover") && !brokerUri.startsWith("vm://localhost?broker.persistent=false")){
				  String prefix = "";
  		          if ( brokerUri.indexOf("?") > -1) {
		            prefix = "&";
		          } else {
		            prefix ="?";
		          }
		          String extraParams = "";
		          // check if maxInactivityDuration exists in the given url
		          if ( brokerUri.indexOf("wireFormat.maxInactivityDuration") == -1 ) {
		            // turn off activemq inactivity monitor
		            extraParams = prefix+"wireFormat.maxInactivityDuration=0";
		          }
				  brokerUri += extraParams;
		        }

		        if (!isOpen()) {
		          Connection conn = null;
		          //  Check connection status and create a new one (if necessary) as an atomic operation
		          try {
		            connectionSemaphore.acquire();
		            
		            if (connectionClosedOrFailed(brokerDestinations)) {
		              // Create one shared connection per unique brokerURL.
		              if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
		                UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
		                        "openChannel", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
		                        "UIMAJMS_activemq_open__FINE",
		                        new Object[] { aController.getComponentName(), anEndpointName, brokerUri });
		              }
		              if ( brokerDestinations.getConnection() != null ) {
		                try {
		                  //  Close the connection to avoid leaks in the broker
		                  brokerDestinations.getConnection().close();
		                } catch( Exception e) {
		                  //  Ignore exceptions on a close of a bad connection
		                }
		              }
		              // log connectivity problem once and retry
		              boolean logConnectionProblem=true;
		              
		              // recover lost connection indefinitely while the service is running
		              while( !controller.isStopped() ) {

		            	  try {
				              ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUri);
				              // White list packages for deserialization 
				              ConnectionFactoryIniter cfIniter =
				                   new ConnectionFactoryIniter(factory);
				              cfIniter.whiteListPackages();

				              factory.setWatchTopicAdvisories(false);
				              //  Create shared jms connection to a broker
				              conn = factory.createConnection();
				              factory.setDispatchAsync(true);
				              factory.setUseAsyncSend(true);
				              factory.setCopyMessageOnSend(false);
				              conn.start();
				              //  Cache the connection. There should only be one connection in the jvm
				              //  per unique broker url. 
				              brokerDestinations.setConnection(conn);
				              // Close and invalidate all sessions previously created from the old connection
				              Iterator<Map.Entry<Object, JmsEndpointConnection_impl>> it = brokerDestinations.endpointMap
				                      .entrySet().iterator();
				              while (it.hasNext()) {
				                Map.Entry<Object, JmsEndpointConnection_impl> entry = it.next();
				                if (entry.getValue().producerSession != null) {
				                  // Close session
				                  entry.getValue().producerSession.close();
				                  // Since we created a new connection invalidate session that
				                  // have been created with the old connection
				                  entry.getValue().producerSession = null;
				                }
				              }
		            		  break; // Got the connection, break out of the while-loop
		            		  
		            	  } catch( JMSException jex) {
		            		  if ( conn != null  ) {
		            			  try {
		            				  conn.close();
		            			  } catch( Exception ee) {}
		            		  }
		            		//  if ( logConnectionProblem ) {
		            			  logConnectionProblem = false;   // log once
			            		  // Check if unable to connect to the broker and retry ...
		            			  if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
		            		          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
		            		                  "openChannel", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
		            		                  "UIMAEE_service_lost_connectivity_WARNING",
		            		                  new Object[] { controller.getComponentName(), brokerUri});
		            		          
		            		          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
		            		                  "openChannel", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
		            		                  "UIMAJMS_exception__WARNING", jex);
		            		        }
		            	//	  } 
		            		 this.wait(1000);  // wait between retries 
		            	  } catch ( Exception ee) {
		            		  ee.printStackTrace();
		            		  if ( conn != null  ) {
		            			  try {
		            				  conn.close();
		            			  } catch( Exception eee) {}
		            		  }
		            	  }
		              } //while
		              if ( logConnectionProblem == false )  { // we had conectivity problem. Log the fact that it was recovered
		            	  if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
            		          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
            		                  "openChannel", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
            		                  "UIMAEE_service_regained_connectivity_INFO",
            		                  new Object[] { controller.getComponentName(), brokerUri});
		            	  }
		            	  
		              }
		              }
		          } catch( Exception exc) {
		        	  if ( conn != null  ) {
	          			  try {
	          				  conn.close();
	          			  } catch( Exception ee) {}
		        	  }
		             throw exc; // rethrow
		          } finally {
		            connectionSemaphore.release();
		           
		          }
		          
		          connectionCreationTimestamp = System.nanoTime();
		          failed = false;
		        } else {
		        	System.out.println("...... Reusing Existing Broker Connetion");
		        }
		        Connection conn = brokerDestinations.getConnection();
		        if (failed) {
		          // Unable to create a connection
		          return;
		        }

		        producerSession = conn.createSession(false, Session.DUPS_OK_ACKNOWLEDGE);
		        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
	            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
                      "openChannel", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAJMS_session_open__FINE",
                      new Object[] { aComponentName, anEndpointName, brokerUri });
		        }

		        if ((delegateEndpoint.getCommand() == AsynchAEMessage.Stop || isReplyEndpoint)
		                && delegateEndpoint.getDestination() != null) {
		          producer = producerSession.createProducer(null);
		          if (aController != null) {
		            if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
		              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
		                      "openChannel", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
		                      "UIMAJMS_temp_conn_starting__FINE",
		                      new Object[] { aComponentName, anEndpointName, brokerUri });
		            }
		          }
		        } else {
		          destination = producerSession.createQueue(getEndpoint());
		          producer = producerSession.createProducer(destination);
		          if (controller != null) {
		            if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
		              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
		                      "openChannel", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
		                      "UIMAJMS_conn_starting__FINE",
		                      new Object[] { aComponentName, anEndpointName, brokerUri });
		            }
		          }
		        }
		        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		        // Since the connection is shared, start it only once
		        if (!((ActiveMQConnection) brokerDestinations.getConnection()).isStarted()) {
		          brokerDestinations.getConnection().start();
		        }
		        if (controller != null) {
		          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
		            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
		                    "openChannel", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
		                    "UIMAJMS_conn_started__FINE", new Object[] { endpoint, brokerUri });
		            if (controller.getInputChannel() != null) {
		              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
		                      "openChannel", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
		                      "UIMAJMS_connection_open_to_endpoint__FINE",
		                      new Object[] { aComponentName, getEndpoint(), brokerUri });
		            }
		          }
		        }
		        failed = false;
		      } catch (Exception e) {
		        boolean rethrow = true;
		        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
		          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
		                  "openChannel", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
		                  "UIMAEE_service_exception_WARNING", controller.getComponentName());
		          
		          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
		                  "openChannel", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
		                  "UIMAJMS_exception__WARNING", e);
		        }

		        if (e instanceof JMSException) {
		          rethrow = handleJmsException((JMSException) e);

		        }
		        if (rethrow) {
		          throw new AsynchAEException(e);
		        }
		      }
		  
	  }
  }

  public synchronized void open() throws AsynchAEException, ServiceShutdownException {
    open(delegateEndpoint.getEndpoint(), serverUri);
  }

  public synchronized void open( String anEndpointName, String brokerUri) throws AsynchAEException,
          ServiceShutdownException {
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "open",
              JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_open__FINE",
              new Object[] { anEndpointName, brokerUri });
    }
    if (!connectionAborted) {
      openChannel();
    }

  }

  public synchronized void abort() {
    connectionAborted = true;
    brokerDestinations.getConnectionTimer().stopTimer();
    try {
      this.close();
    } catch (Exception e) {
    }
  }

	public void close() throws Exception {
		synchronized (lock) {
			if (producer != null) {
				try {
					producer.close();
				} catch (Exception e) {
					// Ignore we are shutting down
				}
			}
			if (producerSession != null) {
				try {
					producerSession.close();
				} catch (Exception e) {
					// Ignore we are shutting down
				}
				producerSession = null;
			}
			if (destination != null) {
				destination = null;
			}
		}
	}

  protected String getEndpoint() {
    return endpoint;
  }

  protected void setEndpoint(String endpoint) {
    this.endpoint = endpoint;
  }
  protected void setDelegateEndpoint(Endpoint delegateEndpoint) {
    this.delegateEndpoint = delegateEndpoint;
  }

  protected synchronized String getServerUri() {
    return serverUri;
  }

  protected synchronized void setServerUri(String serverUri) {
    this.serverUri = serverUri;
  }

  public TextMessage produceTextMessage(String aTextMessage) throws AsynchAEException {
	synchronized( lock ) {
		  if ( producerSession == null ) {
		      throw new AsynchAEException("Controller:"+controller.getComponentName()+" Unable to create JMS Message. Producer Session Not Initialized (Null)");
		    }
		    try {
		       if (aTextMessage == null) {
		          return producerSession.createTextMessage("");
		       } else {
		          return producerSession.createTextMessage(aTextMessage);
		       }
		     } catch (javax.jms.IllegalStateException e) {
		        try {
		          open();
		        } catch (ServiceShutdownException ex) {
		          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
		            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
		                    "produceTextMessage", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
		                    "UIMAEE_service_exception_WARNING", controller.getComponentName());
		            
		            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
		                    "produceTextMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
		                    "UIMAJMS_exception__WARNING", ex);
		          }
		        } catch (AsynchAEException ex) {
		          throw ex;
		        }
		      } catch (Exception e) {
		        throw new AsynchAEException(e);
		    }
		    throw new AsynchAEException(new InvalidMessageException("Unable to produce Message Object"));
	}
  }

  public BytesMessage produceByteMessage(byte[] aSerializedCAS) throws AsynchAEException {
    synchronized( lock ) {
        if ( producerSession == null ) {
            throw new AsynchAEException("Controller:"+controller.getComponentName()+" Unable to create JMS Message. Producer Session Not Initialized (Null)");
          }
          int retryCount = 1;
          while (retryCount > 0) {
            try {
              retryCount--;
              BytesMessage bm = producerSession.createBytesMessage();
              bm.writeBytes(aSerializedCAS);

              return bm;

            } catch (javax.jms.IllegalStateException e) {
              try {
                open();
              } catch (ServiceShutdownException ex) {
                if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
                  UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                          "produceByteMessage", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                          "UIMAEE_service_exception_WARNING", controller.getComponentName());
                  
                  UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                          "produceByteMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                          "UIMAJMS_exception__WARNING", ex);
                }
              }

            } catch (Exception e) {
              throw new AsynchAEException(e);
            }
          }
          throw new AsynchAEException(
                  new InvalidMessageException("Unable to produce BytesMessage Object"));
    }
  }

  public ObjectMessage produceObjectMessage() throws AsynchAEException {
    synchronized( lock ) {
  	  if ( producerSession == null ) {
  	      throw new AsynchAEException("Controller:"+controller.getComponentName()+" Unable to create JMS Message. Producer Session Not Initialized (Null)");
  	    }
  	    try {
  	      if (!((ActiveMQSession) producerSession).isRunning()) {
  	        open();
  	      }
  	      return producerSession.createObjectMessage();
  	    } catch (Exception e) {
  	      throw new AsynchAEException(e);
  	    }
    }
  }

  private boolean delayCasDelivery(int msgType, Message aMessage, int command) throws Exception {

    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "recoverSession",
              JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_open_connection_to_endpoint__FINE",
              new Object[] { getEndpoint() });
    }
    openChannel();
    // The connection has been successful. Now check if we need to create a new listener
    // and a temp queue to receive replies. A new listener will be created only if the
    // endpoint for the delegate is marked as FAILED. This will be the case if the listener
    // on the reply queue for the endpoint has failed.
    String endpointName = delegateEndpoint.getEndpoint();
    synchronized (lock) {
      if (controller instanceof AggregateAnalysisEngineController) {
        // Using the queue name lookup the delegate key
        String key = ((AggregateAnalysisEngineController) controller)
                .lookUpDelegateKey(endpointName);
        if (key != null && destination != null && !isReplyEndpoint) {
          // For Process Requests check the state of the delegate that is to receive
          // the CAS. If the delegate state = TIMEOUT_STATE, push the CAS id onto
          // delegate's list of delayed CASes. The state of the delegate was
          // changed to TIMEOUT when a previous CAS timed out.
          if (msgType != AsynchAEMessage.Request && command == AsynchAEMessage.Process) {
            String casReferenceId = aMessage.getStringProperty(AsynchAEMessage.CasReference);
            if (casReferenceId != null ) {
              CAS cas = ((AggregateAnalysisEngineController) controller).getInProcessCache().getCasByReference(casReferenceId);
              if ( cas != null && ((AggregateAnalysisEngineController) controller)
                            .delayCasIfDelegateInTimedOutState(casReferenceId, endpointName, cas.hashCode())) {
                      return true;
               }
            }
          }
        }
      }
    }
    return false;

  }
  public boolean send(final Message aMessage, long msgSize, boolean startTimer) {
	 return send( aMessage, msgSize, startTimer, false);
  }

  public boolean send(final Message aMessage, long msgSize, boolean startTimer, boolean failOnJMSException) {
    String destinationName = "";
    String target = "Delegate";
    int msgType = 0;
    int command = 0;
    try {
      msgType = aMessage.getIntProperty(AsynchAEMessage.MessageType);
      command = aMessage.getIntProperty(AsynchAEMessage.Command);
      boolean newCAS = false;
      if ( aMessage.propertyExists(AsynchAEMessage.CasSequence) &&
              aMessage.getLongProperty(AsynchAEMessage.CasSequence) > 0 ) {
        newCAS = true;
      }
      
      if ( msgType == AsynchAEMessage.Response || (msgType == AsynchAEMessage.Request && newCAS) ) {
        target = "Client";
      }
      Endpoint masterEndpoint = null;
      if ( delegateEndpoint != null && delegateEndpoint.getDelegateKey() != null ) {
        masterEndpoint = ((AggregateAnalysisEngineController) controller).lookUpEndpoint(
                delegateEndpoint.getDelegateKey(), false);
        // Endpoint is marked as FAILED by the aggregate when it detects broker connection
        // failure. In such an event the aggregate stops the listener on the delegate
        // reply queue.
        if ( msgType == AsynchAEMessage.Request && command == AsynchAEMessage.Process &&
             masterEndpoint != null && masterEndpoint.getStatus() == Endpoint.FAILED) {
          HashMap<Object, Object> map = new HashMap<Object, Object>();
          Delegate delegate = ((AggregateAnalysisEngineController) controller).lookupDelegate(delegateEndpoint.getDelegateKey());
          //  Cancel Delegate timer before entering Error Handler
          if ( delegate != null ) {
            delegate.cancelDelegateTimer();
          }
          //  Handle the Connection error in the ProcessErrorHandler
          map.put(AsynchAEMessage.Command, AsynchAEMessage.Process);
          map.put(AsynchAEMessage.CasReference, aMessage.getStringProperty(AsynchAEMessage.CasReference));
          map.put(AsynchAEMessage.Endpoint, masterEndpoint);
          Exception e = new DelegateConnectionLostException("Controller:"+controller.getComponentName()+" Lost Connection to "+target+ ":"+masterEndpoint.getDelegateKey());
          //  Handle error in ProcessErrorHandler
          ((BaseAnalysisEngineController)controller).handleError(map, e);
          return true; // return true as if this was successful send 
        }
      }

      if ( !isOpen() ) {
        if (delayCasDelivery(msgType, aMessage, command)) {
          // Return true as if the CAS was sent
          return true;
        }
      }
      // Stop messages and replies are sent to the endpoint provided in the destination object
      if ((command == AsynchAEMessage.Stop || command == AsynchAEMessage.ReleaseCAS || isReplyEndpoint)
              && delegateEndpoint.getDestination() != null) {
        destinationName = ((ActiveMQDestination) delegateEndpoint.getDestination())
                .getPhysicalName();
        if (UIMAFramework.getLogger().isLoggable(Level.FINE)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "send",
                  JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_sending_msg_to_endpoint__FINE",
                  new Object[] { destinationName });
        }
        logMessageSize(aMessage, msgSize, destinationName);
        synchronized (producer) {
          producer.send((Destination) delegateEndpoint.getDestination(), aMessage);
        }
      } else {
        destinationName = ((ActiveMQQueue) producer.getDestination()).getPhysicalName();
        if (UIMAFramework.getLogger().isLoggable(Level.FINE)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "send",
                  JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_sending_msg_to_endpoint__FINE",
                  new Object[] { destinationName });
        }
        logMessageSize(aMessage, msgSize, destinationName);
  	  // If in ParallelStep its possible to receive a reply from one of the delegates in parallel 
  	  // step *before* a CAS is dispatched to all of the delegates. This can cause a problem
  	  // as replies are merged which causes the CAS to be in an inconsistent state.
  	  // The following code calls dispatchCasToParallelDelegate() which count down
  	  // a java latch. The same latch is used when receiving replies. If the latch is non zero
  	  // the code blocks a thread from performing deserialization.
  	  if ( msgType == AsynchAEMessage.Request && command == AsynchAEMessage.Process ) {
  		  String casReferenceId = aMessage.getStringProperty(AsynchAEMessage.CasReference);
  		  CasStateEntry casStateEntry = controller.getLocalCache().lookupEntry(casReferenceId);
  		  if ( casStateEntry.getNumberOfParallelDelegates() > 0) {
  			  casStateEntry.dispatchedCasToParallelDelegate();
  		  }
  	  }

        synchronized (producer) {
          producer.send(aMessage);
        }

      }
      // Starts a timer on a broker connection. Every time a new message
      // is sent to a destination managed by the broker the timer is
      // restarted. The main purpose of the timer is to close connections
      // that are not used.
      if (startTimer) {
//        brokerDestinations.getConnectionTimer().startTimer(connectionCreationTimestamp,
//                delegateEndpoint);
      }
      // record the time when this dispatches sent a message. This time will be used
      // to find inactive sessions.
	  lastDispatchTimestamp.set(System.currentTimeMillis());
      // Succeeded sending the CAS
      return true;
    } catch (Exception e) {
    	
    	 // if a client terminates with an outstanding request, the service will not
        // be able to deliver a reply. Just log the fact that the reply queue is
        // no longer available.
      if ( e instanceof InvalidDestinationException && "Client".equals(target) ) {
          if ( delegateEndpoint != null ) {
            endpointName = ((ActiveMQDestination) delegateEndpoint.getDestination())
            		.getPhysicalName();
          }
    	  UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                  "send", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAJMS_invalid_destination__INFO",
                  new Object[] { controller.getComponentName(),endpointName });
    	  if ( command == AsynchAEMessage.ServiceInfo ) {
    		  return false;
    	  }
    	  if ( (msgType == AsynchAEMessage.Response || msgType == AsynchAEMessage.Request ) &&
    			  command == AsynchAEMessage.Process ) {
    		  String casReferenceId="";
    		  try {
    		     casReferenceId = aMessage.getStringProperty(AsynchAEMessage.CasReference);
    		  } catch( Exception exx ) {
    		        String key = "";
    		        String endpointName = "";
    		        if ( delegateEndpoint != null ) {
    		          delegateEndpoint.getDelegateKey();
    		          endpointName = ((ActiveMQDestination) delegateEndpoint.getDestination())
    		          .getPhysicalName();
    		        }
    	          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
    	                  "send", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
    	                  "UIMAEE_service_delivery_exception__WARNING",new Object[] { controller.getComponentName(), key, endpointName});
    		  }
   		      CasStateEntry casStateEntry = controller.getLocalCache().lookupEntry(casReferenceId);
   		      // Mark the CAS as failed so that the CAS is released and cache cleaned up
    		  casStateEntry.setDeliveryToClientFailed();
    	  }
    	  return true;  // expect the client can go away at any time. Not an error
      }    	
    	
    	
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        
        String key = "";
        String endpointName = "";
        if ( delegateEndpoint != null ) {
          delegateEndpoint.getDelegateKey();
          endpointName = ((ActiveMQDestination) delegateEndpoint.getDestination())
          .getPhysicalName();
        }
        if ( "Client".equals(target) ) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                  "send", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_service_delivery_to_client_exception__WARNING",
                  new Object[] { controller.getComponentName(),endpointName });
        } else {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                  "send", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_service_delivery_exception__WARNING",new Object[] { controller.getComponentName(), key, endpointName});
        }

        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                "send", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAJMS_exception__WARNING", e);
      }
      // If the controller has been stopped no need to send messages
      if (controller.isStopped()) {
        return false;
       } else {
        if (e instanceof JMSException) {
          handleJmsException((JMSException) e);
          //	whoever called this method is interested in knowing that there was JMS Exception
          if ( failOnJMSException ) {
        	  return false;
          }
        } else {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                    "send", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_service_exception_WARNING", controller.getComponentName());

            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), "send",
                    JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_exception__WARNING",
                    e);
          }
        }

      }
    }
    //brokerDestinations.getConnectionTimer().stopTimer();
    // Failed here
    return false;
  }

  private void logMessageSize(Message aMessage, long msgSize, String destinationName) {
    if (UIMAFramework.getLogger().isLoggable(Level.FINE)) {
      boolean isReply = false;
      if (isReplyEndpoint) {
        isReply = true;
      }
      String type = "Text";
      if (aMessage instanceof BytesMessage) {
        type = "Binary";
        UIMAFramework.getLogger(CLASS_NAME).logrb(
                Level.FINE,
                CLASS_NAME.getName(),
                "logMessageSize",
                JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAJMS_msg_size__FINE",
                new Object[] { componentName, isReply == true ? "Reply" : "Request", "Binary",
                    destinationName, msgSize });
      } else if (aMessage instanceof TextMessage) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(
                Level.FINE,
                CLASS_NAME.getName(),
                "logMessageSize",
                JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAJMS_msg_size__FINE",
                new Object[] { componentName, isReply == true ? "Reply" : "Request", "XMI",
                    destinationName, msgSize });
      }
    }
  }

  /**
   * This method is called during recovery of failed connection. It is only called if the endpoint
   * associated with a given delegate is marked as FAILED. It is marked that way when a listener
   * attached to the reply queue fails. This method creates a new listener and a new temp queue.
   * 
   * @param delegateKey
   * @throws Exception
   */
  private void createListener(String delegateKey) throws Exception {
    if (controller instanceof AggregateAnalysisEngineController) {
      // Fetch an InputChannel that handles messages for a given delegate
      InputChannel iC = controller.getReplyInputChannel(delegateKey);
      // Create a new Listener, new Temp Queue and associate the listener with the Input Channel
      iC.createListener(delegateKey, null);
    }
  }

  private synchronized boolean handleJmsException(JMSException ex) {
    if (!failed) {
      failed = true;
    }
    try {
      // Check if the exception is due to deleted queue. ActiveMQ does not identify
      // this condition in the cause, so we need to parse the exception message and
      // compare against "Cannot publish to a deleted Destination" text. If match is
      // found, extract the name of the deleted queue from the exception and log it.
      if (ex.getMessage() != null
              && ex.getMessage().startsWith("Cannot publish to a deleted Destination")) {
        String destName = endpointName;
        int startPos = ex.getMessage().indexOf(':');
        if (startPos > 0) {
          destName = ex.getMessage().substring(startPos);
        }
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                  "handleJmsException", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAJMS_send_failed_deleted_queue_INFO",
                  new Object[] { componentName, destName, serverUri });
        }
        return false;

      }
      if (ex instanceof ConnectionFailedException && isReplyEndpoint) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                "handleJmsException", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAJMS_connection_failure__INFO",
                new Object[] { componentName, serverUri, delegateEndpoint.getDestination() });
        return false;

      } else {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
          
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                  "handleJmsException", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_service_exception_WARNING", controller.getComponentName());

          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                  "handleJmsException", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAJMS_exception__WARNING", ex);
        }
      }
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                "handleJmsException", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_service_exception_WARNING", controller.getComponentName());

        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), "handleJmsException",
                JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_exception__WARNING",
                e);
      }
    }

    return true;
  }

  public void onConsumerEvent(ConsumerEvent arg0) {
    if (controller != null) {
      controller.handleDelegateLifeCycleEvent(getEndpoint(), arg0.getConsumerCount());
    }
  }

  protected synchronized void finalize() throws Throwable {
//    brokerDestinations.getConnectionTimer().stopTimer();
  }

}