| /* |
| * 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.net.ConnectException; |
| import java.util.ArrayList; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.ThreadFactory; |
| import java.util.concurrent.TimeUnit; |
| |
| import javax.jms.Connection; |
| import javax.jms.ConnectionFactory; |
| import javax.jms.Destination; |
| import javax.jms.ExceptionListener; |
| import javax.jms.JMSException; |
| import javax.jms.TemporaryQueue; |
| |
| import org.apache.activemq.ActiveMQConnectionFactory; |
| import org.apache.activemq.ActiveMQPrefetchPolicy; |
| import org.apache.activemq.command.ActiveMQDestination; |
| import org.apache.uima.UIMAFramework; |
| import org.apache.uima.aae.InputChannel; |
| import org.apache.uima.aae.UIMAEE_Constants; |
| import org.apache.uima.aae.UimaAsThreadFactory; |
| 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.PrimitiveAnalysisEngineController; |
| import org.apache.uima.aae.controller.BaseAnalysisEngineController.ServiceState; |
| import org.apache.uima.aae.delegate.Delegate; |
| import org.apache.uima.aae.error.ErrorHandler; |
| import org.apache.uima.aae.error.Threshold; |
| import org.apache.uima.aae.error.handler.GetMetaErrorHandler; |
| import org.apache.uima.adapter.jms.JmsConstants; |
| import org.apache.uima.resource.ResourceInitializationException; |
| import org.apache.uima.util.Level; |
| import org.springframework.core.task.TaskExecutor; |
| import org.springframework.jms.JmsException; |
| import org.springframework.jms.listener.DefaultMessageListenerContainer; |
| import org.springframework.jms.support.destination.DestinationResolver; |
| import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; |
| |
| public class UimaDefaultMessageListenerContainer extends DefaultMessageListenerContainer implements |
| ExceptionListener { |
| private static final Class CLASS_NAME = UimaDefaultMessageListenerContainer.class; |
| |
| private String destinationName = ""; |
| |
| private Endpoint endpoint; |
| |
| private volatile boolean freeCasQueueListener; |
| |
| private AnalysisEngineController controller; |
| |
| private volatile boolean failed = false; |
| |
| private Object mux = new Object(); |
| |
| private final UimaDefaultMessageListenerContainer __listenerRef; |
| |
| private TaskExecutor taskExecutor = null; |
| |
| private ConnectionFactory connectionFactory = null; |
| |
| private Object mux2 = new Object(); |
| |
| private ThreadGroup threadGroup = null; |
| |
| private ThreadFactory tf = null; |
| |
| // stores number of consumer threads |
| private int cc = 0; |
| |
| // stores message listener plugged in by Spring |
| private Object ml = null; |
| |
| // A new listener will be injected between |
| // spring and JmsInputChannel Pojo Listener. This |
| // listener purpose is to increment number of children for |
| // an input CAS. |
| private ConcurrentMessageListener concurrentListener = null; |
| |
| private volatile boolean awaitingShutdown = false; |
| |
| public UimaDefaultMessageListenerContainer() { |
| super(); |
| UIMAFramework.getLogger(CLASS_NAME).setLevel(Level.WARNING); |
| __listenerRef = this; |
| setRecoveryInterval(5); |
| setAcceptMessagesWhileStopping(false); |
| setExceptionListener(this); |
| threadGroup = new ThreadGroup("ListenerThreadGroup_" |
| + Thread.currentThread().getThreadGroup().getName()); |
| } |
| |
| public UimaDefaultMessageListenerContainer(boolean freeCasQueueListener) { |
| this(); |
| this.freeCasQueueListener = freeCasQueueListener; |
| } |
| |
| public void setController(AnalysisEngineController aController) { |
| controller = aController; |
| } |
| |
| /** |
| * |
| * @param t |
| * @return |
| */ |
| private boolean disableListener(Throwable t) { |
| System.out.println(t.toString()); |
| if (t.toString().indexOf("SharedConnectionNotInitializedException") > 0 |
| || (t instanceof JMSException && t.getCause() != null && t.getCause() instanceof ConnectException)) |
| return true; |
| return false; |
| } |
| |
| /** |
| * Stops this Listener |
| */ |
| private void handleListenerFailure() { |
| // If shutdown already, nothing to do |
| if (awaitingShutdown) { |
| return; |
| } |
| try { |
| if (controller instanceof AggregateAnalysisEngineController) { |
| String delegateKey = ((AggregateAnalysisEngineController) controller) |
| .lookUpDelegateKey(endpoint.getEndpoint()); |
| InputChannel iC = null; |
| String queueName = null; |
| if (endpoint.getDestination() != null) { |
| queueName = endpoint.getDestination().toString(); |
| } else { |
| queueName = endpoint.getEndpoint(); |
| } |
| iC = ((AggregateAnalysisEngineController) controller).getInputChannel(queueName); |
| if (iC != null) { |
| iC.destroyListener(queueName, delegateKey); |
| } else { |
| System.out.println(">>> Listener Unable To LookUp InputChannel For Queue:" + queueName); |
| } |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "handleListenerFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| } |
| |
| /** |
| * Handles failure on a temp queue |
| * |
| * @param t |
| */ |
| private void handleTempQueueFailure(Throwable t) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(), |
| "handleTempQueueFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_jms_listener_failed_WARNING", |
| new Object[] { endpoint.getDestination(), getBrokerUrl(), t }); |
| } |
| // Check if the failure is due to the failed connection. Spring (and ActiveMQ) dont seem to |
| // provide |
| // the cause. Just the top level IllegalStateException with a text message. This is what we need |
| // to |
| // check for. |
| if (t instanceof javax.jms.IllegalStateException |
| && t.getMessage().equals("The Consumer is closed")) { |
| if (controller != null && controller instanceof AggregateAnalysisEngineController) { |
| String delegateKey = ((AggregateAnalysisEngineController) controller) |
| .lookUpDelegateKey(endpoint.getEndpoint()); |
| try { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| this.getClass().getName(), |
| "handleTempQueueFailure", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_stopping_listener_INFO", |
| new Object[] { controller.getComponentName(), endpoint.getDestination(), |
| delegateKey }); |
| } |
| // Stop current listener |
| handleListenerFailure(); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, this.getClass().getName(), |
| "handleTempQueueFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_stopped_listener_INFO", |
| new Object[] { controller.getComponentName(), endpoint.getDestination() }); |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "handleTempQueueFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| } |
| } else if (disableListener(t)) { |
| handleQueueFailure(t); |
| } |
| } |
| |
| private ErrorHandler fetchGetMetaErrorHandler() { |
| ErrorHandler handler = null; |
| Iterator it = controller.getErrorHandlerChain().iterator(); |
| // Find the error handler for GetMeta in the Error Handler List provided in the |
| // deployment descriptor |
| while (it.hasNext()) { |
| handler = (ErrorHandler) it.next(); |
| if (handler instanceof GetMetaErrorHandler) { |
| return handler; |
| } |
| } |
| return null; |
| } |
| |
| /** |
| * Handles failures on non-temp queues |
| * |
| * @param t |
| */ |
| private void handleQueueFailure(Throwable t) { |
| final String endpointName = (getDestination() == null) ? "" |
| : ((ActiveMQDestination) getDestination()).getPhysicalName(); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(), |
| "handleQueueFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_jms_listener_failed_WARNING", |
| new Object[] { endpointName, getBrokerUrl(), t }); |
| } |
| boolean terminate = true; |
| // Check if the failure is severe enough to disable this listener. Whether or not this listener |
| // is actully |
| // disabled depends on the action associated with GetMeta Error Handler. If GetMeta Error |
| // Handler is |
| // configured to terminate the service on failure, this listener will be terminated and the |
| // entire service |
| // will be stopped. |
| if (disableListener(t)) { |
| endpoint.setReplyDestinationFailed(); |
| // If this is a listener attached to the Aggregate Controller, use GetMeta Error |
| // Thresholds defined to determine what to do next after failure. Either terminate |
| // the service or disable the delegate with which this listener is associated with |
| if (controller != null && controller instanceof AggregateAnalysisEngineController) { |
| ErrorHandler handler = fetchGetMetaErrorHandler(); |
| // Fetch a Map containing thresholds for GetMeta for each delegate. |
| Map thresholds = handler.getEndpointThresholdMap(); |
| // Lookup delegate's key using delegate's endpoint name |
| String delegateKey = ((AggregateAnalysisEngineController) controller) |
| .lookUpDelegateKey(endpoint.getEndpoint()); |
| // If the delegate has a threshold defined on GetMeta apply Action defined |
| if (delegateKey != null && thresholds.containsKey(delegateKey)) { |
| // Fetch the Threshold object containing error configuration |
| Threshold threshold = (Threshold) thresholds.get(delegateKey); |
| // Check if the delegate needs to be disabled |
| if (threshold.getAction().equalsIgnoreCase(ErrorHandler.DISABLE)) { |
| // The disable delegate method takes a list of delegates |
| List list = new ArrayList(); |
| // Add the delegate to disable to the list |
| list.add(delegateKey); |
| try { |
| System.out.println(">>>> Controller:" + controller.getComponentName() |
| + " Disabling Listener On Queue:" + endpoint.getEndpoint() + ". Component's " |
| + delegateKey + " Broker:" + getBrokerUrl() + " is Invalid"); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME) |
| .logrb( |
| Level.INFO, |
| this.getClass().getName(), |
| "handleQueueFailure", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_disabled_delegate_bad_broker__INFO", |
| new Object[] { controller.getComponentName(), delegateKey, |
| getBrokerUrl() }); |
| } |
| // Remove the delegate from the routing table. |
| ((AggregateAnalysisEngineController) controller).disableDelegates(list); |
| terminate = false; // just disable the delegate and continue |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "handleQueueFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| terminate = true; |
| } |
| } |
| } |
| } |
| } |
| System.out.println("****** Unable To Connect Listener To Broker:" + getBrokerUrl()); |
| System.out.println("****** Closing Listener on Queue:" + endpoint.getEndpoint()); |
| setRecoveryInterval(0); |
| |
| // Spin a shutdown thread to terminate listener. |
| new Thread() { |
| public void run() { |
| try { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(), |
| "handleQueueFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_disable_listener__WARNING", |
| new Object[] { endpointName, getBrokerUrl() }); |
| } |
| shutdown(); |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "handleQueueFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| } |
| }.start(); |
| |
| if (terminate) { |
| terminate(t); |
| } |
| |
| } |
| |
| /** |
| * This method is called by Spring when a listener fails |
| */ |
| protected void handleListenerSetupFailure(Throwable t, boolean alreadyHandled) { |
| // If shutdown already, nothing to do |
| if (awaitingShutdown) { |
| return; |
| } |
| // If controller is stopping not need to recover the connection |
| if (controller != null && controller.isStopped()) { |
| return; |
| } |
| if ( controller != null ) { |
| controller.changeState(ServiceState.FAILED); |
| } |
| if (endpoint == null) { |
| super.handleListenerSetupFailure(t, true); |
| String controllerId = ""; |
| if (controller != null) { |
| controllerId = "Uima AS Service:" + controller.getComponentName(); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(), |
| "handleListenerSetupFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_listener_connection_failure__WARNING", |
| new Object[] { controllerId, getBrokerUrl() }); |
| } |
| System.out.println(controllerId + " Listener Unable to Connect to Broker:" + getBrokerUrl() |
| + " Retrying ...."); |
| // This code executes during initialization of the service. The Endpoint is not yet |
| // available. The connection to a broker cannot be established. Keep trying until |
| // the broker becomes available. |
| refreshConnectionUntilSuccessful(); |
| if ( controller != null ) { |
| controller.changeState(ServiceState.RUNNING); |
| } |
| System.out.println(controllerId + " Listener Established Connection to Broker:" |
| + getBrokerUrl()); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(), |
| "handleListenerSetupFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_listener_connection_recovered__WARNING", |
| new Object[] { controllerId, getBrokerUrl() }); |
| } |
| return; |
| } |
| |
| // Connection failure that occurs AFTER the service initialized. |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "handleListenerSetupFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), t }); |
| } |
| |
| synchronized (mux) { |
| if (!failed) { |
| // Check if this listener is attached to a temp queue. If so, this is a listener |
| // on a reply queue. Handle temp queue listener failure differently than an |
| // input queue listener. |
| if (endpoint.isTempReplyDestination()) { |
| handleTempQueueFailure(t); |
| } else { |
| // Handle non-temp queue failure |
| handleQueueFailure(t); |
| } |
| } |
| failed = true; |
| } |
| } |
| |
| private void terminate(Throwable t) { |
| // **************************************** |
| // terminate the service |
| // **************************************** |
| System.out.println(">>>> Terminating Controller:" + controller.getComponentName() |
| + " Unable To Initialize Listener Due to Invalid Broker URL:" + getBrokerUrl()); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(), |
| "terminate", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_terminate_service_dueto_bad_broker__WARNING", |
| new Object[] { controller.getComponentName(), getBrokerUrl() }); |
| } |
| controller.notifyListenersWithInitializationStatus(new ResourceInitializationException(t)); |
| if (!controller.isStopped() && !controller.isAwaitingCacheCallbackNotification()) { |
| controller.stop(); |
| } |
| } |
| |
| protected void handleListenerException(Throwable t) { |
| // Already shutdown, nothing to do |
| if (awaitingShutdown) { |
| return; |
| } |
| String endpointName = (getDestination() == null) ? "" |
| : ((ActiveMQDestination) getDestination()).getPhysicalName(); |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(), |
| "handleListenerException", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_jms_listener_failed_WARNING", |
| new Object[] { endpointName, getBrokerUrl(), t }); |
| } |
| super.handleListenerException(t); |
| } |
| |
| private void allPropertiesSet() { |
| super.afterPropertiesSet(); |
| } |
| |
| private void injectConnectionFactory() { |
| while (connectionFactory == null) { |
| try { |
| Thread.sleep(50); |
| } catch (Exception e) { |
| } |
| } |
| super.setConnectionFactory(connectionFactory); |
| } |
| |
| private void injectTaskExecutor() { |
| super.setTaskExecutor(taskExecutor); |
| } |
| |
| private boolean isGetMetaListener() { |
| return getMessageSelector() != null |
| && __listenerRef.getMessageSelector().equals("Command=2001"); |
| } |
| |
| private boolean isActiveMQDestination() { |
| return getDestination() != null && getDestination() instanceof ActiveMQDestination; |
| } |
| |
| public void initializeContainer() { |
| try { |
| |
| injectConnectionFactory(); |
| initializeTaskExecutor(); |
| injectTaskExecutor(); |
| super.initialize(); |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "initializeContainer", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| } |
| |
| /** |
| * Intercept Spring call to increment number of consumer threads. If the value > 1, don't |
| * propagate to Spring. A new listener will be injected and it will use provided number of |
| * consumer threads. |
| **/ |
| public void setConcurrentConsumers(int concurrentConsumers) { |
| cc = concurrentConsumers; |
| if (this.freeCasQueueListener) { |
| super.setConcurrentConsumers(concurrentConsumers); |
| } |
| } |
| |
| /** |
| * Intercept Spring call to inject application Pojo listener. Don't propagate the listener up to |
| * Spring just yet. If more than one consumer thread is used, a different listener will be |
| * injected. |
| **/ |
| public void setMessageListener(Object messageListener) { |
| ml = messageListener; |
| if (this.freeCasQueueListener) { |
| super.setMessageListener(messageListener); |
| } |
| } |
| |
| /** |
| * Called by Spring and some Uima AS components when all properties have been set. This method |
| * spins a thread in which the listener is initialized. |
| */ |
| public void afterPropertiesSet() { |
| if (endpoint != null) { |
| // Endpoint has been plugged in from spring xml. This means this is a listener |
| // for a reply queue. We need to rewire things a bit. First make Spring use |
| // one thread to make sure we receive messages in order. To fix a race condition |
| // where a parent CAS is processed first instead of its last child, we need to |
| // assure that we get the child first. We need to update the counter of the |
| // parent CAS to reflect that there is another child. In the race condition that |
| // was observed, the parent was being processed first in one thread. The parent |
| // reached the final step and subsequently was dropped. Subsequent to that, a |
| // child CAS processed on another thread begun executing and failed since a look |
| // on its parent resulted in CAS Not Found In Cache Exception. |
| // Make sure Spring uses one thread |
| super.setConcurrentConsumers(1); |
| if (cc > 1) { |
| try { |
| concurrentListener = new ConcurrentMessageListener(cc, ml); |
| super.setMessageListener(concurrentListener); |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "afterPropertiesSet", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| return; |
| } |
| } else { |
| super.setMessageListener(ml); |
| } |
| } else { |
| super.setMessageListener(ml); |
| super.setConcurrentConsumers(cc); |
| } |
| Thread t = new Thread(threadGroup, new Runnable() { |
| public void run() { |
| Destination destination = __listenerRef.getDestination(); |
| try { |
| // Wait until the connection factory is injected by Spring |
| while (connectionFactory == null) { |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException ex) { |
| } |
| } |
| System.setProperty("BrokerURI", ((ActiveMQConnectionFactory) connectionFactory) |
| .getBrokerURL()); |
| boolean done = false; |
| // Wait for controller to be injected by Uima AS |
| if (isActiveMQDestination() && !isGetMetaListener() |
| && !((ActiveMQDestination) destination).isTemporary()) { |
| // Add self to InputChannel |
| connectWithInputChannel(); |
| // Wait for InputChannel to plug in a controller |
| done = true; |
| while (controller == null) |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException ex) { |
| } |
| ; |
| } |
| // Plug in connection Factory to Spring's Listener |
| __listenerRef.injectConnectionFactory(); |
| // Initialize the TaskExecutor. This call injects a custom Thread Pool into the |
| // TaskExecutor provided in the spring xml. The custom thread pool initializes |
| // an instance of AE in a dedicated thread |
| initializeTaskExecutor(); |
| // Plug in TaskExecutor to Spring's Listener |
| __listenerRef.injectTaskExecutor(); |
| // Notify Spring Listener that all properties are ready |
| __listenerRef.allPropertiesSet(); |
| if (isActiveMQDestination() && destination != null) { |
| destinationName = ((ActiveMQDestination) destination).getPhysicalName(); |
| } |
| if (!done) { |
| connectWithInputChannel(); |
| done = true; |
| } |
| if (concurrentListener != null) { |
| concurrentListener.setAnalysisEngineController(controller); |
| } |
| // Save number of concurrent consumers on the temp reply queue in case we need to |
| // recreate a new listener on a new temp queue created during recovery |
| if (endpoint != null && controller instanceof AggregateAnalysisEngineController) { |
| Delegate delegate = ((AggregateAnalysisEngineController) controller) |
| .lookupDelegate(endpoint.getDelegateKey()); |
| if (delegate != null) { |
| delegate.getEndpoint().setConcurrentReplyConsumers(cc); |
| } |
| } |
| // Show ready message on the console only if this listener is *not* listening |
| // on an input queue. Input queue listeners are not started until the service |
| // is fully initialized |
| if (__listenerRef.getMessageListener() == null && getDestination() != null) { |
| System.out.println("Service:" + controller.getComponentName() |
| + " Listener Ready. Broker:" + getBrokerUrl() + " Queue:" + getDestination()); |
| } |
| |
| } catch (Exception e) { |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(), |
| "afterPropertiesSet", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_jms_listener_failed_WARNING", |
| new Object[] { destination, getBrokerUrl(), e }); |
| } |
| } |
| }); |
| t.start(); |
| } |
| |
| /** |
| * Inject instance of this listener into the InputChannel |
| * |
| * @throws Exception |
| */ |
| private void connectWithInputChannel() throws Exception { |
| Object pojoListener = getPojoListener(); |
| |
| if (pojoListener instanceof JmsInputChannel) { |
| // Wait until InputChannel has a valid controller. The controller will be plug in |
| // by Spring on a different thread |
| while ((((JmsInputChannel) pojoListener).getController()) == null) { |
| try { |
| Thread.currentThread().sleep(50); |
| } catch (Exception e) { |
| } |
| } |
| ((JmsInputChannel) pojoListener).setListenerContainer(__listenerRef); |
| } else if (pojoListener instanceof ModifiableListener) { |
| ((ModifiableListener) pojoListener).setListener(__listenerRef); |
| } |
| } |
| |
| public String getDestinationName() { |
| |
| return destinationName; |
| } |
| |
| public String getEndpointName() { |
| if (getDestination() != null) { |
| return ((ActiveMQDestination) getDestination()).getPhysicalName(); |
| } |
| return null; |
| } |
| |
| public String getBrokerUrl() { |
| return ((ActiveMQConnectionFactory) connectionFactory).getBrokerURL(); |
| } |
| |
| /* |
| * Overrides specified Connection Factory. Need to append maxInactivityDuration=0 to the broker |
| * URL. The Connection Factory is immutable thus we need to intercept the one provided in the |
| * deployment descriptor and create a new one with rewritten Broker URL. We will inject the |
| * prefetch policy to the new CF based on what is found in the CF in the deployment descriptor. |
| */ |
| |
| public void setConnectionFactory(ConnectionFactory aConnectionFactory) { |
| connectionFactory = aConnectionFactory; |
| super.setConnectionFactory(connectionFactory); |
| } |
| |
| public void setDestinationResolver(DestinationResolver resolver) { |
| ((TempDestinationResolver) resolver).setListener(this); |
| super.setDestinationResolver(resolver); |
| } |
| |
| public void closeConnection() throws Exception { |
| try { |
| setRecoveryInterval(0); |
| setAcceptMessagesWhileStopping(false); |
| setAutoStartup(false); |
| getSharedConnection().close(); |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(), |
| "closeConnection", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { Thread.currentThread().getId(), e }); |
| } |
| } |
| } |
| |
| public void setDestination(Destination aDestination) { |
| super.setDestination(aDestination); |
| if (endpoint != null) { |
| endpoint.setDestination(aDestination); |
| if (aDestination instanceof TemporaryQueue) { |
| endpoint.setTempReplyDestination(true); |
| Object pojoListener = getPojoListener(); |
| if (pojoListener != null && pojoListener instanceof InputChannel) { |
| ((JmsInputChannel) pojoListener).setListenerContainer(this); |
| } |
| } |
| endpoint.setServerURI(getBrokerUrl()); |
| } |
| } |
| |
| private Object getPojoListener() { |
| Object pojoListener = null; |
| if (ml != null) { |
| pojoListener = ml; |
| } else if (getMessageListener() != null) { |
| pojoListener = getMessageListener(); |
| } |
| return pojoListener; |
| } |
| |
| public Destination getListenerEndpoint() { |
| return getDestination(); |
| } |
| |
| public void onException(JMSException arg0) { |
| if (awaitingShutdown) { |
| return; |
| } |
| String endpointName = (getDestination() == null) ? "" |
| : ((ActiveMQDestination) getDestination()).getPhysicalName(); |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(), |
| "onException", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_jms_listener_failed_WARNING", |
| new Object[] { endpointName, getBrokerUrl(), arg0 }); |
| } |
| } |
| |
| public void setTargetEndpoint(Endpoint anEndpoint) { |
| endpoint = anEndpoint; |
| } |
| |
| public boolean isFreeCasQueueListener() { |
| return freeCasQueueListener; |
| } |
| |
| protected void setModifiedTaskExecutor(TaskExecutor taskExecutor) { |
| super.setTaskExecutor(taskExecutor); |
| System.out.println("Injected Updated Task Executor Into Listener For Destination:" |
| + getDestination()); |
| } |
| |
| /** |
| * Delegate shutdown to the super class |
| */ |
| public void doDestroy() { |
| super.destroy(); |
| } |
| public void setMessageSelector( String messageSelector) { |
| super.setMessageSelector(messageSelector); |
| // turn off auto startup. Selectors are only used on input queues. We dont |
| // want listeners on this queue to start now. Once the service initializes |
| // we will start listeners on input queue. |
| this.setAutoStartup(false); |
| } |
| /** |
| * Spins a shutdown thread and stops Sprint and ActiveMQ threads. |
| * |
| */ |
| public void destroy() { |
| if (awaitingShutdown) { |
| return; |
| } |
| awaitingShutdown = true; |
| // Spin a thread that will wait until all threads complete. This is needed to avoid |
| // memory leak caused by the fact that we did not wait to collect the threads |
| Thread threadGroupDestroyer = new Thread(threadGroup.getParent().getParent(), |
| "threadGroupDestroyer") { |
| public void run() { |
| try { |
| // stop Spring listener and ActiveMQ threads |
| __listenerRef.stop(); |
| __listenerRef.closeConnection(); |
| } catch (Exception e) { |
| } |
| // If using non-default TaskExecutor, stop its threads |
| if (taskExecutor != null && taskExecutor instanceof ThreadPoolTaskExecutor) { |
| ((ThreadPoolTaskExecutor) taskExecutor).getThreadPoolExecutor().shutdown(); |
| // Since the calling thread may be one of those managed by the executor allow |
| // for one open thread when checking active thread count. |
| while (((ThreadPoolTaskExecutor) taskExecutor).getThreadPoolExecutor().getActiveCount() > 1 |
| && !((ThreadPoolTaskExecutor) taskExecutor).getThreadPoolExecutor() |
| .isTerminated()) { |
| try { |
| ((ThreadPoolTaskExecutor) taskExecutor).getThreadPoolExecutor().awaitTermination(200, |
| TimeUnit.MILLISECONDS); |
| } catch (Exception e) { |
| } |
| } |
| } else if (concurrentListener != null) { |
| // Stop internal Executor |
| concurrentListener.stop(); |
| } |
| // if ( taskExecutor != null ) { |
| // System.out.println(">>>>> Thread:"+Thread.currentThread().getId()+ |
| // " +++++++++ Listener:"+getDestination()+" Controller ThreadPoolExecutor Stopped ..."); |
| // } |
| // Shutdown the listener |
| __listenerRef.shutdown(); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) { |
| threadGroup.getParent().list(); |
| } |
| // System.out.println(">>>>> Thread:"+Thread.currentThread().getId()+ |
| // " ThreadGroupDestroyer waiting for threads to stop. Active thread count:"+threadGroup.activeCount()+" Active thread group count:"+threadGroup.activeGroupCount()); |
| // Wait until all threads are accounted for |
| while (threadGroup.activeCount() > 0) { |
| try { |
| Thread[] threads = new Thread[threadGroup.activeCount()]; |
| threadGroup.enumerate(threads); |
| boolean foundExpectedThreads = true; |
| |
| for (Thread t : threads) { |
| try { |
| String tName = t.getName(); |
| // The following is necessary to account for the AMQ threads |
| // Any threads not named in the list below will cause a wait |
| // and retry until all non-amq threads are stopped |
| if (!tName.startsWith("main") && !tName.equalsIgnoreCase("timer-0") |
| && !tName.equals("ReaderThread") && !tName.equals("BrokerThreadGroup") |
| && !tName.startsWith("ActiveMQ")) { |
| foundExpectedThreads = false; |
| break; // from for |
| } |
| } catch (Exception e) { |
| } |
| } |
| if (foundExpectedThreads) { |
| break; // from while |
| } |
| Thread.sleep(100); |
| } catch (InterruptedException e) { |
| } |
| } |
| // System.out.println(">>>>> Thread:"+Thread.currentThread().getId()+ |
| // " ThreadGroupDestroyer all threads stopped"); |
| |
| try { |
| synchronized (threadGroup) { |
| if (!threadGroup.isDestroyed()) { |
| threadGroup.destroy(); |
| } |
| } |
| // System.out.println(">>>>> Thread:"+Thread.currentThread().getId()+ |
| // " >>>>>>>>>>>> Listener:"+getDestinationName()+" Thread Group Destroyed"); |
| } catch (Exception e) { |
| } // Ignore |
| } |
| }; |
| threadGroupDestroyer.start(); |
| } |
| |
| /** |
| * Called by Spring to inject TaskExecutor |
| */ |
| public void setTaskExecutor(TaskExecutor aTaskExecutor) { |
| taskExecutor = aTaskExecutor; |
| } |
| |
| /** |
| * This method initializes ThreadPoolExecutor with a custom ThreadPool. Each thread produced by |
| * the ThreadPool is used to first initialize an instance of the AE before the thread is added to |
| * the pool. From this point on, a thread used to initialize the AE will also be used to call this |
| * AE's process() method. |
| * |
| * @throws Exception |
| */ |
| private void initializeTaskExecutor() throws Exception { |
| // TaskExecutor is only used with primitives |
| if (controller instanceof PrimitiveAnalysisEngineController) { |
| // in case the taskExecutor is not plugged in yet, wait until one |
| // becomes available. The TaskExecutor is plugged in by Spring |
| synchronized (mux2) { |
| while (taskExecutor == null) { |
| mux2.wait(20); |
| } |
| } |
| // Create a Custom Thread Factory. Provide it with an instance of |
| // PrimitiveController so that every thread can call it to initialize |
| // the next available instance of a AE. |
| tf = new UimaAsThreadFactory(threadGroup, (PrimitiveAnalysisEngineController) controller); |
| // This ThreadExecutor will use custom thread factory instead of defult one |
| ((ThreadPoolTaskExecutor) taskExecutor).setThreadFactory(tf); |
| // Initialize the thread pool |
| ((ThreadPoolTaskExecutor) taskExecutor).initialize(); |
| // Make sure all threads are started. This forces each thread to call |
| // PrimitiveController to initialize the next instance of AE |
| ((ThreadPoolTaskExecutor) taskExecutor).getThreadPoolExecutor().prestartAllCoreThreads(); |
| // Change the state of a collocated service |
| if ( !controller.isTopLevelComponent() ) { |
| controller.changeState(ServiceState.RUNNING); |
| } |
| } |
| } |
| |
| public void stop() throws JmsException { |
| setAcceptMessagesWhileStopping(false); |
| destroy(); |
| } |
| } |
