svn commit: r891266 - /incubator/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/error/ErrorHandlerBase.java

[cw...@apache.org]

Author: cwiklik
Date: Wed Dec 16 14:53:25 2009
New Revision: 891266

URL: http://svn.apache.org/viewvc?rev=891266&view=rev
Log:
UIMA-1704 Modified retryLastCommand() to force creation of a new temp queue and a listener on a remote delegate before retrying CAS 

Modified:
    incubator/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/error/ErrorHandlerBase.java

Modified: incubator/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/error/ErrorHandlerBase.java
URL: http://svn.apache.org/viewvc/incubator/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/error/ErrorHandlerBase.java?rev=891266&r1=891265&r2=891266&view=diff
==============================================================================
--- incubator/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/error/ErrorHandlerBase.java (original)
+++ incubator/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/error/ErrorHandlerBase.java Wed Dec 16 14:53:25 2009
@@ -23,12 +23,12 @@
 import java.util.Map;
 
 import org.apache.uima.UIMAFramework;
+import org.apache.uima.aae.InputChannel;
 import org.apache.uima.aae.UIMAEE_Constants;
 import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
 import org.apache.uima.aae.controller.AnalysisEngineController;
 import org.apache.uima.aae.controller.Endpoint;
 import org.apache.uima.aae.controller.PrimitiveAnalysisEngineController;
-import org.apache.uima.aae.error.handler.GetMetaErrorHandler;
 import org.apache.uima.aae.message.AsynchAEMessage;
 import org.apache.uima.aae.monitor.Monitor;
 import org.apache.uima.aae.monitor.statistics.LongNumericStatistic;
@@ -211,6 +211,19 @@
               break;
 
             case AsynchAEMessage.Process:
+              if ( anErrorContext.containsKey(AsynchAEMessage.Endpoint)) {
+                Endpoint masterEndpoint = (Endpoint)anErrorContext.get(AsynchAEMessage.Endpoint);
+
+                if (aController instanceof AggregateAnalysisEngineController && (masterEndpoint != null && masterEndpoint.getStatus() == Endpoint.FAILED)) {
+                  // Fetch an InputChannel that handles messages for a given delegate
+                  InputChannel iC = aController.getReplyInputChannel(masterEndpoint.getDelegateKey());
+                  // Create a new Listener, new Temp Queue and associate the listener with the Input Channel
+                  iC.createListener(masterEndpoint.getDelegateKey(), null);
+                  iC.removeDelegateFromFailedList(masterEndpoint.getDelegateKey());
+                  anEndpoint.setDestination(masterEndpoint.getDestination());
+                }
+                
+              }
               String casReferenceId = (String) anErrorContext.get(AsynchAEMessage.CasReference);
               System.out.println("Controller:" + aController.getComponentName()
                       + " >>>>>>>>>> Retrying Process For Cas Id:" + casReferenceId+" Delegate:"+anEndpoint.getDelegateKey());