/*

 * 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.

* Analysis Engine service wrapper implementation based on 
* Active MQ C++ client.
*/

#include "ActiveMQAnalysisEngineService.hpp"
#include "deployCppService.hpp"

#include <activemq/core/ActiveMQConnectionFactory.h>
#include <activemq/core/ActiveMQProducer.h>
#include <activemq/core/ActiveMQConstants.h>
#include <activemq/core/ActiveMQConnection.h>

#include "uima/xmlwriter.hpp"
#include "uima/xcasdeserializer.hpp"
#include "uima/xmiwriter.hpp"
#include "uima/xmideserializer.hpp"
#include "uima/xmlerror_handler.hpp"

using namespace activemq::core;
using namespace uima;

enum traceLevels {NONE, INFO, FINE, FINER, FINEST };
traceLevels uimacpp_ee_tracelevel=INFO;
#define MSGHEADER   apr_time_now() << " ThreadId: " << apr_os_thread_current() << __FILE__ << ":" << __LINE__ 
#define FORMATMSG(x)  stringstream lstr; lstr << MSGHEADER << " " << x;

#define LOGINFO(n,x) { if (n > uimacpp_ee_tracelevel) {} else { FORMATMSG(x); logMessage(lstr.str().c_str()); } }
#define LOGERROR(x){ FORMATMSG(x); logError(lstr.str().c_str()); }
#define LOGWARN(x) { FORMATMSG(x); logWarning(lstr.str().c_str());}

static void listener_signal_handler(int signum) {
    stringstream str;
    str << __FILE__ << __LINE__ << " Received Signal: " << signum;
    cerr << str.str() << endl;    
}


//=================================================================
//
//Message processing function executed by message handling threads.
//-----------------------------------------------------------------
static void* APR_THREAD_FUNC handleMessages(apr_thread_t *thd, void *data) {  
   //cout << __FILE__ << __LINE__ << Thread::getId() << " handleMessages start " << endl;
   AMQListener * handler = (AMQListener*) data;
   handler->receiveAndProcessMessages(thd);
   apr_thread_exit(thd, APR_SUCCESS);
   return NULL;
}


//===================================================
//AMQConnection
//---------------------------------------------------
 ConnectionFactory * AMQConnection::createConnectionFactory(ServiceParameters & params) {

   //encode prefetch size in the broker url as there is no api to set it.
   stringstream str;
   str << params.getBrokerURL();
   if (str.str().find("?") == std::string::npos)
     str << "?jms.prefetchPolicy.queuePrefetch=" << params.getPrefetchSize() ;
   else 
      str << "&jms.prefetchPolicy.queuePrefetch=" << params.getPrefetchSize() ;
  
   //str << params.getBrokerURL() << "?consumer.prefetchSize=" << params.getPrefetchSize() << endl;
   //str << "tcp://127.0.0.1:61616";
   ConnectionFactory * pConnFactory = ConnectionFactory::createCMSConnectionFactory(params.getBrokerURL());
   cout << "AMQConnection()::createConnectionFactory " <<  str.str() << endl;
   return pConnFactory;
 };

 AMQConnection::AMQConnection(  ConnectionFactory * connFact, 
                                      string brokerURL, Monitor * pMonitor, int id) :
                                      iv_id(id),
                                      iv_pConnFact(connFact),
                                      //iv_pMonitor(0),
                                      //iv_brokerURL(((ActiveMQConnectionFactory*)connFact)->getBrokerURL()), 
									  iv_brokerURL(brokerURL),
                                      iv_pConnection(0), 
                                      iv_pConsumerSession(0),
                                      iv_pConsumer(0),
                                      iv_inputQueueName(),
                                      iv_pInputQueue(0),
                                      iv_pListener(0),
                                      iv_pProducerSession(0),
                                      iv_pProducer(0),
                                      iv_pReplyMessage(0),
                                      iv_replyDestinations(),
                                      iv_valid(false),
                                      iv_started(false),
                                      iv_reconnecting(false)  {
    int pos  = iv_brokerURL.find_first_of("?");
    if (pos != string::npos) {
      iv_brokerURL = iv_brokerURL.substr(0,pos);
    }
    iv_pMonitor = pMonitor;
    initialize();
 }
 void AMQConnection::initialize( ) {

    try {
  
      LOGINFO(INFO, "AMQConnection() connecting to " + iv_brokerURL);
      
      // Create a Connection
      if (iv_pConnFact == NULL) {	
        LOGERROR("AMQConnection() invalid connection factory");
        ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
        msg.addParam("AMQConnection() invalid create connection factory");
        ErrorInfo errInfo;
        errInfo.setMessage(msg);
        UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
          UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
          errInfo.getMessage(),
          errInfo.getMessage().getMessageID(),
          ErrorInfo::unrecoverable);
      }

      bool retrying = false;
      while (iv_pConnection == NULL) {
        try {
          iv_pConnection = iv_pConnFact->createConnection();
          if (iv_pConnection == NULL ) {
            if (!retrying) {
              stringstream str;
              str << " AMQConnection::initialize() Connection object is null. Failed to connect to " << iv_brokerURL
                         << ". Retrying..." << endl;
              LOGWARN(str.str());
              retrying = true;
              apr_sleep(30000000); //wait 30 seconds to reconnect
            }
          }
        } catch (cms::CMSException& e) {
          if (!retrying) {
            stringstream str;
            str << "AMQConnection::initialize() Failed to connect to " << iv_brokerURL
                         << e.getMessage() << ". Retrying..." << endl;
			cout << e.getMessage() << endl;
            LOGWARN(str.str());
            retrying = true;
            apr_sleep(30000000); //wait 30 seconds to reconnect
          }
        } 
      }
     
      if (retrying) {
        LOGWARN("AMQConnection::initialize() Connected to " + iv_brokerURL);
      }

      //default exception listener
      this->iv_pConnection->setExceptionListener(this);	
      ActiveMQConnection* amqConnection = dynamic_cast<ActiveMQConnection*>( this->iv_pConnection );
      if( amqConnection != NULL ) {
        amqConnection->addTransportListener( this );
      }
      // Create a Producer Session
      LOGINFO(FINEST,"AMQConnection() create Producer Session " + iv_brokerURL);
      this->iv_pProducerSession = this->iv_pConnection->createSession( Session::AUTO_ACKNOWLEDGE );

      if  (this->iv_pProducerSession == NULL) {
        LOGERROR("AMQConnection() createSession() failed."); 
        ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
        msg.addParam("AMQConnection() createSession failed." );
        ErrorInfo errInfo;
        errInfo.setMessage(msg);
        UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
          UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
          errInfo.getMessage(),
          errInfo.getMessage().getMessageID(),
          ErrorInfo::unrecoverable);
      } 
      this->iv_pProducer = this->iv_pProducerSession->createProducer(NULL);
      if (this->iv_pProducer == NULL) {
        LOGERROR("AMQConnection() could not create MessageProducer ");
        ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
        msg.addParam("AMQConnection() create MessageProducer failed.");
        ErrorInfo errInfo;
        errInfo.setMessage(msg);
        UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
          UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
          errInfo.getMessage(),
          errInfo.getMessage().getMessageID(),
          ErrorInfo::unrecoverable);
      }
      this->iv_pProducer->setDeliveryMode( DeliveryMode::NON_PERSISTENT );
	  
	  //TODO set sendTimeout ?
      ////((ActiveMQProducer*)this->iv_pProducer)->setSendTimeout(3000);
      
	  //create TextMessage
      this->iv_pReplyMessage = this->iv_pProducerSession->createTextMessage();
      if (this->iv_pReplyMessage == NULL) {
        LOGERROR("AMQConnection() create textMessage failed. ");
        ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
        msg.addParam("AMQConnection() failed to create message.");
        ErrorInfo errInfo;
        errInfo.setMessage(msg);
        UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
          UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
          errInfo.getMessage(),
          errInfo.getMessage().getMessageID(),
          ErrorInfo::unrecoverable);
      }
      this->iv_valid = true;
      LOGINFO(0, "AMQConnection() connected successfully to " + iv_brokerURL);
    } catch (cms::CMSException& e) {
      LOGERROR("AMQConnection(): " + e.getMessage());  
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam(e.getMessage());
      ErrorInfo errInfo;
      errInfo.setErrorId(UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE);
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    } catch (...) {
      cout << "... " << endl;
      LOGERROR("AMQConnection() failed to create a connection");  
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam("AMQConnection() create connection failed");
      ErrorInfo errInfo;
      errInfo.setErrorId(UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE);
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }
  }

/* destructor */
  AMQConnection::~AMQConnection() {
    if (this->iv_pConsumer != NULL) {
      delete this->iv_pConsumer;
      this->iv_pConsumer = NULL;
    }
    if (this->iv_pProducer != NULL) {
      delete this->iv_pProducer;
      this->iv_pProducer = NULL;
    }
    if (this->iv_pConsumerSession != NULL) {
      delete this->iv_pConsumerSession;
      this->iv_pConsumerSession = NULL;
    }
    if (this->iv_pProducerSession != NULL) {
      delete this->iv_pProducerSession;
      this->iv_pProducerSession = NULL;
    }
    if (this->iv_pInputQueue != NULL) {
      delete this->iv_pInputQueue;
      this->iv_pInputQueue=NULL;
    }
    if (this->iv_pConnection != NULL) {
      this->iv_pConnection->close();
      delete this->iv_pConnection;
      this->iv_pConnection = NULL;
    }
    if (this->iv_pReplyMessage != NULL) {
      delete this->iv_pReplyMessage;
      this->iv_pReplyMessage=NULL;
    } 
    
    //destinations
    map<string, cms::Destination*>::iterator ite; 
    for (ite= iv_replyDestinations.begin();ite !=  iv_replyDestinations.end();ite++) {
      delete ite->second;
    }
  }

/* create a MessageConsumer session and register a MessageListener
   to receive messages from the input queue. */
  void AMQConnection::createMessageConsumer(string queueName, string selector, int prefetchSize) {
    LOGINFO(FINEST, "AMQConnection::createMessageConsumer() consumer start " + queueName);
    this->iv_inputQueueName = queueName;
    stringstream str;
    str << queueName;
   
    if (this->iv_pConsumerSession != NULL || iv_pConsumer != NULL) {
      LOGERROR("AMQConnection::createMessageConsumer() A session already exists. ");
      ErrorInfo errInfo;
      errInfo.setMessage(ErrorMessage(UIMA_MSG_ID_LOG_ERROR, "A session already exists."));
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }

    this->iv_pConsumerSession = this->iv_pConnection->createSession( Session::AUTO_ACKNOWLEDGE );		
    if  (this->iv_pConsumerSession == NULL) {
      LOGERROR("AMQConnection() createSession failed."); 
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam("AMQConneciton() createSession failed." );
      ErrorInfo errInfo;
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    } 

    this->iv_pInputQueue = this->iv_pConsumerSession->createQueue(this->iv_inputQueueName);

    if (this->iv_pInputQueue == NULL) {
      LOGERROR("AMQConnection::createMessageConsumer() createQueue failed. " + queueName);
      stringstream str;
      str << "AMQConnection::createMessageConsumer() createQueue failed. " << queueName << endl;
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam(str.str());
      ErrorInfo errInfo;
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }

    iv_selector = selector;
	this->iv_prefetchSize = prefetchSize;
    if (selector.length() > 0) {
      this->iv_pConsumer = this->iv_pConsumerSession->createConsumer(this->iv_pInputQueue, iv_selector);
    } else {
      this->iv_pConsumer = this->iv_pConsumerSession->createConsumer(this->iv_pInputQueue);
    }

    if (this->iv_pConsumer == NULL) {
      LOGERROR("AMQConnection::createMessageConsumer() createConsumer failed. " + queueName);
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam("AMQConnection::createMessageConsumer() createConsumer() failed.");
      ErrorInfo errInfo;
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }
    //set prefetch size
	//no api to set prefetchsize so encode in broker url;

    LOGINFO(FINEST, "AMQConnection::createMessageConsumer() " + queueName + " successful.");
  }


  void AMQConnection::onException(const CMSException & ex) {
    //mark endpoint as broken.
    this->iv_valid = false;
    //log that connection is invalid.
    stringstream str;
    str << "AMQConnection()::onException() Connection to " 
                       << iv_brokerURL << " is broken. Reconnecting ... " << ex.getMessage() << endl;
    LOGWARN(str.str());
  }
  void AMQConnection::transportInterrupted() {
        std::cout << "The Connection's Transport has been Interrupted." << std::endl;
		stringstream str;
        str << "AMQConnection()::transportInterrupted() Connection to " 
                       << iv_brokerURL << " has been interrupted. Reconnecting ... " << endl;
        LOGWARN(str.str());
    }

  void AMQConnection::transportResumed() {
        std::cout << "The Connection's Transport has been Restored." << std::endl;
		stringstream str;
        str << "AMQConnection()::transportResumed() Reconnected to " 
                       << iv_brokerURL  << endl;
        LOGWARN(str.str());
    }

  TextMessage * AMQConnection::getTextMessage() {
    if (this->iv_pReplyMessage == NULL) {
      LOGERROR("AMQConnection::getTextMessage() failed. ");
      ErrorInfo errInfo;
      errInfo.setMessage(ErrorMessage(UIMA_MSG_ID_LOG_ERROR, "TextMessage could not be created."));
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }
    iv_pReplyMessage->clearProperties();
    iv_pReplyMessage->clearBody();
    return this->iv_pReplyMessage;
  }

  void AMQConnection::sendMessage(string queuename) {
    LOGINFO(FINEST, "AMQConnection::sendMessage() to " + queuename);
    if (this->iv_pProducer == NULL) {
      LOGERROR("AMQConnection::sendMessage() Invalid Message producer. ");
      ErrorInfo errInfo;
      errInfo.setMessage(ErrorMessage(UIMA_MSG_ID_LOG_ERROR, "No message producer."));
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }
    //look up destination queue
    cms::Destination * pDest;
    map<string,Destination*>::iterator ite = this->iv_replyDestinations.find(queuename);
    if (ite != this->iv_replyDestinations.end()) {
      pDest = ite->second;
    } else {
      pDest = this->iv_pProducerSession->createQueue(queuename);
      if (pDest != NULL) {
        this->iv_replyDestinations[queuename] = pDest;
      }
    }

    if (pDest == NULL) {
      LOGERROR("AMQConnection::sendMessage() invalid destination " + queuename);
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam("AMQConnection::sendMessage() invalid destination.");
      ErrorInfo errInfo;
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }

    this->iv_pProducer->send(pDest,this->iv_pReplyMessage);

    //cout << "producer->send elapsed time " << (apr_time_now() - stime) << endl;
    this->iv_pReplyMessage->clearBody();
    this->iv_pReplyMessage->clearProperties();
    LOGINFO(FINEST, "AMQConnection::sendMessage() successful to " + queuename);
  }

  void AMQConnection::sendMessage(const Destination * cmsReplyTo) {
    LOGINFO(FINEST, "AMQConnection::sendMessage() to " +
 ((Queue*)cmsReplyTo)->getQueueName() );
    //LOGINFO(FINEST, "AMQConnection::sendMessage() to " + cmsReplyTo->toProviderString());
    if (this->iv_pProducer == NULL) {
      LOGERROR("AMQConnection::getTextMessage Invalid message producer. ");
      ErrorInfo errInfo;
      errInfo.setMessage(ErrorMessage(UIMA_MSG_ID_LOG_ERROR, "A MessageProducer does not exist."));
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }
    this->iv_pProducer->send(cmsReplyTo,this->iv_pReplyMessage);
    //cout << "producer->send elapsed time " << (apr_time_now() - stime) << endl;
    this->iv_pReplyMessage->clearBody();
    this->iv_pReplyMessage->clearProperties();
    LOGINFO(4, "AMQConnection::sendMessage() successful to " + ((Queue*)cmsReplyTo)->getQueueName());
	//LOGINFO(4, "AMQConnection::sendMessage() successful to " + cmsReplyTo->toProviderString());
  }

// must be called to start receiving messages 
  void AMQConnection::start() {
    if (this->iv_pConnection != NULL) {
      this->iv_pConnection->start();
      LOGINFO(0,"AMQConnection::start() Start receiving messages.");
    } else {
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam("AMQConnection::start() failed. A connection does not exist.");
      ErrorInfo errInfo;
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }	
  }

//stops receiving messages
  void AMQConnection::stop() {
    if (this->iv_pConnection != NULL) {
      this->iv_pConnection->stop();
    } else {
      LOGERROR("AMQConnection::stop() invalid connection. ");
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam("stop() invalid connection.");
      ErrorInfo errInfo;
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }
  }

  void AMQConnection::resetBeforeReconnect() {
    this->iv_reconnecting = true;

    if (this->iv_pConsumer != NULL) {
      delete this->iv_pConsumer;
      this->iv_pConsumer = NULL;
    }
    if (this->iv_pProducer != NULL) {
      delete this->iv_pProducer;
      this->iv_pProducer = NULL;
    }
    if (this->iv_pConsumerSession != NULL) {
      delete this->iv_pConsumerSession;
      this->iv_pConsumerSession = NULL;
    }
    if (this->iv_pProducerSession != NULL) {
      delete this->iv_pProducerSession;
      this->iv_pProducerSession = NULL;
    }
    if (this->iv_pConnection != NULL) {
      delete this->iv_pConnection;
      this->iv_pConnection = NULL;
    }

    this->iv_started = false;
    this->iv_valid = false;

    if (this->iv_pReplyMessage != NULL) {
      delete this->iv_pReplyMessage;
      this->iv_pReplyMessage=NULL;
    } 
    
    //destinations
    map<string, cms::Destination*>::iterator ite; 
    for (ite= iv_replyDestinations.begin();ite !=  iv_replyDestinations.end();ite++) {
      delete ite->second;
    }
  }

  void AMQConnection::reconnect() {
    try {
        this->iv_reconnecting = true;
        this->iv_pMonitor->reconnecting(iv_id);
        apr_sleep(30000000); //wait 30 seconds to reconnect
        //cout << "AMQConnection::reconnect() calling initialize >>>" << endl;
        this->initialize();
        this->createMessageConsumer(this->iv_inputQueueName,this->iv_selector, this->iv_prefetchSize);
        this->start();
        stringstream str;
        str << "AMQConnection::reconnect() Successfully reconnected to >>> " <<  iv_brokerURL << endl;
        LOGWARN(str.str());
        this->iv_pMonitor->reconnectionSuccess(iv_id);
        this->iv_reconnecting = false;
     } catch (uima::Exception e) {
       cerr << "AMQConnection::reconnect() " << e << endl;	
       resetBeforeReconnect();
     } catch (...) {
       cerr << "AMQConnection::reconnect() catch ... " << endl;	
       resetBeforeReconnect();
     }

  }
  

  Message * AMQConnection::receive(const int delay) {
      if (iv_valid) {
        return iv_pConsumer->receive(delay);
      } else {
        this->resetBeforeReconnect();
        this->reconnect();
        this->iv_valid = true;
        return NULL;
      }
  }

//===================================================
//AMQConnectionCache
//---------------------------------------------------
  AMQConnectionsCache::AMQConnectionsCache(ConnectionFactory * pConnFact,Monitor * stats) {
    iv_pConnFact = pConnFact;
    iv_pMonitor = stats;
  }

  AMQConnectionsCache::~AMQConnectionsCache() {
    map<string,AMQConnection*>::iterator ite;
    for (ite = this->iv_connections.begin(); ite != this->iv_connections.end(); ite++) {
      if (ite->second != NULL) {
        delete ite->second;
      }	
    }	
  }

//Retrieves a Connection from the cache if it 
//exists or establishes a connection to the
//the specified broker and adds to the cache 
//and returns the new connection.
  AMQConnection * AMQConnectionsCache::getConnection(string brokerURL) {
    LOGINFO(FINE,"AMQConnectionCache::getConnection() looking up connection to " 
      + brokerURL );	 
    AMQConnection * connection = NULL;

    try {	    	
      map<string,AMQConnection*>::iterator ite;
      ite = this->iv_connections.find(brokerURL);
      if (ite == iv_connections.end()) {
        LOGINFO(FINE,"AMQConnectionsCache::getConnection() create new connection to " +
          brokerURL);
        connection = new AMQConnection(iv_pConnFact, brokerURL, iv_pMonitor, iv_connections.size());
        if (connection == NULL) {
          LOGERROR("AMQConnectionCache::getConnection Could not create a endpoint connection to " +
            brokerURL);
        } else {								    
          this->iv_connections[brokerURL] = connection;
        }
      } else {
        connection = ite->second;
        //if not a valid connection, reconnect
        if (connection == NULL) {
          LOGERROR("AMQConnectionCache::getConnection() could not connect to "
            + brokerURL);
        } else {
          if (!connection->isValid()) {
            LOGWARN("AMQConnectionCache::getEndPoint() Existing connection invalid. Reconnecting to " + brokerURL );
            delete connection;
            this->iv_connections.erase(brokerURL);
            connection = new AMQConnection(iv_pConnFact, brokerURL, iv_pMonitor, iv_connections.size());
            if (connection == NULL) {
              LOGERROR("AMQConnectionCache::getConnection() could not connect to "
                + brokerURL );
            } else {
              LOGWARN("AMQConnectionCache::getConnection() reconnected to " + 
                brokerURL);
              this->iv_connections[brokerURL] = connection; 
            }
          }
        }
      }
      return connection;		
    } catch (cms::CMSException & e) {
      LOGERROR("AMQConnectionCache::getConnection() " + e.getMessage());
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam("AMQConnectionCache::getConnection() " + e.getMessage());
      ErrorInfo errInfo;
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);

    } catch (...) {
      LOGERROR("AMQConnectionCache::getConnection() Unknown Exception. ");
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam("AMQConnectionCache::getConnection Unknown Exception");
      ErrorInfo errInfo;
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }
  }


//===================================================
//Listener that process UIMA requests. 
//---------------------------------------------------

  AMQListener::AMQListener(int id,
                           AMQConnection * connection,
                           AnalysisEngine * ae,
                           CAS * cas,
                           Monitor * stats) : 
                            iv_id(id),
                            iv_pConnection(connection),
                            iv_pEngine(ae),
                            iv_pCas(cas),
                            //iv_pMonitor(stats),
                            iv_replyToConnections(connection->iv_pConnFact,stats),
                            iv_timeLastRequestCompleted(0),
                            iv_busy(false),
                            iv_stopProcessing(false),
                            iv_count(0),
                            iv_aeDescriptor(),
                            iv_brokerURL(connection->getBrokerURL()),
                            iv_inputQueueName(connection->getInputQueueName()) {

    iv_pMonitor = stats;
    getMetaData(iv_pEngine); 
 
  }

  AMQListener::AMQListener(int id,
                           AMQConnection * connection,
                           AnalysisEngine * ae,
                           Monitor * stats) : 
                            iv_id(id),
                            iv_pConnection(connection),
                            iv_pEngine(ae),
                            //iv_pMonitor(stats),
                            iv_replyToConnections(connection->iv_pConnFact, stats),
                            iv_timeLastRequestCompleted(0),
                            iv_busy(false),
                            iv_count(0),
                            iv_aeDescriptor(),
                            iv_brokerURL(connection->getBrokerURL()),
                            iv_inputQueueName(connection->getInputQueueName()) {

    iv_pMonitor = stats;
    getMetaData(iv_pEngine); 
 
  }

  AMQListener::~AMQListener() {

  }

  //extract AE descriptor as XML to use when processing GETMETA requests.
 void AMQListener::getMetaData(AnalysisEngine * pEngine) {
    if (pEngine == NULL) {
      LOGERROR("AMQListener::getMetaData() Invalid handle to AnalysisEngine.");
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam("AMQListener::getMetaData() Invalid handle to AnalysisEngine.");
      ErrorInfo errInfo;
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(Uima_runtime_error, 
            UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
            errInfo.getMessage(),
            errInfo.getMessage().getMessageID(),
            ErrorInfo::unrecoverable);
    }

    CAS * mcas = pEngine->newCAS();
    delete mcas;
    const AnalysisEngineMetaData & aeMetaData = pEngine->getAnalysisEngineMetaData();			
    icu::UnicodeString xmlBuffer;
    xmlBuffer.insert(0, "<?xml version=\"1.0\"?>");
    pEngine->getAnnotatorContext().getTaeSpecifier().toXMLBuffer(aeMetaData,
      false, xmlBuffer);
    UnicodeStringRef uref(xmlBuffer.getBuffer(), xmlBuffer.length());

    this->iv_aeDescriptor = uref.asUTF8();
    //cout << this->iv_aeDescriptor << endl;
  }


bool AMQListener::validateRequest(const TextMessage * textMessage, string & errmsg) {
  bool valid = true; 
  
  if ( textMessage->getCMSReplyTo() == NULL) 
    LOGWARN("AMQListener::validateRequest() JMSReplyTo is not set. " );    
  
  if (textMessage->getCMSReplyTo() == NULL &&
    !textMessage->propertyExists("MessageFrom") ) {
     errmsg = "Reply to destination not set.";
     return false;
  }  

  if (!textMessage->propertyExists("Command") ) {
    errmsg = "Required property 'Command' is not set.";
    LOGERROR("AMQListener::validateRequest " + errmsg);
    valid = false;  
  } else {
    int command = textMessage->getIntProperty("Command");   
    if (command != PROCESS_CAS_COMMAND &&
        command != GET_META_COMMAND &&
        command != CPC_COMMAND) {
        stringstream str;
        str << "Unexpected value for 'Command' " << command;
        errmsg = str.str(); 
        LOGERROR("AMQListener::validateRequest " + errmsg);
        valid=false;
    } else if (command == CPC_COMMAND) {
      if (iv_pEngine == NULL) {
        errmsg = "CPC request received but AnalysisEngine not available.";
        LOGERROR("AMQListener::validateRequest() " + errmsg);
        valid = false;
      }
    } else if (command == PROCESS_CAS_COMMAND) {
      if (iv_pCas == NULL || iv_pEngine == NULL) {
        errmsg = "Process Cas request but an AnalysisEngine and CAS not available.";
        LOGERROR("AMQListener::validateRequest() " + errmsg);
        valid = false;
      }
      if (!textMessage->propertyExists("Payload") ) {
        errmsg = "Required property 'Payload' is not set.";
        LOGERROR("AMQListener::validateRequest " + errmsg);
        valid = false;  
      } else {
        int payload = textMessage->getIntProperty("Payload");
        if (payload != XCAS_PAYLOAD && payload != XMI_PAYLOAD) {
           stringstream str;
           str << "Unexpected value for 'Payload' " << payload;
           errmsg = str.str();
           LOGERROR("AMQListener::validateRequest " + errmsg);
           valid=false;
        } 
		
		try {
          string text = textMessage->getText().c_str();
          if (text.length() == 0) {
            errmsg = "There is no payload data. Nothing to process.";
            LOGERROR("AMQListener::validateRequest " + errmsg);
            valid = false;  
          }
		} catch (CMSException& e) {
			errmsg = e.getMessage();
			LOGERROR("AMQListener::validateRequest " + e.getMessage());
			valid = false;
		}
      }
    }
  }
  return valid;
}


void AMQListener::receiveAndProcessMessages(apr_thread_t * thd) {
 
  try {
    this->thd = thd;
    cout << "Instance: " << iv_id << " ThreadId: " << apr_os_thread_current() <<  " started." << endl;
    //start receiving messages
    //this->iv_pConnection->start();

    this->iv_stopProcessing = false;
    apr_time_t lastStatsTime = apr_time_now(); 
    while (!iv_stopProcessing) {

      stringstream astr;
     
      //cout << "receiveAndProcess going to call recieve" << endl; 
      Message * msg = this->iv_pConnection->receive(2000);

      if (msg != NULL) {
         iv_busy = true;
         astr << this->iv_id;
         astr << " *****Message#: "<< ++iv_count << "*****";
         //cerr << astr.str() << endl;         
         //LOGINFO(FINE,astr.str());
         this->handleRequest(msg);
         ///this->testHandleRequest(msg);
         delete msg;
         msg = 0;
         //cerr << iv_id << " ****Message#: " << iv_count << " deleted " << endl;
         iv_busy = false;
         this->iv_timeLastRequestCompleted = apr_time_now();
         //cout << iv_id << " processMessages done: getnext " << iv_count  << endl;
      }
    
    }
    LOGWARN("AMQListener::receiveAndProcessMessage() stopped receiving messages.");
  } catch (uima::Exception ex ) {
     LOGERROR("AMQListener::receiveAndProcessMessages() " + 
       ex.getErrorInfo().getMessage().asString());
     this->iv_stopProcessing = true;
     //tell the monitor that the thread has stopped processing
     iv_pMonitor->listenerStopped(this->iv_id);
  } catch (...) {
     LOGERROR("AMQListener::receiveAndProcessMessages UnExpected error.");
     iv_stopProcessing = true;
     //tell the monitor that the thread has stopped processing
     iv_pMonitor->listenerStopped(this->iv_id);
  }
}


/*
* Receive a TextMessage and examine the header properties
* to determine the type of request and payload.  Processes
* one request at a time and is blocked till the requestis 
* handled and the response sent.
*/
  void AMQListener::handleRequest( const Message* message ) {


    apr_time_t startTime = apr_time_now();
    apr_time_t timeIdle = 0;
    // Idle time is computed as the interval from time last request 
    // was processed. If this is the first request, idle time is 
    // computed from the time the service was started.  
    if (iv_timeLastRequestCompleted != 0)
      timeIdle =  startTime - iv_timeLastRequestCompleted;
    else 
      timeIdle = startTime - iv_pMonitor->getServiceStartTime(); 

    this->iv_pMonitor->processingStarted(this->iv_id, startTime, timeIdle );
    stringstream astr;
    astr << this->iv_id;
    astr << " ****handleRequest(): "<< iv_count << " start"<< endl;
    LOGINFO(FINE,astr.str());
    ///cout << astr.str() << endl;
    ///LOGWARN(astr.str());
    apr_time_t endTime = 0;
    apr_time_t startSerialize = 0;
    apr_time_t startDeserialize = 0;
    apr_time_t startAnnotatorProcess = 0; 
    apr_time_t startSendResponse = 0;
    apr_time_t timeToDeserializeCAS = 0;
    apr_time_t timeToSerializeCAS = 0;
    apr_time_t timeInAnalytic = 0;

    const TextMessage* textMessage=0;
    int command = 0;

    try {	
    
      textMessage = dynamic_cast< const TextMessage* >( message );
      if (textMessage==0) {
        LOGERROR("AMQListener::handleRequest() invalid pointer to TextMessage");
        endTime = apr_time_now();
        iv_pMonitor->processingComplete(iv_id, 0,false,endTime-startTime,0,0,0,endTime-startTime);
        this->iv_timeLastRequestCompleted = apr_time_now();
        return;
      }
      if (textMessage->propertyExists("MessageFrom")) {	
        LOGINFO(FINER,"Received from " + textMessage->getStringProperty("MessageFrom"));
      } 

      //validate request properties
      string errormessage;
      if (!validateRequest(textMessage, errormessage)) {
        LOGERROR("Listener::handleRequest() " + errormessage);
        endTime = apr_time_now();
        sendResponse(textMessage, 0,0,0,timeIdle, endTime-startTime,
          errormessage ,true);
        endTime = apr_time_now();
        iv_pMonitor->processingComplete(iv_id, 0,false,endTime-startTime,0,0,0,endTime-startTime);
        return;
      }

      command = textMessage->getIntProperty("Command");
      astr.str("");
      astr << "Received request Command: " << command ;
      if (textMessage->propertyExists("CasReference")) {
        astr << " CasReference " << textMessage->getStringProperty("CasReference");
      }
   
      astr << "Received request Command: " << command ;
      if (textMessage->propertyExists("CasReference")) {
        astr << " CasReference " << textMessage->getStringProperty("CasReference");
      }
      LOGINFO(INFO,astr.str());

      if (command == PROCESS_CAS_COMMAND) { //process CAS
        LOGINFO(FINE,"Process CAS request start.");
        int payload = textMessage->getIntProperty("Payload");
        //get the text in the payload 
        string text = textMessage->getText().c_str();
        UnicodeString utext(text.c_str());
		text = UnicodeStringRef(utext).asUTF8();
        astr.str("");
        astr << "Payload: " << payload << " Content: " << text ;
        LOGINFO(FINER, astr.str());

        //InputSource
        MemBufInputSource memIS((XMLByte const *)text.c_str(),
          text.length(),
          "sysID");

        //reset the CAS
        iv_pCas->reset();
        ios::openmode mode = ios::binary;
        stringstream xmlstr;  
        //deserialize payload data into the CAS,
        //call AE process method and serialize
        //the CAS which will be sent with the
        //response.
        if (payload == XCAS_PAYLOAD) {
          LOGINFO(FINEST, "AMQListener::handleRequest() XCAS serialization.");
          startDeserialize = apr_time_now();
          XCASDeserializer deserializer;
          deserializer.deserialize(memIS, *iv_pCas); 
          startAnnotatorProcess=apr_time_now();
          timeToDeserializeCAS = startAnnotatorProcess-startDeserialize;

          iv_pEngine->process(*iv_pCas);
          startSerialize=apr_time_now();
          timeInAnalytic = startSerialize - startAnnotatorProcess;

          XCASWriter xcaswriter(*iv_pCas, true);
          xcaswriter.write(xmlstr);
          timeToSerializeCAS = apr_time_now() - startSerialize;
        } else if (payload == XMI_PAYLOAD) {
          //deserialize incoming xmi CAS data.
          LOGINFO(FINEST, "AMQListener::handleRequest() XMI serialization.");
          
          startDeserialize = apr_time_now();
          XmiSerializationSharedData sharedData;
          XmiDeserializer deserializer;
          deserializer.deserialize(memIS,*iv_pCas,sharedData);
          startAnnotatorProcess=apr_time_now();
          timeToDeserializeCAS = startAnnotatorProcess-startDeserialize;
          LOGINFO(FINEST, "AMQListener::handleRequest() calling process.");
          iv_pEngine->process(*iv_pCas);
          startSerialize=apr_time_now();
          timeInAnalytic = startSerialize - startAnnotatorProcess;

          //serialize CAS 
          LOGINFO(FINEST, "AMQListener::handleRequest() calling serialize.");
          XmiWriter xmiwriter(*iv_pCas, true, &sharedData);
          xmiwriter.write(xmlstr);
          //cout << "SERIALIZED CAS " << xmlstr.str() << endl;
          timeToSerializeCAS = apr_time_now() - startSerialize;
          LOGINFO(FINEST, "AMQListener::handleRequest() done processing CAS.");
        }
        //done with this CAS.
        iv_pCas->reset(); 
        //record end time
        endTime = apr_time_now();
        //send reply
        LOGINFO(FINER,"AnalysisEngine::process() completed successfully. Sending reply.");
        sendResponse(textMessage, timeToDeserializeCAS,
                      timeToSerializeCAS, timeInAnalytic,
                      timeIdle, endTime-startTime,
                      xmlstr.str(),false);
        endTime=apr_time_now();
        iv_pMonitor->processingComplete(iv_id, command,true,endTime-startTime,
          timeToDeserializeCAS, timeInAnalytic, timeToSerializeCAS,
          endTime-startSendResponse);
        LOGINFO(FINE,"Process CAS finished.");		
      } else if (command ==  GET_META_COMMAND ) { //get Meta 	 
        LOGINFO(FINE, "Process getMeta request start.");
        endTime = apr_time_now();
        startSendResponse = apr_time_now();
        sendResponse(textMessage, timeToDeserializeCAS,
                  timeToSerializeCAS, timeInAnalytic,
                  timeIdle, endTime-startTime,this->iv_aeDescriptor,false);			
        endTime=apr_time_now();
        //record timing
        iv_pMonitor->processingComplete(iv_id, command,true,endTime-startTime,0,0,0,endTime-startSendResponse);
        LOGINFO(FINE,"Process getMeta request finished.");			
      } else if (command ==  CPC_COMMAND ) { //CPC       			
        LOGINFO(FINE, "Processing CollectionProcessComplete request start");				
        iv_pEngine->collectionProcessComplete();
        endTime = apr_time_now();
        startSendResponse = apr_time_now();
        sendResponse(textMessage, 0,
                    0, timeInAnalytic,
                    timeIdle,endTime-startTime,"CPC completed.",false);
        endTime=apr_time_now();
        iv_pMonitor->processingComplete(iv_id, command,true,endTime-startTime,0,0,0,endTime-startSendResponse);
        LOGINFO(FINE, "Processing CollectionProcessComplete request finished.");	
      } 
    } catch (XMLException& e) {
      stringstream str;
      str << "AMQListener::handleRequest XMLException." << e.getMessage();
      LOGERROR(str.str());
      endTime = apr_time_now();
      startSendResponse = apr_time_now();
      sendResponse(textMessage, timeToDeserializeCAS,
                    timeToSerializeCAS, timeInAnalytic,
                    timeIdle, endTime-startTime,str.str(),true);
      endTime = apr_time_now();
      iv_pMonitor->processingComplete(iv_id, command,false,endTime-startTime,0,0,0,endTime-startSendResponse);
    } catch (uima::Exception e) {
      LOGERROR("AMQListener::handleRequest UIMA Exception " + e.asString());
      endTime = apr_time_now();
      startSendResponse = apr_time_now();
      sendResponse(textMessage, timeToDeserializeCAS,
                      timeToSerializeCAS, timeInAnalytic,
                      timeIdle,endTime-startTime,e.asString(),true);
      endTime = apr_time_now();
      iv_pMonitor->processingComplete(iv_id, command,false,endTime-startTime,0,0,0,endTime-startSendResponse);
    }	catch(...) {
      LOGERROR("AMQListener::handleRequest Unknown exception ");
      //TODO: log / shurdown ?}   
      endTime = apr_time_now(); 
      iv_pMonitor->processingComplete(iv_id, command,false,endTime-startTime,0,0,0,endTime-startTime);

    }
  }

  void AMQListener::sendResponse(const TextMessage * request, 
                               apr_time_t timeToDeserialize,
                               apr_time_t timeToSerialize, 
                               apr_time_t timeInAnalytic, 
                               apr_time_t idleTime,
                               apr_time_t elapsedTime, 
                               string msgContent, bool isExceptionMsg) {

     //TODO retry
     string serverURI;
     AMQConnection * replyTo=NULL;
     TextMessage * reply=NULL;

     try {
       const Destination * cmsReplyTo = request->getCMSReplyTo();
       if (cmsReplyTo !=  NULL)  {
         reply = this->iv_pConnection->getTextMessage();
       } else {
         LOGINFO(2, "AMQListener::sendResponse() start");
         if (!request->propertyExists("ServerURI")  ) {
           LOGERROR("AMQListener::sendResponse() ServerURI header property does not exist.");
           return;
         }
         if (!request->propertyExists("MessageFrom")  ) {
           LOGERROR("AMQListener::sendResponse() MessageFrom header property does not exist.");
           return;
         }

         //get handle to a connection
         string tmp = request->getStringProperty("ServerURI");
         LOGINFO(FINEST,"replyTo ServerURI: " +  tmp);

         //special handling when protocol is http.
         //HTTP protocol not supported by ActiveMQ C++ client.
         //replace the http broker URL with the broker URL of
         //the queue this listener is attached to.
         if (tmp.find("http:") != string::npos) { 
           serverURI=this->iv_brokerURL;
           LOGINFO(FINER,"HTTP reply address: " + tmp);
         } else {  //send reply via tcp
           size_t begpos = tmp.find("tcp:",0);
           tmp = tmp.substr(begpos);
           size_t endpos = tmp.find_first_of(",");

           if (begpos == string::npos) { 
             LOGERROR("AMQListener::sendResponse() Could not find tcp URL in ServerURI header property.");	 
             return;
           }
           if (endpos == string::npos) {
             serverURI = tmp;
           } else {
             serverURI = tmp.substr(0, endpos);
           }
         }
         LOGINFO(FINER,"ReplyTo BrokerURL " + serverURI);

         //look up the endpoint connection
         replyTo = this->iv_replyToConnections.getConnection(serverURI);
         if (replyTo == NULL) {
           LOGERROR("Could not get a connection to " + serverURI);
           return;
         }
         //get message object
         reply = replyTo->getTextMessage();
       }

       if (reply == NULL) {
         LOGERROR("AMQListener::sendResponse() invalid textMessage object " );
         return;
       }

       //construct the reply message
       reply->setStringProperty("MessageFrom", this->iv_pConnection->getInputQueueName() );
       reply->setStringProperty("ServerURI", this->iv_brokerURL);

       reply->setIntProperty("MessageType",RESPONSE);
       reply->setLongProperty("TimeInService", elapsedTime*1000);
       reply->setLongProperty("IdleTime", idleTime*1000);
       if (request->propertyExists("CasReference")  ) {
         reply->setStringProperty("CasReference", request->getStringProperty("CasReference"));
       }
       if (request->propertyExists("Command")  ) {
         reply->setIntProperty("Command", request->getIntProperty("Command"));
         if (request->getIntProperty("Command") == PROCESS_CAS_COMMAND) {
           reply->setLongProperty("TimeToSerializeCAS", timeToSerialize*1000);
           reply->setLongProperty("TimeToDeserializeCAS", timeToDeserialize*1000);
           reply->setLongProperty("TimeInAnalytic", timeInAnalytic*1000);
           reply->setLongProperty("TimeInProcessCAS", timeInAnalytic*1000);
         }
       }
       if (isExceptionMsg) {
         reply->setIntProperty("Payload",EXC_PAYLOAD);
       } else {
         if (request->propertyExists("Payload") ) {
           reply->setIntProperty("Payload",request->getIntProperty("Payload"));
         } else {
           reply->setIntProperty("Payload",NO_PAYLOAD);
         }
       }
       //cargo
       reply->setText(msgContent);	
       //log the reply message content
       stringstream str;
       str << "Sending Reply Command: " << reply->getIntProperty("Command") << " MessageType: " << reply->getIntProperty("MessageType") << " ";
       if (reply->propertyExists("CasReference") ) {
         str << "CasReference: " << reply->getStringProperty("CasReference");
       }
       LOGINFO(INFO,str.str());

       if (cmsReplyTo != NULL) {
         str << " to " << ((Queue*)cmsReplyTo)->getQueueName();
       } else {
         str << " to " << request->getStringProperty("MessageFrom") 
           << " at " << serverURI;
       }
       str << " Message text: " << msgContent;
       //std::cout << "****Reply" << msgContent << std::endl;
       LOGINFO(FINEST,"PRINT Reply message:\n" + str.str());	

       //send
       if (cmsReplyTo != NULL) {
         //cout << "cmsReplyTo=" << cmsReplyTo->toProviderString() << endl;
         iv_pConnection->sendMessage(cmsReplyTo);
       } else {
         replyTo->sendMessage(request->getStringProperty("MessageFrom"));
       }
       LOGINFO(FINER,"AMQListener::sendResponse DONE");

     } catch (CMSException& ex ) {
       LOGERROR("AMQListener::handleMessage()" +  ex.getMessage());
     } catch (...)  { 
       LOGERROR("AMQListener::handleRequest() UnExpected error sending reply.");
     }
}

//===================================================
//AMQAnalysisEngineService
//---------------------------------------------------
  AMQAnalysisEngineService::~AMQAnalysisEngineService() {
    //stop();
    cleanup();
  }

  AMQAnalysisEngineService::AMQAnalysisEngineService(ServiceParameters & desc,
                                                     Monitor * stats,  apr_pool_t * pool)  : 
                                        //iv_pMonitor(stats),
                                        iv_pool(pool),
                                        iv_numInstances(desc.getNumberOfInstances()),
                                        iv_brokerURL(desc.getBrokerURL()),
                                        iv_inputQueueName(desc.getQueueName()),
                                        iv_aeDescriptor(desc.getAEDescriptor()),
                                        iv_prefetchSize(desc.getPrefetchSize()),
                                        iv_initialFSHeapSize(desc.getInitialFSHeapSize()),
                                        iv_vecpConnections(),
                                        iv_vecpAnalysisEngines(),
                                        iv_vecpCas(),
                                        iv_closed(false),
                                        iv_listeners() {
    try {
      iv_pMonitor = stats;
      initialize(desc);
    } catch (CMSException & e) {
      ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
      msg.addParam(e.getMessage());
      ErrorInfo errInfo;
      errInfo.setMessage(msg);
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }
  }

  void AMQAnalysisEngineService::initialize(ServiceParameters & params) {

    try {
      //create connection factory
      LOGINFO(FINEST,"AMQAnalysisEngineService::initialize() Create connection factory");
      this->iv_pConnFact = AMQConnection::createConnectionFactory(params);
	  
      if (iv_pConnFact == NULL) {
        ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
        msg.addParam("AMQAnalysisEngineService::initialize() Failed to create connection factory.");
        ErrorInfo errInfo;
        errInfo.setMessage(msg);
        UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
          UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
          errInfo.getMessage(),
          errInfo.getMessage().getMessageID(),
          ErrorInfo::unrecoverable);
      }
      if (!uima::ResourceManager::hasInstance()) {
        uima::ResourceManager::createInstance("ActiveMQAnalysisEngineService");
      }
      ErrorInfo errInfo;
      UnicodeString ustr(this->iv_aeDescriptor.c_str());
      UnicodeString ufn = ResourceManager::resolveFilename(ustr,ustr);

      //create a AnalysisEngine and CAS for each instance
      for (int i=0; i < iv_numInstances; i++) {
        //create the connection
        AMQConnection * newConnection = new AMQConnection(this->iv_pConnFact, params.getBrokerURL(), this->iv_pMonitor, i);
        if (newConnection == NULL) {
          LOGERROR("AMQAnalysisEngineService::initialize() Could not create ActiveMQ endpoint connection.");
          ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
          msg.addParam("AMQAnalysisEngineService::initialize() Failed to connect to broker.");
          ErrorInfo errInfo;
          errInfo.setMessage(msg);
          UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
            UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
            errInfo.getMessage(),
            errInfo.getMessage().getMessageID(),
            ErrorInfo::unrecoverable);
        }
        /////newConnection->setExceptionListener(this);
        this->iv_vecpConnections.push_back(newConnection);
        //create AE
        AnalysisEngine * pEngine = uima::TextAnalysisEngine::createTextAnalysisEngine((UnicodeStringRef(ufn).asUTF8().c_str()),
          errInfo);
        if (pEngine) {
          LOGINFO(0,"AMQAnalysisEngineService::initialize() AnalysisEngine initialization successful.");
        } else {
          LOGERROR("AMQAnalysisEngineService::initializer() could not create AE" + errInfo.asString());
          ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
          msg.addParam("AMQAnalysisEngineService::initialize() create AE failed. " + errInfo.getMessage().asString() );
          ErrorInfo errInfo;
          errInfo.setErrorId(UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE),
            errInfo.setMessage(msg);
          UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
            UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
            errInfo.getMessage(),
            errInfo.getMessage().getMessageID(),
            ErrorInfo::unrecoverable);
        }

        this->iv_vecpAnalysisEngines.push_back(pEngine);

        //initial FSHeap size
        if (this->iv_initialFSHeapSize > 0) {
          pEngine->getAnnotatorContext().
            assignValue(UIMA_ENGINE_CONFIG_OPTION_FSHEAP_PAGESIZE,this->iv_initialFSHeapSize);
        }

        //create CAS
        CAS * cas = pEngine->newCAS();
        if (cas == NULL) {
          LOGERROR("AMQAnalysisEngineService::initialize() Could not create CAS");
          ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
          msg.addParam("AMQListener::initialize() create CAS failed.");
          ErrorInfo errInfo;
          errInfo.setErrorId(UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE),
            errInfo.setMessage(msg);
          UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
            UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
            errInfo.getMessage(),
            errInfo.getMessage().getMessageID(),
            ErrorInfo::unrecoverable);
        }
        this->iv_vecpCas.push_back(cas);
      
        //cout << __FILE__ << __LINE__ << "AMQAnalysisEngineService::initialize() create listener " << endl;
        //create listeners and register these
        AMQListener * newListener = new AMQListener(i,iv_vecpConnections.at(i),
          iv_vecpAnalysisEngines.at(i), iv_vecpCas.at(i),
          iv_pMonitor);
        if (newListener == NULL) {
          LOGERROR("AMQAnalysisEngineService::initialize() Could not create listener.");
          ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
          msg.addParam("AnalysisEngineService::initialize() Could not create listener.");
          ErrorInfo errInfo;
          errInfo.setMessage(msg);
          UIMA_EXC_THROW_NEW(Uima_runtime_error, 
            UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
            errInfo.getMessage(),
            errInfo.getMessage().getMessageID(),
            ErrorInfo::unrecoverable);
        }
        this->iv_listeners[i] = newListener;
        //cout << __FILE__ << __LINE__ << "AMQAnalysisEngineService::initialize() create message consumer " << endl;
        //create MessageConsumer session and register Listener       
        this->iv_vecpConnections.at(i)->createMessageConsumer(iv_inputQueueName,annotator_selector, params.getPrefetchSize());  
     } 
     
     iv_pMonitor->setNumberOfInstances(iv_numInstances);
     //cout <<  __FILE__ << __LINE__ << "AMQAnalysisEngineService::initialize() setup getmeta " << endl;
      //Fast GetMeta
      //create connection 
     LOGINFO(FINEST, "AMQAnalysisEngineService::initialize() Setup GETMETA instance.");
     iv_pgetMetaConnection = new AMQConnection(this->iv_pConnFact, params.getBrokerURL(), this->iv_pMonitor, iv_numInstances);

     if (iv_pgetMetaConnection == NULL) {
          LOGERROR("AMQAnalysisEngineService::initialize() Could not create fast getmeta ActiveMQ endpoint connection.");
          ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
          msg.addParam("AMQAnalysisEngineService::initialize() Failed to connect to broker.");
          ErrorInfo errInfo;
          errInfo.setMessage(msg);
          UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
            UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
            errInfo.getMessage(),
            errInfo.getMessage().getMessageID(),
            ErrorInfo::unrecoverable);
     }

     //cout <<  __FILE__ << __LINE__ << "AMQAnalysisEngineService::initialize() setup getmeta listener" << endl;
     //create a MessageListener MessageConsumer to handle getMeta requests only.
     AMQListener * newListener = new AMQListener(iv_numInstances,iv_pgetMetaConnection,
          iv_vecpAnalysisEngines.at(0), 
          this->iv_pMonitor);
     if (newListener == NULL) {
          LOGERROR("AMQAnalysisEngineService::initialize() Could not create Fast getMeta listener.");
          ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR);
          msg.addParam("AMQAnalysisEngineService::initialize() Could not create listener.");
          ErrorInfo errInfo;
          errInfo.setMessage(msg);
          UIMA_EXC_THROW_NEW(Uima_runtime_error, 
            UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
            errInfo.getMessage(),
            errInfo.getMessage().getMessageID(),
            ErrorInfo::unrecoverable);
      }
      this->iv_listeners[this->iv_numInstances] = newListener; 
      iv_pgetMetaConnection->createMessageConsumer(iv_inputQueueName, 
                      getmeta_selector,0); 
      this->iv_pMonitor->setGetMetaListenerId(iv_numInstances);
      //cout <<  __FILE__ << __LINE__ << "AMQAnalysisEngineService::initialize() done" << endl;
    } catch (uima::Exception e) {
      cout << __FILE__ << __LINE__ <<  "AMQAnalysisEngineService::initialize() failed " << e.getErrorInfo().asString() << endl;
      LOGERROR("AMQAnalysisEngineService::initialize() " + e.asString());
      throw e;
    }
  }

  void AMQAnalysisEngineService::setTraceLevel(int level) {
    if (level < 0) {
    uimacpp_ee_tracelevel=NONE;
    } else if (level == 0) {
      uimacpp_ee_tracelevel = INFO;
    } else if (level == 1) {
      uimacpp_ee_tracelevel = FINE;
    } else if (level == 2) {
      uimacpp_ee_tracelevel = FINER;
    } else if (level > 2) {
      uimacpp_ee_tracelevel = FINEST;
    } else {
      uimacpp_ee_tracelevel = INFO;
    }
    cout << "tracelevel=" << uimacpp_ee_tracelevel << endl;
  }

  void AMQAnalysisEngineService::startProcessingThreads() {
    LOGINFO(FINER,"AMQAnalysisEngineService::start() create listener threads.");
    //create the listener threads
    thd_attr=0;
    apr_status_t rv;
    rv = apr_threadattr_create(&thd_attr, iv_pool);
    assert(rv == APR_SUCCESS);
    map<int, AMQListener*>::iterator ite; 
    int i=0;
    for (ite= iv_listeners.begin();ite !=  iv_listeners.end();ite++) {
      apr_thread_t *thread=0;
      rv = apr_thread_create(&thread, thd_attr, handleMessages, ite->second, iv_pool);
      assert(rv == APR_SUCCESS);
      iv_listenerThreads.push_back(thread);
    }
    this->iv_pMonitor->setStartTime();
}


void AMQAnalysisEngineService::start() {
  
  if (iv_pgetMetaConnection != NULL) {
    cerr << "Startinging GetMetaData instance" << endl;
    this->iv_pgetMetaConnection->start();
  }
  for (size_t i=0; i < iv_vecpConnections.size(); i++) {
    cerr << "Starting Annotator instance " << i << endl;
    AMQConnection * connection = iv_vecpConnections.at(i);
    if (connection != NULL) {
      connection->start();
    } else {
      LOGERROR("AMQAnalysisEngineService::start() Connection object is NULL.");
      ErrorInfo errInfo;
      errInfo.setMessage(ErrorMessage(UIMA_MSG_ID_LOG_ERROR, "connection object is NULL."));
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }
  } 
}	

void AMQAnalysisEngineService::shutdown() {
  //LOGWARN("AMQAnalysisEngineService::shutdown()");

  stop();

  cout << "AMQAnalysisEngineService::shutdown() going to terminate threads " << endl;;
  //terminate the threads
  apr_status_t rv;
  for (size_t i=0; i < this->iv_listenerThreads.size(); i++) {
    //cout << "wait for thread " << i << " to end " << endl;
    this->iv_listeners[i]->stopProcessing();
    if (!iv_listeners[i]->isReconnecting()) {
      apr_thread_join(&rv, this->iv_listenerThreads.at(i));
    }
  }
  
  cout << "AMQAnalysisEngineService::shutdown stopped all connection" << endl;
  cleanup();
  cout << "AMQAnalysisEngineService::shutdown shutdown done" << endl;
}

int AMQAnalysisEngineService::stop() {
  
  //TODO: let listeners finish processing first
  //stop messages notification

  if (iv_pgetMetaConnection != NULL) {
    cerr << "Stopping GetMetaData instance" << endl;
    this->iv_pgetMetaConnection->stop();
  }
  for (size_t i=0; i < iv_vecpConnections.size(); i++) {
    cerr << "Stopping Annotator instance " << i << endl;
    AMQConnection * connection = iv_vecpConnections.at(i);
    if (connection != NULL) {
      connection->stop();
    } else {
      LOGERROR("AMQAnalysisEngineService::stop() Connection object is NULL.");
      ErrorInfo errInfo;
      errInfo.setMessage(ErrorMessage(UIMA_MSG_ID_LOG_ERROR, "connection object is NULL."));
      UIMA_EXC_THROW_NEW(uima::Uima_runtime_error, 
        UIMA_ERR_RESMGR_COULD_NOT_INITIALIZE_RESOURCE,
        errInfo.getMessage(),
        errInfo.getMessage().getMessageID(),
        ErrorInfo::unrecoverable);
    }
  } 
  return 0;
}	

void  AMQAnalysisEngineService::quiesceAndStop() {
  
  quiesce();
  
  //shutdown worker threads.
   cout << "AMQAnalysisEngineService::quiesceAndStop() going to terminate threads " << endl;;
  if (this->iv_pMonitor->getQuiesceAndStop() ) {
    for (size_t i=0; i < this->iv_listenerThreads.size(); i++) {
      //cout << "wait for thread " << i << " to end " << endl;
      this->iv_listeners[i]->stopProcessing();
      apr_thread_join(&rv, this->iv_listenerThreads.at(i));
    }
  }

  cleanup();
}

void  AMQAnalysisEngineService::quiesce() {
  //stop connections - does not work
  stop();

  //check whether processing threads are finished.
  bool allfinished = false;
  while (!allfinished) {
    allfinished = true;
    map<int, AMQListener*>::iterator ite; 
    for (ite= iv_listeners.begin();ite !=  iv_listeners.end();ite++) {
      if (ite->second->isBusy()) { 
        allfinished  = false;
        break;
      }
    }
    //Thread::sleep(1000);
	apr_sleep(1000000);
  }
}

  void AMQAnalysisEngineService::cleanup() {
    // Destroy resources.
    try { 
      if (iv_closed) {
         return;
      }
      if (iv_pgetMetaConnection != NULL) {
        delete this->iv_pgetMetaConnection;
        this->iv_pgetMetaConnection = 0;
      }

      for (size_t i=0; i < iv_vecpConnections.size(); i++) {
        //cout << "deleting consumerConnection " << i << endl;	
        if (iv_vecpConnections.at(i) != NULL) {
          iv_vecpConnections.at(i)->stop();
          delete iv_vecpConnections.at(i);
        }
      }  
      
      for (size_t i=0; i < iv_vecpAnalysisEngines.size(); i++) {
         delete iv_vecpAnalysisEngines.at(i);
      }

      for (size_t i=0; i < iv_vecpCas.size(); i++) {
         delete iv_vecpCas.at(i);
      }

      iv_vecpAnalysisEngines.clear();
      iv_vecpCas.clear();
      iv_vecpConnections.clear();

      map<int, AMQListener*>::iterator ite; 
      for (ite= iv_listeners.begin();ite !=  iv_listeners.end();ite++) {
        delete ite->second;
      }

      iv_listeners.clear();

      if (iv_pConnFact != NULL) {
        delete iv_pConnFact;
        iv_pConnFact = NULL;
      }
      iv_closed = true;

    } catch (CMSException& e) {
      cerr << "AMQAnalysisEngineService::cleanup() "  << e.getMessage() << endl;;
    }  
  }           


//===================================================
//CommonUtils
//---------------------------------------------------
void CommonUtils::logError(string msg) {
    this->iv_pMonitor->logError(msg);
    //cerr << "ERROR: " << msg << endl;
  }
void CommonUtils::logWarning(string msg) {
    this->iv_pMonitor->logWarning(msg);
    //cout << "WARN: " << msg << endl;
  }
void CommonUtils::logMessage(string msg) {
  cout << "INFO: " << msg << endl;
    this->iv_pMonitor->logMessage(msg);
    //cout << "INFO: " << msg << endl;
  }