| /* |
| * 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.CopyOnWriteArrayList; |
| 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.UIMAEE_Constants; |
| import org.apache.uima.aae.controller.AggregateAnalysisEngineController; |
| import org.apache.uima.aae.controller.AnalysisEngineController; |
| import org.apache.uima.aae.controller.BaseAnalysisEngineController.ServiceState; |
| 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.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.RemoteJMXServer; |
| 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 Handler}. 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 transient RemoteJMXServer remoteJMXServer = null; |
| // synchronizes initialization of RemotBroker |
| private Object brokerMux = new Object(); |
| |
| private ConcurrentHashMap<String, UimaDefaultMessageListenerContainer> failedListenerMap = new ConcurrentHashMap<String, UimaDefaultMessageListenerContainer>(); |
| |
| // A global flag that determines if we should create a connection to broker's MBeanServer to be |
| // able to determine if client's reply queue exists before processing a CAS. |
| public static transient boolean attachToBrokerMBeanServer=true; |
| |
| private static Map<String, List<UimaDefaultMessageListenerContainer>> listenerMap = |
| new ConcurrentHashMap<String, List<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 - true if message Type is valid, false otherwise |
| * @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"; |
| // 5/2013 xcas not used |
| // 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; |
| |
| } |
| } |
| /** |
| * Checks if the incoming request requires response. |
| * |
| * @param aMessage - incoming message to check |
| * @return - true if reply is required, false otherwise |
| */ |
| private boolean isReplyRequired(Message aMessage ) { |
| try { |
| int command = aMessage.getIntProperty(AsynchAEMessage.Command); |
| if (aMessage.getIntProperty(AsynchAEMessage.MessageType) == AsynchAEMessage.Request && |
| ( command == AsynchAEMessage.Process || |
| command == AsynchAEMessage.GetMeta || |
| command == AsynchAEMessage.CollectionProcessComplete ) ) { |
| return true; |
| } |
| } catch( Exception e) { |
| // ignore |
| } |
| return false; |
| } |
| private boolean validEndpoint(JmsMessageContext messageContext) { |
| return (messageContext.getEndpoint() != null |
| // Reply destination must be present |
| && messageContext.getEndpoint().getDestination() != null); |
| } |
| /** |
| * Determines if a given message should be processed or not. If the message |
| * requires response, the code checks if the client's temp reply queue still |
| * exists. If so, the message must be processed. If the reply queue doesnt |
| * exist, the client must have terminated and there is no need to process |
| * the message. The code consults the broker (via JMX) to see if the queue |
| * exists. |
| * |
| * @param aMessage |
| * @param messageContext |
| * @return true if the message is of type Process, false otherwise |
| * @throws Exception |
| */ |
| private boolean processRequestMessage(Message aMessage, JmsMessageContext messageContext) throws Exception { |
| // check if we can reply to the client. The code uses connection to |
| // broker's JMX server to lookup client's temp reply queue. If the |
| // queue does not exist the incoming message is dropped. No need to |
| // waste cycles when we know that the client has terminated. |
| if ( isReplyRequired(aMessage) |
| // Exclude request messages from a Cas Multiplier. We are not replying |
| // to CM. Only CM adds CasSequence to a request message |
| && !messageContext.propertyExists(AsynchAEMessage.CasSequence) |
| // Reply destination must be present |
| && validEndpoint(messageContext) ) { |
| // replace ':' with '_' to enable JMX query to work. ':' is an invalid char for queries |
| String queueName = remoteJMXServer.normalize(messageContext.getEndpoint().getDestination().toString()); |
| // Check if reply queue provided in a message still exists in broker's |
| // JMX Server registry. In case the server is not available the call returns true. |
| return remoteJMXServer.isClientReplyQueueAvailable(queueName); |
| } |
| return true; // Default, PROCESS THE MESSAGE |
| } |
| private boolean ackClient( JmsMessageContext messageContext ) throws Exception { |
| return ( !messageContext.propertyExists(AsynchAEMessage.CasSequence) && |
| messageContext.getMessageStringProperty(UIMAMessage.ServerURI).startsWith("vm") == false && |
| validEndpoint(messageContext) && |
| isReplyRequired((Message)messageContext.getRawMessage()) ); |
| } |
| /** |
| * 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 aJmsSession |
| * - 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 = ""; |
| } |
| } |
| 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 }); |
| } |
| } |
| } |
| |
| private String extractNodeName(String url) { |
| int startPos = url.indexOf("//"); |
| // Strip the protocol |
| String temp = url.substring(startPos+2); |
| int endPos = temp.indexOf(":"); |
| // extract hostname from Broker URL |
| return temp.substring(0, endPos).trim(); |
| } |
| public int getSessionAckMode() { |
| return sessionAckMode; |
| } |
| |
| public String getServerUri() { |
| return brokerURL; |
| } |
| public List<UimaDefaultMessageListenerContainer> registerListener(UimaDefaultMessageListenerContainer messageListener) { |
| List<UimaDefaultMessageListenerContainer> ll = null; |
| if ( listenerMap.containsKey(getController().getComponentName()+getController().hashCode()) ) { |
| ll = listenerMap.get(getController().getComponentName()+getController().hashCode()); |
| } else { |
| ll = new CopyOnWriteArrayList<UimaDefaultMessageListenerContainer>(); |
| listenerMap.put(getController().getComponentName()+getController().hashCode(), ll); |
| } |
| if (!ll.contains(messageListener)) { |
| ll.add(messageListener); |
| } else { |
| } |
| for(Object container: ll ) { |
| UimaDefaultMessageListenerContainer c = (UimaDefaultMessageListenerContainer) container; |
| } |
| return ll; |
| } |
| public List<UimaDefaultMessageListenerContainer> getListeners() { |
| List<UimaDefaultMessageListenerContainer> ll = null; |
| if ( listenerMap.containsKey(getController().getComponentName()+getController().hashCode()) ) { |
| ll = listenerMap.get(getController().getComponentName()+getController().hashCode()); |
| } |
| return ll; |
| } |
| 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); |
| } |
| // Add new message listener to the list |
| registerListener(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) { |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "ackMessage", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", controller.getComponentName()); |
| |
| 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", |
| 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, boolean shutdownNow) 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(), "stopChannel", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_stopping_jms_transport__INFO", |
| new Object[] { eName, shutdownNow }); |
| } |
| } |
| mL.delegateStop(); |
| if (shutdownNow) { |
| mL.destroy(shutdownNow); |
| } |
| |
| String selector = ""; |
| if (mL.getMessageSelector() != null) { |
| selector = " Selector:" + mL.getMessageSelector(); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "stopChannel", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_stopped_jms_transport__INFO", |
| new Object[] {getController().getComponentName(), mL.getDestination(), selector }); |
| } |
| } |
| |
| 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(boolean shutdownNow) throws Exception { |
| stop(InputChannel.CloseAllChannels, shutdownNow); |
| listenerContainerList.clear(); |
| List<UimaDefaultMessageListenerContainer> ll = getListeners(); |
| if ( ll != null ) { |
| ll.clear(); |
| } |
| failedListenerMap.clear(); |
| if ( remoteJMXServer != null ) { |
| remoteJMXServer.disconnect(); |
| remoteJMXServer = null; |
| } |
| } |
| public void disconnectListenersFromQueue() throws Exception { |
| List<UimaDefaultMessageListenerContainer> ll = getListeners(); |
| for (UimaDefaultMessageListenerContainer listenerObject : ll) { |
| stopChannel(listenerObject, false); |
| } |
| |
| } |
| public void setTerminating() { |
| List<UimaDefaultMessageListenerContainer> ll = getListeners(); |
| for (UimaDefaultMessageListenerContainer listenerObject : ll) { |
| listenerObject.setTerminating(); |
| } |
| |
| } |
| public void terminate() { |
| try { |
| List<UimaDefaultMessageListenerContainer> ll = getListeners(); |
| for (UimaDefaultMessageListenerContainer listenerObject : ll) { |
| listenerObject.closeConnection(); |
| } |
| |
| } catch( Exception e) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "terminate", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| |
| } |
| public synchronized void stop(int channelsToClose, boolean shutdownNow) throws Exception { |
| List<UimaDefaultMessageListenerContainer> listenersToRemove = new ArrayList<UimaDefaultMessageListenerContainer>(); |
| List<UimaDefaultMessageListenerContainer> ll = getListeners(); |
| |
| for (UimaDefaultMessageListenerContainer listenerObject : ll) { |
| |
| if (listenerObject != null && doCloseChannel(listenerObject, channelsToClose)) { |
| listenerObject.setRecoveryInterval(0); // https://issues.apache.org/jira/browse/UIMA-3437 |
| listenerObject.setTerminating(); |
| listenerObject.setAutoStartup(false); |
| stopChannel(listenerObject, shutdownNow); |
| |
| } 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() }); |
| } |
| } |
| } |
| } |
| |
| 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); |
| newListener.setAcceptMessagesWhileStopping(true); |
| 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) { |
| newListener.wait(100); |
| } |
| } |
| newListener.afterPropertiesSet(); |
| if ( controller != null && controller.isStopped() ) { |
| newListener.destroy(true); // shutdownNow |
| // 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); |
| } |
| } |
| } |
| public boolean isListenerActiveOnDestination(Destination destination ) { |
| List<UimaDefaultMessageListenerContainer> ll = getListeners(); |
| for (UimaDefaultMessageListenerContainer mListener : ll ) { |
| 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> ll = getListeners(); |
| |
| |
| List<UimaDefaultMessageListenerContainer> listeners = new ArrayList<UimaDefaultMessageListenerContainer>(); |
| for (UimaDefaultMessageListenerContainer mListener : ll) { |
| 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() }); |
| } |
| mListener.stop(); |
| if (getController() != null) { |
| Endpoint endpoint = ((AggregateAnalysisEngineController) getController()).lookUpEndpoint( |
| aDelegateKey, false); |
| endpoint.setStatus(Endpoint.FAILED); |
| List<UimaDefaultMessageListenerContainer> ll = null; |
| if ( listenerMap.containsKey(getController().getComponentName()+getController().hashCode()) ) { |
| ll = listenerMap.get(getController().getComponentName()+getController().hashCode()); |
| } |
| if (mListener.getConnectionFactory() != null) { |
| if (getController() instanceof AggregateAnalysisEngineController) { |
| if (!failedListenerMap.containsKey(aDelegateKey)) { |
| failedListenerMap.put(aDelegateKey, mListener); |
| ll.remove(mListener); |
| } |
| } |
| } |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "destroyListener", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", controller.getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "destroyListener", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", e); |
| } |
| |
| } |
| } |
| } |
| |
| public boolean isFailed(String aDelegateKey) { |
| return failedListenerMap.containsKey(aDelegateKey); |
| } |
| public void removeDelegateFromFailedList( String aDelegateKey ) { |
| if ( failedListenerMap.containsKey(aDelegateKey ) ) { |
| failedListenerMap.remove(aDelegateKey); |
| } |
| } |
| public boolean isListenerForDestination(String anEndpointName) { |
| UimaDefaultMessageListenerContainer[] mListeners = getListenersForEndpoint(anEndpointName); |
| if (mListeners == null) { |
| return false; |
| } |
| return true; |
| } |
| |
| } |
