| /* |
| |
| * 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/concurrent/Thread.h> |
| #include <activemq/concurrent/Runnable.h> |
| #include <activemq/concurrent/Concurrent.h> |
| #include <activemq/core/ActiveMQConnectionFactory.h> |
| #include <activemq/core/ActiveMQConstants.h> |
| #include <activemq/util/Integer.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 activemq::util; |
| using namespace activemq::exceptions; |
| using namespace activemq::concurrent; |
| using namespace uima; |
| |
| enum traceLevels {NONE, INFO, FINE, FINER, FINEST }; |
| traceLevels uimacpp_ee_tracelevel=INFO; |
| #define MSGHEADER apr_time_now() << " ThreadId: " << Thread::getId() << __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());} |
| //=================================================== |
| //AMQConnection |
| //--------------------------------------------------- |
| AMQConnection::AMQConnection( const char * aBrokerURL, Monitor * pStatistics) : |
| iv_brokerURL(aBrokerURL), |
| iv_pConnection(0), |
| iv_pConsumerSession(0), |
| iv_pConsumer(0), |
| iv_inputQueueName(), |
| iv_pInputQueue(0), |
| iv_pListener(0), |
| iv_pProducerSession(0), |
| iv_pProducer(0), |
| iv_pTextMessage(0), |
| iv_replyDestinations(), |
| iv_valid(false) { |
| |
| try { |
| iv_pMonitor = pStatistics; |
| LOGINFO(INFO, "AMQConnection() connecting to " + iv_brokerURL); |
| // Create a ConnectionFactory |
| ActiveMQConnectionFactory* connectionFactory = |
| new ActiveMQConnectionFactory(iv_brokerURL); |
| |
| // Create a Connection |
| if (connectionFactory == NULL) { |
| LOGERROR("AMQConnection() could not create connection factory"); |
| ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR); |
| msg.addParam("AMQConnection() could not 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); |
| } |
| this->iv_pConnection = connectionFactory->createConnection(); |
| if (this->iv_pConnection == NULL) { |
| LOGERROR("AMQConnection() could not create connection."); |
| ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR); |
| msg.addParam("AMQConnection() could not create connection to " + iv_brokerURL); |
| 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); |
| } else { |
| //default exception listener |
| this->iv_pConnection->setExceptionListener(this); |
| } |
| |
| delete connectionFactory; |
| // 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 ); |
| |
| //create TextMessage |
| this->iv_pTextMessage = this->iv_pProducerSession->createTextMessage(); |
| if (this->iv_pTextMessage == 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_pTextMessage != NULL) { |
| delete this->iv_pTextMessage; |
| this->iv_pTextMessage=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, |
| MessageListener * listener, |
| int prefetch) { |
| LOGINFO(FINEST, "AMQConnection::createMessageConsumer() consumer start " + queueName); |
| this->iv_inputQueueName = queueName; |
| stringstream str; |
| //we add one to prefetch size to get the same behavior as the |
| //Spring listener used in UIMA Java SDK which calls the |
| //ActiveMQMessageConsumer.receive() api. |
| //create endpoint destination |
| str << queueName; |
| str << "?consumer.prefetchSize=" << prefetch+1 << endl; |
| |
| 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); |
| } |
| |
| 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); |
| } |
| //register listener |
| this->iv_pConsumer->setMessageListener(listener); //caller owns listener |
| this->iv_pListener = listener; |
| LOGINFO(FINEST, "AMQConnection::createMessageConsumer() " + queueName + " successful."); |
| } |
| |
| //caller owns the ExceptionListener |
| void AMQConnection::setExceptionListener(ExceptionListener * el) { |
| this->iv_pConnection->setExceptionListener(el); |
| } |
| |
| void AMQConnection::onException(const CMSException & ex) { |
| //mark endpoint as broken. |
| this->iv_valid = false; |
| //log that connection is invalid. |
| LOGWARN("AMQConnection()::onException() Connection to " + iv_brokerURL |
| + " may be broken. " + ex.getMessage()); |
| } |
| |
| |
| TextMessage * AMQConnection::getTextMessage() { |
| if (this->iv_pTextMessage == 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_pTextMessage->clearProperties(); |
| iv_pTextMessage->clearBody(); |
| return this->iv_pTextMessage; |
| } |
| |
| 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_pTextMessage); |
| |
| //cout << "producer->send elapsed time " << (apr_time_now() - stime) << endl; |
| this->iv_pTextMessage->clearBody(); |
| this->iv_pTextMessage->clearProperties(); |
| LOGINFO(FINEST, "AMQConnection::sendMessage() successful to " + queuename); |
| } |
| |
| void AMQConnection::sendMessage(const Destination * cmsReplyTo) { |
| 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_pTextMessage); |
| |
| //cout << "producer->send elapsed time " << (apr_time_now() - stime) << endl; |
| this->iv_pTextMessage->clearBody(); |
| this->iv_pTextMessage->clearProperties(); |
| 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() { |
| LOGINFO(0,"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); |
| } |
| } |
| |
| |
| |
| //=================================================== |
| //AMQConnectionCache |
| //--------------------------------------------------- |
| AMQConnectionsCache::AMQConnectionsCache(Monitor * stats) { |
| 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(brokerURL.c_str(), iv_pMonitor); |
| 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(brokerURL.c_str(), iv_pMonitor); |
| 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(stats), |
| iv_timeLastRequestCompleted(0), |
| iv_busy(false), |
| iv_count(0), |
| iv_aeDescriptor(), |
| iv_brokerURL(connection->getBrokerURL()), |
| iv_inputQueueName(connection->getInputQueueName()) { |
| |
| iv_pMonitor = stats; |
| //get AE descriptor as XML to use when processing GETMETA requests. |
| const AnalysisEngineMetaData & aeMetaData = iv_pEngine->getAnalysisEngineMetaData(); |
| icu::UnicodeString xmlBuffer; |
| xmlBuffer.insert(0, "<?xml version=\"1.0\" encoding=\"UTF-8\"?>"); |
| iv_pEngine->getAnnotatorContext().getTaeSpecifier().toXMLBuffer(aeMetaData, |
| false, xmlBuffer); |
| UnicodeStringRef uref(xmlBuffer.getBuffer(), xmlBuffer.length()); |
| |
| this->iv_aeDescriptor = uref.asUTF8(); |
| } |
| |
| AMQListener::~AMQListener() { |
| |
| } |
| |
| /* |
| * 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::onMessage( const Message* message ) { |
| |
| |
| apr_time_t startTime = apr_time_now(); |
| apr_time_t endTime; |
| |
| apr_time_t startSerialize; |
| apr_time_t startDeserialize; |
| apr_time_t startAnnotatorProcess; |
| |
| apr_time_t timeToDeserializeCAS = 0; |
| apr_time_t timeToSerializeCAS = 0; |
| apr_time_t timeInAnalytic = 0; |
| apr_time_t timeIdle = apr_time_now() - iv_timeLastRequestCompleted; |
| |
| const TextMessage* textMessage=0; |
| int command = 0; |
| |
| stringstream astr; |
| astr << "*****Message#: " << ++iv_count << "*****" << endl; |
| LOGINFO(INFO,astr.str()); |
| astr.seekp(0); |
| |
| try { |
| iv_busy=true; |
| textMessage = dynamic_cast< const TextMessage* >( message ); |
| if (textMessage==0) { |
| LOGERROR("AMQListener::onMessage() invalid pointer to TextMessage"); |
| endTime = apr_time_now(); |
| iv_pMonitor->processingComplete(0,false,endTime-startTime); |
| this->iv_timeLastRequestCompleted = apr_time_now(); |
| return; |
| } |
| if (textMessage->propertyExists("MessageFrom")) { |
| LOGINFO(FINER,"Received from " + textMessage->getStringProperty("MessageFrom")); |
| } else { |
| LOGERROR("AMQListener::onMessage() ERROR MessageFrom not set."); |
| endTime = apr_time_now(); |
| iv_pMonitor->processingComplete(0,false,endTime-startTime); |
| this->iv_timeLastRequestCompleted = apr_time_now(); |
| return; |
| } |
| |
| if (textMessage->propertyExists("Command") ){ |
| command = textMessage->getIntProperty("Command"); |
| } else { |
| LOGERROR("AMQListener::onMessage() Required property Command not set."); |
| endTime = apr_time_now(); |
| sendResponse(textMessage, timeToDeserializeCAS, |
| timeToSerializeCAS, timeInAnalytic, |
| timeIdle, endTime-startTime, |
| "Required property 'Command' not set." ,true); |
| endTime = apr_time_now(); |
| iv_pMonitor->processingComplete(0,false,endTime-startTime); |
| this->iv_timeLastRequestCompleted = apr_time_now(); |
| return; |
| } |
| astr.seekp(0); |
| astr << "Received request Command: " << command ; |
| if (textMessage->propertyExists("CasReference")) { |
| astr << " CasReference " << textMessage->getStringProperty("CasReference"); |
| } |
| LOGINFO(INFO,astr.str()); |
| astr.seekp(0); |
| if (command == PROCESS_CAS_COMMAND) { //process CAS |
| //get the payload property |
| if (!textMessage->propertyExists("Payload")) { |
| LOGERROR("AMQListener::onMessage() Required property Payload not set."); |
| endTime = apr_time_now(); |
| sendResponse(textMessage, timeToDeserializeCAS, |
| timeToSerializeCAS, timeInAnalytic, |
| timeIdle,endTime-startTime, |
| "Required property 'Command' not set." ,true); |
| |
| endTime = apr_time_now(); |
| iv_pMonitor->processingComplete(command,false,endTime-startTime); |
| this->iv_timeLastRequestCompleted = apr_time_now(); |
| return; |
| } |
| int payload = textMessage->getIntProperty("Payload"); |
| //get the text in the payload |
| string text = textMessage->getText(); |
| if (text.length() == 0) { |
| LOGERROR("AMQListener::onMessage() There is no payload data. Nothing to process."); |
| text = "There is no payload data. Nothing to process."; |
| endTime = apr_time_now(); |
| sendResponse(textMessage, timeToDeserializeCAS, |
| timeToSerializeCAS, timeInAnalytic, |
| timeIdle, endTime-startTime, |
| text ,true); |
| endTime=apr_time_now(); |
| iv_pMonitor->processingComplete(command,false,endTime-startTime); |
| this->iv_timeLastRequestCompleted = apr_time_now(); |
| return; |
| } |
| |
| astr << "Payload: " << payload; |
| LOGINFO(FINER, astr.str()); |
| LOGINFO(FINER, " Content: " + text); |
| astr.seekp(0); |
| |
| //InputSource |
| MemBufInputSource memIS((XMLByte const *)text.c_str(), |
| text.length(), |
| "sysID"); |
| |
| //reset the CAS |
| iv_pCas->reset(); |
| 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) { |
| 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. |
| startDeserialize = apr_time_now(); |
| XmiSerializationSharedData sharedData; |
| XmiDeserializer deserializer; |
| deserializer.deserialize(memIS,*iv_pCas,sharedData); |
| startAnnotatorProcess=apr_time_now(); |
| timeToDeserializeCAS = startAnnotatorProcess-startDeserialize; |
| |
| iv_pEngine->process(*iv_pCas); |
| startSerialize=apr_time_now(); |
| timeInAnalytic = startSerialize - startAnnotatorProcess; |
| |
| //serialize CAS |
| XmiWriter xmiwriter(*iv_pCas, true, &sharedData); |
| xmiwriter.write(xmlstr); |
| timeToSerializeCAS = apr_time_now() - startSerialize; |
| } else { |
| xmlstr << "Invalid Payload " << payload; |
| LOGERROR("AMQListener::onMessage() " + xmlstr.str()); |
| endTime=apr_time_now(); |
| sendResponse(textMessage, timeToDeserializeCAS, |
| timeToSerializeCAS, timeInAnalytic, |
| timeIdle,endTime-startTime,xmlstr.str() ,true); |
| endTime=apr_time_now(); |
| iv_pMonitor->processingComplete(command,false,endTime-startTime); |
| this->iv_timeLastRequestCompleted = apr_time_now(); |
| return; |
| } |
| //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(command,true,endTime-startTime, |
| timeToDeserializeCAS, timeInAnalytic, timeToSerializeCAS); |
| LOGINFO(FINE,"Process CAS finished."); |
| } else if (command == GET_META_COMMAND ) { //get Meta |
| LOGINFO(FINE, "Process getMeta request start."); |
| endTime = apr_time_now(); |
| sendResponse(textMessage, timeToDeserializeCAS, |
| timeToSerializeCAS, timeInAnalytic, |
| timeIdle, endTime-startTime,this->iv_aeDescriptor,false); |
| endTime=apr_time_now(); |
| iv_pMonitor->processingComplete(command,true,endTime-startTime); |
| 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(); |
| sendResponse(textMessage, timeToDeserializeCAS, |
| timeToSerializeCAS, timeInAnalytic, |
| timeIdle,endTime-startTime,"CPC completed.",false); |
| endTime=apr_time_now(); |
| iv_pMonitor->processingComplete(command,true,endTime-startTime); |
| LOGINFO(FINE, "Processing CollectionProcessComplete request finished."); |
| } else { |
| endTime = apr_time_now(); |
| stringstream str; |
| str << " Invalid Request " << command << endl; |
| LOGERROR(str.str()); |
| sendResponse(textMessage, |
| timeToDeserializeCAS, |
| timeToSerializeCAS, |
| timeInAnalytic, |
| timeIdle, |
| endTime-startTime,str.str(),true); |
| endTime = apr_time_now(); |
| iv_pMonitor->processingComplete(0,false,endTime-startTime); |
| } |
| iv_busy=false; |
| iv_timeLastRequestCompleted = apr_time_now(); |
| } catch (XMLException& e) { |
| stringstream str; |
| str << "AMQListener::onMessage() XMLException." << e.getMessage(); |
| LOGERROR(str.str()); |
| endTime = apr_time_now(); |
| sendResponse(textMessage, timeToDeserializeCAS, |
| timeToSerializeCAS, timeInAnalytic, |
| timeIdle, endTime-startTime,str.str(),true); |
| endTime = apr_time_now(); |
| iv_pMonitor->processingComplete(command,false,endTime-startTime); |
| iv_timeLastRequestCompleted = apr_time_now(); |
| iv_busy=false; |
| } catch (CMSException& e) { |
| LOGERROR("AMQListener::onMessage()" + e.getMessage()); |
| endTime = apr_time_now(); |
| iv_pMonitor->processingComplete(0,false,endTime-startTime); |
| iv_timeLastRequestCompleted = apr_time_now(); |
| iv_busy=false; |
| } catch (uima::Exception e) { |
| LOGERROR("AMQListener::onMessage() UIMA Exception " + e.asString()); |
| endTime = apr_time_now(); |
| sendResponse(textMessage, timeToDeserializeCAS, |
| timeToSerializeCAS, timeInAnalytic, |
| timeIdle,endTime-startTime,e.asString(),true); |
| endTime = apr_time_now(); |
| iv_pMonitor->processingComplete(command,false,endTime-startTime); |
| iv_timeLastRequestCompleted = apr_time_now(); |
| iv_busy=false; |
| } catch(...) { |
| LOGERROR("AMQListener::onMessage() Unknown exception "); |
| //TODO: log / shurdown ?} |
| endTime = apr_time_now(); |
| iv_pMonitor->processingComplete(command,false,endTime-startTime); |
| iv_timeLastRequestCompleted = apr_time_now(); |
| iv_busy=false; |
| } |
| iv_timeLastRequestCompleted = apr_time_now(); |
| } |
| |
| 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 " << cmsReplyTo->toProviderString(); |
| } else { |
| str << " to " << request->getStringProperty("MessageFrom") |
| << " at " << serverURI; |
| } |
| str << " Message text: " << msgContent; |
| 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::onMessage()" + ex.getMessage()); |
| } catch (...) { |
| LOGERROR("AMQListener::onMessage() UnExpected error sending reply."); |
| } |
| } |
| |
| //=================================================== |
| //AMQAnalysisEngineService |
| //--------------------------------------------------- |
| AMQAnalysisEngineService::~AMQAnalysisEngineService() { |
| cleanup(); |
| } |
| |
| AMQAnalysisEngineService::AMQAnalysisEngineService(ServiceParameters & desc, |
| Monitor * stats) : |
| //iv_pMonitor(stats), |
| 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_listeners() { |
| try { |
| iv_pMonitor = stats; |
| initialize(); |
| } 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() { |
| |
| try { |
| //create a connection for each instance |
| for (int i=0; i < iv_numInstances; i++) { |
| //create a connection to MQ Broker |
| AMQConnection * newConnection = new AMQConnection(this->iv_brokerURL.c_str(), iv_pMonitor); |
| 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); |
| } |
| |
| 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 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("AMQListener::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); |
| } |
| |
| //create listeners and register these |
| for (int i=0; i < iv_numInstances; i++) { |
| 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("AnalysisEngineServcie::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; |
| |
| //create MessageConsumer session and register Listener |
| this->iv_vecpConnections.at(i)->createMessageConsumer(iv_inputQueueName, |
| newListener,this->iv_prefetchSize); |
| } |
| iv_pMonitor->setNumberOfInstances(iv_numInstances); |
| } catch (uima::Exception e) { |
| cout << __LINE__ << "got a uima exception " << 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::onException( const CMSException& ex ) { |
| LOGERROR("CMS Exception occured. Shutting down the service." + ex.getMessage()); |
| cerr << "Broken connection. Stopped receiving messages." << endl; |
| stop(); |
| this->iv_pMonitor->shutdown(); |
| } |
| |
| void AMQAnalysisEngineService::onMessage( const Message* message ) { |
| LOGINFO(0,"AMQAnalsisEngineService::onMessage() Got a message."); |
| } |
| |
| void AMQAnalysisEngineService::start() { |
| for (size_t i=0; i < iv_vecpConnections.size(); i++) { |
| cout << "Starting listener " << i << endl; |
| AMQConnection * connection = iv_vecpConnections.at(i); |
| if (connection != NULL) { |
| connection->start(); |
| } else { |
| LOGERROR("AMQAnalysisServiceEngine::start() Invalid connection object."); |
| ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR); |
| msg.addParam("AMQAnalysisServiceEngine::start() 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); |
| } |
| } |
| //update the start time |
| iv_pMonitor->setStartTime(); |
| } |
| |
| int AMQAnalysisEngineService::stop() { |
| //TODO: let listeners finish processing first |
| //stop messages notification |
| for (size_t i=0; i < iv_vecpConnections.size(); i++) { |
| AMQConnection * connection = iv_vecpConnections.at(i); |
| if (connection != NULL) { |
| connection->stop(); |
| } else { |
| LOGERROR("AMQAnalysisEngineService::stop() Connection object is NULL."); |
| ErrorMessage msg(UIMA_MSG_ID_LOG_ERROR); |
| msg.addParam("AMQAnalysisServiceEngine::start() 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); |
| } |
| } |
| return 0; |
| } |
| void AMQAnalysisEngineService::cleanup() { |
| // Destroy resources. |
| try { |
| //cout << "num consumerConnections " << consumerConnections.size() << endl; |
| 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); |
| } |
| if (iv_vecpAnalysisEngines.at(i) != NULL) { |
| delete iv_vecpAnalysisEngines.at(i); |
| } |
| if (iv_vecpCas.at(i) != NULL) { |
| 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(); |
| } catch (CMSException& e) { |
| LOGERROR("AMQAnalysisEngineService::cleanup() " + e.getMessage()); |
| } |
| } |
| |
| |
| //=================================================== |
| //CommonUtils |
| //--------------------------------------------------- |
| void CommonUtils::logError(string msg) { |
| iv_pMonitor->logError(msg); |
| //cerr << "ERROR: " << msg << endl; |
| } |
| void CommonUtils::logWarning(string msg) { |
| iv_pMonitor->logWarning(msg); |
| //cout << "WARN: " << msg << endl; |
| } |
| void CommonUtils::logMessage(string msg) { |
| iv_pMonitor->logMessage(msg); |
| //cout << "INFO: " << msg << endl; |
| } |
| |
| |
| |
| |
| |
| |
| |
