| /* |
| * 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.client; |
| |
| import java.io.File; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.concurrent.Semaphore; |
| |
| import javax.jms.BytesMessage; |
| import javax.jms.Connection; |
| import javax.jms.Destination; |
| import javax.jms.JMSException; |
| import javax.jms.Message; |
| import javax.jms.MessageListener; |
| import javax.jms.MessageProducer; |
| import javax.jms.Queue; |
| import javax.jms.Session; |
| import javax.jms.TextMessage; |
| import javax.management.ObjectName; |
| import javax.naming.Context; |
| import javax.naming.InitialContext; |
| |
| import org.apache.activemq.ActiveMQConnection; |
| import org.apache.activemq.ActiveMQConnectionFactory; |
| import org.apache.activemq.ActiveMQPrefetchPolicy; |
| import org.apache.activemq.RedeliveryPolicy; |
| import org.apache.activemq.command.ActiveMQBytesMessage; |
| import org.apache.activemq.command.ActiveMQDestination; |
| import org.apache.activemq.command.ActiveMQTextMessage; |
| import org.apache.uima.UIMAFramework; |
| import org.apache.uima.UIMA_IllegalArgumentException; |
| import org.apache.uima.UIMA_IllegalStateException; |
| import org.apache.uima.aae.AsynchAECasManager_impl; |
| import org.apache.uima.aae.UIMAEE_Constants; |
| import org.apache.uima.aae.UimaAsVersion; |
| import org.apache.uima.aae.client.UimaASStatusCallbackListener; |
| import org.apache.uima.aae.client.UimaAsynchronousEngine; |
| import org.apache.uima.aae.controller.AnalysisEngineController; |
| import org.apache.uima.aae.controller.ControllerCallbackListener; |
| import org.apache.uima.aae.controller.ControllerLifecycle; |
| import org.apache.uima.aae.controller.Endpoint; |
| import org.apache.uima.aae.controller.UimacppServiceController; |
| import org.apache.uima.aae.delegate.Delegate; |
| import org.apache.uima.aae.delegate.Delegate.DelegateEntry; |
| import org.apache.uima.aae.error.AsynchAEException; |
| import org.apache.uima.aae.error.UimaASMetaRequestTimeout; |
| import org.apache.uima.aae.jmx.JmxManager; |
| import org.apache.uima.aae.message.AsynchAEMessage; |
| import org.apache.uima.aae.message.UIMAMessage; |
| import org.apache.uima.adapter.jms.JmsConstants; |
| import org.apache.uima.adapter.jms.activemq.SpringContainerDeployer; |
| import org.apache.uima.adapter.jms.activemq.UimaEEAdminSpringContext; |
| import org.apache.uima.adapter.jms.client.BaseUIMAAsynchronousEngineCommon_impl.ClientRequest; |
| import org.apache.uima.adapter.jms.service.Dd2spring; |
| import org.apache.uima.analysis_engine.AnalysisEngineDescription; |
| import org.apache.uima.analysis_engine.metadata.AnalysisEngineMetaData; |
| import org.apache.uima.cas.CAS; |
| import org.apache.uima.internal.util.UUIDGenerator; |
| import org.apache.uima.resource.Resource; |
| import org.apache.uima.resource.ResourceConfigurationException; |
| import org.apache.uima.resource.ResourceInitializationException; |
| import org.apache.uima.resource.ResourceManager; |
| import org.apache.uima.resource.ResourceProcessException; |
| import org.apache.uima.util.Level; |
| import org.springframework.context.ApplicationContext; |
| import org.springframework.context.support.FileSystemXmlApplicationContext; |
| |
| public class BaseUIMAAsynchronousEngine_impl extends BaseUIMAAsynchronousEngineCommon_impl |
| implements UimaAsynchronousEngine, MessageListener, ControllerCallbackListener { |
| private static final Class CLASS_NAME = BaseUIMAAsynchronousEngine_impl.class; |
| |
| private MessageSender sender = null; |
| |
| private MessageProducer producer; |
| |
| private String brokerURI = null; |
| |
| private Session session = null; |
| |
| private Session consumerSession = null; |
| |
| private volatile boolean serviceInitializationException; |
| |
| private volatile boolean serviceInitializationCompleted; |
| |
| private Semaphore serviceSemaphore = new Semaphore(1); |
| |
| private Queue consumerDestination = null; |
| |
| private Session producerSession = null; |
| |
| private JmxManager jmxManager = null; |
| |
| private String applicationName = "UimaASClient"; |
| |
| |
| protected static Semaphore sharedConnectionSemaphore = new Semaphore(1); |
| |
| private Object stopMux = new Object(); |
| |
| private static final UimaAsVersion uimaAsVersion = new UimaAsVersion(); |
| |
| protected InitialContext jndiContext; |
| |
| private ObjectName clientJmxObjectName = null; |
| public BaseUIMAAsynchronousEngine_impl() { |
| UIMAFramework.getLogger(CLASS_NAME).log(Level.INFO, |
| "UIMA-AS version " + UIMAFramework.getVersionString()); |
| } |
| |
| protected TextMessage createTextMessage() throws ResourceInitializationException { |
| return new ActiveMQTextMessage(); |
| } |
| |
| protected BytesMessage createBytesMessage() throws ResourceInitializationException { |
| return new ActiveMQBytesMessage(); |
| } |
| |
| /** |
| * Called at the end of collectionProcessingComplete - WAS closes receiving thread here |
| */ |
| protected void cleanup() throws Exception { |
| } |
| |
| /** |
| * Return a name of the queue to which the JMS Producer is connected to. |
| */ |
| public String getEndPointName() throws ResourceProcessException { |
| try { |
| return clientSideJmxStats.getEndpointName(); |
| } catch (Exception e) { |
| throw new ResourceProcessException(e); |
| } |
| } |
| |
| protected void setMetaRequestMessage(Message msg) throws Exception { |
| msg.setStringProperty(AsynchAEMessage.MessageFrom, consumerDestination.getQueueName()); |
| |
| msg.setStringProperty(UIMAMessage.ServerURI, brokerURI); |
| msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.GetMeta); |
| msg.setJMSReplyTo(consumerDestination); |
| if (msg instanceof TextMessage) { |
| ((ActiveMQTextMessage) msg).setText(""); |
| } |
| } |
| |
| /** |
| * Initialize JMS Message with properties relevant to Process CAS request. |
| */ |
| protected void setCASMessage(String aCasReferenceId, CAS aCAS, Message msg) |
| throws ResourceProcessException { |
| try { |
| setCommonProperties(aCasReferenceId, msg, "xmi"); |
| ((TextMessage) msg).setText(serializeCAS(aCAS)); |
| } catch (Exception e) { |
| throw new ResourceProcessException(e); |
| } |
| } |
| |
| protected void setCASMessage(String aCasReferenceId, String aSerializedCAS, Message msg) |
| throws ResourceProcessException { |
| try { |
| setCommonProperties(aCasReferenceId, msg, "xmi"); |
| ((TextMessage) msg).setText(aSerializedCAS); |
| } catch (Exception e) { |
| throw new ResourceProcessException(e); |
| } |
| } |
| |
| protected void setCASMessage(String aCasReferenceId, byte[] aSerializedCAS, Message msg) |
| throws ResourceProcessException { |
| try { |
| setCommonProperties(aCasReferenceId, msg, "binary"); |
| ((BytesMessage) msg).writeBytes(aSerializedCAS); |
| } catch (Exception e) { |
| throw new ResourceProcessException(e); |
| } |
| } |
| |
| protected void setCommonProperties(String aCasReferenceId, Message msg, |
| String aSerializationStrategy) throws ResourceProcessException { |
| try { |
| msg.setStringProperty(AsynchAEMessage.MessageFrom, consumerDestination.getQueueName()); |
| |
| msg.setStringProperty(UIMAMessage.ServerURI, brokerURI); |
| msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.Process); |
| msg.setStringProperty(AsynchAEMessage.CasReference, aCasReferenceId); |
| |
| if (aSerializationStrategy.equals("binary")) { |
| msg.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.BinaryPayload); |
| } else if (aSerializationStrategy.equals("xmi")) { |
| msg.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.XMIPayload); |
| } |
| |
| msg.setBooleanProperty(AsynchAEMessage.AcceptsDeltaCas, true); |
| msg.setJMSReplyTo(consumerDestination); |
| |
| } catch (Exception e) { |
| throw new ResourceProcessException(e); |
| } |
| |
| } |
| |
| public void stop() { |
| if (!running) { |
| return; |
| } |
| super.stop(); |
| synchronized (stopMux) { |
| running = false; |
| if (super.serviceDelegate != null) { |
| // Cancel all timers and purge lists |
| super.serviceDelegate.cleanup(); |
| } |
| try { |
| // SharedConnection object manages a single JMS connection to the |
| // broker. If the client is scaled out in the same JVM, the connection |
| // is shared by all instances of the client to reduce number of threads |
| // in the broker. The SharedConnection object also maintains the number |
| // of client instances to determine when it is ok to close the connection. |
| // The connection is closed when the last client calls stop(). |
| try { |
| sharedConnectionSemaphore.acquire(); |
| if (sharedConnection != null) { |
| // Remove a client from registry |
| sharedConnection.unregisterClient(this); |
| // The destroy method closes the JMS connection when the number of |
| // clients becomes 0, otherwise it is a no-op |
| if (sharedConnection.destroy()) { |
| // This needs to be done to invalidate the object for JUnit tests |
| sharedConnection = null; |
| } |
| } |
| } catch (InterruptedException ex) { |
| System.out |
| .println("UIMA AS Client Interrupted While Acquiring sharedConnectioSemaphore to Close Shared Connection"); |
| } finally { |
| sharedConnectionSemaphore.release(); |
| } |
| if (sender != null) { |
| sender.doStop(); |
| } |
| // Undeploy all containers |
| undeploy(); |
| System.out.println("UIMA AS Client Undeployed All Containers"); |
| if (initialized) { |
| try { |
| consumerSession.close(); |
| consumer.close(); |
| } catch (JMSException exx) { |
| } |
| } |
| // unregister client |
| if (jmxManager != null) { |
| jmxManager.unregisterMBean(clientJmxObjectName); |
| jmxManager.destroy(); |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "stop", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| } |
| } |
| |
| public void setCPCMessage(Message msg) throws Exception { |
| msg.setStringProperty(AsynchAEMessage.MessageFrom, consumerDestination.getQueueName()); |
| msg.setStringProperty(UIMAMessage.ServerURI, brokerURI); |
| msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.CollectionProcessComplete); |
| msg.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.None); |
| msg.setBooleanProperty(AsynchAEMessage.RemoveEndpoint, true); |
| msg.setJMSReplyTo(consumerDestination); |
| if (msg instanceof TextMessage) { |
| ((TextMessage) msg).setText(""); |
| } |
| } |
| private boolean connectionClosedOrInvalid() { |
| if ( sharedConnection == null || |
| sharedConnection.getConnection() == null || |
| ((ActiveMQConnection)sharedConnection.getConnection()).isClosed() || |
| ((ActiveMQConnection)sharedConnection.getConnection()).isClosing() || |
| ((ActiveMQConnection)sharedConnection.getConnection()).isTransportFailed()) { |
| return true; |
| } |
| return false; |
| } |
| protected void createSharedConnection(String aBrokerURI) throws Exception { |
| try { |
| // Acquire global static semaphore |
| sharedConnectionSemaphore.acquire(); |
| // check the state of a connection |
| if ( connectionClosedOrInvalid() ) { |
| if ( sharedConnection != null && sharedConnection.getConnection() != null ) { |
| try { |
| // Cleanup so that we dont leak connections in a broker |
| sharedConnection.getConnection().close(); |
| } catch( Exception ex) { |
| // Ignore exception while closing a bad connection |
| } |
| } |
| // This only effects Consumer |
| // Create AMQ specific connection validator. It uses |
| // AMQ specific approach to test the state of the connection |
| ActiveMQConnectionValidator connectionValidator = |
| new ActiveMQConnectionValidator(); |
| // Create a singleton shared connection object |
| sharedConnection = new SharedConnection(jndiContext, aBrokerURI); |
| // Add AMQ specific connection validator |
| sharedConnection.setConnectionValidator(connectionValidator); |
| // Connect to broker. Throws exception if unable to connect |
| sharedConnection.create(); |
| addPrefetch((ActiveMQConnection) sharedConnection.getConnection()); |
| sharedConnection.start(); |
| System.out.println("UIMA AS Client Created Shared Connection To Broker:" + aBrokerURI); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "setupConnection", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_client_connection_setup_INFO", new Object[] { aBrokerURI }); |
| } |
| } |
| |
| } catch (Exception e) { |
| throw e; |
| } finally { |
| // Release global static semaphore |
| sharedConnectionSemaphore.release(); |
| } |
| } |
| |
| private void addPrefetch(ActiveMQConnection aConnection) { |
| ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy(); |
| prefetchPolicy.setQueuePrefetch(5); |
| ((ActiveMQConnection) aConnection).setPrefetchPolicy(prefetchPolicy); |
| } |
| |
| private void validateConnection(String aBrokerURI) throws Exception { |
| // checks if a sharedConnection exists and if not creates a new one |
| createSharedConnection(aBrokerURI); |
| } |
| |
| protected Session getSession(String aBrokerURI) throws Exception { |
| validateConnection(aBrokerURI); |
| return getSession(sharedConnection.getConnection()); |
| } |
| |
| protected Session getSession(Connection aConnection) throws Exception { |
| session = aConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| return session; |
| } |
| |
| protected MessageProducer lookupProducerForEndpoint(Endpoint anEndpoint) throws Exception { |
| if (sharedConnection == null || producerSession == null) { |
| throw new ResourceInitializationException(); |
| } |
| Destination dest = producerSession.createQueue(anEndpoint.getEndpoint()); |
| return producerSession.createProducer(dest); |
| } |
| |
| protected void initializeProducer(String aBrokerURI, String aQueueName) throws Exception { |
| // Check if a sharedConnection exists. If not it creates one |
| createSharedConnection(aBrokerURI); |
| initializeProducer(aBrokerURI, aQueueName, sharedConnection.getConnection()); |
| } |
| |
| protected void initializeProducer(String aBrokerURI, String aQueueName, Connection aConnection) |
| throws Exception { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "initializeProducer", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_init_jms_producer_INFO", new Object[] { aBrokerURI, aQueueName }); |
| } |
| brokerURI = aBrokerURI; |
| // Create a worker thread for sending messages. Jms sessions are single threaded |
| // and it is illegal (per JMS spec) to use the same sesssion from multiple threads. |
| // The worker thread solves this problem. As it is the only thread that owns the |
| // session and uses it to create message producer. |
| // The worker thread blocks waiting for messages from application threads. The |
| // application threads add messages to the shared "queue" (in-memory queue not |
| // jms queue) and the worker thread consumes them. The worker thread is not |
| // serialializing CASes. This work is done in application threads. |
| |
| // create a Message Dispatcher object. In its constructor it acquires a shared |
| // semaphore producerSemaphore and holds it until the producer is created an |
| // and initialized. Once this happens or there is an error, the semaphore is |
| // released. |
| sender = new ActiveMQMessageSender(aConnection, aQueueName, this); |
| producerInitialized = false; |
| Thread t = new Thread((BaseMessageSender) sender); |
| // Start the worker thread. The jms session and message producer are created. Once |
| // the message producer is created, the worker thread notifies this thread by |
| // calling onProducerInitialized() where the global flag 'producerInitialized' is |
| // set to true. After the notification, the worker thread notifies this instance |
| // that the producer is fully initialized and finally begins to wait for messages |
| // in pendingMessageList. Upon arrival, each message is removed from |
| // pendingMessageList and it is sent to a destination. |
| |
| t.start(); |
| |
| try { |
| // Block waiting for the Sender to complete initializing the Producer. |
| // The sender will release the lock once it instantiates and initializes |
| // the Producer object or if there is an error |
| producerSemaphore.acquire(); |
| } catch (InterruptedException ex) { |
| |
| } finally { |
| producerSemaphore.release(); |
| } |
| // Check if the worker thread failed to initialize. |
| if (sender.failed()) { |
| // Worker thread failed to initialize. Log the reason and stop the uima ee client |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "initializeProducer", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_worker_thread_failed_to_initialize__WARNING", |
| new Object[] { sender.getReasonForFailure() }); |
| } |
| stop(); |
| return; |
| } |
| } |
| |
| /** |
| * Create a JMS Consumer on a temporary queue. Service replies will be handled by this consumer. |
| * |
| * @param aBrokerURI |
| * @throws Exception |
| */ |
| protected void initializeConsumer(String aBrokerURI) throws Exception { |
| createSharedConnection(aBrokerURI); |
| initializeConsumer(aBrokerURI, sharedConnection.getConnection()); |
| } |
| |
| protected void initializeConsumer(String aBrokerURI, Connection connection) throws Exception { |
| // In case we are recovering from a bad broker connection, invalidate old |
| // JMS objects first. |
| if ( consumerSession != null ) { |
| try { |
| consumer.close(); |
| consumerSession.close(); |
| } catch( Exception e) { |
| // ignore, creating a new Session below |
| } |
| } |
| consumerSession = getSession(connection); |
| consumerDestination = consumerSession.createTemporaryQueue(); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "initializeConsumer", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_init_jms_consumer_INFO", |
| new Object[] { aBrokerURI, consumerDestination.getQueueName() }); |
| } |
| consumer = consumerSession.createConsumer(consumerDestination); |
| consumer.setMessageListener(this); |
| System.out.println(">>>> Client Activated Temp Reply Queue:" |
| + consumerDestination.getQueueName()); |
| } |
| |
| /** |
| * Replaces place holder with syntax ${pname} with a system property whose name is pname |
| * |
| * @param aPlaceholder to resolve |
| * @return - the actual broker URL |
| * @throws ResourceInitializationException |
| */ |
| private String replacePlaceholder(String aPlaceholder) throws ResourceInitializationException { |
| // find placeholder starting and endpoint positions |
| int startPos = aPlaceholder.indexOf("{"); |
| int endPos = aPlaceholder.indexOf("}"); |
| // extract the name |
| String placeholder = aPlaceholder.substring(startPos+1, endPos); |
| // using the name, find the broker URL. This property must exist or exception is thrown |
| String url = System.getProperty(placeholder); |
| // the property is missing |
| if ( url == null ) { |
| throw new ResourceInitializationException(new Exception("UIMA AS Client Initialization Exception. Value for placeholder:"+placeholder+" is not defined in the environment. Set System property:"+placeholder+" to the broker URL.")); |
| } |
| return url; |
| } |
| /** |
| * Initialize the uima ee client. Takes initialization parameters from the |
| * <code>anApplicationContext</code> map. |
| */ |
| public synchronized void initialize(Map anApplicationContext) |
| throws ResourceInitializationException { |
| // Check UIMA AS version againg the UIMA Core version. If not the same throw Exception |
| if (!uimaAsVersion.getVersionString().equals(UIMAFramework.getVersionString())) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.WARNING, |
| CLASS_NAME.getName(), |
| "BaseAnalysisEngineController", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_incompatible_version_WARNING", |
| new Object[] { "UIM AS Client", uimaAsVersion.getVersionString(), |
| UIMAFramework.getVersionString() }); |
| throw new ResourceInitializationException(new AsynchAEException( |
| "Version of UIMA-AS is Incompatible with a Version of UIMA Core. UIMA-AS Version:" |
| + uimaAsVersion.getVersionString() + " Core UIMA Version:" |
| + UIMAFramework.getVersionString())); |
| } |
| |
| if (running) { |
| throw new ResourceInitializationException(new UIMA_IllegalStateException()); |
| } |
| reset(); |
| Properties performanceTuningSettings = null; |
| |
| if (!anApplicationContext.containsKey(UimaAsynchronousEngine.ServerUri)) { |
| throw new ResourceInitializationException(); |
| } |
| if (!anApplicationContext.containsKey(UimaAsynchronousEngine.Endpoint)) { |
| throw new ResourceInitializationException(); |
| } |
| ResourceManager rm = null; |
| if (anApplicationContext.containsKey(Resource.PARAM_RESOURCE_MANAGER)) { |
| rm = (ResourceManager) anApplicationContext.get(Resource.PARAM_RESOURCE_MANAGER); |
| } else { |
| rm = UIMAFramework.newDefaultResourceManager(); |
| } |
| if (anApplicationContext.containsKey(UIMAFramework.CAS_INITIAL_HEAP_SIZE)) { |
| String cas_initial_heap_size = (String) anApplicationContext |
| .get(UIMAFramework.CAS_INITIAL_HEAP_SIZE); |
| performanceTuningSettings = new Properties(); |
| performanceTuningSettings.put(UIMAFramework.CAS_INITIAL_HEAP_SIZE, cas_initial_heap_size); |
| } |
| asynchManager = new AsynchAECasManager_impl(rm); |
| |
| brokerURI = (String) anApplicationContext.get(UimaAsynchronousEngine.ServerUri); |
| |
| // Check if a placeholder is passed in instead of actual Broker URL. The placeholder |
| // has this syntax ${placeholderName}. A system property with placeholderName |
| // must exist for successful placeholder resolution. |
| if ( brokerURI.startsWith("${")) { |
| // resolve placeholder |
| // throws ResourceInitializationException if placeholder is not defined in |
| // System properties |
| brokerURI = replacePlaceholder(brokerURI); |
| } |
| |
| |
| String endpoint = (String) anApplicationContext.get(UimaAsynchronousEngine.Endpoint); |
| clientSideJmxStats.setEndpointName(endpoint); |
| int casPoolSize = 1; |
| |
| if (anApplicationContext.containsKey(UimaAsynchronousEngine.CasPoolSize)) { |
| casPoolSize = ((Integer) anApplicationContext.get(UimaAsynchronousEngine.CasPoolSize)) |
| .intValue(); |
| clientSideJmxStats.setCasPoolSize(casPoolSize); |
| } |
| |
| if (anApplicationContext.containsKey(UimaAsynchronousEngine.Timeout)) { |
| processTimeout = ((Integer) anApplicationContext.get(UimaAsynchronousEngine.Timeout)) |
| .intValue(); |
| } |
| |
| if (anApplicationContext.containsKey(UimaAsynchronousEngine.GetMetaTimeout)) { |
| metadataTimeout = ((Integer) anApplicationContext.get(UimaAsynchronousEngine.GetMetaTimeout)) |
| .intValue(); |
| } |
| |
| if (anApplicationContext.containsKey(UimaAsynchronousEngine.CpcTimeout)) { |
| cpcTimeout = ((Integer) anApplicationContext.get(UimaAsynchronousEngine.CpcTimeout)) |
| .intValue(); |
| } |
| if (anApplicationContext.containsKey(UimaAsynchronousEngine.ApplicationName)) { |
| applicationName = (String) anApplicationContext.get(UimaAsynchronousEngine.ApplicationName); |
| } |
| if (anApplicationContext.containsKey(UimaAsynchronousEngine.SerializationStrategy)) { |
| super.serializationStrategy = (String) anApplicationContext |
| .get(UimaAsynchronousEngine.SerializationStrategy); |
| clientSideJmxStats.setSerialization(super.serializationStrategy); |
| } |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.CONFIG)) { |
| UIMAFramework.getLogger(CLASS_NAME) |
| .logrb( |
| Level.CONFIG, |
| CLASS_NAME.getName(), |
| "initialize", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_init_uimaee_client__CONFIG", |
| new Object[] { brokerURI, 0, casPoolSize, processTimeout, metadataTimeout, |
| cpcTimeout }); |
| } |
| super.serviceDelegate = new ClientServiceDelegate(endpoint, applicationName, this); |
| super.serviceDelegate.setCasProcessTimeout(processTimeout); |
| super.serviceDelegate.setGetMetaTimeout(metadataTimeout); |
| try { |
| // Generate unique identifier |
| String uuid = UUIDGenerator.generate(); |
| // JMX does not allow ':' in the ObjectName so replace these with underscore |
| uuid = uuid.replaceAll(":", "_"); |
| uuid = uuid.replaceAll("-", "_"); |
| applicationName += "_" + uuid; |
| jmxManager = new JmxManager("org.apache.uima"); |
| clientSideJmxStats.setApplicationName(applicationName); |
| clientJmxObjectName = new ObjectName("org.apache.uima:name=" + applicationName); |
| jmxManager.registerMBean(clientSideJmxStats, clientJmxObjectName); |
| |
| Properties props = new Properties(); |
| props.setProperty(Context.INITIAL_CONTEXT_FACTORY,"org.apache.activemq.jndi.ActiveMQInitialContextFactory"); |
| props.setProperty(Context.PROVIDER_URL,brokerURI); |
| jndiContext = new InitialContext(props); |
| |
| // Check if sharedConnection exists. If not create a new one. The sharedConnection |
| // is static and shared by all instances of UIMA AS client in a jvm. The check |
| // is made in a critical section by first acquiring a global static semaphore to |
| // prevent a race condition. |
| createSharedConnection(brokerURI); |
| |
| // Reuse existing JMS connection if available |
| if (sharedConnection != null) { |
| initializeProducer(brokerURI, endpoint, sharedConnection.getConnection()); |
| initializeConsumer(brokerURI, sharedConnection.getConnection()); |
| } else { |
| initializeProducer(brokerURI, endpoint); |
| initializeConsumer(brokerURI); |
| } |
| |
| // Increment number of client instances. SharedConnection object is a static |
| // and is used to share a single JMS connection. The connection is closed |
| // when the last client finishes processing and calls stop(). |
| if (sharedConnection != null) { |
| sharedConnection.registerClient(this); |
| } |
| running = true; |
| // Acquire GetMeta Semaphore Before Sending a GetMeta Request. This will force |
| // the client to block in waitForMetadataReply() until GetMeta reply is received |
| try { |
| getMetaSemaphore.acquire(); |
| } catch (InterruptedException e) { |
| System.out.println("UIMA AS Client Interrupted While Waiting On GetMetaSemaphore"); |
| } |
| sendMetaRequest(); |
| waitForMetadataReply(); |
| if (abort || !running) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "initialize", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_aborting_as_WARNING", new Object[] { "Metadata Timeout" }); |
| } |
| throw new ResourceInitializationException(new UimaASMetaRequestTimeout()); |
| } else { |
| if (collectionReader != null) { |
| asynchManager.addMetadata(collectionReader.getProcessingResourceMetaData()); |
| } |
| |
| asynchManager.initialize(casPoolSize, "ApplicationCasPoolContext", |
| performanceTuningSettings); |
| |
| // Create a special CasPool of size 1 to be used for deserializing CASes from a Cas |
| // Multiplier |
| if (super.resourceMetadata != null |
| && super.resourceMetadata instanceof AnalysisEngineMetaData) { |
| if (((AnalysisEngineMetaData) super.resourceMetadata).getOperationalProperties() |
| .getOutputsNewCASes()) { |
| // Create a Shadow CAS Pool used to de-serialize CASes produced by a CAS Multiplier |
| asynchManager.initialize(1, SHADOW_CAS_POOL, performanceTuningSettings); |
| } |
| } |
| initialized = true; |
| remoteService = true; |
| // running = true; |
| |
| for (int i = 0; listeners != null && i < listeners.size(); i++) { |
| ((UimaASStatusCallbackListener) listeners.get(i)).initializationComplete(null); |
| } |
| } |
| |
| } catch (ResourceInitializationException e) { |
| state = ClientState.FAILED; |
| notifyOnInitializationFailure(e); |
| throw e; |
| } catch (Exception e) { |
| state = ClientState.FAILED; |
| notifyOnInitializationFailure(e); |
| throw new ResourceInitializationException(e); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "initialize", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_as_initialized__INFO", |
| new Object[] { super.serializationStrategy }); |
| } |
| // Acquire cpcReady semaphore to block sending CPC request until |
| // ALL outstanding CASes are received. |
| super.acquireCpcReadySemaphore(); |
| state = ClientState.RUNNING; |
| } |
| |
| /** |
| * First generates a Spring context from a given deploy descriptor and than deploys the context |
| * into a Spring Container. |
| * |
| * @param aDeploymentDescriptor |
| * - deployment descriptor to generate Spring Context from |
| * @param anApplicationContext |
| * - a Map containing properties required by dd2spring |
| * |
| * @return - a unique spring container id |
| * |
| */ |
| public String deploy(String aDeploymentDescriptor, Map anApplicationContext) throws Exception { |
| String springContext = generateSpringContext(aDeploymentDescriptor, anApplicationContext); |
| |
| SpringContainerDeployer springDeployer = new SpringContainerDeployer(springContainerRegistry); |
| try { |
| String id = springDeployer.deploy(springContext); |
| if ( springDeployer.isInitialized() ) { |
| springDeployer.startListeners(); |
| } |
| return id; |
| } catch (ResourceInitializationException e) { |
| running = true; |
| throw e; |
| } |
| } |
| |
| /** |
| * |
| */ |
| public String deploy(String[] aDeploymentDescriptorList, Map anApplicationContext) |
| throws Exception { |
| if (aDeploymentDescriptorList == null) { |
| throw new ResourceConfigurationException(UIMA_IllegalArgumentException.ILLEGAL_ARGUMENT, |
| new Object[] { "Null", "DeploymentDescriptorList", "deploy()" }); |
| } |
| |
| if (aDeploymentDescriptorList.length == 0) { |
| throw new ResourceConfigurationException( |
| ResourceConfigurationException.MANDATORY_VALUE_MISSING, |
| new Object[] { "DeploymentDescriptorList" }); |
| } |
| String[] springContextFiles = new String[aDeploymentDescriptorList.length]; |
| |
| for (int i = 0; i < aDeploymentDescriptorList.length; i++) { |
| springContextFiles[i] = generateSpringContext(aDeploymentDescriptorList[i], |
| anApplicationContext); |
| } |
| |
| SpringContainerDeployer springDeployer = new SpringContainerDeployer(springContainerRegistry); |
| try { |
| String id = springDeployer.deploy(springContextFiles); |
| if ( springDeployer.isInitialized() ) { |
| springDeployer.startListeners(); |
| } |
| return id; |
| } catch (ResourceInitializationException e) { |
| running = true; |
| throw e; |
| } |
| |
| } |
| |
| public void undeploy() throws Exception { |
| Iterator containerIterator = springContainerRegistry.keySet().iterator(); |
| while (containerIterator.hasNext()) { |
| String containerId = (String) containerIterator.next(); |
| undeploy(containerId); |
| } |
| } |
| |
| public void undeploy(String aSpringContainerId) throws Exception { |
| this.undeploy(aSpringContainerId, SpringContainerDeployer.STOP_NOW); |
| } |
| |
| /** |
| * Undeploys Spring container with a given container Id. All deployed Spring containers are |
| * registered in the local registry under a unique id. |
| * |
| */ |
| public void undeploy(String aSpringContainerId, int stop_level) throws Exception { |
| if (aSpringContainerId == null) { |
| return; |
| } |
| |
| UimaEEAdminSpringContext adminContext = null; |
| if (!springContainerRegistry.containsKey(aSpringContainerId)) { |
| return; |
| // throw new InvalidContainerException("Invalid Spring container Id:" + aSpringContainerId + |
| // ". Unable to undeploy the Spring container"); |
| } |
| // Fetch an administrative context which contains a Spring Container |
| adminContext = (UimaEEAdminSpringContext) springContainerRegistry.get(aSpringContainerId); |
| if (adminContext == null) { |
| throw new InvalidContainerException( |
| "Spring Container Does Not Contain Valid UimaEEAdminSpringContext Object"); |
| } |
| // Fetch instance of the Container from its context |
| ApplicationContext ctx = adminContext.getSpringContainer(); |
| // Query the container for objects that implement |
| // ControllerLifecycle interface. These |
| // objects are typically of type AnalysisEngineController or |
| // UimacppServiceController. |
| String[] asyncServiceList = ctx |
| .getBeanNamesForType(org.apache.uima.aae.controller.ControllerLifecycle.class); |
| // Given a valid list of controllers select the first from the list |
| // and |
| // initiate a shutdown. We don't care which controller will be |
| // invoked. In case of |
| // AggregateAnalysisEngineController the terminate event will |
| // propagate all the way |
| // to the top controller in the hierarchy and the shutdown will take |
| // place from there. |
| // If the controller is of kind UimecppServiceController or |
| // PrimitiveAnalysisController |
| // the termination logic will be immediately triggered in the |
| // terminate() method. |
| if (asyncServiceList != null && asyncServiceList.length > 0) { |
| boolean topLevelController = false; |
| ControllerLifecycle ctrer = null; |
| int indx = 0; |
| while (!topLevelController) { |
| ctrer = (ControllerLifecycle) ctx.getBean(asyncServiceList[indx++]); |
| if (ctrer instanceof UimacppServiceController |
| || ((AnalysisEngineController) ctrer).isTopLevelComponent()) { |
| topLevelController = true; |
| } |
| } |
| // Send a trigger to initiate shutdown. |
| if (ctrer != null) { |
| if (ctrer instanceof AnalysisEngineController) { |
| ((AnalysisEngineController) ctrer).getControllerLatch().release(); |
| } |
| switch (stop_level) { |
| case SpringContainerDeployer.QUIESCE_AND_STOP: |
| ((AnalysisEngineController) ctrer).quiesceAndStop(); |
| break; |
| case SpringContainerDeployer.STOP_NOW: |
| ((AnalysisEngineController) ctrer).terminate(); |
| break; |
| } |
| } |
| } |
| if (ctx instanceof FileSystemXmlApplicationContext) { |
| ((FileSystemXmlApplicationContext) ctx).destroy(); |
| } |
| // Remove the container from a local registry |
| springContainerRegistry.remove(aSpringContainerId); |
| } |
| |
| /** |
| * Use dd2spring to generate Spring context file from a given deployment descriptor file. |
| * |
| * @param aDeploymentDescriptor |
| * - deployment descriptor to generate Spring Context from |
| * @param anApplicationContext |
| * - a Map containing properties required by dd2spring |
| * @return - an absolute path to the generated Spring Context file |
| * |
| * @throws Exception |
| * - if failure occurs |
| */ |
| private String generateSpringContext(String aDeploymentDescriptor, Map anApplicationContext) |
| throws Exception { |
| |
| String dd2SpringXsltFilePath = null; |
| String saxonClasspath = null; |
| |
| if (anApplicationContext.containsKey(UimaAsynchronousEngine.DD2SpringXsltFilePath)) { |
| dd2SpringXsltFilePath = (String) anApplicationContext |
| .get(UimaAsynchronousEngine.DD2SpringXsltFilePath); |
| } else { |
| throw new ResourceConfigurationException( |
| ResourceConfigurationException.MANDATORY_VALUE_MISSING, |
| new Object[] { "Xslt File Path" }); |
| } |
| if (anApplicationContext.containsKey(UimaAsynchronousEngine.SaxonClasspath)) { |
| saxonClasspath = (String) anApplicationContext.get(UimaAsynchronousEngine.SaxonClasspath); |
| } else { |
| throw new ResourceConfigurationException( |
| ResourceConfigurationException.MANDATORY_VALUE_MISSING, |
| new Object[] { "Saxon Classpath" }); |
| } |
| |
| Dd2spring dd2Spring = new Dd2spring(); |
| File springContextFile = dd2Spring.convertDd2Spring(aDeploymentDescriptor, |
| dd2SpringXsltFilePath, saxonClasspath, (String) anApplicationContext |
| .get(UimaAsynchronousEngine.UimaEeDebug)); |
| |
| return springContextFile.getAbsolutePath(); |
| } |
| |
| /** |
| * Deploys provided context files ( and beans) in a new Spring container. |
| * |
| */ |
| protected String deploySpringContainer(String[] springContextFiles) |
| throws ResourceInitializationException { |
| |
| SpringContainerDeployer springDeployer = new SpringContainerDeployer(); |
| try { |
| return springDeployer.deploy(springContextFiles); |
| } catch (ResourceInitializationException e) { |
| // turn on the global flag so that the stop() can do the cleanup |
| running = true; |
| throw e; |
| } |
| } |
| |
| protected void waitForServiceNotification() throws Exception { |
| while (!serviceInitializationCompleted) { |
| if (serviceInitializationException) { |
| throw new ResourceInitializationException(); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "waitForServiceNotification", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_awaiting_container_init__INFO", new Object[] {}); |
| } |
| try { |
| serviceSemaphore.acquire(); |
| } catch (InterruptedException e) { |
| } finally { |
| serviceSemaphore.release(); |
| } |
| if (serviceInitializationException) { |
| throw new ResourceInitializationException(); |
| } |
| } |
| } |
| |
| protected void deployEmbeddedBroker() throws Exception { |
| // TBI |
| } |
| |
| public static void main(String[] args) { |
| try { |
| |
| BaseUIMAAsynchronousEngineCommon_impl uimaee = new BaseUIMAAsynchronousEngine_impl(); |
| |
| Map appContext = new HashMap(); |
| appContext.put(UimaAsynchronousEngine.DD2SpringXsltFilePath, args[1]); |
| appContext.put(UimaAsynchronousEngine.SaxonClasspath, args[2]); |
| String containerId = uimaee.deploy(args[0], appContext); // args[1], |
| // args[2]); |
| |
| uimaee.undeploy(containerId); |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "main", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| } |
| |
| public void setReleaseCASMessage(TextMessage msg, String aCasReferenceId) throws Exception { |
| msg.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.None); |
| msg.setStringProperty(AsynchAEMessage.CasReference, aCasReferenceId); |
| msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.ReleaseCAS); |
| msg.setStringProperty(UIMAMessage.ServerURI, brokerURI); |
| msg.setJMSReplyTo(consumerDestination); |
| } |
| |
| public void notifyOnInitializationFailure(Exception e) { |
| notifyOnInitializationFailure(null, e); |
| } |
| |
| public void notifyOnInitializationSuccess() { |
| notifyOnInitializationSuccess(null); |
| } |
| |
| public void notifyOnInitializationFailure(AnalysisEngineController aController, Exception e) { |
| |
| // Initialization exception. Notify blocking thread and indicate a problem |
| serviceInitializationException = true; |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "notifyOnInitializationFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_container_init_exception__WARNING", new Object[] { e }); |
| } |
| serviceSemaphore.release(); |
| } |
| |
| public void notifyOnInitializationSuccess(AnalysisEngineController aController) { |
| serviceInitializationCompleted = true; |
| serviceSemaphore.release(); |
| } |
| |
| public void notifyOnTermination(String message) { |
| |
| } |
| |
| protected MessageProducer getMessageProducer(Destination destination) throws Exception { |
| return sender.getMessageProducer(destination); |
| } |
| |
| /** |
| * Request Uima AS client to initiate sending Stop requests to a service for all outstanding CASes |
| * awaiting reply. |
| * |
| */ |
| public void stopProducingCases() { |
| List<DelegateEntry> outstandingCasList = serviceDelegate.getDelegateCasesPendingReply(); |
| for (DelegateEntry entry : outstandingCasList) { |
| // The Cas is still being processed |
| ClientRequest clientCachedRequest = (ClientRequest) clientCache |
| .get(entry.getCasReferenceId()); |
| if (clientCachedRequest != null && !clientCachedRequest.isMetaRequest() |
| && clientCachedRequest.getCasReferenceId() != null) { |
| stopProducingCases(clientCachedRequest); |
| } |
| } |
| } |
| |
| /** |
| * Request Uima AS client to initiate sending Stop request to a service for a given CAS id If the |
| * service is a Cas Multiplier, it will stop producing new CASes, will wait until all child CASes |
| * finish and finally returns the input CAS. |
| * |
| */ |
| public void stopProducingCases(String aCasReferenceId) { |
| // The Cas is still being processed |
| ClientRequest clientCachedRequest = (ClientRequest) clientCache.get(aCasReferenceId); |
| if (clientCachedRequest != null) { |
| stopProducingCases(clientCachedRequest); |
| } |
| } |
| |
| private void stopProducingCases(ClientRequest clientCachedRequest) { |
| try { |
| if (clientCachedRequest.getFreeCasNotificationQueue() != null) { |
| TextMessage msg = createTextMessage(); |
| msg.setText(""); |
| msg.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.None); |
| msg |
| .setStringProperty(AsynchAEMessage.CasReference, clientCachedRequest |
| .getCasReferenceId()); |
| msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.Stop); |
| msg.setStringProperty(UIMAMessage.ServerURI, brokerURI); |
| try { |
| MessageProducer msgProducer = getMessageProducer(clientCachedRequest |
| .getFreeCasNotificationQueue()); |
| if (msgProducer != null) { |
| System.out.println(">>> Client Sending Stop to Service for CAS:" |
| + clientCachedRequest.getCasReferenceId() + " Destination:" |
| + clientCachedRequest.getFreeCasNotificationQueue()); |
| // Send STOP message to Cas Multiplier Service |
| msgProducer.send(msg); |
| } else { |
| System.out |
| .println(">>> UIMA AS Client Unable to Send Stop To Service. Message Producer Not Initialized"); |
| } |
| |
| } catch (Exception ex) { |
| System.out.println("Client Unable to send STOP Request to Service. Reason:"); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "notifyOnInitializationFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", |
| new Object[] { Thread.currentThread().getId(), ex }); |
| } |
| } |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "notifyOnInitializationFailure", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { Thread.currentThread().getId(), e }); |
| } |
| } |
| } |
| protected MessageSender getDispatcher() { |
| return sender; |
| } |
| } |
