uima-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sc...@apache.org
Subject svn commit: r702174 - in /incubator/uima/sandbox/trunk/uima-as: uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/ uimaj-as-activemq/src/test/java/org/apache/uima/ee/test/utils/ uimaj-as-core/src/main/java/org/apache/uima/aae/ uimaj-...
Date Mon, 06 Oct 2008 15:22:04 GMT
Author: schor
Date: Mon Oct  6 08:22:03 2008
New Revision: 702174

URL: http://svn.apache.org/viewvc?rev=702174&view=rev
Log:
[UIMA-1140] Jerry's changes to use internal java queues instead of activemq ones.  Applied
3 patches dated 2008-10-6

Modified:
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsOutputChannel.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/SpringContainerDeployer.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/TempDestinationResolver.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/test/java/org/apache/uima/ee/test/utils/BaseTestSupport.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/UimaSerializer.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint_impl.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController_impl.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/error/handler/ProcessCasErrorHandler.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java
    incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/jmx/ServicePerformance.java

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java
Mon Oct  6 08:22:03 2008
@@ -42,12 +42,14 @@
 import org.apache.activemq.advisory.ConsumerListener;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.pool.PooledConnectionFactory;
 import org.apache.uima.UIMAFramework;
 import org.apache.uima.aae.controller.AnalysisEngineController;
 import org.apache.uima.aae.controller.Endpoint;
 import org.apache.uima.aae.error.AsynchAEException;
 import org.apache.uima.aae.error.InvalidMessageException;
 import org.apache.uima.aae.error.ServiceShutdownException;
+import org.apache.uima.aae.message.AsynchAEMessage;
 import org.apache.uima.adapter.jms.JmsConstants;
 import org.apache.uima.util.Level;
 import org.springframework.util.Assert;
@@ -78,11 +80,11 @@
 
 	private Map connectionMap;
 
-	private boolean retryEnabled;
+	private volatile boolean retryEnabled;
 
 	private AnalysisEngineController controller = null;
 
-	private boolean connectionAborted = false;
+	private volatile boolean connectionAborted = false;
 
 	private long connectionCreationTimestamp = 0L;
 
@@ -162,12 +164,24 @@
 			UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
 	                "openChannel", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_activemq_open__FINE",
 	                new Object[] { endpoint, serverUri });
-
+//			PooledConnectionFactory factory = new PooledConnectionFactory(brokerUri);
+//			factory.setMaximumActive(20);
 			ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUri);
+
+			factory.setDispatchAsync(true);
+      factory.setUseAsyncSend(true);
+  
+      factory.setCopyMessageOnSend(false);
+/*
+			factory.setDispatchAsync(true);
+			factory.setOptimizeAcknowledge(true);
+			factory.setUseAsyncSend(true);
+			factory.setUseCompression(false);
 			factory.setCopyMessageOnSend(false);
+	*/
 			conn = factory.createConnection();
 			connectionCreationTimestamp = System.nanoTime();
-			producerSession = conn.createSession(false, 0);
+			producerSession = conn.createSession(false, Session.DUPS_OK_ACKNOWLEDGE);
 			if ( isReplyEndpoint && delegateEndpoint.getDestination() != null )
 			{
 				producer = producerSession.createProducer(null); 
@@ -307,10 +321,13 @@
 			try
 			{
 				retryCount--;
+/*				
 				if (!((ActiveMQSession) producerSession).isRunning())
 				{
 					open();
 				}
+*/
+				
 				if (aTextMessage == null)
 				{
 					return producerSession.createTextMessage();
@@ -418,7 +435,7 @@
 		}
 	}
 
-	public boolean send(Message aMessage, boolean startTimer) 
+	public boolean send(final Message aMessage, boolean startTimer) 
 	{
 		String destinationName = "";
 
@@ -440,14 +457,28 @@
 					if ( isReplyEndpoint && delegateEndpoint.getDestination() != null  )
 					{
 						destinationName = ((ActiveMQDestination)delegateEndpoint.getDestination()).getPhysicalName();
-						producer.send((Destination)delegateEndpoint.getDestination(), aMessage);
+						if ( UIMAFramework.getLogger().isLoggable(Level.FINE))
+						{
+	            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
"send", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_sending_msg_to_endpoint__FINE", new
Object[] {destinationName });
+						}
+						synchronized(producer)
+						{
+              producer.send((Destination)delegateEndpoint.getDestination(), aMessage);
+						}
 					}
 					else
 					{
 						destinationName = ((ActiveMQQueue) producer.getDestination()).getPhysicalName();
-						UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "send",
JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_sending_msg_to_endpoint__FINE", new Object[]
{destinationName });
-						producer.send(aMessage);
+            if ( UIMAFramework.getLogger().isLoggable(Level.FINE))
+            {
+              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
"send", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_sending_msg_to_endpoint__FINE", new
Object[] {destinationName });
+            }
+            synchronized(producer)
+            {
+              producer.send(aMessage);
+            }
 					}
+					
 				if (startTimer)
 				{
 					startTimer(connectionCreationTimestamp);
@@ -457,13 +488,17 @@
 			}
 			catch ( Exception e)
 			{
-				UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "send", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAJMS_not_ableto_send_msg_INFO", new Object[] { controller.getComponentName(), destinationName,
i+1, 10 });
 				lastException = e;
 				//	If the controller has been stopped no need to send messages
 				if ( controller.isStopped())
 				{
 					return true;
 				}
+				else
+				{
+	        e.printStackTrace();
+	        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "send",
JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_not_ableto_send_msg_INFO", new Object[] { controller.getComponentName(),
destinationName, i+1, 10 });
+				}
 			}
 			try
 			{

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsInputChannel.java
Mon Oct  6 08:22:03 2008
@@ -87,8 +87,8 @@
 	
 	private ServiceInfo serviceInfo = null;
 	
-	private boolean stopped = false;
-	private boolean	channelRegistered = false;
+	private volatile boolean stopped = false;
+	private volatile boolean	channelRegistered = false;
 	
 	private List listenerContainerList = new ArrayList();
 	
@@ -531,29 +531,6 @@
 		{
 			//	Wrap JMS Message in MessageContext
 			messageContext = new JmsMessageContext( aMessage, endpointName );
-			if ( jmsSession == null )
-			{
-				jmsSession = aJmsSession;
-				sessionAckMode = jmsSession.getAcknowledgeMode();
-				if ( sessionAckMode == Session.AUTO_ACKNOWLEDGE)
-				{
-					UIMAFramework.getLogger(CLASS_NAME).logrb(Level.CONFIG, CLASS_NAME.getName(),
-			                "onMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_ack_mode__CONFIG",
-			                new Object[] { endpointName, "AUTO_ACKNOWLEDGE" });
-				}
-				else if ( sessionAckMode == Session.CLIENT_ACKNOWLEDGE)
-				{
-					UIMAFramework.getLogger(CLASS_NAME).logrb(Level.CONFIG, CLASS_NAME.getName(),
-			                "onMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_ack_mode__CONFIG",
-			                new Object[] { endpointName, "CLIENT_ACKNOWLEDGE" });
-				}
-				else 
-				{
-					UIMAFramework.getLogger(CLASS_NAME).logrb(Level.CONFIG, CLASS_NAME.getName(),
-			                "onMessage", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_ack_mode__CONFIG",
-			                new Object[] { endpointName, sessionAckMode });
-				}
-			}
 			if ( aMessage.getStringProperty(AsynchAEMessage.CasReference) == null )
 			{
 				casRefId = "CasReferenceId Not In Message";

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsOutputChannel.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsOutputChannel.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsOutputChannel.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsOutputChannel.java
Mon Oct  6 08:22:03 2008
@@ -90,9 +90,21 @@
 	
 	private String serviceProtocolList ="";
 
-	private boolean aborting = false;
+	private volatile boolean aborting = false;
 	
 	private Destination freeCASTempQueue;
+
+  private String hostIP = null;
+
+  public JmsOutputChannel()
+  {
+    try
+    {
+      hostIP = InetAddress.getLocalHost().getHostAddress();
+    }
+    catch ( Exception e) {  /* silently deal with this error */ }
+
+  }
 	/**
 	 * Sets the ActiveMQ Broker URI 
 	 */
@@ -1066,7 +1078,11 @@
 				
 				aTextMessage.setLongProperty(AsynchAEMessage.TimeToSerializeCAS, casStats.getRawCasSerializationTime());
 				aTextMessage.setLongProperty(AsynchAEMessage.TimeToDeserializeCAS, casStats.getRawCasDeserializationTime());
-				aTextMessage.setLongProperty(AsynchAEMessage.TimeInProcessCAS, casStats.getRawAnalysisTime());
+
+//				long t = getAnalysisEngineController().getServicePerformance().getRawAnalysisTime();
+//				aTextMessage.setLongProperty(AsynchAEMessage.TimeInProcessCAS, t);
+        aTextMessage.setLongProperty(AsynchAEMessage.TimeInProcessCAS, casStats.getRawAnalysisTime());
+				
 				long iT =getAnalysisEngineController().getIdleTimeBetweenProcessCalls(AsynchAEMessage.Process);

 				aTextMessage.setLongProperty(AsynchAEMessage.IdleTime, iT );
 				String lookupKey = getAnalysisEngineController().getName();
@@ -1170,12 +1186,6 @@
 		if (anEndpoint.isRemote())
 		{
 			aMessage.setStringProperty(UIMAMessage.ServerURI, getServerURI());
-			String hostIP = null;
-			try
-			{
-				hostIP = InetAddress.getLocalHost().getHostAddress();
-			}
-			catch ( Exception e) {  /* silently deal with this error */ }
 			if ( hostIP != null )
 			{
 				aMessage.setStringProperty(AsynchAEMessage.ServerIP,hostIP);
@@ -1301,6 +1311,7 @@
 			{
 				return;
 			}
+			
 			//	Get the connection object for a given endpoint
 			JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint);
 			//	Create empty JMS Text Message
@@ -1386,12 +1397,10 @@
 			{
 				startConnectionTimer = false;
 			}
-			
 			// ----------------------------------------------------
 			//	Send Request Messsage to the Endpoint
 			// ----------------------------------------------------
 			endpointConnection.send(tm, startConnectionTimer);
-
 //			if ( getAnalysisEngineController().isTopLevelComponent() )
 //			{
 //				getAnalysisEngineController().getInProcessCache().dumpContents(getAnalysisEngineController().getComponentName());

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/SpringContainerDeployer.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/SpringContainerDeployer.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/SpringContainerDeployer.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/SpringContainerDeployer.java
Mon Oct  6 08:22:03 2008
@@ -203,10 +203,7 @@
 			int concurrentConsumerCountOnReplies = getConcurrentConsumerCount(ctx);
 			// Configure and initialize vm transport in the top level aggregate.
 			// The aggregate will initialize all delegates with the vm transport.
-			if ( System.getProperty("UseVmTransport") != null )
-			{
-	      ((AggregateAnalysisEngineController) cntlr).initializeVMTransport(concurrentConsumerCountOnReplies);
-			}
+      ((AggregateAnalysisEngineController) cntlr).initializeVMTransport(concurrentConsumerCountOnReplies);
 			// Complete initialization of the aggregate by sending
 			// getMeta requests to
 			// all remote delegates (if any). Collocated delegates

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/TempDestinationResolver.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/TempDestinationResolver.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/TempDestinationResolver.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/TempDestinationResolver.java
Mon Oct  6 08:22:03 2008
@@ -48,11 +48,11 @@
 			if ( destination == null )
 			{
 				destination = session.createTemporaryQueue();
+        System.out.println(">>> Created New Temp Queue:"+destination+" Listener
Hash:"+listener.hashCode());
 				if ( listener != null )
 				{
 					listener.setDestination(destination);
 				}
-				System.out.println(">>> Created New Temp Queue:"+destination+" Listener Hash:"+listener.hashCode());
 			}
 		}
 		return destination;

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/test/java/org/apache/uima/ee/test/utils/BaseTestSupport.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/test/java/org/apache/uima/ee/test/utils/BaseTestSupport.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/test/java/org/apache/uima/ee/test/utils/BaseTestSupport.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-activemq/src/test/java/org/apache/uima/ee/test/utils/BaseTestSupport.java
Mon Oct  6 08:22:03 2008
@@ -51,7 +51,7 @@
   private static final char FS = System.getProperty("file.separator").charAt(0);
 	protected String text = "IBM today elevated five employees to the title of IBM Fellow\n
-- its most prestigious technical honor.\n The company also presented more than $2.8 million
in cash awards to employees whose technical innovation have yielded exceptional value to the
company and its customers.\nIBM conferred the accolades and awards at its 2003 Corporate Technical
Recognition Event (CTRE) in Scottsdale, Ariz. CTRE is a 40-year tradition at IBM, established
to recognize exceptional technical employees and reward them for extraordinary achievements
and contributions to the company's technology leadership.\n Our technical employees are among
the best and brightest innovators in the world.\n They share a passion for excellence that
defines their work and permeates the products and services IBM delivers to its customers,
said Nick Donofrio, senior vice president, technology and manufacturing for IBM.\n CTRE provides
the means for us to honor those who have distinguished th
 emselves as exceptional leaders among their peers.\nAmong the special honorees at the 2003
CTRE are five employees who earned the coveted distinction of IBM Fellow:- David Ferrucci
aka Dave, Grady Booch, chief scientist of Rational Software, IBM Software Group.\n Recognized
internationally for his innovative work on software architecture, modeling, and software engineering
process. \nMr. Booch is one of the original authors of the Unified Modeling Language (UML),
the industry-standard language of blueprints for software-intensive systems.- Dr. Donald Chamberlin,
researcher, IBM Almaden Research Center. An expert in relational database languages, Dr. Chamberlin
is co- inventor of SQL, the language that energized the relational database market. He has
also";
 	protected String doubleByteText = null;
-	protected boolean unexpectedException = false;
+	protected volatile boolean unexpectedException = false;
 	private static final boolean SEND_CAS_ASYNCHRONOUSLY = true;
 
 	protected static final int CPC_LATCH = 1;
@@ -328,7 +328,9 @@
 	{
 		Thread t1 = null;
 		Thread t2 = null;
-
+		serviceShutdownException = false;
+		unexpectedException = false;
+		
 		if (appCtx == null)
 		{
 			appCtx = buildContext(aBrokerURI, aTopLevelServiceQueueName, 0);
@@ -373,6 +375,10 @@
 						// Send CPC
 						aUimaEeEngine.collectionProcessingComplete();
 					}
+					else
+					{
+            System.out.println(">>>>>>>>>>>>>>>>
Not Sending CPC Due To Exception [serviceShutdownException="+serviceShutdownException+"] [isStopped="+isStopped+"]
[unexpectedException="+unexpectedException+"]");
+					}
 				}
 
 				// If have skipped CPC trip the latch
@@ -566,10 +572,6 @@
 			}
 			else if (processCountLatch != null)
 			{
-			  
-			  
-			  
-			  
 				if ( !expectedException && !(serviceShutdownException && expectingServiceShutdownException)
)
 				{
 				unexpectedException = true;
@@ -669,20 +671,40 @@
 	{
 		if (aStatus != null && aStatus.isException())
 		{
-			System.out.println(" Received CPC Reply Containing Exception");
+		  
+		  List list = aStatus.getExceptions();
+      boolean expectedException = false;
+      for( int i=0; i < list.size(); i++)
+      {
+        Exception e = (Exception)list.get(i);
+        if ( e instanceof ServiceShutdownException || 
+           (e.getCause() != null && e.getCause() instanceof ServiceShutdownException
))
+        {
+          serviceShutdownException = true;
+        }
+        else if ( ignoreException( e.getClass()))
+        {
+          expectedException = true;
+        }
+        if ( !expectedException && !expectingServiceShutdownException )
+        {
+          e.printStackTrace();
+        }
+      }
+      if ( !expectedException && !(serviceShutdownException && expectingServiceShutdownException)
)
+      {
+        System.out.println(" Received CPC Reply Containing Exception");
+        System.out.println(" ... when expecting normal CPC reply!");
+        unexpectedException = true;
+      }
 			if (exceptionCountLatch != null)
 			{
 				exceptionCountLatch.countDown();
 			}
-			else
-			{
-				if (cpcLatch != null)
-				{
-					System.out.println(" ... when expecting normal CPC reply!");
-					unexpectedException = true;
-					cpcLatch.countDown();
-				}
-			}
+      if (cpcLatch != null)
+      {
+        cpcLatch.countDown();
+      }
 		}
 		else
 		{

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/UimaSerializer.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/UimaSerializer.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/UimaSerializer.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/UimaSerializer.java
Mon Oct  6 08:22:03 2008
@@ -39,6 +39,7 @@
 import org.apache.uima.cas.TypeSystem;
 import org.apache.uima.cas.impl.AllowPreexistingFS;
 import org.apache.uima.cas.impl.OutOfTypeSystemData;
+import org.apache.uima.cas.impl.Serialization;
 import org.apache.uima.cas.impl.XCASDeserializer;
 import org.apache.uima.cas.impl.XCASSerializer;
 import org.apache.uima.cas.impl.XmiCasDeserializer;
@@ -167,17 +168,6 @@
 			writer.close();
 		}
 	}
-/*
-	public static  String serializeCasToXmi(CAS cas, XmiSerializationSharedData serSharedData)
throws IOException, SAXException
-	{
-		ByteArrayOutputStream baos = new ByteArrayOutputStream();
-		XmiCasSerializer.serialize(cas, null, baos, false, serSharedData);
-		
-		baos.close();
-		String xmiStr = new String(baos.toByteArray(), "UTF-8"); 
-		return xmiStr;
-	}
-*/
 	
 	/** Utility method for deserializing a CAS from an XMI String */
 	public static void deserializeCasFromXmi(String anXmlStr, CAS aCAS, XmiSerializationSharedData
aSharedData, 
@@ -187,31 +177,11 @@
 		
 		Reader reader = new StringReader(anXmlStr);
 		XMLReader xmlReader = XMLReaderFactory.createXMLReader();
-	    XmiCasDeserializer deser = new XmiCasDeserializer(aCAS.getTypeSystem());
+    XmiCasDeserializer deser = new XmiCasDeserializer(aCAS.getTypeSystem());
 	    ContentHandler handler = deser.getXmiCasHandler(aCAS, aLenient, aSharedData, aMergePoint);
 	    xmlReader.setContentHandler(handler);
 	    xmlReader.parse(new InputSource(reader));
 
-	    
-	    
-	    
-	    
-	    /*		
-		byte[] bytes = xmlStr.getBytes("UTF-8"); // this assumes the encoding
-													// is UTF-8, which is the
-													// default output encoding
-													// of the XmiCasSerializer
-		ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
-		try
-		{
-			XmiCasDeserializer.deserialize(bais, cas, lenient, sharedData, mergePoint);
-		}
-		finally
-		{
-			bais.close();
-		}
-*/
-		
 	}
 
 	public static void deserializeCasFromXmi(String anXmlStr, CAS aCAS, XmiSerializationSharedData
aSharedData, 
@@ -225,8 +195,7 @@
 	    ContentHandler handler = deser.getXmiCasHandler(aCAS, aLenient, aSharedData, aMergePoint,
allow);
 	    xmlReader.setContentHandler(handler); 
 	    xmlReader.parse(new InputSource(reader));
+	}
 
-
-}
 	
 }	

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java
Mon Oct  6 08:22:03 2008
@@ -20,7 +20,6 @@
 package org.apache.uima.aae.controller;
 
 import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
@@ -298,7 +297,6 @@
 		FlowContainer flow = lookupFlow(aCasReferenceId);
 			if ( flow != null )
 			{
-				//flow.aborted();
 				synchronized( flowMap )
 				{
 					flowMap.remove(aCasReferenceId);
@@ -401,10 +399,10 @@
 			UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "processCollectionCompleteReplyFromDelegate",

 					UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_recvd_cpc_reply__FINE", new Object[]
{ key });
 			
-			
-			if (sendReply && allDelegatesCompletedCollection() && getClientEndpoint()
!= null)
+			Endpoint cEndpoint = null;
+			if (sendReply && allDelegatesCompletedCollection() && (( cEndpoint = getClientEndpoint())
!= null) )
 			{
-				sendCpcReply();
+				sendCpcReply(cEndpoint);
 			}
 		}
 		catch ( Exception e)
@@ -412,7 +410,7 @@
 			throw new AsynchAEException(e);
 		}
 	}
-	private void sendCpcReply() throws Exception
+	private void sendCpcReply(Endpoint aClientEndpoint) throws Exception
 	{
 		Iterator destIterator = destinationMap.keySet().iterator();
 		while(destIterator.hasNext())
@@ -431,20 +429,22 @@
 		logStats(getComponentName(),servicePerformance);
 		
 		endProcess(AsynchAEMessage.Process);
-
-    if ( !getClientEndpoint().isRemote() && System.getProperty("UseVmTransport")
!= null)
+		if ( aClientEndpoint == null )
+		{
+		  aClientEndpoint = getClientEndpoint();
+		}
+    if ( !aClientEndpoint.isRemote())
     {
         UimaMessage message = 
-          getTransport(getClientEndpoint().getEndpoint()).produceMessage(AsynchAEMessage.CollectionProcessComplete,AsynchAEMessage.Response,getName());
+          getTransport(aClientEndpoint.getEndpoint()).produceMessage(AsynchAEMessage.CollectionProcessComplete,AsynchAEMessage.Response,getName());
         //  Send reply back to the client. Use internal (non-jms) transport
         getTransport(getName()).getUimaMessageDispatcher().dispatch(message);
     }
     else
     {
-      getOutputChannel().sendReply(AsynchAEMessage.CollectionProcessComplete, getClientEndpoint());
+      getOutputChannel().sendReply(AsynchAEMessage.CollectionProcessComplete, aClientEndpoint);
     }
 		
-		clientEndpoint = null;
 		clearStats();
 	}
 	/**
@@ -533,7 +533,7 @@
 		{
 			try
 			{
-				sendCpcReply();
+				sendCpcReply(null);
 			}
 			catch(Exception e)
 			{
@@ -550,7 +550,7 @@
 				if (endpoint != null )
 				{
 				  
-			    if ( !endpoint.isRemote() && System.getProperty("UseVmTransport") != null)
+			    if ( !endpoint.isRemote())
 			    {
 			      try
 			      {
@@ -1033,28 +1033,18 @@
 	      delegateEndpoints[i].initialize();
 	      delegateEndpoints[i].setController(this);
 	      
-	      // Dispatch GetMeta to the collocated component
-	      if ( System.getProperty("UseVmTransport") != null)
-	      {
-	        delegateEndpoints[i].setWaitingForResponse(true);
-	        try
-	        {
-	          UimaMessage message = 
-	            getTransport(delegateEndpoints[i].getEndpoint()).produceMessage(AsynchAEMessage.GetMeta,AsynchAEMessage.Request,getName());
-
-	          getTransport(delegateEndpoints[i].getEndpoint()).getUimaMessageDispatcher().dispatch(message);
-	        }
-	        catch( Exception e)
-	        {
-	          throw new AsynchAEException(e);
-	        }
-	      }
-	      else
-	      {
-	        ((AggregateAnalysisEngineController) this).dispatchMetadataRequest(delegateEndpoints[i]);
-	      }
-	      
-			}
+        delegateEndpoints[i].setWaitingForResponse(true);
+        try
+        {
+          UimaMessage message = 
+            getTransport(delegateEndpoints[i].getEndpoint()).produceMessage(AsynchAEMessage.GetMeta,AsynchAEMessage.Request,getName());
+          getTransport(delegateEndpoints[i].getEndpoint()).getUimaMessageDispatcher().dispatch(message);
+        }
+        catch( Exception e)
+        {
+          throw new AsynchAEException(e);
+        }
+      }
 		}
 	}
 
@@ -1289,13 +1279,6 @@
 		else
 		{
 		  endpoint = getReplyEndpoint( cacheEntry );
-/*
-		  endpoint = getMessageOrigin(cacheEntry.getCasReferenceId());
-			if ( endpoint == null && cacheEntry.getInputCasReferenceId() != null)
-			{
-			  endpoint = getMessageOrigin(cacheEntry.getInputCasReferenceId());
-			}
-*/	
 			dropFlow(cacheEntry.getCasReferenceId(), false);
 		}
 		if ( endpoint != null )
@@ -1330,21 +1313,20 @@
 					      cmOutstandingCASes.put(cacheEntry.getCasReferenceId(),cacheEntry.getCasReferenceId());
 	            }
 						}
-				    if ( !endpoint.isRemote() && System.getProperty("UseVmTransport") != null)
+				    if ( !endpoint.isRemote() )
 				    {
 				      sendVMMessage(AsynchAEMessage.Request, endpoint, cacheEntry);
 				    }
 				    else
 				    {
 	            // Send response to a given endpoint
-	            //getOutputChannel().sendReply(cacheEntry.getCas(), cacheEntry.getInputCasReferenceId(),
aCasReferenceId, endpoint, cacheEntry.getCasSequence());
 	            getOutputChannel().sendReply(cacheEntry, endpoint);
 
 				    }
 					}
 					else
 					{
-            if ( !endpoint.isRemote() && System.getProperty("UseVmTransport") !=
null)
+            if ( !endpoint.isRemote())
             {
               sendVMMessage(AsynchAEMessage.Response, endpoint, cacheEntry);
             }
@@ -1485,7 +1467,7 @@
 
 	private void dispatch(String aCasReferenceId, Endpoint anEndpoint) throws AsynchAEException
 	{
-    if ( !anEndpoint.isRemote() && System.getProperty("UseVmTransport") != null)
+    if ( !anEndpoint.isRemote() )
     {
       try
       {
@@ -1606,11 +1588,6 @@
 			}
 		}
 
-//		if (destinationToKeyMap.containsKey(anEndpointName))
-//		{
-//			return (String) destinationToKeyMap.get(anEndpointName);
-//		}
-
 		return key;
 	}
 
@@ -1664,6 +1641,7 @@
     }
     public void mergeTypeSystem(String aTypeSystem, String fromDestination, String fromServer)
throws AsynchAEException
 	{
+      
 		try
 		{
 			Set keys = destinationMap.keySet();
@@ -2044,13 +2022,19 @@
 		if ( serviceInfo == null )
 		{
 			serviceInfo = new AggregateServiceInfo(isCasMultiplier());
-			if ( getInputChannel() != null )
+			// if this is a top level service and the input channel not yet initialized
+			// block in getInputChannel() on the latch
+			if ( isTopLevelComponent() && getInputChannel() != null )
 			{
-				serviceInfo.setInputQueueName(getInputChannel().getName());
-				serviceInfo.setBrokerURL(getInputChannel().getServerUri());
-				serviceInfo.setDeploymentDescriptor("");
-
+	      serviceInfo.setInputQueueName(getInputChannel().getName());
+	      serviceInfo.setBrokerURL(getInputChannel().getServerUri());
 			}
+			else
+			{
+	      serviceInfo.setInputQueueName(getName());
+	      serviceInfo.setBrokerURL("vm://localhost");
+			}
+			serviceInfo.setDeploymentDescriptor("");
 			serviceInfo.setState("Running");
 		}
 		return serviceInfo;
@@ -2154,7 +2138,7 @@
 	 * Force all collocated delegates to perform any post-initialization steps.
 	 */
 	public void onInitialize()
-    {
+  {
 		//	For each collocated delegate
 		for( int i=0; i < childControllerList.size(); i++ )
 		{
@@ -2163,6 +2147,6 @@
 			//	notify the delegate 
 			delegateController.onInitialize();
 		}
-    }
+  }
 	
 }

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
Mon Oct  6 08:22:03 2008
@@ -146,21 +146,21 @@
 	
 	private JmxManagement jmxManagement = null;
 	
-	protected boolean stopped = false;
+	protected volatile boolean stopped = false;
 	
 	protected String delegateKey = null;
 	
 	protected List unregisteredDelegateList = new ArrayList();
 	
-	protected boolean allDelegatesAreRemote = false;
+	protected volatile boolean allDelegatesAreRemote = false;
 	
 	protected List controllerListeners = new ArrayList();
 	
-	protected boolean serviceInitialized = false;
+	protected volatile boolean serviceInitialized = false;
 	
 	protected ConcurrentHashMap perCasStatistics = new ConcurrentHashMap();
 
-	private boolean casMultiplier = false;
+	private volatile boolean casMultiplier = false;
 	
 	protected Object syncObject = new Object();
 	
@@ -171,7 +171,7 @@
 	
 	private Object waitmux = new Object();
 	
-	private boolean waitingForCAS = false;
+	private volatile boolean waitingForCAS = false;
 	
 	private long startTime = System.nanoTime();
 	
@@ -292,7 +292,7 @@
 			}
 		}
 		paramsMap.put(AnalysisEngine.PARAM_MBEAN_NAME_PREFIX, jmxManagement.getJmxDomain()); 
-		if ( isTopLevelComponent() && System.getProperty("UseVmTransport") != null )
+		if ( isTopLevelComponent())
 		{
 		  System.out.println("Top Level Service:"+getComponentName()+ " Configured to Use Java
VM Transport For Internal Messaging");
       UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
@@ -358,7 +358,7 @@
 			jmxManagement.registerMBean(inProcessCache, on);
 		}
 		initializeServiceStats();
-		
+
 	}
   public UimaTransport getTransport(String aKey) throws Exception
   {
@@ -391,11 +391,6 @@
 	  */
 	 public void initializeVMTransport(int parentControllerReplyConsumerCount) throws Exception
 	 {
-	   //  Check if internal messaging is enabled by checking system properties.
-	   if ( System.getProperty("UseVmTransport") == null)
-	   {
-	     return;
-	   }
 	   //  If this controller is an Aggregate Controller, force delegates to initialize
 	   //  their internal transports.
 	   if ( this instanceof AggregateAnalysisEngineController )
@@ -415,12 +410,27 @@
 	   if ( parentController != null )
      {
        UimaAsContext uimaAsContext = new UimaAsContext();
-       InputChannel ic = getInputChannel(endpointName);
-       int serviceRequestConsumerCount = ic.getConcurrentConsumerCount();
+       if ( !registeredWithJMXServer )
+       {
+         registeredWithJMXServer = true;
+         registerServiceWithJMX(jmxContext, false);
+       }
+       // Determine how many consumer threads to create. First though use the parent Aggregate
Controller
+       // to lookup this delegate key. Next fetch the delegate endpoint which contains 
+       // concurrentConsumers property.
+       String key = ((AggregateAnalysisEngineController)parentController).lookUpDelegateKey(getName());
+       int concurrentRequestConsumers = 1;
+       int concurrentReplyConsumers = 1;
+       if ( key != null )
+       {
+         Endpoint e = ((AggregateAnalysisEngineController)parentController).lookUpEndpoint(key,
false);
+         concurrentRequestConsumers = e.getConcurrentRequestConsumers();
+         concurrentReplyConsumers = e.getConcurrentReplyConsumers();
+       }
        
-       System.out.println("Controller:"+getComponentName()+" Starting Request Listener With
"+serviceRequestConsumerCount+" Concurrent Consumers. Reply Listener Configured With "+parentControllerReplyConsumerCount+"
Concurrent Consumers");
+       System.out.println("Controller:"+getComponentName()+" Starting Request Listener With
"+concurrentRequestConsumers+" Concurrent Consumers. Reply Listener Configured With "+concurrentReplyConsumers+"
Concurrent Consumers");
        
-       uimaAsContext.setConcurrentConsumerCount(serviceRequestConsumerCount);
+       uimaAsContext.setConcurrentConsumerCount(concurrentRequestConsumers);
        uimaAsContext.put("EndpointName", endpointName);
 
        UimaTransport vmTransport = getTransport(uimaAsContext);
@@ -431,7 +441,9 @@
        // Creates parent controller dispatcher for this delegate. The dispatcher is wired
        // with this delegate's listener.
        UimaAsContext uimaAsContext2 = new UimaAsContext();
-       uimaAsContext2.setConcurrentConsumerCount(parentControllerReplyConsumerCount);
+       // Set up as many reply threads as there are threads to process requests
+       uimaAsContext2.setConcurrentConsumerCount(concurrentReplyConsumers);
+//       uimaAsContext2.setConcurrentConsumerCount(parentControllerReplyConsumerCount);
        uimaAsContext2.put("EndpointName", endpointName);
        UimaTransport parentVmTransport = parentController.getTransport(uimaAsContext2, endpointName);
        parentVmTransport.produceUimaMessageDispatcher(this, vmTransport);
@@ -614,8 +626,18 @@
 		
 		registerWithAgent(servicePerformance, name );
 		servicePerformance.setIdleTime(System.nanoTime());
-		
-		ServiceInfo serviceInfo = getInputChannel().getServiceInfo();
+    ServiceInfo serviceInfo = null;
+		if ( remote )
+		{
+	    serviceInfo = getInputChannel().getServiceInfo();
+		}
+		else
+		{
+      serviceInfo = new ServiceInfo();
+      serviceInfo.setBrokerURL(getBrokerURL());
+      serviceInfo.setInputQueueName(getName());
+      serviceInfo.setState("Active");
+		}
 		ServiceInfo pServiceInfo = null;
 
 		if ( this instanceof PrimitiveAnalysisEngineController )
@@ -1475,7 +1497,13 @@
 				}
 			}
 		}
+		/*
+		 * Commented this block. It generates ShutdownException which causes problems
+		 * The shutdown of services happens ad hoc and not orderly. This whole logic
+		 * needs to be revisited.
+		 * 
 		//	Send an exception to the client if this is a top level service
+		 */ 
 		if (getOutputChannel() != null && isTopLevelComponent() )
 		{
 			Endpoint clientEndpoint = null;
@@ -1494,16 +1522,15 @@
 				}
 			}
 		}
-		//	Stop output channel
-		getOutputChannel().stop();
 		
-		adminContext = null;
 		if ( !isTopLevelComponent() )
 		{
 			adminContext = null;
 		}
 		else
 		{
+	    //  Stop output channel
+	    getOutputChannel().stop();
 			try
 			{
 				//	Remove all MBeans registered by this service
@@ -2262,7 +2289,7 @@
 	      if ( metadata != null )
 	      {
 	        
-	        if ( !anEndpoint.isRemote() && System.getProperty("UseVmTransport") != null)
+	        if ( !anEndpoint.isRemote())
 	        {
 	          ByteArrayOutputStream bos = new ByteArrayOutputStream();
 	          try

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint.java
Mon Oct  6 08:22:03 2008
@@ -140,4 +140,12 @@
 	public void setEndpointServer( String anEndpointServer );
 	
 	public String getEndpointServer();
+	
+  public void setConcurrentRequestConsumers(int aConsumerCount);
+  
+  public int getConcurrentRequestConsumers();
+
+  public void setConcurrentReplyConsumers(int aConsumerCount);
+  
+  public int getConcurrentReplyConsumers();
 }

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint_impl.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint_impl.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint_impl.java
Mon Oct  6 08:22:03 2008
@@ -105,6 +105,10 @@
 	
 	private long idleTime=0;
 	
+	private int concurrentRequestConsumers = 1;
+	
+	private int concurrentReplyConsumers = 1;
+	
 	//	This is supplied by the remote client. It needs to be
 	//	echoed back to the client. 
 	private String endpointServer = null;
@@ -562,5 +566,22 @@
 	public String getEndpointServer() {
 		return endpointServer;
 	}
-
+	
+  public void setConcurrentRequestConsumers(int aConsumerCount) {
+    concurrentRequestConsumers = aConsumerCount;
+  }
+  
+  public int getConcurrentRequestConsumers() {
+    return concurrentRequestConsumers;
+  }
+
+  public void setConcurrentReplyConsumers(int aConsumerCount)  {
+    concurrentReplyConsumers = aConsumerCount;
+    
+  }
+  
+  public int getConcurrentReplyConsumers() {
+    return concurrentReplyConsumers;
+  }
+  
 }
\ No newline at end of file

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController_impl.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController_impl.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController_impl.java
Mon Oct  6 08:22:03 2008
@@ -237,7 +237,7 @@
 			getServicePerformance().incrementAnalysisTime(super.getCpuTime()-start);
 
 			
-	    if ( !anEndpoint.isRemote() && System.getProperty("UseVmTransport") != null)
+	    if ( !anEndpoint.isRemote())
 	    {
 	        UimaMessage message = 
 	          getTransport(anEndpoint.getEndpoint()).produceMessage(AsynchAEMessage.CollectionProcessComplete,AsynchAEMessage.Response,getName());
@@ -409,8 +409,8 @@
 					}
 	        //  Increment number of CASes processed by this service
 	        sequence++;
-				}
-		      if ( !anEndpoint.isRemote() && System.getProperty("UseVmTransport") != null)
+			  }
+		      if ( !anEndpoint.isRemote())
 		      {
 		          UimaMessage message = 
 		            getTransport(anEndpoint.getEndpoint()).produceMessage(AsynchAEMessage.Process,AsynchAEMessage.Request,getName());
@@ -444,7 +444,7 @@
 			
 			// Store total time spent processing this input CAS
 			getCasStatistics(aCasReferenceId).incrementAnalysisTime(totalProcessTime);
-      if ( !anEndpoint.isRemote() && System.getProperty("UseVmTransport") != null)
+      if ( !anEndpoint.isRemote())
       {
           inputCASReturned = true;
           UimaMessage message = 
@@ -630,7 +630,7 @@
 		{
 			serviceInfo = new PrimitiveServiceInfo();
 		}
-		if ( getInputChannel() != null )
+		if ( isTopLevelComponent() && getInputChannel() != null )
 		{
 			serviceInfo.setInputQueueName(getInputChannel().getServiceInfo().getInputQueueName());
 			serviceInfo.setBrokerURL(getInputChannel().getServiceInfo().getBrokerURL());

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/error/handler/ProcessCasErrorHandler.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/error/handler/ProcessCasErrorHandler.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/error/handler/ProcessCasErrorHandler.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/error/handler/ProcessCasErrorHandler.java
Mon Oct  6 08:22:03 2008
@@ -109,7 +109,7 @@
 		{
 			try
 			{
-			   if ( !anEndpoint.isRemote() && System.getProperty("UseVmTransport") != null
)
+			   if ( !anEndpoint.isRemote())
 			   {
 			      anEndpoint.setReplyEndpoint(true);
 			      UimaTransport vmTransport = aController.getTransport(aController.getName()) ;//anEndpoint.getEndpoint());

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java
Mon Oct  6 08:22:03 2008
@@ -313,7 +313,9 @@
 				{
 					if ( delegateServicePerformance != null )
 					{
-						delegateServicePerformance.incrementAnalysisTime(timeInProcessCAS);
+           delegateServicePerformance.incrementAnalysisTime(timeInProcessCAS);
+            //  The remote delegate returns the actual analysis time not the delta
+//					  delegateServicePerformance.setAnalysisTime(timeInProcessCAS);
 					}
 				}
 				else 
@@ -328,7 +330,8 @@
 					// Update processing time for this CAS
 					if ( inputCasStats != null )
 					{
-						inputCasStats.incrementAnalysisTime(timeInProcessCAS);
+//            inputCasStats.setAnalysisTime(timeInProcessCAS);
+            inputCasStats.incrementAnalysisTime(timeInProcessCAS);
 					}
 				}
 			}

Modified: incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/jmx/ServicePerformance.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/jmx/ServicePerformance.java?rev=702174&r1=702173&r2=702174&view=diff
==============================================================================
--- incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/jmx/ServicePerformance.java
(original)
+++ incubator/uima/sandbox/trunk/uima-as/uimaj-as-core/src/main/java/org/apache/uima/aae/jmx/ServicePerformance.java
Mon Oct  6 08:22:03 2008
@@ -138,6 +138,17 @@
 		}
 	}
 
+	/**
+	 * Adjust the analysis time. This method is called when a reply is received from a remote
+	 * delegate. Each reply message containing a CAS include the current actual analysis time
+	 * This is not a delta, its the running analysis time.
+	 * 
+	 * @param anAnalysisTime
+	 */
+	public synchronized void setAnalysisTime( long anAnalysisTime )
+	{
+	  analysisTime = anAnalysisTime;
+	}
 	public void incrementAnalysisTime( long anAnalysisTime )
 	{
 		synchronized(sem)
@@ -172,7 +183,17 @@
 	
 	public long getRawAnalysisTime()
 	{
-		return analysisTime;
+    if ( controller != null )
+    {
+      return controller.getAnalysisTime();
+    }
+    else
+    {
+      synchronized( sem )
+      {
+        return analysisTime;
+      }
+    }
 	}
 
 	public long getNumberOfCASesProcessed()



Mime
View raw message