| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| package org.apache.uima.adapter.jms.activemq; |
| |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.CountDownLatch; |
| |
| import javax.jms.Connection; |
| import javax.jms.Destination; |
| import javax.jms.JMSException; |
| import javax.jms.Message; |
| import javax.jms.Session; |
| |
| import org.apache.activemq.ActiveMQConnectionFactory; |
| import org.apache.activemq.command.ActiveMQDestination; |
| import org.apache.activemq.command.ActiveMQMessage; |
| import org.apache.uima.UIMAFramework; |
| import org.apache.uima.aae.InputChannel; |
| import org.apache.uima.aae.controller.AggregateAnalysisEngineController; |
| import org.apache.uima.aae.controller.AnalysisEngineController; |
| import org.apache.uima.aae.controller.Endpoint; |
| import org.apache.uima.aae.controller.Endpoint_impl; |
| import org.apache.uima.aae.controller.PrimitiveAnalysisEngineController; |
| import org.apache.uima.aae.controller.BaseAnalysisEngineController.ServiceState; |
| import org.apache.uima.aae.delegate.Delegate; |
| import org.apache.uima.aae.error.InvalidMessageException; |
| import org.apache.uima.aae.handler.Handler; |
| import org.apache.uima.aae.handler.HandlerBase; |
| import org.apache.uima.aae.jmx.ServiceInfo; |
| import org.apache.uima.aae.message.AsynchAEMessage; |
| import org.apache.uima.aae.message.MessageContext; |
| import org.apache.uima.aae.message.UIMAMessage; |
| import org.apache.uima.adapter.jms.JmsConstants; |
| import org.apache.uima.adapter.jms.message.JmsMessageContext; |
| import org.apache.uima.util.Level; |
| import org.springframework.jms.listener.SessionAwareMessageListener; |
| |
| /** |
| * Thin adapter for receiving JMS messages from Spring. It delegates processing of all messages to |
| * the {@link MessageHandler}. Each JMS Message is wrapped in transport neutral MessageContext |
| * wrapper. |
| * |
| */ |
| public class JmsInputChannel implements InputChannel, JmsInputChannelMBean, |
| SessionAwareMessageListener { |
| /** |
| * |
| */ |
| private static final long serialVersionUID = -3318400773113552290L; |
| |
| private static final Class CLASS_NAME = JmsInputChannel.class; |
| |
| private transient final CountDownLatch msgHandlerLatch = new CountDownLatch(1); |
| |
| private transient final CountDownLatch controllerLatch = new CountDownLatch(1); |
| |
| // Reference to the first Message Handler in the Chain. |
| private transient Handler handler; |
| |
| // The name of the queue this Listener is expecting to receive messages from |
| private String endpointName; |
| |
| // Reference to the Controller Object |
| private transient AnalysisEngineController controller; |
| |
| private int sessionAckMode; |
| |
| private transient UimaDefaultMessageListenerContainer messageListener; |
| |
| private transient Session jmsSession; |
| |
| private String brokerURL = ""; |
| |
| private ServiceInfo serviceInfo = null; |
| |
| private volatile boolean stopped = false; |
| |
| private volatile boolean channelRegistered = false; |
| |
| private List listenerContainerList = new ArrayList(); |
| |
| private Object mux = new Object(); |
| |
| private ConcurrentHashMap<String, UimaDefaultMessageListenerContainer> failedListenerMap = new ConcurrentHashMap<String, UimaDefaultMessageListenerContainer>(); |
| |
| public AnalysisEngineController getController() { |
| return controller; |
| } |
| |
| public String getName() { |
| return endpointName; |
| } |
| |
| public void setController(AnalysisEngineController aController) throws Exception { |
| |
| this.controller = aController; |
| if (!channelRegistered) { |
| controller.addInputChannel(this); |
| } |
| controller.setInputChannel(this); |
| controllerLatch.countDown(); |
| } |
| |
| public void setMessageHandler(Handler aHandler) { |
| handler = aHandler; |
| msgHandlerLatch.countDown(); |
| } |
| |
| public void setEndpointName(String anEndpointName) { |
| endpointName = anEndpointName; |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "setEndpointName", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_service_listening__INFO", new Object[] { anEndpointName }); |
| |
| } |
| } |
| |
| /** |
| * Validate message type contained in the JMS header. |
| * |
| * @param aMessage |
| * - jms message retrieved from queue |
| * @param properties |
| * - map containing message properties |
| * @return |
| * @throws Exception |
| */ |
| |
| private boolean validMessageType(Message aMessage, Map properties) throws Exception { |
| if (properties.containsKey(AsynchAEMessage.MessageType)) { |
| int msgType = aMessage.getIntProperty(AsynchAEMessage.MessageType); |
| if (msgType != AsynchAEMessage.Response && msgType != AsynchAEMessage.Request) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "validMessageType", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_invalid_msgtype_in_message__INFO", |
| new Object[] { msgType, endpointName }); |
| } |
| return false; |
| |
| } |
| } else { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "validMessageType", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_msgtype_notin_message__INFO", new Object[] { endpointName }); |
| } |
| return false; |
| } |
| |
| return true; |
| } |
| |
| private boolean isProcessRequest(Message aMessage) throws Exception { |
| Map properties = ((ActiveMQMessage) aMessage).getProperties(); |
| if (properties.containsKey(AsynchAEMessage.MessageType) |
| && properties.containsKey(AsynchAEMessage.Command)) { |
| int msgType = aMessage.getIntProperty(AsynchAEMessage.MessageType); |
| int command = aMessage.getIntProperty(AsynchAEMessage.Command); |
| |
| if (msgType != AsynchAEMessage.Request || command != AsynchAEMessage.Process) { |
| return false; |
| } |
| return true; |
| } |
| return false; |
| } |
| |
| private boolean isRemoteRequest(Message aMessage) throws Exception { |
| |
| // Dont do checkpoints if a message was sent from a Cas Multiplier |
| if (aMessage.propertyExists(AsynchAEMessage.CasSequence)) { |
| return false; |
| } |
| |
| Map properties = ((ActiveMQMessage) aMessage).getProperties(); |
| if (properties.containsKey(AsynchAEMessage.MessageType) |
| && properties.containsKey(AsynchAEMessage.Command) |
| && properties.containsKey(UIMAMessage.ServerURI)) { |
| int msgType = aMessage.getIntProperty(AsynchAEMessage.MessageType); |
| int command = aMessage.getIntProperty(AsynchAEMessage.Command); |
| boolean isRemote = aMessage.getStringProperty(UIMAMessage.ServerURI).startsWith("vm") == false; |
| if (isRemote |
| && msgType == AsynchAEMessage.Request |
| && (command == AsynchAEMessage.Process || command == AsynchAEMessage.CollectionProcessComplete)) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| private boolean acceptsDeltaCas(Message aMessage) throws Exception { |
| Map properties = ((ActiveMQMessage) aMessage).getProperties(); |
| boolean acceptsDeltaCas = false; |
| if (properties.containsKey(AsynchAEMessage.AcceptsDeltaCas)) { |
| acceptsDeltaCas = aMessage.getBooleanProperty(AsynchAEMessage.AcceptsDeltaCas); |
| } |
| return acceptsDeltaCas; |
| } |
| |
| /** |
| * Validate command contained in the header of the JMS Message |
| * |
| * @param aMessage |
| * - JMS Message received |
| * @param properties |
| * - Map containing header properties |
| * @return - true if the command received is a valid one, false otherwise |
| * @throws Exception |
| */ |
| private boolean validCommand(Message aMessage, Map properties) throws Exception { |
| if (properties.containsKey(AsynchAEMessage.Command)) { |
| int command = aMessage.getIntProperty(AsynchAEMessage.Command); |
| if (command != AsynchAEMessage.Process && command != AsynchAEMessage.GetMeta |
| && command != AsynchAEMessage.ReleaseCAS && command != AsynchAEMessage.Stop |
| && command != AsynchAEMessage.Ping && command != AsynchAEMessage.ServiceInfo |
| && command != AsynchAEMessage.CollectionProcessComplete) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "validCommand", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_invalid_command_in_message__INFO", |
| new Object[] { command, endpointName }); |
| } |
| return false; |
| } |
| } else { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "validCommand", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_command_notin_message__INFO", |
| new Object[] { endpointName }); |
| } |
| return false; |
| } |
| |
| return true; |
| } |
| |
| /** |
| * Validates payload in the JMS Message. |
| * |
| * @param aMessage |
| * - JMS Message received |
| * @param properties |
| * - Map containing header properties |
| * @return - true if the payload is valid, false otherwise |
| * @throws Exception |
| */ |
| private boolean validPayload(Message aMessage, Map properties) throws Exception { |
| if (properties.containsKey(AsynchAEMessage.Command)) { |
| int command = aMessage.getIntProperty(AsynchAEMessage.Command); |
| if (command == AsynchAEMessage.GetMeta |
| || command == AsynchAEMessage.CollectionProcessComplete |
| || command == AsynchAEMessage.Stop || command == AsynchAEMessage.Ping |
| || command == AsynchAEMessage.ServiceInfo || command == AsynchAEMessage.ReleaseCAS) { |
| // Payload not included in GetMeta Request |
| return true; |
| } |
| } |
| |
| if (properties.containsKey(AsynchAEMessage.Payload)) { |
| int payload = aMessage.getIntProperty(AsynchAEMessage.Payload); |
| if (payload != AsynchAEMessage.XMIPayload && payload != AsynchAEMessage.BinaryPayload |
| && payload != AsynchAEMessage.CASRefID && payload != AsynchAEMessage.Exception |
| && payload != AsynchAEMessage.Metadata) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "validPayload", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_invalid_payload_in_message__INFO", |
| new Object[] { payload, endpointName }); |
| } |
| |
| return false; |
| } |
| } else { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "validPayload", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_payload_notin_message__INFO", |
| new Object[] { endpointName }); |
| } |
| return false; |
| } |
| |
| return true; |
| } |
| |
| private boolean isStaleMessage(Message aMessage) throws JMSException { |
| int command = aMessage.getIntProperty(AsynchAEMessage.Command); |
| int msgType = aMessage.getIntProperty(AsynchAEMessage.MessageType); |
| if (isStopped() || getController() == null || getController().getInProcessCache() == null) { |
| // Shutting down |
| return true; |
| } |
| if (command == AsynchAEMessage.Process && msgType == AsynchAEMessage.Response) { |
| String casReferenceId = aMessage.getStringProperty(AsynchAEMessage.CasReference); |
| if (!getController().getInProcessCache().entryExists(casReferenceId)) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "isStaleMessage", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_stale_message__FINE", |
| new Object[] { endpointName, casReferenceId, |
| aMessage.getStringProperty(AsynchAEMessage.MessageFrom) }); |
| } |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Validates contents of the message. It checks if command, payload and message types contain |
| * valid data. |
| * |
| * @param aMessage |
| * - JMS Message to validate |
| * @return - true if message is valid, false otherwise |
| * @throws Exception |
| */ |
| public boolean validMessage(Message aMessage) throws Exception { |
| if (aMessage instanceof ActiveMQMessage) { |
| Map properties = ((ActiveMQMessage) aMessage).getProperties(); |
| if (!validMessageType(aMessage, properties)) { |
| int msgType = 0; |
| if (properties.containsKey(AsynchAEMessage.MessageType)) { |
| msgType = aMessage.getIntProperty(AsynchAEMessage.MessageType); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "validMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_invalid_msg_type__INFO", |
| new Object[] { getController().getComponentName(), msgType }); |
| } |
| return false; |
| } |
| if (!validCommand(aMessage, properties)) { |
| int command = 0; |
| if (properties.containsKey(AsynchAEMessage.Command)) { |
| command = aMessage.getIntProperty(AsynchAEMessage.Command); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "validMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_invalid_cmd_type__INFO", |
| new Object[] { getController().getComponentName(), command }); |
| } |
| return false; |
| } |
| if (!validPayload(aMessage, properties)) { |
| int payload = 0; |
| if (properties.containsKey(AsynchAEMessage.Payload)) { |
| payload = aMessage.getIntProperty(AsynchAEMessage.Payload); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "validMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_invalid_payload_type__INFO", |
| new Object[] { getController().getComponentName(), payload }); |
| } |
| return false; |
| } |
| |
| if (isStaleMessage(aMessage)) { |
| if (sessionAckMode == Session.CLIENT_ACKNOWLEDGE) { |
| aMessage.acknowledge(); |
| } |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| public void abort() { |
| } |
| |
| private String decodeIntToString(String aTypeToDecode, int aValueToDecode) { |
| if (AsynchAEMessage.MessageType.equals(aTypeToDecode)) { |
| switch (aValueToDecode) { |
| case AsynchAEMessage.Request: |
| return "Request"; |
| case AsynchAEMessage.Response: |
| return "Response"; |
| } |
| } else if (AsynchAEMessage.Command.equals(aTypeToDecode)) { |
| switch (aValueToDecode) { |
| case AsynchAEMessage.Process: |
| return "Process"; |
| case AsynchAEMessage.GetMeta: |
| return "GetMetadata"; |
| case AsynchAEMessage.CollectionProcessComplete: |
| return "CollectionProcessComplete"; |
| case AsynchAEMessage.ReleaseCAS: |
| return "ReleaseCAS"; |
| case AsynchAEMessage.Stop: |
| return "Stop"; |
| case AsynchAEMessage.Ping: |
| return "Ping"; |
| case AsynchAEMessage.ServiceInfo: |
| return "ServiceInfo"; |
| } |
| |
| } else if (AsynchAEMessage.Payload.equals(aTypeToDecode)) { |
| switch (aValueToDecode) { |
| case AsynchAEMessage.XMIPayload: |
| return "XMIPayload"; |
| case AsynchAEMessage.BinaryPayload: |
| return "BinaryPayload"; |
| case AsynchAEMessage.CASRefID: |
| return "CASRefID"; |
| case AsynchAEMessage.Metadata: |
| return "Metadata"; |
| case AsynchAEMessage.Exception: |
| return "Exception"; |
| case AsynchAEMessage.XCASPayload: |
| return "XCASPayload"; |
| case AsynchAEMessage.None: |
| return "None"; |
| } |
| } |
| return "UNKNOWN"; |
| } |
| |
| private boolean ackMessageNow(Message aMessage) throws JMSException { |
| if (sessionAckMode != Session.CLIENT_ACKNOWLEDGE) { |
| return false; |
| } |
| |
| if (aMessage.getIntProperty(AsynchAEMessage.Command) == AsynchAEMessage.GetMeta |
| || aMessage.getIntProperty(AsynchAEMessage.Command) == AsynchAEMessage.CollectionProcessComplete |
| || aMessage.getIntProperty(AsynchAEMessage.Command) == AsynchAEMessage.ReleaseCAS |
| || aMessage.getIntProperty(AsynchAEMessage.Command) == AsynchAEMessage.ACK) { |
| return true; |
| } |
| return false; |
| } |
| |
| private boolean isCheckpointWorthy(Message aMessage) throws Exception { |
| synchronized (mux) { |
| // Dont do checkpoints if a message was sent from a Cas Multiplier |
| if (aMessage.propertyExists(AsynchAEMessage.CasSequence)) { |
| return false; |
| } |
| Map properties = ((ActiveMQMessage) aMessage).getProperties(); |
| if (properties.containsKey(AsynchAEMessage.MessageType) |
| && properties.containsKey(AsynchAEMessage.Command) |
| && properties.containsKey(UIMAMessage.ServerURI)) { |
| int msgType = aMessage.getIntProperty(AsynchAEMessage.MessageType); |
| int command = aMessage.getIntProperty(AsynchAEMessage.Command); |
| if (msgType == AsynchAEMessage.Request |
| && (command == AsynchAEMessage.Process || command == AsynchAEMessage.CollectionProcessComplete)) { |
| return true; |
| } |
| } |
| return false; |
| |
| } |
| } |
| |
| /** |
| * Receives Messages from the JMS Provider. It checks the message header to determine the type of |
| * message received. Based on the type, a MessageContext is created to facilitate access to the |
| * transport specific message. Once the MessageContext is determined this routine delegates |
| * handling of the message to the chain of MessageHandlers. |
| * |
| * @param aMessage |
| * - JMS Message containing header and payload |
| * @param aSession |
| * - JMSSession object |
| */ |
| 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 = ""; |
| } |
| } |
| |
| 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_msg__FINE", new Object[] { eN }); |
| } |
| 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); |
| } |
| if (validMessage(aMessage)) { |
| String command = decodeIntToString(AsynchAEMessage.Command, aMessage |
| .getIntProperty(AsynchAEMessage.Command)); |
| // Request or Response |
| String messageType = decodeIntToString(AsynchAEMessage.MessageType, aMessage |
| .getIntProperty(AsynchAEMessage.MessageType)); |
| // Check if the request message contains a reply destination known to have |
| // been deleted. This could be the case if the client has been shutdown after |
| // sending requests. Previous attempt to deliver a reply failed and the client's |
| // reply endpoint was placed on the DoNotProcess List. For optimization, all |
| // requests from a dead client will be dropped. |
| if (aMessage.getIntProperty(AsynchAEMessage.MessageType) == AsynchAEMessage.Request |
| && messageContext != null |
| && messageContext.getEndpoint() != null |
| && messageContext.getEndpoint().getDestination() != null |
| && getController().isEndpointOnDontProcessList( |
| messageContext.getEndpoint().getDestination().toString())) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| CLASS_NAME.getName(), |
| "onMessage", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_dropping_msg_client_is_dead__INFO", |
| new Object[] { controller.getComponentName(), |
| messageContext.getEndpoint().getDestination(), casRefId }); |
| } |
| return; |
| } |
| |
| if (ackMessageNow(aMessage)) { |
| aMessage.acknowledge(); |
| } |
| 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) { |
| 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 (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "onMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), 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 }); |
| } |
| } |
| } |
| |
| public int getSessionAckMode() { |
| return sessionAckMode; |
| } |
| |
| public String getServerUri() { |
| return brokerURL; |
| } |
| |
| public synchronized void setListenerContainer(UimaDefaultMessageListenerContainer messageListener) { |
| this.messageListener = messageListener; |
| System.setProperty("BrokerURI", messageListener.getBrokerUrl()); |
| if ( messageListener.getMessageSelector() !=null && messageListener.getMessageSelector().equals("Command=2001") ) { |
| brokerURL = messageListener.getBrokerUrl(); |
| getController().getOutputChannel().setServerURI(brokerURL); |
| } |
| if (!listenerContainerList.contains(messageListener)) { |
| listenerContainerList.add(messageListener); |
| } |
| if (getController() != null) { |
| try { |
| getController().addInputChannel(this); |
| messageListener.setController(getController()); |
| } catch (Exception e) { |
| } |
| } |
| } |
| |
| public ActiveMQConnectionFactory getConnectionFactory() { |
| if (messageListener == null) { |
| return null; |
| } else { |
| return (ActiveMQConnectionFactory) messageListener.getConnectionFactory(); |
| } |
| } |
| |
| public void ackMessage(MessageContext aMessageContext) { |
| if (aMessageContext != null && sessionAckMode == Session.CLIENT_ACKNOWLEDGE) { |
| try { |
| ((Message) aMessageContext.getRawMessage()).acknowledge(); |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "ackMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_exception__WARNING", |
| new Object[] { Thread.currentThread().getName(), e }); |
| } |
| } |
| } |
| } |
| |
| public String getBrokerURL() { |
| return brokerURL; |
| } |
| |
| public String getInputQueueName() { |
| if (messageListener != null) |
| if (messageListener.getDestination() != null) { |
| return messageListener.getDestination().toString(); |
| } else { |
| return messageListener.getDestinationName();// getEndpointName(); |
| } |
| else { |
| return ""; |
| } |
| } |
| |
| public ServiceInfo getServiceInfo() { |
| if (serviceInfo == null) { |
| serviceInfo = new ServiceInfo(false,controller); |
| serviceInfo.setBrokerURL(getBrokerURL()); |
| serviceInfo.setInputQueueName(getName()); |
| if ( controller == null ) { |
| serviceInfo.setState(ServiceState.INITIALIZING.name()); |
| } else { |
| if ( controller.isCasMultiplier()) { |
| serviceInfo.setCASMultiplier(); |
| } |
| } |
| } |
| return serviceInfo; |
| } |
| |
| public void setServerUri(String serverUri) { |
| brokerURL = serverUri; |
| if (getController() != null && getController() instanceof AggregateAnalysisEngineController) { |
| ((AggregateAnalysisEngineController) getController()).getServiceInfo() |
| .setBrokerURL(brokerURL); |
| } else { |
| ((PrimitiveAnalysisEngineController) getController()).getServiceInfo() |
| .setBrokerURL(brokerURL); |
| } |
| } |
| |
| private void stopChannel(UimaDefaultMessageListenerContainer mL) throws Exception { |
| String eName = mL.getEndpointName(); |
| if (eName != null) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "stop", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_stopping_jms_transport__INFO", |
| new Object[] { eName }); |
| } |
| } |
| mL.stop(); |
| |
| String selector = ""; |
| if (mL.getMessageSelector() != null) { |
| selector = " Selector:" + mL.getMessageSelector(); |
| } |
| System.out.println("Service:" + getController().getComponentName() + " Message Channel:" |
| + mL.getDestination() + selector + " Stopped"); |
| } |
| |
| private boolean doCloseChannel(UimaDefaultMessageListenerContainer mL, int channelsToClose) { |
| // Check if we are closing just the input channel |
| if (channelsToClose == InputChannel.InputChannels) { |
| // Fetch the listener object |
| ActiveMQDestination destination = (ActiveMQDestination) mL.getListenerEndpoint(); |
| // if this is a listener on a temp queue return false. We need to keep all temp |
| // channels open to receive replies and/or notifications to free CASes |
| // Keep the temp reply channel open |
| if (destination != null && destination.isTemporary()) { |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| public void stop() throws Exception { |
| stop(InputChannel.CloseAllChannels); |
| listenerContainerList.clear(); |
| failedListenerMap.clear(); |
| } |
| |
| public synchronized void stop(int channelsToClose) throws Exception { |
| |
| List<UimaDefaultMessageListenerContainer> listenersToRemove = new ArrayList<UimaDefaultMessageListenerContainer>(); |
| for (Object listenerObject : listenerContainerList) { |
| final UimaDefaultMessageListenerContainer mL = (UimaDefaultMessageListenerContainer) listenerObject; |
| if (mL != null && mL.isRunning() && doCloseChannel(mL, channelsToClose)) { |
| stopChannel(mL); |
| // Just in case check if the container still in the list. If so, add it to |
| // another list that container listeners that have been stopped and need |
| // to be removed from the listenerContainerList. Removing the listener from |
| // the listenerContainerList in this iterator loop is not working. If for |
| // example the iterator has two elements, after the first remove from the |
| // listenerContainerList, the iterator stops event though there is still |
| // one element left. Process removal of listeners outside of the iterator |
| // loop |
| if (listenerContainerList.contains(mL)) { |
| listenersToRemove.add(mL); |
| } |
| } else { |
| if (getController() != null) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "stop", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_transport_not_stopped__INFO", |
| new Object[] { getController().getComponentName() }); |
| } |
| } |
| } |
| } |
| // Remove listeners from the listenerContainerList |
| for (UimaDefaultMessageListenerContainer mL : listenersToRemove) { |
| listenerContainerList.remove(mL); |
| } |
| listenersToRemove.clear(); |
| if (channelsToClose == InputChannel.CloseAllChannels) { |
| stopped = true; |
| } |
| } |
| |
| public boolean isStopped() { |
| return stopped; |
| } |
| |
| public int getConcurrentConsumerCount() { |
| return messageListener.getConcurrentConsumers(); |
| } |
| |
| private void testIfBrokerRunning(String aBrokerUrl) throws Exception { |
| ActiveMQConnectionFactory f = new ActiveMQConnectionFactory(aBrokerUrl); |
| // Try to create a test connection to make sure that the broker is available |
| Connection testConnection = null; |
| try { |
| testConnection = f.createConnection(); |
| } catch (Exception e) { |
| throw e; |
| } finally { |
| if (testConnection != null) { |
| // close test connection. Broker is running |
| testConnection.close(); |
| } |
| } |
| } |
| |
| public void createListener(String aDelegateKey, Endpoint endpointToUpdate) throws Exception { |
| if (getController() instanceof AggregateAnalysisEngineController) { |
| Delegate delegate = ((AggregateAnalysisEngineController) getController()) |
| .lookupDelegate(aDelegateKey); |
| if (delegate != null) { |
| |
| UimaDefaultMessageListenerContainer newListener = new UimaDefaultMessageListenerContainer(); |
| |
| testIfBrokerRunning(delegate.getEndpoint().getServerURI()); |
| ActiveMQConnectionFactory f = new ActiveMQConnectionFactory(delegate.getEndpoint() |
| .getServerURI()); |
| newListener.setConnectionFactory(f); |
| newListener.setMessageListener(this); |
| newListener.setController(getController()); |
| |
| TempDestinationResolver resolver = new TempDestinationResolver(); |
| resolver.setConnectionFactory(f); |
| resolver.setListener(newListener); |
| newListener.setDestinationResolver(resolver); |
| |
| org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor executor = new org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor(); |
| executor.setCorePoolSize(delegate.getEndpoint().getConcurrentReplyConsumers()); |
| executor.setMaxPoolSize(delegate.getEndpoint().getConcurrentReplyConsumers()); |
| executor.setQueueCapacity(delegate.getEndpoint().getConcurrentReplyConsumers()); |
| executor.initialize(); |
| newListener.setConcurrentConsumers(delegate.getEndpoint().getConcurrentReplyConsumers()); |
| newListener.setTaskExecutor(executor); |
| newListener.initialize(); |
| newListener.start(); |
| // Wait until the resolver plugs in the destination |
| while (newListener.getDestination() == null) { |
| synchronized (newListener) { |
| |
| System.out |
| .println(".... Waiting For Spring Resolver to Create New Temp Reply Queue ..."); |
| newListener.wait(100); |
| } |
| } |
| newListener.afterPropertiesSet(); |
| if ( controller != null && controller.isStopped() ) { |
| System.out.println("Controller:"+controller.getComponentName()+" Stopping New Listener. The Service is stopping"); |
| newListener.stop(); |
| // we are aborting, the controller has been stopped |
| return; |
| } |
| // Get the endpoint object for a given delegate key from the Aggregate |
| Endpoint endpoint = ((AggregateAnalysisEngineController) getController()).lookUpEndpoint( |
| aDelegateKey, false); |
| // Override the reply destination. |
| endpoint.setDestination(newListener.getDestination()); |
| if ( endpointToUpdate != null) { |
| endpointToUpdate.setDestination(newListener.getDestination()); |
| } |
| Object clone = ((Endpoint_impl) endpoint).clone(); |
| newListener.setTargetEndpoint((Endpoint) clone); |
| endpoint.setStatus(Endpoint.OK); |
| System.out |
| .println(".... Listener Started on New Temp Reply Queue ..."); |
| } |
| } |
| } |
| public boolean isListenerActiveOnDestination(Destination destination ) { |
| for (int i = 0; i < listenerContainerList.size(); i++) { |
| UimaDefaultMessageListenerContainer mListener = (UimaDefaultMessageListenerContainer) listenerContainerList |
| .get(i); |
| if ( mListener.getDestination() != null && |
| mListener.getDestination() == destination && |
| mListener.isRunning()) { |
| return true; |
| } |
| } |
| return false; |
| } |
| /** |
| * Given an endpoint name returns all listeners attached to this endpoint. There can be multiple |
| * listeners on an endpoint each with a different selector to receive targeted messages like |
| * GetMeta and Process. |
| * |
| * @param anEndpointName |
| * - name of the endpoint that is used to find associated listener(s) |
| * |
| * @return - list of listeners |
| */ |
| private UimaDefaultMessageListenerContainer[] getListenersForEndpoint(String anEndpointName) { |
| List<UimaDefaultMessageListenerContainer> listeners = new ArrayList<UimaDefaultMessageListenerContainer>(); |
| for (int i = 0; i < listenerContainerList.size(); i++) { |
| UimaDefaultMessageListenerContainer mListener = (UimaDefaultMessageListenerContainer) listenerContainerList |
| .get(i); |
| if (mListener.getDestinationName() != null |
| && mListener.getDestinationName().equals(anEndpointName)) { |
| listeners.add(mListener); |
| } else if (mListener.getDestination() != null |
| && mListener.getDestination().toString().equals(anEndpointName)) { |
| listeners.add(mListener); |
| } |
| } |
| if (listeners.size() > 0) { |
| UimaDefaultMessageListenerContainer[] listenerArray = new UimaDefaultMessageListenerContainer[listeners |
| .size()]; |
| listeners.toArray(listenerArray); |
| return listenerArray; |
| } |
| return null; |
| } |
| |
| /** |
| * |
| */ |
| public void destroyListener(final String anEndpointName, String aDelegateKey) { |
| // check if delegate listener has already been placed in the failed listeners list |
| // If so, nothing else to do here |
| if (failedListenerMap.containsKey(aDelegateKey)) { |
| return; |
| } |
| // Fetch all associated listeners. |
| final UimaDefaultMessageListenerContainer[] mListeners = getListenersForEndpoint(anEndpointName); |
| if (mListeners == null) { |
| return; |
| } |
| // Stop each listener |
| for (final UimaDefaultMessageListenerContainer mListener : mListeners) { |
| if (!mListener.isRunning()) { |
| continue; // Already Stopped |
| } |
| |
| try { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO) |
| && mListener.getDestination() != null) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "destroyListener", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_stop_listener__INFO", |
| new Object[] { mListener.getDestination().toString() }); |
| } |
| // Spin a thread that will stop the listener and wait for its shutdown |
| Thread stopThread = new Thread("InputChannelStopThread") { |
| public void run() { |
| mListener.stop(); |
| // wait until the listener shutsdown |
| while (mListener.isRunning()) |
| ; |
| System.out.println("Thread:" + Thread.currentThread().getId() |
| + "++++ Listener on Queue:" + anEndpointName + " Has Been Stopped..."); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO) |
| && mListener.getDestination() != null) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| CLASS_NAME.getName(), |
| "destroyListener", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_stopped_listener_INFO", |
| new Object[] { controller.getComponentName(), |
| mListener.getDestination().toString() }); |
| } |
| } |
| }; |
| stopThread.start(); |
| |
| if (getController() != null) { |
| Endpoint endpoint = ((AggregateAnalysisEngineController) getController()).lookUpEndpoint( |
| aDelegateKey, false); |
| endpoint.setStatus(Endpoint.FAILED); |
| if (mListener.getConnectionFactory() != null) { |
| if (getController() instanceof AggregateAnalysisEngineController) { |
| if (!failedListenerMap.containsKey(aDelegateKey)) { |
| failedListenerMap.put(aDelegateKey, mListener); |
| listenerContainerList.remove(mListener); |
| } |
| } |
| } |
| } |
| // } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "destroyListener", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| |
| } |
| } |
| } |
| |
| public boolean isFailed(String aDelegateKey) { |
| return failedListenerMap.containsKey(aDelegateKey); |
| } |
| |
| public boolean isListenerForDestination(String anEndpointName) { |
| UimaDefaultMessageListenerContainer[] mListeners = getListenersForEndpoint(anEndpointName); |
| if (mListeners == null) { |
| return false; |
| } |
| return true; |
| } |
| |
| } |
