You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uima.apache.org by sc...@apache.org on 2018/12/20 14:40:58 UTC
svn commit: r1849399 [1/2] - in
/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq: ./
src/main/java/org/apache/uima/adapter/jms/activemq/
src/main/java/org/apache/uima/adapter/jms/client/
src/main/java/org/apache/uima/examples/as/ src/test/java/org/apache/u...
Author: schor
Date: Thu Dec 20 14:40:58 2018
New Revision: 1849399
URL: http://svn.apache.org/viewvc?rev=1849399&view=rev
Log:
[UIMA-5937] update pom version and scm, merge v2 updates, record merge
Added:
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/PriorityMessageHandler.java
- copied unchanged from r1846917, uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/PriorityMessageHandler.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/resources/deployment/Deploy_AggregateAnnotatorWithDOCTYPE.xml
- copied unchanged from r1846917, uima/uima-as/trunk/uimaj-as-activemq/src/test/resources/deployment/Deploy_AggregateAnnotatorWithDOCTYPE.xml
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/resources/deployment/Deploy_AsyncAggregateWithJmsService.xml
- copied unchanged from r1846917, uima/uima-as/trunk/uimaj-as-activemq/src/test/resources/deployment/Deploy_AsyncAggregateWithJmsService.xml
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/resources/deployment/Deploy_NoOpAnnotatorWithTargetingSupport.xml
- copied unchanged from r1846917, uima/uima-as/trunk/uimaj-as-activemq/src/test/resources/deployment/Deploy_NoOpAnnotatorWithTargetingSupport.xml
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/resources/descriptors/analysis_engine/NoOpAnnotatorWithTargetingSupport.xml
- copied unchanged from r1846917, uima/uima-as/trunk/uimaj-as-activemq/src/test/resources/descriptors/analysis_engine/NoOpAnnotatorWithTargetingSupport.xml
Modified:
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/ (props changed)
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/pom.xml
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/ConcurrentMessageListener.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsOutputChannel.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/UimaDefaultMessageListenerContainer.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/ActiveMQMessageSender.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngine_impl.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/RunRemoteAsyncAE.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/java/org/apache/uima/ae/noop/NoOpAnnotator.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/java/org/apache/uima/ae/noop/NoOpCC.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/java/org/apache/uima/ee/test/TestUimaASExtended.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/java/org/apache/uima/ee/test/utils/ActiveMQSupport.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/java/org/apache/uima/ee/test/utils/BaseTestSupport.java
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/resources/descriptors/analysis_engine/NoOpAnnotatorWith2SecDelay.xml
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/resources/descriptors/analysis_engine/NoOpAnnotatorWith5SecDelay.xml
uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/test/resources/descriptors/analysis_engine/SimpleTestAggregate.xml
Propchange: uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Thu Dec 20 14:40:58 2018
@@ -1,2 +1,3 @@
/uima/uima-as/branches/depend-on-parent-pom-4/uimaj-as-activemq:961335-961760
/uima/uima-as/branches/mavenAlign/uimaj-as-activemq:941450-944450
+/uima/uima-as/trunk/uimaj-as-activemq:1786000-1846917
Modified: uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/pom.xml
URL: http://svn.apache.org/viewvc/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/pom.xml?rev=1849399&r1=1849398&r2=1849399&view=diff
==============================================================================
--- uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/pom.xml (original)
+++ uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/pom.xml Thu Dec 20 14:40:58 2018
@@ -15,7 +15,7 @@
<parent>
<groupId>org.apache.uima</groupId>
<artifactId>uima-as-parent</artifactId>
- <version>3.0.0-SNAPSHOT</version>
+ <version>3.0.1-SNAPSHOT</version>
<relativePath>../uima-as-parent/pom.xml</relativePath>
</parent>
@@ -31,13 +31,13 @@
cutting/pasting the <scm> element, and just changing the following two properties -->
<scm>
<connection>
- scm:svn:http://svn.apache.org/repos/asf/uima/uima-as/trunk/uimaj-as-activemq
+ scm:svn:http://svn.apache.org/repos/asf/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq
</connection>
<developerConnection>
- scm:svn:https://svn.apache.org/repos/asf/uima/uima-as/trunk/uimaj-as-activemq
+ scm:svn:https://svn.apache.org/repos/asf/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq
</developerConnection>
<url>
- http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-activemq
+ http://svn.apache.org/viewvc/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq
</url>
</scm>
Modified: uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/ConcurrentMessageListener.java
URL: http://svn.apache.org/viewvc/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/ConcurrentMessageListener.java?rev=1849399&r1=1849398&r2=1849399&view=diff
==============================================================================
--- uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/ConcurrentMessageListener.java (original)
+++ uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/ConcurrentMessageListener.java Thu Dec 20 14:40:58 2018
@@ -30,12 +30,10 @@ import javax.jms.Message;
import javax.jms.Session;
import org.apache.uima.UIMAFramework;
-import org.apache.uima.aae.InProcessCache.CacheEntry;
import org.apache.uima.aae.UIMAEE_Constants;
import org.apache.uima.aae.UimaAsThreadFactory;
import org.apache.uima.aae.UimaBlockingExecutor;
import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
-import org.apache.uima.aae.controller.AggregateAnalysisEngineController_impl;
import org.apache.uima.aae.controller.AnalysisEngineController;
import org.apache.uima.aae.controller.LocalCache.CasStateEntry;
import org.apache.uima.aae.delegate.Delegate;
Modified: uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java
URL: http://svn.apache.org/viewvc/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java?rev=1849399&r1=1849398&r2=1849399&view=diff
==============================================================================
--- uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java (original)
+++ uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java Thu Dec 20 14:40:58 2018
@@ -227,7 +227,7 @@ public class JmsEndpointConnection_impl
}
if ( brokerDestinations.getConnection() != null ) {
try {
- // Close the connection to avoid leaks in the broker
+ // Close the connection to avoid leaks in the broker
brokerDestinations.getConnection().close();
} catch( Exception e) {
// Ignore exceptions on a close of a bad connection
@@ -249,7 +249,9 @@ public class JmsEndpointConnection_impl
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUri);
// White list packages for deserialization
factory.setTrustAllPackages(true);
+ factory.setConnectionIDPrefix("JmsOutputChannel");
factory.setWatchTopicAdvisories(false);
+
// Create shared jms connection to a broker
conn = factory.createConnection();
factory.setDispatchAsync(true);
@@ -755,7 +757,7 @@ public class JmsEndpointConnection_impl
if ( command == AsynchAEMessage.ServiceInfo ) {
return false;
}
- if ( (msgType == AsynchAEMessage.Response || msgType == AsynchAEMessage.Request ) &&
+ if ( (msgType == AsynchAEMessage.Response || msgType == AsynchAEMessage.Request ) &&
command == AsynchAEMessage.Process ) {
String casReferenceId="";
try {
@@ -764,7 +766,7 @@ public class JmsEndpointConnection_impl
String key = "";
String endpointName = "";
if ( delegateEndpoint != null ) {
- delegateEndpoint.getDelegateKey();
+ key = delegateEndpoint.getDelegateKey();
endpointName = ((ActiveMQDestination) delegateEndpoint.getDestination())
.getPhysicalName();
}
@@ -781,20 +783,22 @@ public class JmsEndpointConnection_impl
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
-
+
String key = "";
String endpointName = "";
if ( delegateEndpoint != null ) {
- delegateEndpoint.getDelegateKey();
+ key = delegateEndpoint.getDelegateKey();
endpointName = ((ActiveMQDestination) delegateEndpoint.getDestination())
.getPhysicalName();
- }
+
+ }
if ( "Client".equals(target) ) {
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
"send", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAEE_service_delivery_to_client_exception__WARNING",
new Object[] { controller.getComponentName(),endpointName });
} else {
+
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
"send", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAEE_service_delivery_exception__WARNING",new Object[] { controller.getComponentName(), key, endpointName});
@@ -1021,8 +1025,8 @@ public class JmsEndpointConnection_impl
}
} else {
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
- "send", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
- "UIMAEE_service_delivery_exception__WARNING",new Object[] { controller.getComponentName(), "", endpointName});
+ "UimaAsAsyncCallbackListener.onException()", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
+ "UIMAEE_service_delivery_exception_WARNING",new Object[] { controller.getComponentName(), exception} );
}
}
Modified: uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java
URL: http://svn.apache.org/viewvc/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java?rev=1849399&r1=1849398&r2=1849399&view=diff
==============================================================================
--- uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java (original)
+++ uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java Thu Dec 20 14:40:58 2018
@@ -19,6 +19,7 @@
package org.apache.uima.adapter.jms.activemq;
+import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -34,12 +35,12 @@ import javax.jms.Message;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.ActiveMQPrefetchPolicy;
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.client.UimaAsynchronousEngine;
import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
import org.apache.uima.aae.controller.AnalysisEngineController;
import org.apache.uima.aae.controller.BaseAnalysisEngineController.ServiceState;
@@ -54,12 +55,12 @@ import org.apache.uima.aae.jmx.RemoteJMX
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.MessageWrapper;
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;
-import org.springframework.jms.support.destination.DestinationResolver;
/**
* Thin adapter for receiving JMS messages from Spring. It delegates processing of all messages to
@@ -558,6 +559,19 @@ public class JmsInputChannel implements
validEndpoint(messageContext) &&
isReplyRequired((Message)messageContext.getRawMessage()) );
}
+ public void onMessage(MessageWrapper wrapper) {
+ try {
+ onMessage((Message)wrapper.getMessage(), (Session)wrapper.getSession());
+ } finally {
+ // semaphore is only added by target and process listeners. The semaphore is used
+ // to throttle work into the service. A JMS thread blocks if semaphore permits are
+ // exhausted. The blocking is done in PriorityMessageHandler class. The JmsInputChannel
+ // and PriorityMessageHandler are the only classes operating on the semaphore.
+ if ( wrapper.getSemaphore() != null ) {
+ wrapper.getSemaphore().release();
+ }
+ }
+ }
/**
* 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
@@ -810,10 +824,11 @@ public class JmsInputChannel implements
}
return ll;
}
- public synchronized void setListenerContainer(UimaDefaultMessageListenerContainer messageListener) {
- this.messageListener = messageListener;
+ public synchronized void setListenerContainer(UimaDefaultMessageListenerContainer jmsL) {
+ this.messageListener = jmsL;
System.setProperty("BrokerURI", messageListener.getBrokerUrl());
- if ( messageListener.getMessageSelector() !=null && messageListener.getMessageSelector().equals("Command=2001") ) {
+
+ if ( jmsL.isGetMetaListener() ) {
brokerURL = messageListener.getBrokerUrl();
getController().getOutputChannel().setServerURI(brokerURL);
}
@@ -825,8 +840,9 @@ public class JmsInputChannel implements
getController().addInputChannel(this);
messageListener.setController(getController());
} catch (Exception e) {
+ e.printStackTrace();
}
- }
+ }
}
public ActiveMQConnectionFactory getConnectionFactory() {
@@ -1099,6 +1115,66 @@ public class JmsInputChannel implements
new Object[] { getController().getComponentName(), connector.getEndpointName() });
}
}
+ public void createListenerForTargetedMessages() throws Exception {
+ List<UimaDefaultMessageListenerContainer> listeners =
+ getListeners();
+ // the TargetServiceId property value will become part of a jms selector.
+ String targetStringSelector = "";
+ if ( System.getProperty(UimaAsynchronousEngine.TargetSelectorProperty) != null ) {
+ targetStringSelector = System.getProperty(UimaAsynchronousEngine.TargetSelectorProperty);
+ } else {
+ // the default selector is IP:PID
+ String ip = InetAddress.getLocalHost().getHostAddress();
+ targetStringSelector = ip+":"+controller.getPID();
+ }
+ // find a listener instance which handles Process requests. The targeted
+ // listener created here will share a Connection Factory and ThreadFactory.
+ //
+ for(UimaDefaultMessageListenerContainer listener : listeners ) {
+ // Is this a Process listener instance? Check the selector
+ if ( listener.getMessageSelector().endsWith(UimaDefaultMessageListenerContainer.PROCESS_SELECTOR_SUFFIX) ) {
+ // this will be a dedicated listener which handles targeted messages
+ UimaDefaultMessageListenerContainer targetedListener = new UimaDefaultMessageListenerContainer();
+ // setup jms selector
+ if ( getController().isCasMultiplier()) {
+ targetedListener.setMessageSelector(UimaAsynchronousEngine.TargetSelectorProperty+" = '"+targetStringSelector+"' AND"+UimaDefaultMessageListenerContainer.CM_PROCESS_SELECTOR_SUFFIX);//(Command=2000 OR Command=2002)");
+ } else {
+ targetedListener.setMessageSelector(UimaAsynchronousEngine.TargetSelectorProperty+" = '"+targetStringSelector+"' AND"+UimaDefaultMessageListenerContainer.PROCESS_SELECTOR_SUFFIX);//(Command=2000 OR Command=2002)");
+ }
+
+ // use shared ConnectionFactory
+ targetedListener.setConnectionFactory(listener.getConnectionFactory());
+ // mark the listener as a 'Targeted' listener
+ targetedListener.setTargetedListener();
+ targetedListener.setController(getController());
+ // there will only be one AMQ delivery thread. Its job will be to
+ // add a targeted message to a BlockingQueue. Such thread will block
+ // in an enqueue if a dequeue is not available. This will be prevent
+ // the overwhelming the service with messages.
+ targetedListener.setConcurrentConsumers(1);
+ if ( listener.getMessageListener() instanceof PriorityMessageHandler ) {
+ // the targeted listener will use the same message handler as the
+ // Process listener. This handler will add a message wrapper
+ // to enable prioritizing messages.
+ targetedListener.setMessageListener(listener.getMessageListener());
+ }
+ // Same queue as the Process queue
+ targetedListener.setDestination(listener.getDestination());
+ registerListener(targetedListener);
+ targetedListener.afterPropertiesSet();
+ targetedListener.initialize();
+ targetedListener.start();
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "createListenerForTargetedMessages", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
+ "UIMAJMS_TARGET_LISTENER__INFO",
+ new Object[] {targetedListener.getMessageSelector(), controller.getComponentName() });
+ }
+ break;
+
+ }
+ }
+ }
public void createListener(String aDelegateKey, Endpoint endpointToUpdate) throws Exception {
if (getController() instanceof AggregateAnalysisEngineController) {
Delegate delegate = ((AggregateAnalysisEngineController) getController())
Modified: uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsOutputChannel.java
URL: http://svn.apache.org/viewvc/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsOutputChannel.java?rev=1849399&r1=1849398&r2=1849399&view=diff
==============================================================================
--- uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsOutputChannel.java (original)
+++ uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsOutputChannel.java Thu Dec 20 14:40:58 2018
@@ -25,16 +25,16 @@ import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.ConnectException;
import java.net.InetAddress;
-import java.util.ArrayList;
-import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Timer;
-import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -74,8 +74,8 @@ import org.apache.uima.aae.jmx.ServicePe
import org.apache.uima.aae.message.AsynchAEMessage;
import org.apache.uima.aae.message.UIMAMessage;
import org.apache.uima.aae.monitor.Monitor;
-import org.apache.uima.aae.monitor.statistics.LongNumericStatistic;
import org.apache.uima.aae.monitor.statistics.AnalysisEnginePerformanceMetrics;
+import org.apache.uima.aae.monitor.statistics.LongNumericStatistic;
import org.apache.uima.adapter.jms.JmsConstants;
import org.apache.uima.cas.CAS;
import org.apache.uima.cas.SerialFormat;
@@ -83,14 +83,6 @@ import org.apache.uima.cas.impl.XmiSeria
import org.apache.uima.resource.metadata.ProcessingResourceMetaData;
import org.apache.uima.util.Level;
-
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
-import org.apache.uima.resource.ResourceProcessException;
-
-
import com.thoughtworks.xstream.XStream;
public class JmsOutputChannel implements OutputChannel {
@@ -140,6 +132,8 @@ public class JmsOutputChannel implements
private Semaphore connectionSemaphore = new Semaphore(1);
public JmsOutputChannel() {
+ UimaSerializer.initXStream(xstream);
+
try {
if( System.getenv("IP") != null ) {
hostIP = System.getenv("IP");
@@ -364,16 +358,17 @@ public class JmsOutputChannel implements
private void invalidateConnectionAndEndpoints(BrokerConnectionEntry brokerConnectionEntry ) {
Connection conn = brokerConnectionEntry.getConnection();
try {
- if ( conn != null && ((ActiveMQConnection)conn).isClosed()) {
+ if ( conn != null)
+ if ( conn != null && !((ActiveMQConnection)conn).isClosed()) {
for (Entry<Object, JmsEndpointConnection_impl> endpoints : brokerConnectionEntry.endpointMap
.entrySet()) {
endpoints.getValue().close(); // close session and producer
}
- brokerConnectionEntry.getConnection().stop();
brokerConnectionEntry.getConnection().close();
brokerConnectionEntry.setConnection(null);
}
} catch (Exception e) {
+ e.printStackTrace();
// Ignore this for now. Attempting to close connection that has been closed
// Ignore we are shutting down
} finally {
@@ -648,7 +643,6 @@ public class JmsOutputChannel implements
connectionSemaphore.release();
}
- //System.out.println("+++++++++++++++++++++ ConnectionMap Size:"+connectionMap.size());
return endpointConnection;
}
@@ -850,10 +844,21 @@ public class JmsOutputChannel implements
// If this service is a Cas Multiplier add to the message a FreeCasQueue.
// The client may need send Stop request to that queue.
if (aCommand == AsynchAEMessage.ServiceInfo
- && getAnalysisEngineController().isCasMultiplier() && freeCASTempQueue != null) {
- // Attach a temp queue to the outgoing message. This a queue where
- // Free CAS notifications need to be sent from the client
- tm.setJMSReplyTo(freeCASTempQueue);
+ && getAnalysisEngineController().isCasMultiplier() ) {
+ if ( freeCASTempQueue != null ) {
+ // Attach a temp queue to the outgoing message. This a queue where
+ // Free CAS notifications need to be sent from the client
+ tm.setJMSReplyTo(freeCASTempQueue);
+ }
+ // new services will receive FreeCas request via a targeted queue
+ StringBuffer selector = new StringBuffer().
+ append("TargetServiceId = ").
+ append("'").append(hostIP).append(":").
+ append(getAnalysisEngineController().getPID()).
+ append("' AND").
+ append(UimaDefaultMessageListenerContainer.CM_PROCESS_SELECTOR_SUFFIX);
+ tm.setStringProperty(AsynchAEMessage.TargetingSelector,selector.toString());
+
}
// Check if there was a failure while sending a message
if ( !endpointConnection.send(tm, 0, false, notifyOnJmsException) && notifyOnJmsException ) {
@@ -1387,7 +1392,7 @@ public class JmsOutputChannel implements
*/
private void populateHeaderWithRequestContext(Message aMessage, Endpoint anEndpoint, int aCommand)
throws Exception {
- aMessage.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
+ aMessage.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
aMessage.setIntProperty(AsynchAEMessage.Command, aCommand);
// TODO override default based on system property
aMessage.setBooleanProperty(AsynchAEMessage.AcceptsDeltaCas, true);
Modified: uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/UimaDefaultMessageListenerContainer.java
URL: http://svn.apache.org/viewvc/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/UimaDefaultMessageListenerContainer.java?rev=1849399&r1=1849398&r2=1849399&view=diff
==============================================================================
--- uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/UimaDefaultMessageListenerContainer.java (original)
+++ uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/UimaDefaultMessageListenerContainer.java Thu Dec 20 14:40:58 2018
@@ -28,6 +28,7 @@ import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -38,6 +39,9 @@ import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.Session;
import javax.jms.TemporaryQueue;
import org.apache.activemq.ActiveMQConnection;
@@ -46,8 +50,11 @@ import org.apache.activemq.command.Activ
import org.apache.uima.UIMAFramework;
import org.apache.uima.aae.InputChannel;
import org.apache.uima.aae.UIMAEE_Constants;
+import org.apache.uima.aae.UimaAsPriorityBasedThreadFactory;
//import org.apache.uima.aae.UimaASCredentials;
import org.apache.uima.aae.UimaAsThreadFactory;
+import org.apache.uima.aae.client.UimaAsynchronousEngine;
+//import org.apache.uima.aae.UimaAsThreadFactory.UsedFor;
import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
import org.apache.uima.aae.controller.AnalysisEngineController;
import org.apache.uima.aae.controller.BaseAnalysisEngineController.ServiceState;
@@ -57,21 +64,30 @@ import org.apache.uima.aae.delegate.Dele
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.aae.message.AsynchAEMessage;
+import org.apache.uima.aae.message.MessageWrapper;
import org.apache.uima.adapter.jms.JmsConstants;
+import org.apache.uima.adapter.jms.activemq.JmsOutputChannel.BrokerConnectionEntry;
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.AbstractJmsListeningContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.jms.support.JmsUtils;
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 static final Class<?> CLASS_NAME = UimaDefaultMessageListenerContainer.class;
+ public static final String PROCESS_SELECTOR_SUFFIX = "(Command=2000 OR Command=2002)";
+ public static final String CM_PROCESS_SELECTOR_SUFFIX = "(Command=2000 OR Command=2002 OR Command=2005)";
+ public static final String GETMETA_SELECTOR_SUFFIX = "(Command=2001)";
+
+ public static final int HIGH_PRIORITY = 9;
+
private String destinationName = "";
private Endpoint endpoint;
@@ -125,6 +141,10 @@ public class UimaDefaultMessageListenerC
private volatile boolean logListenerFailure=true;
private static CountDownLatch recoveryLatch = new CountDownLatch(4);
+ // indicates if this listener is dedicated to pull targeted messages which are
+ // messages with a selector.
+ private boolean targetedListener = false;
+
public UimaDefaultMessageListenerContainer() {
super();
// reset global static. This only effects unit testing as services are deployed
@@ -143,12 +163,28 @@ public class UimaDefaultMessageListenerC
this();
this.freeCasQueueListener = freeCasQueueListener;
}
+
+ public void setTargetedListener() {
+ targetedListener = true;
+ }
+ private static boolean connectionClosedOrFailed(ActiveMQConnection connection) {
+ if (connection == null
+ || connection.isClosed()
+ || connection.isClosing()
+ || connection.isTransportFailed()) {
+ return true;
+ }
+ return false;
+ }
+
/**
* Overriden Spring's method that tries to recover from lost connection. We dont
* want to recover when the service is stopping.
*/
protected void refreshConnectionUntilSuccessful() {
- boolean doLogFailureMsg = true;
+ // System.out.println("............refreshConnectionUntilSuccessful() called");
+
+ boolean doLogFailureMsg = true;
try {
// Only one listener thread should enter to recover lost connection.
// Seems like spring recovery api is not reentrant. If multiple listeners
@@ -156,6 +192,16 @@ public class UimaDefaultMessageListenerC
// on observing jconsole attached to uima-as service with multiple listeners
// on an endpoint.
synchronized(UimaDefaultMessageListenerContainer.class ) {
+ ActiveMQConnection c = null;//(ActiveMQConnection)super.getSharedConnection();
+ try {
+ c = (ActiveMQConnection)super.getSharedConnection();
+ } catch( SharedConnectionNotInitializedException ee) {
+ // the onnectionClosedOrFailed(c) below will test for null
+ }
+ if ( !connectionClosedOrFailed(c) ) {
+ //System.out.println("............. Thread:"+Thread.currentThread().getId()+" Connection restored - returning");
+ return;
+ }
while (isRunning() && !terminating ) {
Connection tcon = null;
try {
@@ -438,6 +484,7 @@ public class UimaDefaultMessageListenerC
* @param t
*/
private void handleQueueFailure(Throwable t) {
+ // System.out.println("............handleQueueFailure() called");
final String endpointName = (getDestination() == null) ? ""
: ((ActiveMQDestination) getDestination()).getPhysicalName();
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
@@ -556,6 +603,10 @@ public class UimaDefaultMessageListenerC
* This method is called by Spring when a listener fails
*/
protected void handleListenerSetupFailure(Throwable t, boolean alreadyHandled) {
+ if ( t.getCause() instanceof InterruptedException ) {
+// System.out.println("............handleListenerFailure(Throwable t, boolean alreadyHandled) called - Cause:"+t);
+ return;
+ }
// If shutdown already, nothing to do
// If controller is stopping no need to recover the connection
if (awaitingShutdown || terminating || (controller != null && controller.isStopped()) ) {
@@ -646,10 +697,13 @@ public class UimaDefaultMessageListenerC
}
protected void handleListenerException(Throwable t) {
+ // System.out.println("............handleListenerException(Throwable t)");
+
// Already shutdown, nothing to do
if (awaitingShutdown) {
return;
}
+ /*
String endpointName = (getDestination() == null) ? ""
: ((ActiveMQDestination) getDestination()).getPhysicalName();
@@ -659,6 +713,7 @@ public class UimaDefaultMessageListenerC
"UIMAJMS_jms_listener_failed_WARNING",
new Object[] { endpointName, getBrokerUrl(), t });
}
+ */
super.handleListenerException(t);
}
@@ -676,15 +731,14 @@ public class UimaDefaultMessageListenerC
super.setConnectionFactory(connectionFactory);
}
- private void injectTaskExecutor() {
- super.setTaskExecutor(taskExecutor);
- }
- private boolean isGetMetaListener() {
+ public boolean isGetMetaListener() {
+
return getMessageSelector() != null
- && __listenerRef.getMessageSelector().equals("Command=2001");
+ && __listenerRef.getMessageSelector().endsWith(GETMETA_SELECTOR_SUFFIX);
+// && __listenerRef.getMessageSelector().endsWith("(Command=2001)");
}
-
+
private boolean isActiveMQDestination() {
return getDestination() != null && getDestination() instanceof ActiveMQDestination;
}
@@ -728,8 +782,10 @@ public class UimaDefaultMessageListenerC
**/
public void setMessageListener(Object messageListener) {
ml = messageListener;
- if (this.freeCasQueueListener) {
+ if (this.freeCasQueueListener || targetedListener ) {
super.setMessageListener(messageListener);
+ } else if ( endpoint != null && endpoint.isTempReplyDestination()) {
+ super.setMessageListener(messageListener);
}
}
public void afterPropertiesSet() {
@@ -783,13 +839,18 @@ public class UimaDefaultMessageListenerC
pluginThreadPool = true;
}
} else {
- super.setConcurrentConsumers(cc);
+ super.setConcurrentConsumers(1);
+ //if ( targetedListener ) {
+ //super.setConcurrentConsumers(1);
+ //} else {
+ // super.setConcurrentConsumers(cc);
+ //}
pluginThreadPool = true;
}
Thread t = new Thread(threadGroup, new Runnable() {
public void run() {
Destination destination = __listenerRef.getDestination();
- try {
+ try {
// Wait until the connection factory is injected by Spring
while (connectionFactory == null) {
try {
@@ -816,7 +877,7 @@ public class UimaDefaultMessageListenerC
}
// Plug in connection Factory to Spring's Listener
__listenerRef.injectConnectionFactory();
-
+
if ( pluginThreadPool ) {
setUimaASThreadPoolExecutor(cc);
}
@@ -824,13 +885,13 @@ public class UimaDefaultMessageListenerC
// 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
- if ( getMessageSelector() != null && !isGetMetaListener()) {
+ if ( !isGetMetaListener()) {
initializeTaskExecutor(cc);
}
- if ( threadPoolExecutor == null ) {
- // Plug in TaskExecutor to Spring's Listener
- __listenerRef.injectTaskExecutor();
- }
+// if ( threadPoolExecutor == null ) {
+// // Plug in TaskExecutor to Spring's Listener
+// __listenerRef.injectTaskExecutor();
+// }
if ( propagate ) {
// Notify Spring Listener that all properties are ready
__listenerRef.allPropertiesSet();
@@ -863,7 +924,6 @@ public class UimaDefaultMessageListenerC
"UIMAJMS_listener_ready__INFO",
new Object[] {controller.getComponentName(), getBrokerUrl(), getDestination() });
}
-
} catch (Exception e) {
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(),
@@ -892,7 +952,7 @@ public class UimaDefaultMessageListenerC
// by Spring on a different thread
while ((((JmsInputChannel) pojoListener).getController()) == null) {
try {
- Thread.currentThread().sleep(50);
+ Thread.currentThread().wait(50);
} catch (Exception e) {
}
}
@@ -956,7 +1016,7 @@ public class UimaDefaultMessageListenerC
public void closeConnection() throws Exception {
try {
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
- String msg = ".................... ection() Called";
+ String msg = ".................... closeConnection() Called";
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "closeConnection",
JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
new Object[] { msg });
@@ -1041,6 +1101,7 @@ public class UimaDefaultMessageListenerC
if (awaitingShutdown) {
return;
}
+
String endpointName = (getDestination() == null) ? ""
: ((ActiveMQDestination) getDestination()).getPhysicalName();
@@ -1081,13 +1142,49 @@ public class UimaDefaultMessageListenerC
public void doDestroy() {
super.destroy();
}
+ /**
+ * Called by Spring
+ */
public void setMessageSelector( String messageSelector) {
+ if ( !messageSelector.startsWith(UimaAsynchronousEngine.TargetSelectorProperty)) {
+ messageSelector = UimaAsynchronousEngine.TargetSelectorProperty+" is NULL AND("+messageSelector+")";
+ }
+ //this.doInvokeListene
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);
}
+ private boolean isProcessListener() {
+ return getMessageSelector().endsWith(PROCESS_SELECTOR_SUFFIX);
+ }
+ /**
+ * Callback called by Spring when it receives a messages. Its purpose is to assign high
+ * priority for targeted messages.
+ */
+ @SuppressWarnings("unchecked")
+ protected void doInvokeListener(@SuppressWarnings("rawtypes") SessionAwareMessageListener l, Session s, Message m) {
+ try {
+ if ( targetedListener ) {
+ m.setJMSPriority(HIGH_PRIORITY);
+ m.setJMSType("TargetMessage");
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "doInvokeListener",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_MSG_INTERCEPTOR__FINEST",
+ new Object[] {controller.getComponentName(), m.getJMSPriority(), getMessageSelector() });
+
+ }
+ }
+ //System.out.println("................ doInvokeListener() - Thread ID:"+Thread.currentThread().getId()+" Listener Class:"+l.getClass().getName()+" Controller:"+controller.getComponentName());
+ l.onMessage(m, s);
+ } catch( Throwable t ) {
+ t.printStackTrace();
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, this.getClass().getName(),
+ "destroy", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
+ "UIMAJMS_exception__WARNING", t);
+ }
+ }
public void shutdownTaskExecutor(ThreadPoolExecutor tpe, boolean stopImmediate) throws InterruptedException {
tpe.awaitTermination(50, TimeUnit.MILLISECONDS);
@@ -1134,6 +1231,10 @@ public class UimaDefaultMessageListenerC
amqc.stop();
}
awaitingShutdown = true;
+
+ if ( getMessageListener() instanceof PriorityMessageHandler ) {
+ ((PriorityMessageHandler)getMessageListener()).getQueue().put(new MessageWrapper(null, null, null,HIGH_PRIORITY));
+ }
if (taskExecutor != null && taskExecutor instanceof ThreadPoolTaskExecutor) {
// Modify task executor to terminate idle threads. While the thread terminates
// it calls destroy() method on the pinned instance of AE
@@ -1159,6 +1260,12 @@ public class UimaDefaultMessageListenerC
shutdownTaskExecutor(threadPoolExecutor, true);
}
}
+ if ( getTaskExecutor() != null ) {
+
+ if ( getTaskExecutor() instanceof ThreadPoolTaskExecutor ) {
+ ((ThreadPoolTaskExecutor)getTaskExecutor()).shutdown();
+ }
+ }
String controllerName = (__listenerRef.controller == null) ? "" :__listenerRef.controller.getComponentName();
__listenerRef.shutdown();
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
@@ -1228,32 +1335,105 @@ public class UimaDefaultMessageListenerC
}
- private void setUimaASThreadPoolExecutor(int consumentCount) throws Exception{
- super.setMessageListener(ml);
+ private void setUimaASThreadPoolExecutor(int consumerCount) throws Exception {
+ if ( isGetMetaListener() ) {
+ super.setMessageListener(ml);
+ } else if ( isFreeCasQueueListener()) {
+ super.setMessageListener(ml);
+ } else if (endpoint != null && endpoint.isTempReplyDestination()) {
+ super.setMessageListener(ml);
+ } else {
+ if ( isProcessListener() ) { //controller != null && controller instanceof PrimitiveAnalysisEngineController ) {
+
+ // Singleton handler shared by Process CAS listener and a targeted listener. The handler
+ // onMessage() is called by Spring when a message with a matching selector is available.
+ // When onMessage() is called, it adds a message to the Priority Queue
+ // PriorityMessageHandler h = PriorityMessageHandler.getInstance();
+ // System.out.println("+++++++++++++ Controller:"+controller.getComponentName()+" Listener Scaleout="+cc+" Selector:"+super.getMessageSelector());
+
+ //super.setMessageListener(h);
+ // targeted listener should not have its own thread pool because it needs to use
+ // threads created by the Process Cas Listener. Each of these threads is pinned to
+ // a dedicated AE initialized at startup. Contract says that each AE process() will be called
+ // on the same thread that initialized it. The targeted listener and process listener share
+ // the same handler where CASes are pushed onto a Blocking Priority Queue for processing.
+ if (!targetedListener && !isFreeCasQueueListener()) {
+ // System.out.println(">>>>>>>>>>> Controller:"+controller.getComponentName()+" Listener Scaleout="+cc+" Selector:"+super.getMessageSelector());
+ PriorityMessageHandler h = new PriorityMessageHandler(cc);
+
+ super.setMessageListener(h);
+ try {
+ while (controller.getInputChannel() == null) {
+ synchronized (h) {
+ h.wait(100);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ //if (isPrimitiveService()) {
+ latchToCountNumberOfTerminatedThreads = new CountDownLatch(consumerCount);
+ // 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 UimaAsPriorityBasedThreadFactory(threadGroup,
+ controller, latchToCountNumberOfTerminatedThreads);
+ //(PrimitiveAnalysisEngineController) controller, latchToCountNumberOfTerminatedThreads);
+ ((UimaAsPriorityBasedThreadFactory) tf).withQueue(h.getQueue())
+ .withChannel(controller.getInputChannel());
+
+ ((UimaAsPriorityBasedThreadFactory) tf).setDaemon(true);
+ if ( taskExecutor == null ) { // true for aggregates
+ taskExecutor = new ThreadPoolTaskExecutor();
+ }
+ // 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();
+// } else {
+//
+// }
+ }
+ } else {
+ if ( controller.getInputChannel() == null ) {
+ //System.out.println("............. Error - JmsInputChannel not set yet...");
+ } else if ( !(controller.getInputChannel() instanceof MessageListener ||
+ controller.getInputChannel() instanceof SessionAwareMessageListener) ) {
+ //System.out.println("............. Error - wrong MessageListener type - Getting this:"+controller.getInputChannel().getClass().getName());
+ }
+ super.setMessageListener(controller.getInputChannel());
+ }
+ }
// create task executor with custom thread pool for:
// 1) GetMeta request processing
// 2) ReleaseCAS request
- if ( taskExecutor == null ) {
- UimaAsThreadFactory tf = new UimaAsThreadFactory(threadGroup);
- tf.setDaemon(false);
+ if ( !targetedListener && taskExecutor == null ) {
+ UimaAsThreadFactory utf = new UimaAsThreadFactory(threadGroup);
+ utf.setDaemon(false);
+// tf.defineUsageAs(UsedFor.GetMetaHandling);//setForGetMetaHandling();
if ( isFreeCasQueueListener()) {
- tf.setThreadNamePrefix(controller.getComponentName()+" - FreeCASRequest Thread");
+ utf.setThreadNamePrefix(controller.getComponentName()+" - FreeCASRequest Thread");
} else if ( isGetMetaListener() ) {
- tf.setThreadNamePrefix(super.getBeanName()+" - Thread");
+ utf.setThreadNamePrefix(super.getBeanName()+" - Thread");
} else if ( getDestination() != null && getMessageSelector() != null ) {
- tf.setThreadNamePrefix(controller.getComponentName() + " Process Thread");
+ utf.setThreadNamePrefix(controller.getComponentName() + " Process Thread");
} else if ( endpoint != null && endpoint.isTempReplyDestination() ) {
- tf.setThreadNamePrefix(super.getBeanName()+" - Thread");
+ utf.setThreadNamePrefix(super.getBeanName()+" - Thread");
} else {
throw new Exception("Unknown Context Detected in setUimaASThreadPoolExecutor()");
}
- ExecutorService es = Executors.newFixedThreadPool(consumentCount,tf);
+ ExecutorService es = Executors.newFixedThreadPool(consumerCount,utf);
if ( es instanceof ThreadPoolExecutor ) {
threadPoolExecutor = (ThreadPoolExecutor)es;
super.setTaskExecutor(es);
}
- } else {
+ }
+ /*
+ else {
UimaAsThreadFactory tf = new UimaAsThreadFactory(threadGroup);
tf.setDaemon(true);
if ( isFreeCasQueueListener()) {
@@ -1269,8 +1449,13 @@ public class UimaDefaultMessageListenerC
}
}
+ */
}
+ private boolean isPrimitiveService() {
+ return controller != null && controller instanceof PrimitiveAnalysisEngineController &&
+ controller.getInputChannel() != null;
+ }
/**
* Called by Spring to inject TaskExecutor
@@ -1296,30 +1481,14 @@ public class UimaDefaultMessageListenerC
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);
- }
- }
- latchToCountNumberOfTerminatedThreads = new CountDownLatch(consumers);
- // 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, latchToCountNumberOfTerminatedThreads);
- ((UimaAsThreadFactory)tf).setDaemon(true);
- // 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);
+ if ( !targetedListener && !isFreeCasQueueListener() ) {
+ synchronized (mux2) {
+ while (taskExecutor == null) {
+ mux2.wait(20);
+ }
+ }
}
}
-
if ( threadPoolExecutor != null ) {
threadPoolExecutor.prestartAllCoreThreads();
}
Modified: uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/ActiveMQMessageSender.java
URL: http://svn.apache.org/viewvc/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/ActiveMQMessageSender.java?rev=1849399&r1=1849398&r2=1849399&view=diff
==============================================================================
--- uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/ActiveMQMessageSender.java (original)
+++ uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/ActiveMQMessageSender.java Thu Dec 20 14:40:58 2018
@@ -24,191 +24,400 @@ import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
+import javax.jms.IllegalStateException;
import javax.jms.JMSException;
+import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
-import org.apache.activemq.ActiveMQMessageProducer;
-import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.uima.UIMAFramework;
import org.apache.uima.aae.UIMAEE_Constants;
+import org.apache.uima.aae.client.UimaASProcessStatus;
+import org.apache.uima.aae.client.UimaASProcessStatusImpl;
+import org.apache.uima.aae.client.UimaAsynchronousEngine;
+import org.apache.uima.aae.message.AsynchAEMessage;
+import org.apache.uima.aae.message.UimaMessageValidator;
import org.apache.uima.adapter.jms.JmsConstants;
+import org.apache.uima.adapter.jms.client.BaseUIMAAsynchronousEngineCommon_impl.ClientRequest;
import org.apache.uima.adapter.jms.client.BaseUIMAAsynchronousEngineCommon_impl.SharedConnection;
+import org.apache.uima.adapter.jms.message.PendingMessage;
+import org.apache.uima.cas.CAS;
+import org.apache.uima.cas.SerialFormat;
import org.apache.uima.util.Level;
+import org.apache.uima.util.impl.ProcessTrace_impl;
/**
- * Initializes JMS session and creates JMS MessageProducer to be used for sending messages to a
- * given destination. It extends BaseMessageSender which starts the worker thread and is tasked with
- * sending messages. The application threads share a common 'queue' with the worker thread. The
- * application threads add messages to the pendingMessageList 'queue' and the worker thread consumes
- * them.
+ * Initializes JMS session and creates JMS MessageProducer to be used for
+ * sending messages to a given destination. It extends BaseMessageSender which
+ * starts the worker thread and is tasked with sending messages. The application
+ * threads share a common 'queue' with the worker thread. The application
+ * threads add messages to the pendingMessageList 'queue' and the worker thread
+ * consumes them.
*
*/
public class ActiveMQMessageSender extends BaseMessageSender {
- private static final Class CLASS_NAME = ActiveMQMessageSender.class;
+ private static final Class<?> CLASS_NAME = ActiveMQMessageSender.class;
- private volatile Connection connection = null;
+ private volatile Connection connection = null;
- private Session session = null;
+ private Session session = null;
- private MessageProducer producer = null;
+ private MessageProducer producer = null;
- private String destinationName = null;
-
- private ConcurrentHashMap<Destination, MessageProducer> producerMap = new ConcurrentHashMap<Destination, MessageProducer>();
-
- public ActiveMQMessageSender(Connection aConnection, String aDestinationName,
- BaseUIMAAsynchronousEngineCommon_impl engine) throws Exception {
- super(engine);
- connection = aConnection;
- destinationName = aDestinationName;
- }
-
- public synchronized MessageProducer getMessageProducer(Destination destination) throws Exception {
- if (producerMap.containsKey(destination)) {
- return (MessageProducer) producerMap.get(destination);
- }
- createSession();
- MessageProducer mProducer = session.createProducer(destination);
- mProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
- producerMap.put(destination, mProducer);
- return mProducer;
- }
- /**
- * This is called when a new Connection is created after broker is restarted
- */
- public void setConnection(Connection aConnection) {
- connection = aConnection;
- cleanup();
- try {
- initializeProducer();
- } catch( Exception e) {
- if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
- "setConnection", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
- "UIMAEE_exception__WARNING", e);
- }
- }
-
- }
- private String getBrokerURL() {
- try {
- return ((ActiveMQConnection) connection).getBrokerInfo().getBrokerURL();
- } catch (Exception ex) { /* handle silently. */
- }
- return "";
- }
-
- private void createSession() throws Exception {
- String broker = getBrokerURL();
- try {
- if (session == null || engine.producerInitialized == false) {
- session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- }
- } catch (JMSException e) {
- if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
- "createSession", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
- "UIMAJMS_client_failed_creating_session_INFO",
- new Object[] { destinationName, broker });
- }
- if (connection == null) {
- if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
- "createSession", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
- "UIMAJMS_client_connection_not_ready_INFO", new Object[] { broker });
- }
- } else if (((ActiveMQConnection) connection).isClosed()
- || ((ActiveMQConnection) connection).isClosing()) {
- if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
- UIMAFramework.getLogger(CLASS_NAME)
- .logrb(Level.INFO, CLASS_NAME.getName(), "createSession",
- JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
- "UIMAJMS_client_connection_closed_INFO",
- new Object[] { destinationName, broker });
- }
- }
- throw e;
- } catch (Exception e) {
- throw e;
- }
- }
-
- /**
- * Creates a jms session object used to instantiate message producer
- */
- protected void initializeProducer() throws Exception {
- createSession();
- producer = getMessageProducer(session.createQueue(destinationName));
- }
-
- /**
- * Returns the full name of the destination queue
- */
- protected String getDestinationEndpoint() throws Exception {
- return ((ActiveMQDestination) producer.getDestination()).getPhysicalName();
- }
-
- /**
- * Returns jsm MessageProducer
- */
- public MessageProducer getMessageProducer() {
- if ( engine.running && engine.producerInitialized == false ) {
- try {
- SharedConnection con = engine.lookupConnection(getBrokerURL());
- if ( con != null ) {
- setConnection(con.getConnection());
- initializeProducer();
- engine.producerInitialized = true;
- }
- } catch( Exception e) {
- e.printStackTrace();
- if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
- "getMessageProducer", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
- "UIMAEE_exception__WARNING", e);
- }
- }
- }
- return producer;
- }
-
- public TextMessage createTextMessage() throws Exception {
- if (session == null) {
- // Force initialization of Producer
- initializeProducer();
- }
- return session.createTextMessage("");
- }
-
- public BytesMessage createBytesMessage() throws Exception {
- if (session == null) {
- // Force initialization of Producer
- initializeProducer();
- }
- return session.createBytesMessage();
- }
-
- /**
- * Cleanup any jms resources used by the worker thread
- */
- protected void cleanup() {
- try {
- if (session != null) {
- session.close();
- session = null;
- }
- if (producer != null) {
- producer.close();
- producer = null;
- }
- } catch (Exception e) {
- // Ignore we are shutting down
- } finally {
- producerMap.clear();
- }
- }
+ private String destinationName = null;
+
+ private ConcurrentHashMap<Destination, MessageProducer> producerMap = new ConcurrentHashMap<Destination, MessageProducer>();
+
+ public ActiveMQMessageSender(Connection aConnection, String aDestinationName,
+ BaseUIMAAsynchronousEngineCommon_impl engine) throws Exception {
+ super(engine);
+ connection = aConnection;
+ destinationName = aDestinationName;
+ }
+
+ public synchronized MessageProducer getMessageProducer(Destination destination) throws Exception {
+ if (producerMap.containsKey(destination)) {
+ return (MessageProducer) producerMap.get(destination);
+ }
+ createSession();
+ MessageProducer mProducer = session.createProducer(destination);
+ mProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+ producerMap.put(destination, mProducer);
+ return mProducer;
+ }
+
+ /**
+ * This is called when a new Connection is created after broker is restarted
+ */
+ public void setConnection(Connection aConnection) {
+ connection = aConnection;
+ cleanup();
+ try {
+ initializeProducer();
+ } catch (Exception e) {
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), "setConnection",
+ UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING", e);
+ }
+ }
+
+ }
+
+ private String getBrokerURL() {
+ try {
+ return ((ActiveMQConnection) connection).getBrokerInfo().getBrokerURL();
+ } catch (Exception ex) { /* handle silently. */
+ }
+ return "";
+ }
+
+ private void createSession() throws Exception {
+ String broker = getBrokerURL();
+ try {
+ if (session == null || engine.producerInitialized == false) {
+ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ }
+ } catch (JMSException e) {
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "createSession",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_client_failed_creating_session_INFO",
+ new Object[] { destinationName, broker });
+ }
+ if (connection == null) {
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "createSession",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_client_connection_not_ready_INFO",
+ new Object[] { broker });
+ }
+ } else if (((ActiveMQConnection) connection).isClosed() || ((ActiveMQConnection) connection).isClosing()) {
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "createSession",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_client_connection_closed_INFO",
+ new Object[] { destinationName, broker });
+ }
+ }
+ throw e;
+ } catch (Exception e) {
+ throw e;
+ }
+ }
+
+ /**
+ * Returns the full name of the destination queue
+ */
+ protected String getDestinationEndpoint() throws Exception {
+ return ((ActiveMQDestination) producer.getDestination()).getPhysicalName();
+ }
+
+ /**
+ * Creates a jms session object used to instantiate message producer
+ */
+ protected void initializeProducer() throws Exception {
+ createSession();
+ producer = getMessageProducer(session.createQueue(destinationName));
+ }
+
+ /**
+ * Returns jsm MessageProducer
+ */
+ public MessageProducer getMessageProducer() {
+ if (engine.running && engine.producerInitialized == false) {
+ try {
+ SharedConnection con = engine.lookupConnection(getBrokerURL());
+ if (con != null) {
+ setConnection(con.getConnection());
+ initializeProducer();
+ engine.producerInitialized = true;
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), "getMessageProducer",
+ UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING", e);
+ }
+ }
+ }
+ return producer;
+ }
+
+ public TextMessage createTextMessage() throws Exception {
+ synchronized (ActiveMQMessageSender.class) {
+ if (session == null) {
+ // Force initialization of Producer
+ initializeProducer();
+ }
+ // return session.createTextMessage("");
+ TextMessage msg = null;
+ try {
+ msg = session.createTextMessage("");
+ } catch (IllegalStateException e) {
+ // stale Session
+ session = null;
+ initializeProducer();
+ msg = session.createTextMessage("");
+ }
+ return msg;
+ }
+
+ }
+
+ public BytesMessage createBytesMessage() throws Exception {
+ synchronized (ActiveMQMessageSender.class) {
+ if (session == null) {
+ // Force initialization of Producer
+ initializeProducer();
+ }
+ BytesMessage msg = null;
+ try {
+ msg = session.createBytesMessage();
+ } catch (IllegalStateException e) {
+ // stale Session
+ session = null;
+ initializeProducer();
+ msg = session.createBytesMessage();
+ }
+ return msg;
+ }
+
+ // return session.createBytesMessage();
+ }
+
+ /**
+ * Cleanup any jms resources used by the worker thread
+ */
+ protected void cleanup() {
+ try {
+ if (session != null) {
+ session.close();
+ session = null;
+ }
+ if (producer != null) {
+ producer.close();
+ producer = null;
+ }
+ } catch (Exception e) {
+ // Ignore we are shutting down
+ } finally {
+ producerMap.clear();
+ }
+ }
+
+ protected void dispatchMessage(PendingMessage pm, BaseUIMAAsynchronousEngineCommon_impl engine,
+ boolean casProcessRequest) throws Exception {
+ SharedConnection sc = engine.lookupConnection(engine.getBrokerURI());
+ ClientRequest cacheEntry = null;
+ boolean doCallback = false;
+ boolean addTimeToLive = true;
+ Session jmsSession = null;
+
+ // Check the environment for existence of NoTTL tag. If present,
+ // the deployer of the service wants to disable message expiration.
+ if (System.getProperty("NoTTL") != null) {
+ addTimeToLive = false;
+ }
+ try {
+ // long t1 = System.currentTimeMillis();
+ jmsSession = sc.getConnection().createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ // Request JMS Message from the concrete implementation
+ Message message = null;
+ // Determine if this a CAS Process Request
+ // boolean casProcessRequest = isProcessRequest(pm);
+ // Only Process request can be serialized as binary
+ if (casProcessRequest && (engine.getSerialFormat() != SerialFormat.XMI)) {
+ message = jmsSession.createBytesMessage();
+ } else {
+ message = jmsSession.createTextMessage();
+ }
+ // get the producer initialized from a valid connection
+ // producer = getMessageProducer();
+
+ Destination d = null;
+ String selector = null;
+ // UIMA-AS ver 2.10.0 + sends Free Cas request to a service targeted queue
+ // instead of a temp queue. Regular queues can be recovered in case of
+ // a broker restart. The test below will be true for UIMA-AS v. 2.10.0 +.
+ // Code in JmsOutputChannel will add the selector if the service is a CM.
+ if (pm.get(AsynchAEMessage.TargetingSelector) != null) {
+ selector = (String) pm.get(AsynchAEMessage.TargetingSelector);
+ }
+ if (selector == null && (pm.getMessageType() == AsynchAEMessage.ReleaseCAS
+ || pm.getMessageType() == AsynchAEMessage.Stop)) {
+ d = (Destination) pm.get(AsynchAEMessage.Destination);
+
+ } else {
+ d = jmsSession.createQueue(destinationName);
+ }
+ MessageProducer mProducer = jmsSession.createProducer(d);
+ mProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+ // System.out.println(">>>>>>> Time to create and initialize JMS
+ // Sesssion:"+(System.currentTimeMillis()-t1));
+ super.initializeMessage(pm, message);
+ String destination = ((ActiveMQDestination) d).getPhysicalName();
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "run",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_sending_msg_to_endpoint__FINE",
+ new Object[] {
+ UimaMessageValidator.decodeIntToString(AsynchAEMessage.Command,
+ message.getIntProperty(AsynchAEMessage.Command)),
+ UimaMessageValidator.decodeIntToString(AsynchAEMessage.MessageType,
+ message.getIntProperty(AsynchAEMessage.MessageType)),
+ destination });
+ }
+ if (casProcessRequest) {
+ cacheEntry = (ClientRequest) engine.getCache().get(pm.get(AsynchAEMessage.CasReference));
+ if (cacheEntry != null) {
+ // CAS cas = cacheEntry.getCAS();
+ // enable logging
+ if (System.getProperty("UimaAsCasTracking") != null) {
+ message.setStringProperty("UimaAsCasTracking", "enable");
+ }
+ // Target specific service instance if targeting for the CAS is provided
+ // by the client application
+ if (cacheEntry.getTargetServiceId() != null) {
+ // System.out.println("------------Client Sending CAS to Service Instance With
+ // Id:"+cacheEntry.getTargetServiceId());;
+ message.setStringProperty(UimaAsynchronousEngine.TargetSelectorProperty,
+ cacheEntry.getTargetServiceId());
+ }
+ // Use Process Timeout value for the time-to-live property in the
+ // outgoing JMS message. When this time is exceeded
+ // while the message sits in a queue, the JMS Server will remove it from
+ // the queue. What happens with the expired message depends on the
+ // configuration. Most JMS Providers create a special dead-letter queue
+ // where all expired messages are placed. NOTE: In ActiveMQ expired msgs in the
+ // DLQ
+ // are not auto evicted yet and accumulate taking up memory.
+ long timeoutValue = cacheEntry.getProcessTimeout();
+
+ if (timeoutValue > 0 && addTimeToLive) {
+ // Set high time to live value
+ message.setJMSExpiration(10 * timeoutValue);
+ }
+ if (pm.getMessageType() == AsynchAEMessage.Process) {
+ cacheEntry.setCASDepartureTime(System.nanoTime());
+ }
+ cacheEntry.setCASDepartureTime(System.nanoTime());
+
+ doCallback = true;
+
+ } else {
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
+
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), "run",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_failed_cache_lookup__WARNING",
+ new Object[] { pm.get(AsynchAEMessage.CasReference),
+ UimaMessageValidator.decodeIntToString(AsynchAEMessage.Command,
+ message.getIntProperty(AsynchAEMessage.Command)),
+ UimaMessageValidator.decodeIntToString(AsynchAEMessage.MessageType,
+ message.getIntProperty(AsynchAEMessage.MessageType)),
+ destination });
+ }
+ return; // no cacheEntry so just return
+ }
+
+ }
+ // start timers
+ if (casProcessRequest) {
+ CAS cas = cacheEntry.getCAS();
+
+ // Add the cas to a list of CASes pending reply. Also start the timer if
+ // necessary
+ engine.serviceDelegate.addCasToOutstandingList(cacheEntry.getCasReferenceId(), cas.hashCode(),
+ engine.timerPerCAS); // true=timer per cas
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "sendCAS",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_cas_added_to_pending_FINE",
+ new Object[] { cacheEntry.getCasReferenceId(), String.valueOf(cas.hashCode()),
+ engine.serviceDelegate.toString() });
+ }
+
+ } else if (pm.getMessageType() == AsynchAEMessage.GetMeta
+ && engine.serviceDelegate.getGetMetaTimeout() > 0) {
+ // timer for PING has been started in sendCAS()
+ if (!engine.serviceDelegate.isAwaitingPingReply()) {
+ engine.serviceDelegate.startGetMetaRequestTimer();
+ }
+ } else {
+ doCallback = false; // dont call onBeforeMessageSend() callback on CPC
+ }
+ // Dispatch asynchronous request to Uima AS service
+ mProducer.send(message);
+
+ if (doCallback) {
+ UimaASProcessStatus status = new UimaASProcessStatusImpl(new ProcessTrace_impl(), cacheEntry.getCAS(),
+ cacheEntry.getCasReferenceId());
+ // Notify engine before sending a message
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "run",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_calling_onBeforeMessageSend__FINE",
+ new Object[] { pm.get(AsynchAEMessage.CasReference),
+ String.valueOf(cacheEntry.getCAS().hashCode()) });
+ }
+ // Note the callback is a misnomer. The callback is made *after* the send now
+ // Application receiving this callback can consider the CAS as delivere to a
+ // queue
+ engine.onBeforeMessageSend(status);
+
+ }
+ } catch( Exception e) {
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
+ "run", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
+ "UIMAEE_exception__WARNING", e);
+ }
+ } finally {
+ if (jmsSession != null) {
+ try {
+ jmsSession.close();
+ } catch (Exception eee) {
+
+ }
+ }
+ }
+
+ }
}
\ No newline at end of file
Modified: uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngine_impl.java
URL: http://svn.apache.org/viewvc/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngine_impl.java?rev=1849399&r1=1849398&r2=1849399&view=diff
==============================================================================
--- uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngine_impl.java (original)
+++ uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngine_impl.java Thu Dec 20 14:40:58 2018
@@ -78,6 +78,7 @@ import org.apache.uima.adapter.jms.JmsCo
import org.apache.uima.adapter.jms.activemq.ConnectionFactoryIniter;
import org.apache.uima.adapter.jms.activemq.SpringContainerDeployer;
import org.apache.uima.adapter.jms.activemq.UimaEEAdminSpringContext;
+import org.apache.uima.adapter.jms.message.PendingMessage;
import org.apache.uima.adapter.jms.service.Dd2spring;
import org.apache.uima.analysis_engine.metadata.AnalysisEngineMetaData;
import org.apache.uima.cas.CAS;
@@ -99,7 +100,7 @@ public class BaseUIMAAsynchronousEngine_
implements UimaAsynchronousEngine, MessageListener, ControllerCallbackListener, ApplicationListener<ApplicationEvent>{
private static final Class CLASS_NAME = BaseUIMAAsynchronousEngine_impl.class;
- private MessageSender sender = null;
+ private ActiveMQMessageSender sender = null;
private MessageProducer producer;
@@ -137,6 +138,9 @@ public class BaseUIMAAsynchronousEngine_
protected static Lock globalLock = new ReentrantLock();
+ //private String serviceTargetSelector = null;
+
+ protected volatile boolean stopped = false;
public BaseUIMAAsynchronousEngine_impl() {
super();
UIMAFramework.getLogger(CLASS_NAME).log(Level.INFO,
@@ -219,7 +223,10 @@ public class BaseUIMAAsynchronousEngine_
SerialFormat serialFormat) throws ResourceProcessException {
try {
msg.setStringProperty(AsynchAEMessage.MessageFrom, consumerDestination.getQueueName());
-
+// // check if this message should target specific service instance
+// if ( serviceTargetSelector != null ) {
+// msg.setStringProperty(UimaAsynchronousEngine.TargetSelectorProperty,serviceTargetSelector);
+// }
msg.setStringProperty(UIMAMessage.ServerURI, brokerURI);
msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.Process);
@@ -308,12 +315,16 @@ public class BaseUIMAAsynchronousEngine_
JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_debug_msg__FINEST",
new Object[] { msg });
}
- if ( sharedConnection.getClientCount() == 1 ) {
+ if ( sharedConnection.getClientCount() <= 1 ) {
sharedConnection.destroy();
amqc.close();
}
- }
+ } else if ( sharedConnection.getClientCount() <= 1 ) {
+
+ sharedConnection.destroy();
+ amqc.close();
+ }
} catch (Exception exx) {exx.printStackTrace();}
}
// Delete client's temp reply queue from AMQ Broker
@@ -341,7 +352,13 @@ public class BaseUIMAAsynchronousEngine_
}
public void stop() {
try {
- System.out.println(this.getClass().getName()+".stop() - Stopping UIMA-AS Client");
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(
+ Level.INFO)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO,
+ CLASS_NAME.getName(), "stop",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
+ "UIMAJMS_stopping_as_client_INFO");
+ }
stopConnection();
super.doStop();
@@ -390,7 +407,7 @@ public class BaseUIMAAsynchronousEngine_
}
}
- public void setCPCMessage(Message msg) throws Exception {
+ protected void setCPCMessage(Message msg) throws Exception {
msg.setStringProperty(AsynchAEMessage.MessageFrom, consumerDestination.getQueueName());
msg.setStringProperty(UIMAMessage.ServerURI, brokerURI);
msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
@@ -402,7 +419,35 @@ public class BaseUIMAAsynchronousEngine_
((TextMessage) msg).setText("");
}
}
+ protected void setFreeCasMessage(Message msg, String aCasReferenceId, String selector) 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);
+ if ( selector != null ) {
+ msg.setStringProperty(UimaAsynchronousEngine.TargetSelectorProperty,
+ selector);
+ }
+
+ if (msg instanceof TextMessage) {
+ ((TextMessage) msg).setText("");
+ }
+ }
+ protected void setStopMessage(Message 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.Stop);
+ msg.setStringProperty(UIMAMessage.ServerURI, brokerURI);
+ msg.setJMSReplyTo(consumerDestination);
+
+ if (msg instanceof TextMessage) {
+ ((TextMessage) msg).setText("");
+ }
+ }
private boolean connectionClosedOrInvalid() {
SharedConnection sharedConnection = lookupConnection(brokerURI);
if (sharedConnection == null
@@ -419,6 +464,9 @@ public class BaseUIMAAsynchronousEngine_
}
private SharedConnection createAndInitializeAMQConnection( Semaphore semaphore, String aBrokerURI) throws Exception {
+ if ( stopped ) {
+ return null;
+ }
// This only effects Consumer
// Create AMQ specific connection validator. It uses
// AMQ specific approach to test the state of the connection
@@ -697,23 +745,7 @@ public class BaseUIMAAsynchronousEngine_
// throws an exception if verions of UIMA-AS is not compatible with UIMA SDK
VersionCompatibilityChecker.check(CLASS_NAME, "UIMA AS Client", "initialize");
-/*
- // Check for compatibility with a version of uima sdk. Only check major versions.
- if (UimaAsVersion.getMajorVersion() != UimaVersion.getMajorVersion() ) {
- UIMAFramework.getLogger(CLASS_NAME).logrb(
- Level.WARNING,
- CLASS_NAME.getName(),
- "initialize",
- UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
- "UIMAEE_incompatible_version_WARNING",
- new Object[] { "UIMA AS Client", UimaAsVersion.getUimajFullVersionString(),
- UimaVersion.getFullVersionString() });
- throw new ResourceInitializationException(new AsynchAEException(
- "Version of UIMA-AS is Incompatible with a Version of UIMA Core. UIMA-AS Version is built to depend on Core UIMA version:"
- + UimaAsVersion.getUimajFullVersionString() + " but is running with version:"
- + UimaVersion.getFullVersionString()));
- }
-*/
+
if (running) {
throw new ResourceInitializationException(new UIMA_IllegalStateException());
}
@@ -758,7 +790,11 @@ public class BaseUIMAAsynchronousEngine_
.intValue();
clientSideJmxStats.setCasPoolSize(casPoolSize);
}
-
+// if ( anApplicationContext.containsKey(UimaAsynchronousEngine.TargetSelectorProperty) ) {
+// serviceTargetSelector =
+// (String)anApplicationContext.get(UimaAsynchronousEngine.TargetSelectorProperty);
+// }
+
if (anApplicationContext.containsKey(UimaAsynchronousEngine.Timeout)) {
processTimeout = ((Integer) anApplicationContext.get(UimaAsynchronousEngine.Timeout))
.intValue();
@@ -1304,6 +1340,32 @@ public class BaseUIMAAsynchronousEngine_
// private void stopProducingCases(ClientRequest clientCachedRequest) {
private void stopProducingCases(String casReferenceId, Destination cmFreeCasQueue) {
+ PendingMessage msg = new PendingMessage(AsynchAEMessage.Stop);
+ msg.put(AsynchAEMessage.Destination, cmFreeCasQueue);
+ msg.put(AsynchAEMessage.CasReference, casReferenceId);
+ try {
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "stopProducingCases", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
+ "UIMAJMS_client_sending_stop_to_service__INFO", new Object[] {casReferenceId,cmFreeCasQueue});
+ }
+ sender.dispatchMessage(msg, this, false);
+
+ } catch (Exception ex) {
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
+ "stopProducingCases", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
+ "UIMAJMS_exception__WARNING",
+ ex);
+ }
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
+ "stopProducingCases", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
+ "UIMAJMS_client_unable_to_send_stop_to_cm__WARNING");
+ }
+ }
+
+ /*
try {
// if (clientCachedRequest.getFreeCasNotificationQueue() != null) {
if (cmFreeCasQueue != null) {
@@ -1350,6 +1412,18 @@ public class BaseUIMAAsynchronousEngine_
"UIMAJMS_exception__WARNING", e);
}
}
+ */
+ }
+ protected void dispatchFreeCasRequest(String casReferenceId, Message message) throws Exception {
+ PendingMessage msg = new PendingMessage(AsynchAEMessage.ReleaseCAS);
+// if ( message.getStringProperty(AsynchAEMessage.TargetingSelector) != null ) {
+// msg.put(AsynchAEMessage.TargetingSelector,message.getStringProperty(AsynchAEMessage.TargetingSelector) );
+// } else {
+// msg.put(AsynchAEMessage.Destination, message.getJMSReplyTo());
+// }
+ msg.put(AsynchAEMessage.Destination, message.getJMSReplyTo());
+ msg.put(AsynchAEMessage.CasReference, casReferenceId);
+ sender.dispatchMessage(msg, this, false);
}
protected MessageSender getDispatcher() {
return sender;
Modified: uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/RunRemoteAsyncAE.java
URL: http://svn.apache.org/viewvc/uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/RunRemoteAsyncAE.java?rev=1849399&r1=1849398&r2=1849399&view=diff
==============================================================================
--- uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/RunRemoteAsyncAE.java (original)
+++ uima/uv3/uima-as-v3/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/RunRemoteAsyncAE.java Thu Dec 20 14:40:58 2018
@@ -26,6 +26,7 @@ import java.io.FileWriter;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -79,6 +80,9 @@ import org.apache.uima.util.XMLInputSour
* <li>-uimaEeDebug true causes various debugging things to happen, including *not* deleting the
* generated spring file generated by running dd-2-spring. This parameter only affects deployments
* specified using the -d parameter that follow it in the command line sequence.</li>
+ * <li>-TargetServiceId specifies identifier of a service which should process a CAS. This
+ * identifier must match service's identifier. By default a service is launched with an IP:PID
+ * identifier but the identifier can be an arbitrary String. </li>
* </ul>
*/
public class RunRemoteAsyncAE {
@@ -109,6 +113,8 @@ public class RunRemoteAsyncAE {
private BufferedWriter logPerfWriter = null;
private boolean needPerfHeaders = true;
+
+ private String selector=null;
/**
* Start time of the processing - used to compute elapsed time.
@@ -196,6 +202,11 @@ public class RunRemoteAsyncAE {
cpc_timeout = Integer.parseInt(args[++i]);
} else if (args[i].equals(UimaAsynchronousEngine.UimaEeDebug)) {
appCtx.put(UimaAsynchronousEngine.UimaEeDebug, args[++i]);
+ } else if (args[i].equals("-"+UimaAsynchronousEngine.TargetSelectorProperty)) {
+ selector = args[++i];
+ System.out.println("... Target Selector:"+selector);
+ // when a service is internally deployed (-d option) it will use a selector defined below
+ System.setProperty(UimaAsynchronousEngine.TargetSelectorProperty,selector);
} else {
System.err.println("Unknown switch " + args[i]);
printUsageAndExit();
@@ -253,6 +264,11 @@ public class RunRemoteAsyncAE {
appCtx.put(UimaAsynchronousEngine.CasPoolSize, casPoolSize);
appCtx.put(UIMAFramework.CAS_INITIAL_HEAP_SIZE, Integer.valueOf(fsHeapSize / 4).toString());
+ // if configured to use service targeting, tell the client which service to use
+ String target = null;
+ if ( (target = System.getProperty(UimaAsynchronousEngine.TargetSelectorProperty) ) != null ) {
+ appCtx.put(UimaAsynchronousEngine.TargetSelectorProperty, target);
+ }
// initialize
uimaEEEngine.initialize(appCtx);
@@ -266,7 +282,13 @@ public class RunRemoteAsyncAE {
} else {
// send an empty CAS
CAS cas = uimaEEEngine.getCAS();
- uimaEEEngine.sendCAS(cas);
+ if ( selector != null ) {
+ System.out.println("Sending CAS to targeted service using sendAndReceive() - target selector:"+selector);
+ ArrayList<AnalysisEnginePerformanceMetrics> perfList = new ArrayList<AnalysisEnginePerformanceMetrics>();
+ uimaEEEngine.sendAndReceiveCAS(cas,perfList , selector);
+ } else {
+ uimaEEEngine.sendCAS(cas);
+ }
uimaEEEngine.collectionProcessingComplete();
}
if (logPerfWriter != null) {
@@ -319,7 +341,11 @@ public class RunRemoteAsyncAE {
+ "-uimaEeDebug true This is optional. Leave it out for normal operation. If specified, causes"
+ " additional debugging things to happen, including *not* deleting the generated Spring xml file generated"
+ " from running dd2spring. It only affects deployments specified using the -d parameter that follow it on"
- + " the command line");
+ + " the command line\n"
+ + " -TargetServiceId - service identifier. This is optional. If specified, this"
+ + " identifier enables targeting of a specific service. The UIMA-AS client will send CASes to"
+ + " a service instance which was launched with a matching identifier.")
+ ;
System.exit(1);
}