svn commit: r1071337 - /uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java

[cw...@apache.org]

Author: cwiklik
Date: Wed Feb 16 18:06:24 2011
New Revision: 1071337

URL: http://svn.apache.org/viewvc?rev=1071337&view=rev
Log:
UIMA-2055 Modified to work with new/refactored JmsOutputChannel

Modified:
    uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java

Modified: uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java
URL: http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java?rev=1071337&r1=1071336&r2=1071337&view=diff
==============================================================================
--- uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java (original)
+++ uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AggregateAnalysisEngineController_impl.java Wed Feb 16 18:06:24 2011
@@ -481,7 +481,7 @@ public class AggregateAnalysisEngineCont
       // Send reply back to the client. Use internal (non-jms) transport
       transport.getUimaMessageDispatcher(aClientEndpoint.getEndpoint()).dispatch(message);
     } else {
-      getOutputChannel().sendReply(AsynchAEMessage.CollectionProcessComplete, aClientEndpoint);
+      getOutputChannel().sendReply(AsynchAEMessage.CollectionProcessComplete, aClientEndpoint, null, false);
     }
 
     clearStats();
@@ -610,7 +610,7 @@ public class AggregateAnalysisEngineCont
               }
             }
           } else {
-            getOutputChannel().sendRequest(AsynchAEMessage.CollectionProcessComplete, endpoint);
+            getOutputChannel().sendRequest(AsynchAEMessage.CollectionProcessComplete, null, endpoint);
             endpoint.startCollectionProcessCompleteTimer();
           }
         }
@@ -1008,8 +1008,12 @@ public class AggregateAnalysisEngineCont
             // If the delegate CM is a remote, send a Free CAS notification
             if (delegateCM.getEndpoint().isRemote()) {
               parentCasStateEntry.getFreeCasNotificationEndpoint().setCommand(AsynchAEMessage.Stop);
+              Endpoint fcEndpoint = parentCasStateEntry.getFreeCasNotificationEndpoint();
+              fcEndpoint.setReplyEndpoint(true);
+              fcEndpoint.setIsCasMultiplier(true);
+              fcEndpoint.setFreeCasEndpoint(true);
               getOutputChannel().sendRequest(AsynchAEMessage.ReleaseCAS, entry.getCasReferenceId(),
-                      parentCasStateEntry.getFreeCasNotificationEndpoint());
+                      fcEndpoint);
             }
             // Check if a request to stop generation of new CASes from the parent of
             // this CAS has been sent to the CM. The Delegate object keeps track of
@@ -1136,7 +1140,7 @@ public class AggregateAnalysisEngineCont
           // Remove a delegate endpoint from the single step list cached in the CAS entry
           Endpoint endpoint = (Endpoint_impl) entry.getDelayedSingleStepList().remove(0);
           // send the CAS to a collocated delegate from the delayed single step list.
-          dispatchProcessRequest(aCasReferenceId, endpoint, true);
+          dispatchProcessRequest(entry, endpoint, true);
         }
       } catch (Exception e) {
         if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
@@ -1244,13 +1248,13 @@ public class AggregateAnalysisEngineCont
       String analysisEngineKey = aStep.getAnalysisEngineKey();
       // Find the endpoint for the delegate
       endpoint = lookUpEndpoint(analysisEngineKey, true);
+      CacheEntry cacheEntry = getInProcessCache().getCacheEntryForCAS(aCasReferenceId);
       if (endpoint != null) {
         endpoint.setController(this);
         CasStateEntry casStateEntry = getLocalCache().lookupEntry(aCasReferenceId);
         casStateEntry.resetReplyReceived();
         if (enableCasLogMap!=null && enableCasLogMap.containsKey(analysisEngineKey)) {
           //  Get a CAS
-          CacheEntry cacheEntry = getInProcessCache().getCacheEntryForCAS(aCasReferenceId);
           CAS cas = cacheEntry.getCas();
           logCasForEndpoint(analysisEngineKey, cas);
         }
@@ -1287,7 +1291,7 @@ public class AggregateAnalysisEngineCont
                     "UIMAEE_next_step_dispatch__FINEST",
                     new Object[] { getComponentName(), analysisEngineKey, aCasReferenceId  });
           }
-          dispatchProcessRequest(aCasReferenceId, endpoint, true);
+          dispatchProcessRequest(cacheEntry, endpoint, true);
           if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
             UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
                     "simpleStep", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
@@ -1431,7 +1435,7 @@ public class AggregateAnalysisEngineCont
           casStateEntry.setNumberOfParallelDelegates(endpoints.length);
         }
         // Dispatch CAS to remote parallel delegates
-        dispatchProcessRequest(aCasReferenceId, endpoints, true);
+        dispatchProcessRequest(cacheEntry, endpoints, true);
       } else {
         // All delegates in a parallel step are co-located. Send the CAS
         // to the first delegate in the single step list.
@@ -1836,8 +1840,10 @@ public class AggregateAnalysisEngineCont
       }
       // freeCasEndpoint is a special endpoint for sending Free CAS Notification.
       if (casDropped && freeCasEndpoint != null) {
-        freeCasEndpoint.setReplyEndpoint(true);
         try {
+          freeCasEndpoint.setReplyEndpoint(true);
+          freeCasEndpoint.setIsCasMultiplier(true);
+          freeCasEndpoint.setFreeCasEndpoint(true);
           // send Free CAS Notification to a Cas Multiplier
           getOutputChannel().sendRequest(AsynchAEMessage.ReleaseCAS, aCasReferenceId,
                   freeCasEndpoint);
@@ -2343,13 +2349,13 @@ public class AggregateAnalysisEngineCont
     }
   }
 
-  private void dispatch(String aCasReferenceId, Endpoint anEndpoint) throws AsynchAEException {
+  private void dispatch(CacheEntry entry, Endpoint anEndpoint) throws AsynchAEException {
     if (!anEndpoint.isRemote()) {
       try {
         UimaTransport transport = getTransport(anEndpoint.getEndpoint());
         UimaMessage message = transport.produceMessage(AsynchAEMessage.Process,
                 AsynchAEMessage.Request, getName());
-        message.addStringProperty(AsynchAEMessage.CasReference, aCasReferenceId);
+        message.addStringProperty(AsynchAEMessage.CasReference, entry.getCasReferenceId());
         transport.getUimaMessageDispatcher(anEndpoint.getEndpoint()).dispatch(message);
 
       } catch (Exception e) {
@@ -2370,9 +2376,9 @@ public class AggregateAnalysisEngineCont
       // is in this state, delay CASes by placing them on a list of
       // CASes pending dispatch. Once the ping reply is received all
       // delayed CASes will be dispatched to the delegate.
-      if (!delayCasIfDelegateInTimedOutState(aCasReferenceId, anEndpoint.getDelegateKey())) {
+      if (!delayCasIfDelegateInTimedOutState(entry.getCasReferenceId(), anEndpoint.getDelegateKey())) {
         // The delegate is in the normal state so send it this CAS
-        getOutputChannel().sendRequest(aCasReferenceId, anEndpoint);
+        getOutputChannel().sendRequest(AsynchAEMessage.Process, entry.getCasReferenceId(), anEndpoint);
       }
     }
   }
@@ -2406,13 +2412,13 @@ public class AggregateAnalysisEngineCont
     return false; // Cas Not Delayed
   }
 
-  private void dispatchProcessRequest(String aCasReferenceId, Endpoint anEndpoint,
+  private void dispatchProcessRequest(CacheEntry entry, Endpoint anEndpoint,
           boolean addEndpointToCache) throws AsynchAEException {
     if (addEndpointToCache) {
-      getInProcessCache().addEndpoint(anEndpoint, aCasReferenceId);
+      getInProcessCache().addEndpoint(anEndpoint, entry.getCasReferenceId());
     }
     anEndpoint.setController(this);
-    dispatch(aCasReferenceId, anEndpoint);
+    dispatch(entry, anEndpoint);
 
   }
 
@@ -2439,29 +2445,38 @@ public class AggregateAnalysisEngineCont
       endpoint = lookUpEndpoint(key, true);
       getInProcessCache().addEndpoint(endpoint, aCasReferenceId);
     }
-    dispatchProcessRequest(aCasReferenceId, endpoint, addEndpointToCache);
+    if ( getInProcessCache().entryExists(aCasReferenceId) ) {
+      dispatchProcessRequest(getInProcessCache().getCacheEntryForCAS(aCasReferenceId), endpoint, addEndpointToCache);
+    }
   }
 
-  private void dispatchProcessRequest(String aCasReferenceId, Endpoint[] anEndpointList,
+  private void dispatchProcessRequest(CacheEntry entry, Endpoint[] anEndpointList,
           boolean addEndpointToCache) throws AsynchAEException {
     List<Endpoint> endpointList = new ArrayList<Endpoint>();
     for (int i = 0; i < anEndpointList.length; i++) {
       // Check if the delegate previously timed out. If so, add the CAS
       // Id to the list pending dispatch. This list holds CASes that are
       // delayed until the service responds to a Ping.
-      if (delayCasIfDelegateInTimedOutState(aCasReferenceId, anEndpointList[i].getEndpoint())) {
+      if (delayCasIfDelegateInTimedOutState(entry.getCasReferenceId(), anEndpointList[i].getEndpoint())) {
         // The CAS was delayed until the delegate responds to a Ping
         continue;
       } else {
         endpointList.add(anEndpointList[i]);
       }
       if (addEndpointToCache) {
-        getInProcessCache().addEndpoint(anEndpointList[i], aCasReferenceId);
+        getInProcessCache().addEndpoint(anEndpointList[i], entry.getCasReferenceId());
       }
     }
     Endpoint[] endpoints = new Endpoint[endpointList.size()];
     endpointList.toArray(endpoints);
-    getOutputChannel().sendRequest(aCasReferenceId, endpoints);
+    for (Endpoint endpoint : endpoints) {
+      if (endpoint.isRemote()) {
+        //  In Parallel Step serialization must be xmi.Binary serialization doesnt support merge.
+        endpoint.setSerializer("xmi");
+        getOutputChannel().sendRequest(AsynchAEMessage.Process, entry.getCasReferenceId(), endpoint);
+      }    
+    }
+
   }
 
   public boolean isDelegateKeyValid(String aDelegateKey) {
@@ -2813,7 +2828,6 @@ public class AggregateAnalysisEngineCont
       throw new AsynchAEException("Controller:" + getComponentName()
               + " Unable To Dispatch GetMeta Request. Provided Endpoint is Invalid (NULL)");
     }
-    anEndpoint.startMetadataRequestTimer();
     anEndpoint.setController(this);
     anEndpoint.setWaitingForResponse(true);
     String key = lookUpDelegateKey(anEndpoint.getEndpoint());
@@ -2839,7 +2853,7 @@ public class AggregateAnalysisEngineCont
 
       }
     }
-    getOutputChannel().sendRequest(AsynchAEMessage.GetMeta, anEndpoint);
+    getOutputChannel().sendRequest(AsynchAEMessage.GetMeta, null, anEndpoint);
   }
 
   public void retryMetadataRequest(Endpoint anEndpoint) throws AsynchAEException {
@@ -2915,11 +2929,7 @@ public class AggregateAnalysisEngineCont
 
   public void retryLastCommand(int aCommand, Endpoint anEndpoint, String aCasReferenceId) {
     try {
-      if (AsynchAEMessage.Process == aCommand) {
-        getOutputChannel().sendRequest(aCasReferenceId, anEndpoint);
-      } else {
-        getOutputChannel().sendRequest(aCommand, anEndpoint);
-      }
+      getOutputChannel().sendRequest(aCommand, aCasReferenceId, anEndpoint);
     } catch (AsynchAEException e) {
 
     }