in uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java [586:786]
public void onMessage(Message aMessage, Session aJmsSession) {
String casRefId = null;
if (isStopped()) {
return;
}
try {
// wait until message handlers are plugged in
msgHandlerLatch.await();
} catch (InterruptedException e) {
}
try {
// wait until the controller is plugged in
controllerLatch.await();
} catch (InterruptedException e) {
}
long idleTime = 0;
boolean doCheckpoint = false;
String eN = endpointName;
if (getController() != null) {
eN = getController().getComponentName();
if (eN == null) {
eN = "";
}
}
String command = "N/A";
String messageType ="N/A";
JmsMessageContext messageContext = null;
int requestType = 0;
try {
// Wrap JMS Message in MessageContext
messageContext = new JmsMessageContext(aMessage, endpointName);
if (aMessage.getStringProperty(AsynchAEMessage.CasReference) == null) {
casRefId = "CasReferenceId Not In Message";
} else {
casRefId = aMessage.getStringProperty(AsynchAEMessage.CasReference);
}
try {
if ( aMessage.getStringProperty("UimaAsCasTracking") != null ) {
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
UIMAFramework.getLogger(CLASS_NAME).logrb(
Level.FINE,
CLASS_NAME.getName(),
"onMessage",
JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAJMS_service_recvd_new_message__FINE",
new Object[] {casRefId});
}
}
} catch (JMSException e1) {
}
if (validMessage(aMessage)) {
command = decodeIntToString(AsynchAEMessage.Command, aMessage
.getIntProperty(AsynchAEMessage.Command));
// Request or Response
messageType = decodeIntToString(AsynchAEMessage.MessageType, aMessage
.getIntProperty(AsynchAEMessage.MessageType));
String msgSentFromIP = null;
if (aMessage.getIntProperty(AsynchAEMessage.MessageType) == AsynchAEMessage.Response
&& aMessage.propertyExists(AsynchAEMessage.ServerIP)) {
msgSentFromIP = aMessage.getStringProperty(AsynchAEMessage.ServerIP);
}
// System.out.println("***********************************************************************************"
// +
// " \n**CONTROLLER::"+controller.getName()+"**** Received New Message From [ "+aMessage.getStringProperty(AsynchAEMessage.MessageFrom)+" ]**************"
// +
// " \n**MSGTYPE::"+messageType+" COMMAND:"+command +
// " Cas Reference Id::"+casRefId+
// " \n******************************************************************************");
String msgFrom = (String) aMessage.getStringProperty(AsynchAEMessage.MessageFrom);
if (controller != null && msgFrom != null) {
// Send an ACK to a client. This serves dual purpose:
// 1 - tests for existence of temp reply queue
// 2 - notifies client that a CAS it sent is about to be processed by a service
// ACK is sent for every request message
if ( ackClient(messageContext) ) {
try {
// Any exception while sending an ACK results in a dropped request
getController().getOutputChannel().sendReply(AsynchAEMessage.ServiceInfo,
messageContext.getEndpoint(), aMessage.getStringProperty(AsynchAEMessage.CasReference), true);
} catch( Exception ex) {
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
"produceTextMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAJMS_exception__WARNING", ex);
// The exception has already been logged in sendReply() method
return; // DONT PROCESS
}
}
if (msgSentFromIP != null) {
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
UIMAFramework.getLogger(CLASS_NAME).logrb(
Level.FINE,
CLASS_NAME.getName(),
"onMessage",
JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAJMS_recvd_new_message_with_ip__FINE",
new Object[] { controller.getComponentName(), msgFrom, msgSentFromIP,
messageType, command, casRefId });
}
} else {
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
UIMAFramework.getLogger(CLASS_NAME).logrb(
Level.FINE,
CLASS_NAME.getName(),
"onMessage",
JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAJMS_recvd_new_message__FINE",
new Object[] { controller.getComponentName(), msgFrom, messageType, command,
casRefId });
}
}
} else {
}
// Delegate processing of the message contained in the MessageContext to the
// chain of handlers
try {
if (isRemoteRequest(aMessage)) {
// Compute the idle time waiting for this request
idleTime = getController().getIdleTime();
// This idle time is reported to the client thus save it in the endpoint
// object. This value will be fetched and added to the outgoing reply.
messageContext.getEndpoint().setIdleTime(idleTime);
}
} catch (Exception e) {
}
// Determine if this message is a request and either GetMeta, CPC, or Process
doCheckpoint = isCheckpointWorthy(aMessage);
requestType = aMessage.getIntProperty(AsynchAEMessage.Command);
// Checkpoint
if (doCheckpoint) {
getController().beginProcess(requestType);
}
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(CLASS_NAME).logrb(
Level.FINEST,
CLASS_NAME.getName(),
"onMessage",
JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAJMS_new_msg_in__FINEST",
new Object[] { getController().getComponentName(), msgFrom, command, messageType,
casRefId });
}
if (handler != null) {
handler.handle(messageContext);
}
} else {
if (!isStaleMessage(aMessage)) {
controller.getErrorHandlerChain().handle(new InvalidMessageException(),
HandlerBase.populateErrorContext(messageContext), controller);
}
}
} catch (Throwable t) {
if ( t instanceof Error) {
t.printStackTrace();
System.err.println("\n\n\n\tUIMA AS Service Caught Java Error. Exiting via System.exit(2)");
System.err.flush();
System.exit(2);
}
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
"onMessage", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAEE_service_exception_WARNING", controller.getComponentName());
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
"onMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAJMS_exception__WARNING", t);
}
controller.getErrorHandlerChain().handle(t, HandlerBase.populateErrorContext(messageContext),
controller);
} finally {
// Call the end checkpoint for non-aggregates. For primitives the CAS has been fully processed
// if we are here
if (doCheckpoint && getController() instanceof PrimitiveAnalysisEngineController) {
getController().endProcess(requestType);
}
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "onMessage",
JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_msg_processed__FINE",
new Object[] { getController().getComponentName(), casRefId });
}
}
}