/*
 * 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.client;

import java.io.File;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.management.ObjectName;
import javax.naming.InitialContext;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQMessageConsumer;
import org.apache.activemq.ActiveMQPrefetchPolicy;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.command.ActiveMQBytesMessage;
import org.apache.activemq.command.ActiveMQTempDestination;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.apache.uima.UIMAFramework;
import org.apache.uima.UIMA_IllegalArgumentException;
import org.apache.uima.UIMA_IllegalStateException;
import org.apache.uima.aae.AsynchAECasManager_impl;
import org.apache.uima.aae.UIMAEE_Constants;
import org.apache.uima.aae.VersionCompatibilityChecker;
import org.apache.uima.aae.UimaASApplicationEvent.EventTrigger;
import org.apache.uima.aae.UimaASApplicationExitEvent;
import org.apache.uima.aae.UimaAsVersion;
import org.apache.uima.aae.client.UimaASStatusCallbackListener;
import org.apache.uima.aae.client.UimaAsBaseCallbackListener;
import org.apache.uima.aae.client.UimaAsynchronousEngine;
import org.apache.uima.aae.controller.AnalysisEngineController;
import org.apache.uima.aae.controller.ControllerCallbackListener;
import org.apache.uima.aae.controller.ControllerLifecycle;
import org.apache.uima.aae.controller.Endpoint;
import org.apache.uima.aae.controller.UimacppServiceController;
import org.apache.uima.aae.error.AsynchAEException;
import org.apache.uima.aae.error.UimaASMetaRequestTimeout;
import org.apache.uima.aae.jmx.JmxManager;
import org.apache.uima.aae.message.AsynchAEMessage;
import org.apache.uima.aae.message.UIMAMessage;
import org.apache.uima.adapter.jms.JmsConstants;
import org.apache.uima.adapter.jms.activemq.ConnectionFactoryIniter;
import org.apache.uima.adapter.jms.activemq.SpringContainerDeployer;
import org.apache.uima.adapter.jms.activemq.UimaEEAdminSpringContext;
import org.apache.uima.adapter.jms.service.Dd2spring;
import org.apache.uima.analysis_engine.metadata.AnalysisEngineMetaData;
import org.apache.uima.cas.CAS;
import org.apache.uima.cas.SerialFormat;
import org.apache.uima.impl.UimaVersion;
import org.apache.uima.internal.util.UUIDGenerator;
import org.apache.uima.resource.Resource;
import org.apache.uima.resource.ResourceConfigurationException;
import org.apache.uima.resource.ResourceInitializationException;
import org.apache.uima.resource.ResourceManager;
import org.apache.uima.resource.ResourceProcessException;
import org.apache.uima.util.Level;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.support.FileSystemXmlApplicationContext;

public class BaseUIMAAsynchronousEngine_impl extends BaseUIMAAsynchronousEngineCommon_impl
        implements UimaAsynchronousEngine, MessageListener, ControllerCallbackListener, ApplicationListener<ApplicationEvent>{
  private static final Class CLASS_NAME = BaseUIMAAsynchronousEngine_impl.class;

  private MessageSender sender = null;

  private MessageProducer producer;

  private Session session = null;

  private Session consumerSession = null;

  private volatile boolean serviceInitializationException;

  private volatile boolean serviceInitializationCompleted;

  private Semaphore serviceSemaphore = new Semaphore(1);

  private Queue consumerDestination = null;

  private Session producerSession = null;

  private JmxManager jmxManager = null;

  private String applicationName = "UimaASClient";

 private String endpoint;
 
 protected static Semaphore sharedConnectionSemaphore = new Semaphore(1);

  protected static Object connectionMux = new Object();

  protected InitialContext jndiContext;
  
  private ObjectName clientJmxObjectName = null;
  
  private String amqUser = null;
  
  private String amqPassword = null;
  
  protected static Lock globalLock = new ReentrantLock();
  
  public BaseUIMAAsynchronousEngine_impl() {
	  super();
    UIMAFramework.getLogger(CLASS_NAME).log(Level.INFO,
            "UIMA Version " + UIMAFramework.getVersionString() +
    " UIMA-AS Version " + UimaAsVersion.getVersionString());
  }


  protected TextMessage createTextMessage() throws ResourceInitializationException {
    return new ActiveMQTextMessage();
  }

  protected BytesMessage createBytesMessage() throws ResourceInitializationException {
    return new ActiveMQBytesMessage();
  }

  /**
   * Called at the end of collectionProcessingComplete - WAS closes receiving thread here
   */
  protected void cleanup() throws Exception {
  }

  /**
   * Return a name of the queue to which the JMS Producer is connected to.
   */
  public String getEndPointName() throws ResourceProcessException {
    try {
      return clientSideJmxStats.getEndpointName();
    } catch (Exception e) {
      throw new ResourceProcessException(e);
    }
  }

  protected void setMetaRequestMessage(Message msg) throws Exception {
    msg.setStringProperty(AsynchAEMessage.MessageFrom, consumerDestination.getQueueName());

    msg.setStringProperty(UIMAMessage.ServerURI, brokerURI);
    msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
    msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.GetMeta);
    msg.setJMSReplyTo(consumerDestination);
    if (msg instanceof TextMessage) {
      ((ActiveMQTextMessage) msg).setText("");
    }
  }

  /**
   * Initialize JMS Message with properties relevant to Process CAS request.
   */
  protected void setCASMessage(String aCasReferenceId, CAS aCAS, Message msg)
          throws ResourceProcessException {
    try {
      setCommonProperties(aCasReferenceId, msg, SerialFormat.XMI);
      ((TextMessage) msg).setText(serializeCAS(aCAS));
    } catch (Exception e) {
      throw new ResourceProcessException(e);
    }
  }

  protected void setCASMessage(String aCasReferenceId, String aSerializedCAS, Message msg)
          throws ResourceProcessException {
    try {
      setCommonProperties(aCasReferenceId, msg, SerialFormat.XMI);
      ((TextMessage) msg).setText(aSerializedCAS);
    } catch (Exception e) {
      throw new ResourceProcessException(e);
    }
  }

  protected void setCASMessage(String aCasReferenceId, byte[] aSerializedCAS, Message msg)
          throws ResourceProcessException {
    try {
      setCommonProperties(aCasReferenceId, msg, SerialFormat.BINARY);
      ((BytesMessage) msg).writeBytes(aSerializedCAS);
    } catch (Exception e) {
      throw new ResourceProcessException(e);
    }
  }

  protected void setCommonProperties(String aCasReferenceId, Message msg,
          SerialFormat serialFormat) throws ResourceProcessException {
    try {
      msg.setStringProperty(AsynchAEMessage.MessageFrom, consumerDestination.getQueueName());

      msg.setStringProperty(UIMAMessage.ServerURI, brokerURI);
      msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
      msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.Process);
      msg.setStringProperty(AsynchAEMessage.CasReference, aCasReferenceId);
      msg.setIntProperty(AsynchAEMessage.Payload, (serialFormat == SerialFormat.XMI) ? AsynchAEMessage.XMIPayload : AsynchAEMessage.BinaryPayload);
      msg.setBooleanProperty(AsynchAEMessage.AcceptsDeltaCas, true);
      msg.setJMSReplyTo(consumerDestination);

    } catch (Exception e) {
      throw new ResourceProcessException(e);
    }

  }
  private void stopConnection() {
	
	SharedConnection sharedConnection;
    // there is a dedicated shared connection for each broker URI
	if ( brokerURI != null && (sharedConnection = lookupConnection(brokerURI)) != null) {
    	try {
    		if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
        		String msg = ":::::::::::::::: "+Thread.currentThread().getId()+" stopConnection() - acquiring semaphore";
    	          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "stopConnection",
    	                 JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
    	                  new Object[] { msg });
    	    }
    		// use broker dedicated semaphore to lock the code for updates
        	sharedConnection.getSemaphore().acquire();
    		if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
        	  String msg =":::::::::::::::: "+Thread.currentThread().getId()+" stopConnection() - acquired semaphore";
  	          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "stopConnection",
  	                 JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
  	                  new Object[] { msg });
  	        }
        	 
        	// Remove a client from registry
    	      sharedConnection.unregisterClient(this);
    		if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
          		String msg = ":::::::::::::::: "+Thread.currentThread().getId()+" stopConnection() - 1";
    	          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "stopConnection",
    	                 JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
    	                  new Object[] { msg });
   	        }

    	      ActiveMQConnection amqc = (ActiveMQConnection)sharedConnection.getConnection();
           		if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
            		String msg = ":::::::::::::::: "+Thread.currentThread().getId()+" stopConnection() - 2";
        	          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "stopConnection",
        	                 JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
        	                  new Object[] { msg });
       	        }

        		if (initialized) {
               		if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
        	      		String msg = ":::::::::::::::: "+Thread.currentThread().getId()+" stopConnection() - 3";
            	          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "stopConnection",
            	                 JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
            	                  new Object[] { msg });
           	        }

   	    	     try {
   	    	    	if ( amqc != null ) {
   	               		if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
   	   	    	    		String msg = "................ Closing Client Connection client ID:"+amqc.getConnectionInfo().getClientId()+" Client Count:"+sharedConnection.getClientCount();
   	            	          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "stopConnection",
   	            	                 JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
   	            	                  new Object[] { msg });
   	           	        }
   	    	    		
   	    	    	} else {
   	               		if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
   	   	    	      		String msg = ":::::::::::::::: "+Thread.currentThread().getId()+" stopConnection() - 5";
   	            	          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "stopConnection",
   	            	                 JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
   	            	                  new Object[] { msg });
   	           	        }

   	    	    	}
    	    		if ( amqc != null && amqc.isStarted() && 
    	    			 ((ActiveMQSession)consumerSession).isRunning() ) {
   	   	           		consumerSession.close();
     	    	   	    ((ActiveMQMessageConsumer)consumer).stop();
     	    	   	    consumer.close();
   	               		if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
   	     	         		String msg = ":::::::::::::::: "+Thread.currentThread().getId()+" stopConnection() - 6";
   	            	          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "stopConnection",
   	            	                 JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
   	            	                  new Object[] { msg });
   	           	        }
   	   	        		if ( sharedConnection.getClientCount() == 1 ) {
   	   	        			
   	   	        		   sharedConnection.destroy();
   	   	        		   amqc.close();
   	   	        		}
    	    		}
   	   	         } catch (Exception exx) {exx.printStackTrace();}
    	   	  }
    	      // Delete client's temp reply queue from AMQ Broker 
    	      if ( amqc != null && !amqc.isClosed() && !amqc.isClosing() && consumerDestination != null && 
    	           consumerDestination instanceof ActiveMQTempDestination ) {
    	        try {
    	        	if ( !amqc.isClosed() && !amqc.isTransportFailed()) {
    	    	          amqc.deleteTempDestination((ActiveMQTempDestination)consumerDestination);
    	        	}
    	        } catch( Exception e) {
    	          e.printStackTrace();
    	        }
    	      }
    		
    	} catch (Exception e) {
    		
    	} finally {
    		sharedConnection.destroy();
    		if ( sharedConnection != null  ) {
    			sharedConnection.getSemaphore().release();
    		}

    	}
    }
  }
	public void stop() {
		try {
			  System.out.println(this.getClass().getName()+".stop() - Stopping UIMA-AS Client");
			  stopConnection();

		      super.doStop();
		      if (!running) {
		        return;
		      }
		      running = false;
			  if (super.serviceDelegate != null) {
				// Cancel all timers and purge lists
				super.serviceDelegate.cleanup();
			  }
		      if (sender != null) {
		        sender.doStop();
		      }
			  try {
//				  System.out.println(this.getClass().getName()+".stop() - Stopping UIMA-AS Client");
//				stopConnection();
				// Undeploy all containers
				undeploy();
		 	    clientCache.clear();
				if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(
								Level.INFO)) {
					UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO,
									CLASS_NAME.getName(), "stop",
									JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
									"UIMAJMS_undeployed_containers__INFO");
				}
				// unregister client
				if (jmxManager != null) {
					jmxManager.unregisterMBean(clientJmxObjectName);
					jmxManager.destroy();
				}
			  } catch (Exception e) {
				if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(
								Level.WARNING)) {
					UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING,
									CLASS_NAME.getName(), "stop",
									JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
									"UIMAJMS_exception__WARNING", e);
				}
			  }
				
	//		}
		} catch( Exception e) {
			e.printStackTrace();
		} 
	}

  public void setCPCMessage(Message msg) throws Exception {
    msg.setStringProperty(AsynchAEMessage.MessageFrom, consumerDestination.getQueueName());
    msg.setStringProperty(UIMAMessage.ServerURI, brokerURI);
    msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
    msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.CollectionProcessComplete);
    msg.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.None);
    msg.setBooleanProperty(AsynchAEMessage.RemoveEndpoint, true);
    msg.setJMSReplyTo(consumerDestination);
    if (msg instanceof TextMessage) {
      ((TextMessage) msg).setText("");
    }
  }

	private boolean connectionClosedOrInvalid() {
			SharedConnection sharedConnection = lookupConnection(brokerURI);
			if (sharedConnection == null
					|| sharedConnection.getConnection() == null
					|| ((ActiveMQConnection) sharedConnection.getConnection())
							.isClosed()
					|| ((ActiveMQConnection) sharedConnection.getConnection())
							.isClosing()
					|| ((ActiveMQConnection) sharedConnection.getConnection())
							.isTransportFailed()) {
				return true;
			}
		return false;
	}

	private SharedConnection createAndInitializeAMQConnection( Semaphore semaphore, String aBrokerURI) throws Exception {
		// This only effects Consumer
		// Create AMQ specific connection validator. It uses
		// AMQ specific approach to test the state of the connection
		ActiveMQConnectionValidator connectionValidator = new ActiveMQConnectionValidator();
		//Initalize the connection Factory
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(aBrokerURI);
		connectionFactory.setTrustAllPackages(true);
		connectionFactory.setUserName(amqUser);
		connectionFactory.setPassword(amqPassword);
		// Create a singleton shared connection object
		SharedConnection sharedConnection = sharedConnections.get(aBrokerURI);
		sharedConnection.setConnectionFactory(connectionFactory);
/*
		new SharedConnection(
				connectionFactory,
				//new ActiveMQConnectionFactory(aBrokerURI),
				aBrokerURI);
		sharedConnection.setSemaphore(semaphore);
	*/
		// Add AMQ specific connection validator
		sharedConnection
				.setConnectionValidator(connectionValidator);
		// Connect to broker. Throws exception if unable to connect
		sharedConnection.create();
		addPrefetch((ActiveMQConnection) sharedConnection
				.getConnection());
		((ActiveMQConnection) sharedConnection.getConnection()).setExceptionListener(new ClientExceptionListener());
		//System.out.println(">>>>>>>>>>>>>>>> Starting Connection to Broker:"+aBrokerURI);
		sharedConnection.start();
		sharedConnections.put( aBrokerURI, sharedConnection);

		if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(
				Level.INFO)) {
			UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO,
					CLASS_NAME.getName(), "createAndInitializeAMQConnection",
					JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
					"UIMAJMS_client_connection_setup_INFO",
					new Object[] { aBrokerURI });
		}
		return sharedConnection;
	}
	protected SharedConnection createSharedConnection(String aBrokerURI) throws Exception {
		Semaphore perBrokerSemaphore;
		SharedConnection sharedConnection;
		try {
			// Use global sharedConnectionSemaphore to lock the code below to 
			// prevent two threads from creating per broker semaphore in case 
			// sharedConnection equals null. 
			sharedConnectionSemaphore.acquire();
			// fetch shared connection object for a given broker URI
			sharedConnection = sharedConnections.get(aBrokerURI);
			// if null, we need create create a new one and start the AMQ connection.
			// Each shared connection has its own binary semaphore to guard critical
			// code.
			if ( sharedConnection == null  ) {
				// create dedicated semaphore for this broker
				perBrokerSemaphore = new Semaphore(1);
				sharedConnection = new SharedConnection(null,aBrokerURI);
				sharedConnection.setSemaphore(perBrokerSemaphore);
				sharedConnections.put( aBrokerURI, sharedConnection);
				
				//sharedConnection = createAndInitializeAMQConnection(perBrokerSemaphore, aBrokerURI);
		    } else {
		    	// fetch dedicated semaphore from shared connectiion object
		    	perBrokerSemaphore = sharedConnection.getSemaphore();
		    }

		} catch( Exception e) {
			throw e;
		} finally {
			sharedConnectionSemaphore.release();
		}
		try {
			// Used broker specific semaphore to lock critical code 
			if ( !perBrokerSemaphore.tryAcquire(2, TimeUnit.SECONDS) ) {
				throw new TimeoutException("UIMA-AS Client Timed Out Waiting to Acquire Broker Semaphore (2 Seconds) - Broker:"+aBrokerURI);
			}
			
		} catch( Exception e) {
			throw e;
		}
		//System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> THREAD:"+Thread.currentThread().getId()+" LOCK: "+aBrokerURI);
		try {
			// check if AMQ connection is still valid
			if ( connectionClosedOrInvalid() ) {
				// connection either stale or not present. Create new shared connection object for this broker
				sharedConnection = createAndInitializeAMQConnection(perBrokerSemaphore, aBrokerURI);
			}
			// Create and initialize dispatch thread with JMS Producer
			initializeProducer(brokerURI, endpoint, sharedConnection.getConnection());
            // Create JMS Consumer and plug in Listener for processing replies
			initializeConsumer(brokerURI, sharedConnection.getConnection());

		    // Increment number of client instances. SharedConnection object is a static
		    // and is used to share a single JMS connection. The connection is closed
		    // when the last client finishes processing and calls stop().
		    if (sharedConnection != null) {
		       sharedConnection.registerClient(this);
		    }
		} catch( Exception e) {
			throw e;
		} finally {
			//System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>  THREAD:"+Thread.currentThread().getId()+" UN-LOCK: "+aBrokerURI);

			perBrokerSemaphore.release();
		}
		return sharedConnection;
	}

  private void addPrefetch(ActiveMQConnection aConnection) {
    ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();
    prefetchPolicy.setQueuePrefetch(5);
    ((ActiveMQConnection) aConnection).setPrefetchPolicy(prefetchPolicy);
  }

  protected SharedConnection validateConnection(String aBrokerURI) throws Exception {
    // checks if a sharedConnection exists and if not creates a new one
    return createSharedConnection(aBrokerURI);
  }

  protected Session getSession(String aBrokerURI) throws Exception {
	SharedConnection sharedConnection = validateConnection(aBrokerURI);
    return getSession(sharedConnection.getConnection());
  }

  protected Session getSession(Connection aConnection) throws Exception {
    session = aConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    return session;
  }

  protected MessageProducer lookupProducerForEndpoint(Endpoint anEndpoint) throws Exception {
    if (lookupConnection(brokerURI) == null || producerSession == null) {
      throw new ResourceInitializationException();
    }
    Destination dest = producerSession.createQueue(anEndpoint.getEndpoint());
    return producerSession.createProducer(dest);
  }

  protected void initializeProducer(String aBrokerURI, String aQueueName) throws Exception {
    // Check if a sharedConnection exists. If not it creates one
    SharedConnection sharedConnection = createSharedConnection(aBrokerURI);
    initializeProducer(aBrokerURI, aQueueName, sharedConnection.getConnection());
  }

  protected void initializeProducer(String aBrokerURI, String aQueueName, Connection aConnection)
          throws Exception {
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
              "initializeProducer", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
              "UIMAJMS_init_jms_producer_INFO", new Object[] { aBrokerURI, aQueueName });
    }
    brokerURI = aBrokerURI;
    // Create a worker thread for sending messages. Jms sessions are single threaded
    // and it is illegal (per JMS spec) to use the same sesssion from multiple threads.
    // The worker thread solves this problem. As it is the only thread that owns the
    // session and uses it to create message producer.
    // The worker thread blocks waiting for messages from application threads. The
    // application threads add messages to the shared "queue" (in-memory queue not
    // jms queue) and the worker thread consumes them. The worker thread is not
    // serialializing CASes. This work is done in application threads.

    // create a Message Dispatcher object. In its constructor it acquires a shared
    // semaphore producerSemaphore and holds it until the producer is created an
    // and initialized. Once this happens or there is an error, the semaphore is
    // released.
    sender = new ActiveMQMessageSender(aConnection, aQueueName, this);
    producerInitialized = false;
    Thread t = new Thread((BaseMessageSender) sender);
    // Start the worker thread. The jms session and message producer are created. Once
    // the message producer is created, the worker thread notifies this thread by
    // calling onProducerInitialized() where the global flag 'producerInitialized' is
    // set to true. After the notification, the worker thread notifies this instance
    // that the producer is fully initialized and finally begins to wait for messages
    // in pendingMessageList. Upon arrival, each message is removed from
    // pendingMessageList and it is sent to a destination.

    t.start();

    try {
      // Block waiting for the Sender to complete initializing the Producer.
      // The sender will release the lock once it instantiates and initializes
      // the Producer object or if there is an error
      producerSemaphore.acquire();
    } catch (InterruptedException ex) {

    } finally {
      producerSemaphore.release();
    }
    // Check if the worker thread failed to initialize.
    if (sender.failed()) {
      // Worker thread failed to initialize. Log the reason and stop the uima ee client
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                "initializeProducer", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAJMS_worker_thread_failed_to_initialize__WARNING",
                new Object[] { sender.getReasonForFailure() });
      }
      stop();
      return;
    }
  }

  /**
   * Create a JMS Consumer on a temporary queue. Service replies will be handled by this consumer.
   * 
   * @param aBrokerURI
   * @throws Exception
   */
  protected void initializeConsumer(String aBrokerURI) throws Exception {
	  SharedConnection  sharedConnection = createSharedConnection(aBrokerURI);
    initializeConsumer(aBrokerURI, sharedConnection.getConnection());
  }

  protected void initializeConsumer(String aBrokerURI, Connection connection) throws Exception {
    // In case we are recovering from a bad broker connection, invalidate old
    // JMS objects first.
    if ( consumerSession != null ) {
      try {
        consumer.close();
        consumerSession.close();
      } catch( Exception e) {
        //  ignore, creating a new Session below
      }
    }
    
    consumerSession = getSession(connection);
    consumerDestination = consumerSession.createTemporaryQueue();
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
              "initializeConsumer", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
              "UIMAJMS_init_jms_consumer_INFO",
              new Object[] { aBrokerURI, consumerDestination.getQueueName() });
    }
    consumer = consumerSession.createConsumer(consumerDestination);
    consumer.setMessageListener(this);
  }
  
  /**
   * Replaces place holder with syntax ${pname} with a system property whose name is pname
   * 
   * @param aPlaceholder to resolve
   * @return - the actual broker URL
   * @throws ResourceInitializationException
   */
  private String replacePlaceholder(String aPlaceholder) throws ResourceInitializationException {
    //  find placeholder starting and ending positions
    // If no '${' or following '}' leave as-is
    int startPos = aPlaceholder.indexOf("${");
    if (startPos < 0) {
      return aPlaceholder;
    }
    int endPos = aPlaceholder.indexOf("}", startPos);
    if (endPos < 0) {
      return aPlaceholder;
    }
    //  extract the name
    String placeholder = aPlaceholder.substring(startPos+2, endPos);
    //  using the name, find the broker URL. This property must exist or exception is thrown
    String url = System.getProperty(placeholder);
    //  the property is missing 
    if ( url == null ) {
      throw new ResourceInitializationException(new Exception("UIMA AS Client Initialization Exception. Value for placeholder:"+placeholder+" is not defined in the system properties."));
    }
    return aPlaceholder.substring(0, startPos) + url + aPlaceholder.substring(endPos+1);
  }
  /**
   * Initialize the uima ee client. Takes initialization parameters from the
   * <code>anApplicationContext</code> map.
   */
  public synchronized void initialize(Map anApplicationContext)
          throws ResourceInitializationException {
    // Add ShutdownHook to make sure the connection to the
    // broker is always closed on process exit.
    shutdownHookThread = new Thread(new UimaASShutdownHook(this));
    Runtime.getRuntime().addShutdownHook(shutdownHookThread);
           
    // throws an exception if verions of UIMA-AS is not compatible with UIMA SDK
    VersionCompatibilityChecker.check(CLASS_NAME, "UIMA AS Client", "initialize");
/*
    // Check for compatibility with a version of uima sdk. Only check major versions.
    if (UimaAsVersion.getMajorVersion() != UimaVersion.getMajorVersion() ) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(
              Level.WARNING,
              CLASS_NAME.getName(), 
              "initialize",
              UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
              "UIMAEE_incompatible_version_WARNING",
              new Object[] { "UIMA AS Client", UimaAsVersion.getUimajFullVersionString(),
                UimaVersion.getFullVersionString() });
      throw new ResourceInitializationException(new AsynchAEException(
              "Version of UIMA-AS is Incompatible with a Version of UIMA Core. UIMA-AS Version is built to depend on Core UIMA version:"
                      + UimaAsVersion.getUimajFullVersionString() + " but is running with version:"
                      + UimaVersion.getFullVersionString()));
    }
*/
    if (running) {
      throw new ResourceInitializationException(new UIMA_IllegalStateException());
    }
    reset();
    Properties performanceTuningSettings = null;

    if (!anApplicationContext.containsKey(UimaAsynchronousEngine.ServerUri)) {
      throw new ResourceInitializationException();
    }
    if (!anApplicationContext.containsKey(UimaAsynchronousEngine.ENDPOINT)) {
      throw new ResourceInitializationException();
    }
    ResourceManager rm = null;
    if (anApplicationContext.containsKey(Resource.PARAM_RESOURCE_MANAGER)) {
      rm = (ResourceManager) anApplicationContext.get(Resource.PARAM_RESOURCE_MANAGER);
    } else {
      rm = UIMAFramework.newDefaultResourceManager();
    }
    performanceTuningSettings = new Properties();
    if (anApplicationContext.containsKey(UIMAFramework.CAS_INITIAL_HEAP_SIZE)) {
      String cas_initial_heap_size = (String) anApplicationContext
              .get(UIMAFramework.CAS_INITIAL_HEAP_SIZE);
      performanceTuningSettings.put(UIMAFramework.CAS_INITIAL_HEAP_SIZE, cas_initial_heap_size);
    }
    asynchManager = new AsynchAECasManager_impl(rm);

    brokerURI = (String) anApplicationContext.get(UimaAsynchronousEngine.ServerUri);
    endpoint = (String) anApplicationContext.get(UimaAsynchronousEngine.ENDPOINT);
    
    //  Check if a placeholder is passed in instead of actual broker URL or endpoint. 
    //  The placeholder has the syntax ${placeholderName} and may be imbedded in text.
    //  A system property with placeholderName must exist for successful placeholder resolution.
    //  Throws ResourceInitializationException if placeholder is not in the System properties.
    brokerURI = replacePlaceholder(brokerURI); 
    endpoint = replacePlaceholder(endpoint); 

    clientSideJmxStats.setEndpointName(endpoint);
    int casPoolSize = 1;

    if (anApplicationContext.containsKey(UimaAsynchronousEngine.CasPoolSize)) {
      casPoolSize = ((Integer) anApplicationContext.get(UimaAsynchronousEngine.CasPoolSize))
              .intValue();
      clientSideJmxStats.setCasPoolSize(casPoolSize);
    }

    if (anApplicationContext.containsKey(UimaAsynchronousEngine.Timeout)) {
      processTimeout = ((Integer) anApplicationContext.get(UimaAsynchronousEngine.Timeout))
              .intValue();
    }

    if (anApplicationContext.containsKey(UimaAsynchronousEngine.GetMetaTimeout)) {
      metadataTimeout = ((Integer) anApplicationContext.get(UimaAsynchronousEngine.GetMetaTimeout))
              .intValue();
    }

    if (anApplicationContext.containsKey(UimaAsynchronousEngine.CpcTimeout)) {
      cpcTimeout = ((Integer) anApplicationContext.get(UimaAsynchronousEngine.CpcTimeout))
              .intValue();
    }
    if (anApplicationContext.containsKey(UimaAsynchronousEngine.ApplicationName)) {
      applicationName = (String) anApplicationContext.get(UimaAsynchronousEngine.ApplicationName);
    }
    if (anApplicationContext.containsKey(UimaAsynchronousEngine.SERIALIZATION_STRATEGY)) {
      final String serializationStrategy = (String) anApplicationContext.get(UimaAsynchronousEngine.SERIALIZATION_STRATEGY);
      // change this to support compressed filitered as the default
      setSerialFormat((serializationStrategy.equalsIgnoreCase("xmi")) ? SerialFormat.XMI : SerialFormat.BINARY);
//      setSerialFormat((serializationStrategy.equalsIgnoreCase("xmi")) ? SerialFormat.XMI : SerialFormat.COMPRESSED_FILTERED);
      clientSideJmxStats.setSerialization(getSerialFormat());
    }
    if (anApplicationContext.containsKey(UimaAsynchronousEngine.userName)) {
        amqUser = (String) anApplicationContext
                .get(UimaAsynchronousEngine.userName);
    }
    if (anApplicationContext.containsKey(UimaAsynchronousEngine.password)) {
    	amqPassword = (String) anApplicationContext
    	.get(UimaAsynchronousEngine.password);
    }

    if (anApplicationContext.containsKey(UimaAsynchronousEngine.TimerPerCAS)) {
        timerPerCAS = ((Boolean) anApplicationContext.get(UimaAsynchronousEngine.TimerPerCAS))
                .booleanValue();
      }
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.CONFIG)) {
      UIMAFramework.getLogger(CLASS_NAME)
              .logrb(
                      Level.CONFIG,
                      CLASS_NAME.getName(),
                      "initialize",
                      JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAJMS_init_uimaee_client__CONFIG",
                      new Object[] { brokerURI, 0, casPoolSize, processTimeout, metadataTimeout,
                          cpcTimeout,timerPerCAS });
    }
    super.serviceDelegate = new ClientServiceDelegate(endpoint, applicationName, this);
    super.serviceDelegate.setCasProcessTimeout(processTimeout);
    super.serviceDelegate.setGetMetaTimeout(metadataTimeout);
    try {
      // Generate unique identifier
      String uuid = UUIDGenerator.generate();
      // JMX does not allow ':' in the ObjectName so replace these with underscore
      uuid = uuid.replaceAll(":", "_");
      uuid = uuid.replaceAll("-", "_");
      applicationName += "_" + uuid;
      jmxManager = new JmxManager("org.apache.uima");
      clientSideJmxStats.setApplicationName(applicationName);
      clientJmxObjectName = new ObjectName("org.apache.uima:name=" + applicationName);
      jmxManager.registerMBean(clientSideJmxStats, clientJmxObjectName);

      // Check if sharedConnection exists. If not create a new one. The sharedConnection
      // is static and shared by all instances of UIMA AS client in a jvm. The check
      // is made in a critical section by first acquiring a global static semaphore to
      // prevent a race condition.
      createSharedConnection(brokerURI);
      running = true;
      //  This is done to give the broker enough time to 'finalize' creation of
      //  temp reply queue. It's been observed (on MAC OS only) that AMQ
      //  broker QueueSession.createTemporaryQueue() call is not synchronous. Meaning,
      //  return from createTemporaryQueue() does not guarantee immediate availability
      //  of the temp queue. It seems like this operation is asynchronous, causing: 
      //  "InvalidDestinationException: Cannot publish to a deleted Destination..."
      //  on the service side when it tries to reply to the client.
      try {
        wait(100);
      } catch( InterruptedException e) {}
      sendMetaRequest();
      waitForMetadataReply();
      if (abort || !running) {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                  "initialize", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAJMS_aborting_as_WARNING", new Object[] { "Metadata Timeout" });
        }
        throw new ResourceInitializationException(new UimaASMetaRequestTimeout());
      } else {
        if (collectionReader != null) {
          asynchManager.addMetadata(collectionReader.getProcessingResourceMetaData());
        }

        asynchManager.initialize(casPoolSize, "ApplicationCasPoolContext",
                performanceTuningSettings);

        // Create a special CasPool of size 1 to be used for deserializing CASes from a Cas
        // Multiplier
        if (super.resourceMetadata != null
                && super.resourceMetadata instanceof AnalysisEngineMetaData) {
          if (((AnalysisEngineMetaData) super.resourceMetadata).getOperationalProperties()
                  .getOutputsNewCASes()) {
            // Create a Shadow CAS Pool used to de-serialize CASes produced by a CAS Multiplier
            asynchManager.initialize(1, SHADOW_CAS_POOL, performanceTuningSettings);
          }
        }
        initialized = true;
        remoteService = true;
        // running = true;

        for (int i = 0; listeners != null && i < listeners.size(); i++) {
          ((UimaASStatusCallbackListener) listeners.get(i)).initializationComplete(null);
        }
      }

    } catch (ResourceInitializationException e) {
      state = ClientState.FAILED;
      notifyOnInitializationFailure(e);
      throw e;
    } catch (Exception e) {
      state = ClientState.FAILED;
      notifyOnInitializationFailure(e);
      throw new ResourceInitializationException(e);
    }
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "initialize",
              JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_as_initialized__INFO",
              new Object[] { UimaAsynchronousEngine.SERIALIZATION_STRATEGY });
    }
    // Acquire cpcReady semaphore to block sending CPC request until
    // ALL outstanding CASes are received.
    super.acquireCpcReadySemaphore();
    state = ClientState.RUNNING;
  }

  /**
   * First generates a Spring context from a given deploy descriptor and than deploys the context
   * into a Spring Container.
   * 
   * @param aDeploymentDescriptor
   *          - deployment descriptor to generate Spring Context from
   * @param anApplicationContext
   *          - a Map containing properties required by dd2spring
   * 
   * @return - a unique spring container id
   * 
   */
  public String deploy(String aDeploymentDescriptor, Map anApplicationContext) throws Exception {
	  String springContext = null;
	  try {
		springContext = generateSpringContext(aDeploymentDescriptor, anApplicationContext);

        SpringContainerDeployer springDeployer = new SpringContainerDeployer(springContainerRegistry, this);

    	String id = springDeployer.deploy(springContext);
      if ( springDeployer.isInitialized() ) {
        springDeployer.startListeners();
      }
      return id;
    } catch (Exception e) {
      running = true;
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                  "main", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAJMS_exception__WARNING", e);
      }

      throw e;
    } finally {
	  String uimaAsDebug = (String) anApplicationContext.get(UimaAsynchronousEngine.UimaEeDebug);
	  if ( springContext != null && (null == uimaAsDebug || uimaAsDebug.equals("") ) ) {
           disposeContextFiles(springContext);
	  }
    }
  }
  private void disposeContextFiles(String ...contextFiles) {
    for( String contextFile: contextFiles) {
      File file = new File(contextFile);
      if ( file.exists()) {
        file.delete();
      }
    }
  }
  /**
	 * 
	 */
  public String deploy(String[] aDeploymentDescriptorList, Map anApplicationContext)
          throws Exception {
    if (aDeploymentDescriptorList == null) {
      throw new ResourceConfigurationException(UIMA_IllegalArgumentException.ILLEGAL_ARGUMENT,
              new Object[] { "Null", "DeploymentDescriptorList", "deploy()" });
    }

    if (aDeploymentDescriptorList.length == 0) {
      throw new ResourceConfigurationException(
              ResourceConfigurationException.MANDATORY_VALUE_MISSING,
              new Object[] { "DeploymentDescriptorList" });
    }
    String[] springContextFiles = new String[aDeploymentDescriptorList.length];

    for (int i = 0; i < aDeploymentDescriptorList.length; i++) {
      springContextFiles[i] = generateSpringContext(aDeploymentDescriptorList[i],
              anApplicationContext);
    }

    SpringContainerDeployer springDeployer = new SpringContainerDeployer(springContainerRegistry);
    try {
      String id = springDeployer.deploy(springContextFiles);
      if ( springDeployer.isInitialized() ) {
        springDeployer.startListeners();
      }
      return id;
    } catch (ResourceInitializationException e) {
      running = true;
      throw e;
    } finally {
	String uimaAsDebug = (String) anApplicationContext.get(UimaAsynchronousEngine.UimaEeDebug);
	if ( null == uimaAsDebug || uimaAsDebug.equals("")) {
           disposeContextFiles(springContextFiles);
	}

	//      disposeContextFiles(springContextFiles);
    }

  }

  public void undeploy() throws Exception {
    Iterator containerIterator = springContainerRegistry.keySet().iterator();
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
        String msg ="undeploying "+springContainerRegistry.size()+" Containers";
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "undeploy",
               JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
                new Object[] { msg });
    }
    while (containerIterator.hasNext()) {
      String containerId = (String) containerIterator.next();
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
          String msg = "Undeploying Container Id:"+containerId;
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "undeploy",
                 JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
                  new Object[] { msg });
      }
      undeploy(containerId);
    }
  }

  public void undeploy(String aSpringContainerId) throws Exception {
    this.undeploy(aSpringContainerId, SpringContainerDeployer.STOP_NOW);
  }

  /**
   * Undeploys Spring container with a given container Id. All deployed Spring containers are
   * registered in the local registry under a unique id.
   * 
   */
  public void undeploy(String aSpringContainerId, int stop_level) throws Exception {
    if (aSpringContainerId == null  ) {
      return;
    }
    
    UimaEEAdminSpringContext adminContext = null;
    if (!springContainerRegistry.containsKey(aSpringContainerId)) {
        return;
        // throw new InvalidContainerException("Invalid Spring container Id:" + aSpringContainerId +
        // ". Unable to undeploy the Spring container");
      }
      // Fetch an administrative context which contains a Spring Container
      adminContext = (UimaEEAdminSpringContext) springContainerRegistry.get(aSpringContainerId);
      if (adminContext == null) {
        throw new InvalidContainerException(
                "Spring Container Does Not Contain Valid UimaEEAdminSpringContext Object");
      }
      // Fetch instance of the Container from its context
      ApplicationContext ctx = adminContext.getSpringContainer();
      // Query the container for objects that implement
      // ControllerLifecycle interface. These
      // objects are typically of type AnalysisEngineController or
      // UimacppServiceController.
      String[] asyncServiceList = ctx
              .getBeanNamesForType(org.apache.uima.aae.controller.ControllerLifecycle.class);
      // Given a valid list of controllers select the first from the list
      // and
      // initiate a shutdown. We don't care which controller will be
      // invoked. In case of
      // AggregateAnalysisEngineController the terminate event will
      // propagate all the way
      // to the top controller in the hierarchy and the shutdown will take
      // place from there.
      // If the controller is of kind UimecppServiceController or
      // PrimitiveAnalysisController
      // the termination logic will be immediately triggered in the
      // terminate() method.
      if (asyncServiceList != null && asyncServiceList.length > 0) {
        boolean topLevelController = false;
        ControllerLifecycle ctrer = null;
        int indx = 0;
        while (!topLevelController) {
          ctrer = (ControllerLifecycle) ctx.getBean(asyncServiceList[indx++]);
          if (ctrer instanceof UimacppServiceController
                  || ((AnalysisEngineController) ctrer).isTopLevelComponent()) {
            topLevelController = true;
          }
        }
        // Send a trigger to initiate shutdown.
        if (ctrer != null) {
          if (ctrer instanceof AnalysisEngineController &&
        		  ((AnalysisEngineController) ctrer).getControllerLatch() != null ) {
            ((AnalysisEngineController) ctrer).getControllerLatch().release();
          }
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
              String msg = "++++++++++++++++++++++ calling terminate()-service:"+((AnalysisEngineController) ctrer).getComponentName();
              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "undeploy",
                     JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
                      new Object[] { msg });
          }
          switch (stop_level) {
            case SpringContainerDeployer.QUIESCE_AND_STOP:
                if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
                    String msg = "++++++++++++++++++++++ calling quiesceAndStop()";
                    UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "undeploy",
                           JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
                            new Object[] { msg });
                }
              ((AnalysisEngineController) ctrer).quiesceAndStop();

              break;
            case SpringContainerDeployer.STOP_NOW:
                if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
                    String msg = "++++++++++++++++++++++ calling terminate()";
                    UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "undeploy",
                           JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
                            new Object[] { msg });
                }
              ((AnalysisEngineController) ctrer).terminate();
              break;
          }
        }
      }
      if (ctx instanceof FileSystemXmlApplicationContext) {
        ((FileSystemXmlApplicationContext) ctx).destroy();
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
            String msg = "---------------------- Destroying Application Context:"+((FileSystemXmlApplicationContext) ctx).getApplicationName();
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "undeploy",
                   JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
                    new Object[] { msg });
        }
      }
      // Remove the container from a local registry
      springContainerRegistry.remove(aSpringContainerId);
  }

  /**
   * Use dd2spring to generate Spring context file from a given deployment descriptor file.
   * 
   * @param aDeploymentDescriptor
   *          - deployment descriptor to generate Spring Context from
   * @param anApplicationContext
   *          - a Map containing properties required by dd2spring
   * @return - an absolute path to the generated Spring Context file
   * 
   * @throws Exception
   *           - if failure occurs
   */
  private String generateSpringContext(String aDeploymentDescriptor, Map anApplicationContext)
          throws Exception {

    String dd2SpringXsltFilePath = null;
    String saxonClasspath = null;

    if (anApplicationContext.containsKey(UimaAsynchronousEngine.DD2SpringXsltFilePath)) {
      dd2SpringXsltFilePath = (String) anApplicationContext
              .get(UimaAsynchronousEngine.DD2SpringXsltFilePath);
    } else {
      throw new ResourceConfigurationException(
              ResourceConfigurationException.MANDATORY_VALUE_MISSING,
              new Object[] { "Xslt File Path" });
    }
    if (anApplicationContext.containsKey(UimaAsynchronousEngine.SaxonClasspath)) {
      saxonClasspath = (String) anApplicationContext.get(UimaAsynchronousEngine.SaxonClasspath);
    } else {
      throw new ResourceConfigurationException(
              ResourceConfigurationException.MANDATORY_VALUE_MISSING,
              new Object[] { "Saxon Classpath" });
    }

    Dd2spring dd2Spring = new Dd2spring();
    File springContextFile = dd2Spring.convertDd2Spring(aDeploymentDescriptor,
            dd2SpringXsltFilePath, saxonClasspath, (String) anApplicationContext
                    .get(UimaAsynchronousEngine.UimaEeDebug));

    return springContextFile.getAbsolutePath();
  }

  /**
   * Deploys provided context files ( and beans) in a new Spring container.
   * 
   */
  protected String deploySpringContainer(String[] springContextFiles)
          throws ResourceInitializationException {

    SpringContainerDeployer springDeployer = new SpringContainerDeployer(this);
    try {
      return springDeployer.deploy(springContextFiles);
    } catch (ResourceInitializationException e) {
      // turn on the global flag so that the stop() can do the cleanup
      running = true;
      throw e;
    }
  }

  protected void waitForServiceNotification() throws Exception {
    while (!serviceInitializationCompleted) {
      if (serviceInitializationException) {
        throw new ResourceInitializationException();
      }
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                "waitForServiceNotification", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAJMS_awaiting_container_init__INFO", new Object[] {});
      }
      try {
        serviceSemaphore.acquire();
      } catch (InterruptedException e) {
      } finally {
        serviceSemaphore.release();
      }
      if (serviceInitializationException) {
        throw new ResourceInitializationException();
      }
    }
  }

  protected void deployEmbeddedBroker() throws Exception {
    // TBI
  }

  public static void main(String[] args) {
    try {

      BaseUIMAAsynchronousEngineCommon_impl uimaee = new BaseUIMAAsynchronousEngine_impl();

      Map appContext = new HashMap();
      appContext.put(UimaAsynchronousEngine.DD2SpringXsltFilePath, args[1]);
      appContext.put(UimaAsynchronousEngine.SaxonClasspath, args[2]);
      String containerId = uimaee.deploy(args[0], appContext); // args[1],
      // args[2]);

      uimaee.undeploy(containerId);
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                "main", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAJMS_exception__WARNING", e);
      }
    }
  }

  public void setReleaseCASMessage(TextMessage msg, String aCasReferenceId) throws Exception {
    msg.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.None);
    msg.setStringProperty(AsynchAEMessage.CasReference, aCasReferenceId);
    msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
    msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.ReleaseCAS);
    msg.setStringProperty(UIMAMessage.ServerURI, brokerURI);
    msg.setJMSReplyTo(consumerDestination);
  }

  public void notifyOnInitializationFailure(Exception e) {
    notifyOnInitializationFailure(null, e);
  }

  public void notifyOnInitializationSuccess() {
    notifyOnInitializationSuccess(null);
  }

  public void notifyOnInitializationFailure(AnalysisEngineController aController, Exception e) {

    // Initialization exception. Notify blocking thread and indicate a problem
    serviceInitializationException = true;
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
              "notifyOnInitializationFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
              "UIMAJMS_container_init_exception__WARNING", e);
    }
    serviceSemaphore.release();
  }

  public void notifyOnInitializationSuccess(AnalysisEngineController aController) {
    serviceInitializationCompleted = true;
    serviceSemaphore.release();
  }

  
  public void notifyOnTermination(String aServiceName, EventTrigger cause) {
    for (int i = 0; listeners != null && i < listeners.size(); i++) {
        UimaAsBaseCallbackListener statCL = (UimaAsBaseCallbackListener) listeners.get(i);
        statCL.onUimaAsServiceExit(cause);
    }
  }

  public void notifyOnTermination(String aServiceName, String aCasReferenceId, Exception cause) {
//    super.n
  }

  protected MessageProducer getMessageProducer(Destination destination) throws Exception {
    return sender.getMessageProducer(destination);
  }

  /**
   * Request Uima AS client to initiate sending Stop requests to a service for all outstanding CASes
   * awaiting reply.
   * 
   */
  public void stopProducingCases() {
    
    String[] casIdsPendingReply = serviceDelegate.getDelegateCasIdsPendingReply();
    if ( casIdsPendingReply != null && casIdsPendingReply.length > 0 ) {
      for( String casReferenceId : casIdsPendingReply ) {
        // The Cas is still being processed
        ClientRequest clientCachedRequest = (ClientRequest) clientCache
                .get(casReferenceId);
        if (clientCachedRequest != null && !clientCachedRequest.isMetaRequest()
                && clientCachedRequest.getCasReferenceId() != null) {
          stopProducingCases(casReferenceId, clientCachedRequest.getFreeCasNotificationQueue());
        } 
      }
    }
  }

  /**
   * Request Uima AS client to initiate sending Stop request to a service for a given CAS id If the
   * service is a Cas Multiplier, it will stop producing new CASes, will wait until all child CASes
   * finish and finally returns the input CAS.
   * 
   */
  public void stopProducingCases(String aCasReferenceId) {
    // The Cas is still being processed
    ClientRequest clientCachedRequest = (ClientRequest) clientCache.get(aCasReferenceId);
    if (clientCachedRequest != null) {
      stopProducingCases(aCasReferenceId, clientCachedRequest.getFreeCasNotificationQueue());
    }
  }

//  private void stopProducingCases(ClientRequest clientCachedRequest) {
  private void stopProducingCases(String casReferenceId, Destination cmFreeCasQueue) {
    try {
//      if (clientCachedRequest.getFreeCasNotificationQueue() != null) {
      if (cmFreeCasQueue != null) {
        TextMessage msg = createTextMessage();
        msg.setText("");
        msg.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.None);
//        msg.setStringProperty(AsynchAEMessage.CasReference, clientCachedRequest.getCasReferenceId());
        msg.setStringProperty(AsynchAEMessage.CasReference, casReferenceId);
        msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
        msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.Stop);
        msg.setStringProperty(UIMAMessage.ServerURI, brokerURI);
        try {
          MessageProducer msgProducer = getMessageProducer(cmFreeCasQueue);
          if (msgProducer != null) {
        	  
              if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
                  UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                          "stopProducingCases", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                          "UIMAJMS_client_sending_stop_to_service__INFO", new Object[] {casReferenceId,cmFreeCasQueue});
              }
            // Send STOP message to Cas Multiplier Service
            msgProducer.send(msg);
          } else {
              if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
                  UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                          "stopProducingCases", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                          "UIMAJMS_client_unable_to_send_stop_to_cm__WARNING");
              }
          }

        } catch (Exception ex) {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                    "stopProducingCases", 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(),
                "stopProducingCases", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAJMS_exception__WARNING", e);
      }
    }
  }
  protected MessageSender getDispatcher() {
    return sender;
  }
  public void notifyOnReconnecting(String aMessage) {
  
  }
  
  public void notifyOnReconnectionSuccess() {
    
  }


  public void onApplicationEvent(ApplicationEvent event) {
	for (int i = 0; listeners != null && i < listeners.size(); i++) {
	    UimaAsBaseCallbackListener statCL = (UimaAsBaseCallbackListener) listeners.get(i);
	    if ( event instanceof UimaASApplicationExitEvent) {
		   statCL.onUimaAsServiceExit( ((UimaASApplicationExitEvent)event).getEventTrigger());
	    }
    }
  }
  public class ClientExceptionListener implements ExceptionListener {

	@Override
	public void onException(JMSException arg0) {
		arg0.printStackTrace();
		
	}
	  
  }
}