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 2008/01/08 15:38:51 UTC
svn commit: r610003 [6/7] - in
/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src: ./ main/
main/java/ main/java/org/ main/java/org/apache/ main/java/org/apache/uima/
main/java/org/apache/uima/aae/ main/java/org/apache/uima/aae/client/
main/java/o...
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,203 @@
+package org.apache.uima.aae.handler;
+
+import org.apache.uima.UIMAFramework;
+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.PrimitiveAnalysisEngineController;
+import org.apache.uima.aae.error.AsynchAEException;
+import org.apache.uima.aae.error.ErrorContext;
+import org.apache.uima.aae.message.AsynchAEMessage;
+import org.apache.uima.aae.message.MessageContext;
+import org.apache.uima.aae.message.UIMAMessage;
+import org.apache.uima.aae.monitor.Monitor;
+import org.apache.uima.cas.CAS;
+import org.apache.uima.util.Level;
+
+public abstract class HandlerBase implements Handler
+{
+ private static final Class CLASS_NAME = HandlerBase.class;
+
+ private Handler delegateHandler;
+ private Handler parentHandler;
+ private String handlerName;
+ private AnalysisEngineController controller;
+
+ public AnalysisEngineController getController()
+ {
+ return controller;
+ }
+
+ public void setController(AnalysisEngineController controller)
+ {
+ this.controller = controller;
+ }
+ public static ErrorContext populateErrorContext( MessageContext aMessageCtx )
+ {
+ ErrorContext errorContext = new ErrorContext();
+ if ( aMessageCtx != null )
+ {
+ try
+ {
+ if ( aMessageCtx.propertyExists(AsynchAEMessage.Command))
+ {
+ errorContext.add( AsynchAEMessage.Command, aMessageCtx.getMessageIntProperty(AsynchAEMessage.Command));
+ }
+
+ if ( aMessageCtx.propertyExists(AsynchAEMessage.MessageType))
+ {
+ errorContext.add( AsynchAEMessage.MessageType, aMessageCtx.getMessageIntProperty(AsynchAEMessage.MessageType));
+ }
+
+ if ( aMessageCtx.propertyExists(AsynchAEMessage.CasReference))
+ {
+ errorContext.add( AsynchAEMessage.CasReference, aMessageCtx.getMessageIntProperty(AsynchAEMessage.CasReference));
+ }
+ errorContext.add(UIMAMessage.RawMsg, aMessageCtx.getRawMessage());
+ }
+ catch( Exception e) { /*ignore */ }
+ }
+ return errorContext;
+ }
+
+ public void validate(Object anObjectToHandle) throws AsynchAEException
+ {
+ if (anObjectToHandle == null)
+ {
+ throw new AsynchAEException("Nothing to Handle - Input Object is Null");
+ }
+ if ( !(anObjectToHandle instanceof MessageContext) )
+ {
+ throw new AsynchAEException("Invalid Object in Handler. Expected MessageContext Instead Got::"+anObjectToHandle.getClass().getName());
+ }
+ }
+
+ public boolean isHandlerForMessage( MessageContext aMessageContext, int anExpectedMessageType, int anExpectedCommand ) throws AsynchAEException
+ {
+ int messageType = aMessageContext.getMessageIntProperty(AsynchAEMessage.MessageType);
+ int command = aMessageContext.getMessageIntProperty(AsynchAEMessage.Command);
+ if (anExpectedMessageType == messageType && anExpectedCommand == command)
+ {
+ return true;
+ }
+ return false;
+ }
+ public void invokeProcess(CAS aCAS, String anInputCasReferenceId, String aNewCasReferenceId, MessageContext aMessageContext, String aNewCasProducedBy) throws AsynchAEException
+ {
+ try
+ {
+ // Use empty string as key. Top level component stats are stored under this key.
+ controller.getMonitor().incrementCount("", Monitor.ProcessCount);
+
+ if ( controller instanceof AggregateAnalysisEngineController )
+ {
+ boolean isNewCAS = aMessageContext.propertyExists(AsynchAEMessage.CasSequence);
+ if ( isNewCAS )
+ {
+ ((AggregateAnalysisEngineController)controller).process(aCAS, anInputCasReferenceId, aNewCasReferenceId, aNewCasProducedBy);
+ }
+ else
+ {
+ ((AggregateAnalysisEngineController)controller).process(aCAS, anInputCasReferenceId);
+ }
+ }
+ else if ( controller instanceof PrimitiveAnalysisEngineController )
+ {
+ ((PrimitiveAnalysisEngineController)controller).process(aCAS, anInputCasReferenceId, aMessageContext.getEndpoint());
+ }
+ else
+ {
+ throw new AsynchAEException("Invalid Controller. Expected AggregateController or PrimitiveController. Got:"+controller.getClass().getName());
+ }
+ }
+ catch ( AsynchAEException e)
+ {
+ throw e;
+ }
+ catch( Exception e)
+ {
+ throw new AsynchAEException(e);
+ }
+
+
+ }
+ public void delegate( MessageContext aMessageContext) throws AsynchAEException
+ {
+ int messageType = aMessageContext.getMessageIntProperty(AsynchAEMessage.MessageType);
+ if ( hasDelegateHandler() )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "delegate", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_dispatch_msg_to_handler__FINE",
+ new Object[] { messageType, aMessageContext.getEndpoint().getEndpoint(), getDelegate().getName() });
+
+ getDelegate().handle(aMessageContext);
+ }
+ else
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "delegate", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_no_handler_for_message__FINE",
+ new Object[] { messageType, aMessageContext.getEndpoint().getEndpoint() });
+ }
+ }
+ public HandlerBase(String aName )
+ {
+ handlerName = aName;
+ }
+
+ public Handler getDelegate()
+ {
+ return delegateHandler;
+ }
+
+ public void setParent(Handler aHandler)
+ {
+ parentHandler = aHandler;
+ }
+ public String getName()
+ {
+ return handlerName;
+ }
+ public void setDelegate(Handler aHandler)
+ {
+ delegateHandler = aHandler;
+ }
+ public boolean hasDelegateHandler()
+ {
+ return ( delegateHandler != null );
+ }
+
+ public void removeDelegate(String aHandlerName)
+ {
+ if ( handlerName.equals( aHandlerName))
+ {
+ if ( hasDelegateHandler())
+ {
+ // connect the delegate with the parent of this handler
+ ((HandlerBase)delegateHandler).setParent(parentHandler);
+ parentHandler.setDelegate(delegateHandler);
+ }
+ }
+ else
+ {
+ delegateHandler.removeDelegate(aHandlerName);
+ }
+
+ }
+
+ public void resequenceDelegateHandler(String aHandlerName, int aNewPositionInHandlerChain)
+ {
+ // TODO Auto-generated method stub
+
+ }
+ public void handle(Object anObjectToHandle) throws AsynchAEException
+ {
+ getController().getControllerLatch().waitUntilInitialized();
+ }
+
+ public void handle(Object anObjectToHandle, String expectedOutputType) throws AsynchAEException
+ {
+ }
+
+
+private static final String copyright = com.ibm.uima.copyright.Copyright.COPYRIGHT;
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/MetadataRequestHandler_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/MetadataRequestHandler_impl.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/MetadataRequestHandler_impl.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/MetadataRequestHandler_impl.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,89 @@
+package org.apache.uima.aae.handler.input;
+
+import org.apache.uima.UIMAFramework;
+import org.apache.uima.aae.UIMAEE_Constants;
+import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
+import org.apache.uima.aae.controller.Endpoint;
+import org.apache.uima.aae.error.AsynchAEException;
+import org.apache.uima.aae.error.ServiceShutdownException;
+import org.apache.uima.aae.handler.HandlerBase;
+import org.apache.uima.aae.message.AsynchAEMessage;
+import org.apache.uima.aae.message.MessageContext;
+import org.apache.uima.util.Level;
+
+public class MetadataRequestHandler_impl extends HandlerBase
+{
+ private static final Class CLASS_NAME = MetadataRequestHandler_impl.class;
+
+ public MetadataRequestHandler_impl(String aName)
+ {
+ super(aName);
+ }
+
+ public void handle(Object anObjectToHandle)// throws AsynchAEException
+ {
+
+ if (anObjectToHandle instanceof MessageContext)
+ {
+ try
+ {
+ int messageType = ((MessageContext)anObjectToHandle).getMessageIntProperty(AsynchAEMessage.MessageType);
+ int command = ((MessageContext)anObjectToHandle).getMessageIntProperty(AsynchAEMessage.Command);
+
+ // This handler handles request for metadata. Check if this is the request for metadata. If not, pass it on
+ // to the next handler in the chain.
+ if ( AsynchAEMessage.Request == messageType && AsynchAEMessage.GetMeta == command)
+ {
+
+ Endpoint endpoint = ((MessageContext)anObjectToHandle).getEndpoint();
+ if ( getController().isTopLevelComponent() )
+ {
+ endpoint.setCommand( AsynchAEMessage.GetMeta);
+ getController().cacheClientEndpoint(endpoint);
+ }
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
+ "handle", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_handling_metadata_request__FINEST",
+ new Object[] { endpoint.getEndpoint() });
+
+ getController().getControllerLatch().waitUntilInitialized();
+ // Check to see if the controller hasnt been aborted while we were waiting on the latch
+ if ( !getController().isStopped())
+ {
+ getController().sendMetadata(endpoint);
+ }
+ }
+ else
+ {
+ if ( super.hasDelegateHandler() )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
+ "handle", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_msg_for_next_handler__FINEST",
+ new Object[] { messageType });
+
+ super.getDelegate().handle(anObjectToHandle);
+ }
+ else
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handle", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_no_handler_for_message__FINE",
+ new Object[] { messageType });
+
+ }
+ }
+ }
+ catch( Exception e)
+ {
+ getController().getErrorHandlerChain().handle(e, HandlerBase.populateErrorContext( (MessageContext)anObjectToHandle ), getController());
+ }
+ }
+ else
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handle", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_invalid_context_object__INFO",
+ new Object[] {getController().getName(), anObjectToHandle.getClass().getName() });
+ }
+ }
+
+
+private static final String copyright = com.ibm.uima.copyright.Copyright.COPYRIGHT;
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/MetadataResponseHandler_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/MetadataResponseHandler_impl.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/MetadataResponseHandler_impl.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/MetadataResponseHandler_impl.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,76 @@
+package org.apache.uima.aae.handler.input;
+
+import org.apache.uima.UIMAFramework;
+import org.apache.uima.aae.UIMAEE_Constants;
+import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
+import org.apache.uima.aae.error.AsynchAEException;
+import org.apache.uima.aae.handler.HandlerBase;
+import org.apache.uima.aae.message.AsynchAEMessage;
+import org.apache.uima.aae.message.MessageContext;
+import org.apache.uima.util.Level;
+
+public class MetadataResponseHandler_impl extends HandlerBase
+{
+ private static final Class CLASS_NAME = MetadataResponseHandler_impl.class;
+
+ public MetadataResponseHandler_impl(String aName)
+ {
+ super(aName);
+ }
+
+ /**
+ * This method handles both incoming request for metadata and outgoing response
+ * containing metadata.
+ *
+ */
+ public synchronized void handle(Object anObjectToHandle)
+ {
+
+ if (anObjectToHandle instanceof MessageContext)
+ {
+ try
+ {
+ int messageType = ((MessageContext)anObjectToHandle).getMessageIntProperty(AsynchAEMessage.MessageType);
+ int command = ((MessageContext)anObjectToHandle).getMessageIntProperty(AsynchAEMessage.Command);
+ int payload = ((MessageContext)anObjectToHandle).getMessageIntProperty(AsynchAEMessage.Payload);
+
+ if ( AsynchAEMessage.Response == messageType && AsynchAEMessage.GetMeta == command)
+ {
+ // Metadata Response is only applicable to the Aggregate Controller
+ if (getController() instanceof AggregateAnalysisEngineController)
+ {
+ if (AsynchAEMessage.Exception == payload)
+ {
+ return;
+ }
+
+ String analysisEngineMetadata = ((MessageContext)anObjectToHandle).getStringMessage();
+ String fromEndpoint = ((MessageContext)anObjectToHandle).getMessageStringProperty(AsynchAEMessage.MessageFrom);
+ ((AggregateAnalysisEngineController) getController()).mergeTypeSystem(analysisEngineMetadata, fromEndpoint);
+ }
+ }
+ else
+ {
+
+ if ( super.hasDelegateHandler() )
+ {
+ super.getDelegate().handle(anObjectToHandle);
+ }
+ }
+ }
+ catch( Exception e)
+ {
+ getController().getErrorHandlerChain().handle(e, HandlerBase.populateErrorContext( (MessageContext)anObjectToHandle ), getController());
+ }
+ }
+ else
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handle", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_invalid_context_object__INFO",
+ new Object[] {getController().getName(), anObjectToHandle.getClass().getName() });
+
+ }
+ }
+
+private static final String copyright = com.ibm.uima.copyright.Copyright.COPYRIGHT;
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/ProcessRequestHandler_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/ProcessRequestHandler_impl.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/ProcessRequestHandler_impl.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/ProcessRequestHandler_impl.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,667 @@
+package org.apache.uima.aae.handler.input;
+
+import org.apache.uima.UIMAFramework;
+import org.apache.uima.aae.UIMAEE_Constants;
+import org.apache.uima.aae.UimaSerializer;
+import org.apache.uima.aae.InProcessCache.CacheEntry;
+import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
+import org.apache.uima.aae.controller.Endpoint;
+import org.apache.uima.aae.controller.PrimitiveAnalysisEngineController;
+import org.apache.uima.aae.error.AsynchAEException;
+import org.apache.uima.aae.error.ErrorContext;
+import org.apache.uima.aae.error.InvalidMessageException;
+import org.apache.uima.aae.handler.HandlerBase;
+import org.apache.uima.aae.message.AsynchAEMessage;
+import org.apache.uima.aae.message.MessageContext;
+import org.apache.uima.aae.monitor.Monitor;
+import org.apache.uima.aae.monitor.statistics.DelegateStats;
+import org.apache.uima.aae.monitor.statistics.LongNumericStatistic;
+import org.apache.uima.aae.monitor.statistics.TimerStats;
+import org.apache.uima.analysis_engine.asb.impl.FlowContainer;
+import org.apache.uima.cas.CAS;
+import org.apache.uima.cas.impl.XmiSerializationSharedData;
+import org.apache.uima.util.Level;
+
+public class ProcessRequestHandler_impl extends HandlerBase
+{
+ private static final Class CLASS_NAME = ProcessRequestHandler_impl.class;
+
+ public ProcessRequestHandler_impl(String aName)
+ {
+ super(aName);
+ }
+
+ private synchronized void cacheStats(String aCasReferenceId, long aTimeWaitingForCAS, long aTimeToDeserializeCAS )
+ throws Exception
+ {
+ CacheEntry entry = getController().getInProcessCache().getCacheEntryForCAS(aCasReferenceId);
+ entry.incrementTimeWaitingForCAS( aTimeWaitingForCAS);
+ entry.incrementTimeToDeserializeCAS(aTimeToDeserializeCAS);
+ }
+
+ private void handleProcessRequestWithXMI(MessageContext aMessageContext) throws AsynchAEException
+ {
+ CAS cas = null;
+ String casReferenceId = null;
+ long inTime = System.nanoTime();
+ boolean casRegistered = false;
+ boolean requestToFreeCasSent = false;
+ try
+ {
+ boolean isNewCAS = false;
+ String newCASProducedBy = null;
+ String remoteCasReferenceId = null;
+ // This is only used when handling CASes produced by CAS Multiplier
+
+ // Get the CAS Reference Id of the input CAS
+ casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ String inputCasReferenceId = casReferenceId;
+ if ( aMessageContext.propertyExists(AsynchAEMessage.CasSequence) )
+ {
+ if ( aMessageContext.getEndpoint().isRemote())
+ {
+ remoteCasReferenceId = casReferenceId;
+ }
+ isNewCAS = true;
+ inputCasReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.InputCasReference);
+ // This CAS came in from the CAS Multiplier. Treat it differently than the
+ // input CAS. First, in case the Aggregate needs to send this CAS to the
+ // client, retrieve the client destination by looking up the client endpoint
+ // using input CAS reference id. CASes generated by the CAS multiplier will have
+ // the same Cas Reference id.
+ Endpoint replyToEndpoint =
+ getController().
+ getInProcessCache().
+ getCacheEntryForCAS(inputCasReferenceId).getMessageOrigin();
+ //
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+
+ //String key = ((AggregateAnalysisEngineController) getController()).getLastDelegateKeyFromFlow(inputCasReferenceId);
+ newCASProducedBy = getController().
+ getInProcessCache().
+ getCacheEntryForCAS(inputCasReferenceId).getCasMultiplierKey();
+
+ //newCASProducedBy = key;
+// ((AggregateAnalysisEngineController)getController()).lookUpDelegateKey(key); //replyToEndpoint.getEndpoint());
+
+ if ( ((AggregateAnalysisEngineController)getController()).sendRequestToReleaseCas() )
+ {
+ try
+ {
+ // Change the name of the queue where the request to free a CAS will be sent.
+ aMessageContext.getEndpoint().setEndpoint(aMessageContext.getEndpoint().getEndpoint()+"CasSync");
+
+ getController().getOutputChannel().sendRequest(AsynchAEMessage.ReleaseCAS, casReferenceId, aMessageContext.getEndpoint());
+ requestToFreeCasSent = true;
+ }
+ catch( Exception e){}
+ }
+ }
+ // MessageContext contains endpoint set by the CAS Multiplier service. Overwrite
+ // this with the endpoint of the client who sent the input CAS. In case this
+ // aggregate is configured to send new CASes to the client we know where to send them.
+ aMessageContext.getEndpoint().setEndpoint(replyToEndpoint.getEndpoint());
+ aMessageContext.getEndpoint().setServerURI(replyToEndpoint.getServerURI());
+ // Set this to null so that the new CAS gets its own Cas Reference Id below
+ casReferenceId = null;
+ }
+ else if ( getController().isTopLevelComponent())
+ {
+ Endpoint replyToEndpoint = aMessageContext.getEndpoint();
+
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ ((AggregateAnalysisEngineController)getController()).addMessageOrigin(casReferenceId, replyToEndpoint);
+ }
+
+ }
+ //long arrivalTime = System.nanoTime();
+
+ // To prevent processing multiple messages with the same CasReferenceId, check the CAS cache
+ // to see if the message with a given CasReferenceId is already being processed. It is, the
+ // message contains the same request possibly issued by the caller due to timeout. Also this
+ // mechanism helps with dealing with scenario when this service is not up when the client sends
+ // request. The client can keep re-sending the same request until its timeout thresholds are
+ // exceeded. By that time, there may be multiple messages in this service queue with the same
+ // CasReferenceId. When the service finally comes back up, it will have multiple messages in
+ // its queue possibly from the same client. Only the first message for any given CasReferenceId
+ // should be processed.
+ if ( casReferenceId == null || !getController().getInProcessCache().entryExists(casReferenceId) )
+ {
+ String xmi = aMessageContext.getStringMessage();
+
+ // *****************************************************************
+ // ***** NO XMI In Message. Kick this back to sender with exception
+ // *****************************************************************
+ if ( xmi == null )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_message_has_no_cargo__INFO",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+ getController().
+ getOutputChannel().
+ sendReply(new InvalidMessageException("No XMI data in message"), casReferenceId, aMessageContext.getEndpoint(),AsynchAEMessage.Process);
+ // Dont process this empty message
+ return;
+ }
+
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
+ "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas__FINEST",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+ long t1 = System.nanoTime();
+ if ( isNewCAS )
+ {
+ cas = getController().getCasManagerWrapper().getNewCas(newCASProducedBy);
+ }
+ else
+ {
+ cas = getController().getCasManagerWrapper().getNewCas();
+ }
+ long timeWaitingForCAS = System.nanoTime() - t1;
+
+ if ( getController().isStopped() )
+ {
+ // The Controller is in shutdown state.
+ getController().dropCAS(cas);
+ return;
+ }
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas_granted__FINE",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+
+// if ( UIMAFramework.getLogger().isLoggable(Level.FINEST))
+// {
+// UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
+// "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_got_cas__FINEST",
+// new Object[] { aMessageContext.getEndpoint().getEndpoint(), xmi });
+// }
+
+ t1 = System.nanoTime();
+ XmiSerializationSharedData deserSharedData = new XmiSerializationSharedData();
+
+//if ( 1 == 1 && getController() instanceof PrimitiveAnalysisEngineController && getController().isTopLevelComponent())
+//{
+// System.out.println("Simulating Exception in Deserializer");
+// throw new IndexOutOfBoundsException();
+//}
+
+
+ UimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, -1);
+
+ long timeToDeserializeCAS = System.nanoTime() - t1;
+ LongNumericStatistic statistic;
+ if ( (statistic = getController().getMonitor().getLongNumericStatistic("",Monitor.TotalDeserializeTime)) != null )
+ {
+ statistic.increment(timeToDeserializeCAS);
+ }
+
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_deserialize_cas_time_FINE",
+ new Object[] { timeToDeserializeCAS / 1000 });
+
+
+ if (casReferenceId == null)
+ {
+ if (getController() instanceof PrimitiveAnalysisEngineController)
+ {
+ inputCasReferenceId = getController().getInProcessCache().register(cas, aMessageContext, deserSharedData);
+ }
+ else
+ {
+ casReferenceId = getController().getInProcessCache().register(cas, aMessageContext, deserSharedData);
+ if ( inputCasReferenceId == null )
+ {
+ inputCasReferenceId = casReferenceId;
+ }
+ }
+ }
+ else
+ {
+ getController().getInProcessCache().register(cas, aMessageContext, deserSharedData, casReferenceId);
+ }
+ // Set a local flag to indicate that the CAS has been added to the cache. This will be usefull when handling an exception
+ // If an exception happens before the CAS is added to the cache, the CAS needs to be dropped immediately.
+ casRegistered = true;
+ if ( casReferenceId == null )
+ {
+ getController().saveTime(inTime, inputCasReferenceId, getController().getName()); //aMessageContext.getEndpointName());
+ }
+ else
+ {
+ getController().saveTime(inTime, casReferenceId, getController().getName()); //aMessageContext.getEndpointName());
+ }
+
+ CacheEntry entry = null;
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ entry = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId);
+ if ( isNewCAS )
+ {
+ if ( newCASProducedBy != null )
+ {
+ entry.setCasMultiplierKey(newCASProducedBy);
+ }
+ if ( remoteCasReferenceId != null )
+ {
+ entry.setRemoteCMCasReferenceId(remoteCasReferenceId);
+ }
+ if ( requestToFreeCasSent )
+ {
+ entry.setSendRequestToFreeCas(false);
+ }
+ }
+ DelegateStats stats = new DelegateStats();
+ if ( entry.getStat() == null )
+ {
+ entry.setStat(stats);
+ // Add entry for self (this aggregate). MessageContext.getEndpointName()
+ // returns the name of the queue receiving the message.
+ stats.put(getController().getServiceEndpointName(), new TimerStats());
+ }
+ else
+ {
+ if (!stats.containsKey(getController().getServiceEndpointName()))
+ {
+ stats.put(getController().getServiceEndpointName(), new DelegateStats());
+ }
+ }
+ }
+
+ cacheStats( inputCasReferenceId, timeWaitingForCAS, timeToDeserializeCAS);
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_deserialized_cas_ready_to_process_FINE",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+
+
+ cacheProcessCommandInClientEndpoint();
+
+ if ( getController().isStopped() )
+ {
+ if ( entry != null )
+ {
+ // The Controller is in shutdown state.
+ getController().dropCAS( entry.getCasReferenceId(), true);
+ return;
+ }
+ }
+ invokeProcess(cas, inputCasReferenceId, casReferenceId, aMessageContext, newCASProducedBy);
+ }
+ else
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_duplicate_request__INFO",
+ new Object[] { casReferenceId});
+ }
+
+ }
+ catch ( Exception e)
+ {
+ e.printStackTrace();
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), "handleProcessRequestWithXMI",
+ UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING", e);
+
+ ErrorContext errorContext = new ErrorContext();
+ errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
+ errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
+ errorContext.add(AsynchAEMessage.CasReference, casReferenceId );
+ if ( casRegistered == false )
+ {
+ getController().dropCAS(cas);
+ }
+ getController().getErrorHandlerChain().handle(e, errorContext, getController());
+ }
+
+ }
+ private void handleProcessRequestWithXCAS(MessageContext aMessageContext) throws AsynchAEException
+ {
+
+ try
+ {
+ boolean isNewCAS = false;
+ String newCASProducedBy = null;
+
+ // This is only used when handling CASes produced by CAS Multiplier
+
+ // Get the CAS Reference Id of the input CAS
+ String casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ String inputCasReferenceId = casReferenceId;
+
+ if ( aMessageContext.propertyExists(AsynchAEMessage.CasSequence) )
+ {
+ isNewCAS = true;
+ // This CAS came in from the CAS Multiplier. Treat it differently than the
+ // input CAS. First, in case the Aggregate needs to send this CAS to the
+ // client, retrieve the client destination by looking up the client endpoint
+ // using input CAS reference id. CASes generated by the CAS multiplier will have
+ // the same Cas Reference id.
+ Endpoint replyToEndpoint =
+ getController().
+ getInProcessCache().
+ getCacheEntryForCAS(casReferenceId).getMessageOrigin();
+
+ //
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ newCASProducedBy =
+ ((AggregateAnalysisEngineController)getController()).lookUpDelegateKey(replyToEndpoint.getEndpoint());
+ }
+ // MessageContext contains endpoint set by the CAS Multiplier service. Overwrite
+ // this with the endpoint of the client who sent the input CAS. In case this
+ // aggregate is configured to send new CASes to the client we know where to send them.
+ aMessageContext.getEndpoint().setEndpoint(replyToEndpoint.getEndpoint());
+ aMessageContext.getEndpoint().setServerURI(replyToEndpoint.getServerURI());
+ inputCasReferenceId = String.valueOf(casReferenceId);
+ // Set this to null so that the new CAS gets its own Cas Reference Id below
+ casReferenceId = null;
+ }
+
+ long arrivalTime = System.nanoTime();
+ getController().saveTime(arrivalTime, casReferenceId, getController().getName());//aMessageContext.getEndpointName());
+
+ // To prevent processing multiple messages with the same CasReferenceId, check the CAS cache
+ // to see if the message with a given CasReferenceId is already being processed. It is, the
+ // message contains the same request possibly issued by the caller due to timeout. Also this
+ // mechanism helps with dealing with scenario when this service is not up when the client sends
+ // request. The client can keep re-sending the same request until its timeout thresholds are
+ // exceeded. By that time, there may be multiple messages in this service queue with the same
+ // CasReferenceId. When the service finally comes back up, it will have multiple messages in
+ // its queue possibly from the same client. Only the first message for any given CasReferenceId
+ // should be processed.
+ if ( casReferenceId == null || !getController().getInProcessCache().entryExists(casReferenceId) )
+ {
+ String xmi = aMessageContext.getStringMessage();
+
+ // *****************************************************************
+ // ***** NO XMI In Message. Kick this back to sender with exception
+ // *****************************************************************
+ if ( xmi == null )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handleProcessRequestWithXCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_message_has_no_cargo__INFO",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+ getController().
+ getOutputChannel().
+ sendReply(new InvalidMessageException("No XMI data in message"), casReferenceId, aMessageContext.getEndpoint(),AsynchAEMessage.Process);
+ // Dont process this empty message
+ return;
+ }
+
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithXCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas__FINE",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+
+ CAS cas = getController().getCasManagerWrapper().getNewCas();
+
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithXCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas_granted__FINE",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+
+ XmiSerializationSharedData deserSharedData = new XmiSerializationSharedData();
+ UimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, -1);
+
+ if (casReferenceId == null)
+ {
+ casReferenceId = getController().getInProcessCache().register(cas, aMessageContext, deserSharedData);
+ }
+ else
+ {
+ if (getController() instanceof PrimitiveAnalysisEngineController)
+ {
+ getController().getInProcessCache().register(cas, aMessageContext, deserSharedData, casReferenceId);
+ }
+ }
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithXCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_deserialized_cas_ready_to_process_FINE",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+
+ cacheProcessCommandInClientEndpoint();
+
+ invokeProcess(cas, inputCasReferenceId, casReferenceId, aMessageContext, newCASProducedBy);
+ }
+ else
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handleProcessRequestWithXCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_duplicate_request__INFO",
+ new Object[] { casReferenceId});
+ }
+
+ }
+ catch ( AsynchAEException e)
+ {
+ throw e;
+ }
+ catch ( Exception e)
+ {
+ throw new AsynchAEException(e);
+ }
+
+ }
+
+ private void handleProcessRequestWithCASReference(MessageContext aMessageContext) throws AsynchAEException
+ {
+ boolean isNewCAS = false;
+ String newCASProducedBy = null;
+
+ if ( getController().isStopped() )
+ {
+ return;
+ }
+
+ try
+ {
+ String casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+
+ if ( casReferenceId == null )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_message_has_cas_refid__INFO",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+
+ getController().
+ getOutputChannel().
+ sendReply(new InvalidMessageException("No Cas Reference Id Received From Delegate In message"), null, aMessageContext.getEndpoint(),AsynchAEMessage.Process);
+ // Dont process this empty message
+ return;
+
+ }
+ Endpoint replyToEndpoint = aMessageContext.getEndpoint();
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ ((AggregateAnalysisEngineController)getController()).addMessageOrigin(casReferenceId, replyToEndpoint);
+ }
+
+ // This is only used when handling CASes produced by CAS Multiplier
+ String inputCasReferenceId = null;
+ CAS cas = null;
+
+ if ( aMessageContext.propertyExists(AsynchAEMessage.CasSequence) )
+ {
+ isNewCAS = true;
+ if ( replyToEndpoint == null )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_no_endpoint_for_reply__INFO",
+ new Object[] { casReferenceId });
+ }
+ //
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ getController().getInProcessCache().setCasProducer(casReferenceId, replyToEndpoint.getEndpoint());
+ newCASProducedBy =
+ ((AggregateAnalysisEngineController)getController()).lookUpDelegateKey(replyToEndpoint.getEndpoint());
+ replyToEndpoint.setIsCasMultiplier(true);
+ }
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_new_cas__FINE",
+ new Object[] { casReferenceId, newCASProducedBy });
+ // MessageContext contains endpoint set by the CAS Multiplier service. Overwrite
+ // this with the endpoint of the client who sent the input CAS. In case this
+ // aggregate is configured to send new CASes to the client we know where to send them.
+ if ( aMessageContext.getEndpoint() != null )
+ {
+ aMessageContext.getEndpoint().setEndpoint(replyToEndpoint.getEndpoint());
+ aMessageContext.getEndpoint().setServerURI(replyToEndpoint.getServerURI());
+ }
+
+ inputCasReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.InputCasReference);
+ }
+
+ cas = getController().getInProcessCache().getCasByReference(casReferenceId);
+
+ long arrivalTime = System.nanoTime();
+ getController().saveTime(arrivalTime, casReferenceId, getController().getName());//aMessageContext.getEndpointName());
+
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_analyzing_cas__FINE",
+ new Object[] { casReferenceId});
+
+ // Save Process command in the client endpoint.
+ cacheProcessCommandInClientEndpoint();
+
+ if ( getController().isStopped() )
+ {
+ return;
+ }
+
+ if ( isNewCAS )
+ {
+ invokeProcess(cas, inputCasReferenceId, casReferenceId, aMessageContext, newCASProducedBy);
+ }
+ else
+ {
+ invokeProcess(cas, casReferenceId, null, aMessageContext, newCASProducedBy);
+ }
+ }
+ catch ( AsynchAEException e)
+ {
+ throw e;
+ }
+ catch ( Exception e)
+ {
+ throw new AsynchAEException(e);
+ }
+
+ }
+
+
+
+ private void cacheProcessCommandInClientEndpoint()
+ {
+ Endpoint clientEndpoint = getController().getClientEndpoint();
+ if ( clientEndpoint != null )
+ {
+ clientEndpoint.setCommand(AsynchAEMessage.Process);
+ }
+ }
+
+ private void handleCollectionProcessCompleteRequest(MessageContext aMessageContext)
+ throws AsynchAEException
+ {
+ Endpoint replyToEndpoint = aMessageContext.getEndpoint();
+ getController().collectionProcessComplete(replyToEndpoint);
+ }
+
+ private void handleReleaseCASRequest(MessageContext aMessageContext)
+ {
+
+ if ( getController() instanceof PrimitiveAnalysisEngineController )
+ {
+ ( (PrimitiveAnalysisEngineController)getController()).releaseNextCas();
+ }
+/*
+ try
+ {
+ String casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleReleaseCASRequest", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_release_cas_req__FINE",
+ new Object[] { getController().getName(), casReferenceId });
+System.out.println(getController().getName()+" ::::::: Processing Release CAS Request:"+casReferenceId);
+
+ if ( casReferenceId != null && getController().getInProcessCache().entryExists(casReferenceId))
+ {
+ getController().dropCAS(casReferenceId, true);
+ }
+ }
+ catch( Exception e)
+ {
+ getController().getErrorHandlerChain().handle(e, HandlerBase.populateErrorContext( (MessageContext)aMessageContext ), getController());
+ }
+*/
+ }
+
+ private void handleStopRequest(MessageContext aMessageContext)
+ {
+ System.out.println("###################Controller::"+getController().getComponentName()+" Received <<<STOP>>> Request");
+ if ( getController() instanceof PrimitiveAnalysisEngineController )
+ {
+ try
+ {
+ String casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ ( (PrimitiveAnalysisEngineController)getController()).addAbortedCasReferenceId(casReferenceId);
+ }
+ catch( Exception e){}
+ }
+ }
+ /**
+ * Main method called by the predecessor handler.
+ *
+ *
+ */
+ public void handle(Object anObjectToHandle) //throws AsynchAEException
+ {
+ try
+ {
+ super.validate(anObjectToHandle);
+
+ MessageContext messageContext = (MessageContext) anObjectToHandle;
+ if ( isHandlerForMessage(messageContext, AsynchAEMessage.Request, AsynchAEMessage.Process) ||
+ isHandlerForMessage(messageContext, AsynchAEMessage.Request, AsynchAEMessage.CollectionProcessComplete) ||
+ isHandlerForMessage(messageContext, AsynchAEMessage.Request, AsynchAEMessage.ReleaseCAS ) ||
+ isHandlerForMessage(messageContext, AsynchAEMessage.Request, AsynchAEMessage.Stop )
+ )
+ {
+ int payload = ((MessageContext) anObjectToHandle).getMessageIntProperty(AsynchAEMessage.Payload);
+ int command = messageContext.getMessageIntProperty(AsynchAEMessage.Command);
+
+ getController().getControllerLatch().waitUntilInitialized();
+
+
+ if (AsynchAEMessage.CASRefID == payload)
+ {
+ handleProcessRequestWithCASReference(messageContext);
+ }
+ else if (AsynchAEMessage.XMIPayload == payload)
+ {
+ handleProcessRequestWithXMI(messageContext);
+ }
+ else if (AsynchAEMessage.XCASPayload == payload)
+ {
+ handleProcessRequestWithXCAS(messageContext);
+ }
+ else if ( AsynchAEMessage.None == payload && AsynchAEMessage.CollectionProcessComplete == command)
+ {
+ handleCollectionProcessCompleteRequest(messageContext);
+ }
+ else if ( AsynchAEMessage.None == payload && AsynchAEMessage.ReleaseCAS == command)
+ {
+ handleReleaseCASRequest(messageContext);
+ }
+ else if ( AsynchAEMessage.None == payload && AsynchAEMessage.Stop == command)
+ {
+ handleStopRequest(messageContext);
+ }
+ // Handled Request
+ return;
+ }
+ // Not a Request nor Command. Delegate to the next handler in the chain
+ super.delegate(messageContext);
+ }
+ catch( Exception e)
+ {
+ getController().getErrorHandlerChain().handle(e, HandlerBase.populateErrorContext( (MessageContext)anObjectToHandle ), getController());
+ }
+ }
+
+private static final String copyright = com.ibm.uima.copyright.Copyright.COPYRIGHT;
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/ProcessResponseHandler.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/ProcessResponseHandler.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/ProcessResponseHandler.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/handler/input/ProcessResponseHandler.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,674 @@
+package org.apache.uima.aae.handler.input;
+
+import java.util.HashMap;
+import java.util.Iterator;
+
+import org.apache.uima.UIMAFramework;
+import org.apache.uima.aae.UIMAEE_Constants;
+import org.apache.uima.aae.UimaSerializer;
+import org.apache.uima.aae.InProcessCache.CacheEntry;
+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.AsynchAEException;
+import org.apache.uima.aae.error.ErrorContext;
+import org.apache.uima.aae.error.ExpiredMessageException;
+import org.apache.uima.aae.error.ServiceShutdownException;
+import org.apache.uima.aae.error.UimaEEServiceException;
+import org.apache.uima.aae.handler.HandlerBase;
+import org.apache.uima.aae.jmx.ServicePerformance;
+import org.apache.uima.aae.message.AsynchAEMessage;
+import org.apache.uima.aae.message.MessageContext;
+import org.apache.uima.aae.monitor.Monitor;
+import org.apache.uima.aae.monitor.statistics.DelegateStats;
+import org.apache.uima.aae.monitor.statistics.LongNumericStatistic;
+import org.apache.uima.aae.monitor.statistics.TimerStats;
+import org.apache.uima.cas.CAS;
+import org.apache.uima.cas.impl.XmiSerializationSharedData;
+import org.apache.uima.util.Level;
+
+public class ProcessResponseHandler extends HandlerBase
+{
+ private static final Class CLASS_NAME = ProcessResponseHandler.class;
+
+ private Object monitor = new Object();
+
+ public ProcessResponseHandler(String aName)
+ {
+ super(aName);
+ }
+
+ private void computeStats(MessageContext aMessageContext, String aCasReferenceId) throws AsynchAEException
+ {
+ if (aMessageContext.propertyExists(AsynchAEMessage.TimeInService))
+ {
+ long departureTime = getController().getTime(aCasReferenceId, aMessageContext.getEndpoint().getEndpoint());
+ long currentTime = System.nanoTime();
+ long roundTrip = currentTime - departureTime;
+ long timeInService = aMessageContext.getMessageLongProperty(AsynchAEMessage.TimeInService);
+ long totalTimeInComms = currentTime - (departureTime - timeInService);
+
+
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "computeStats", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_show_roundtrip_time__FINE",
+ new Object[] { aCasReferenceId, aMessageContext.getEndpoint(),(double) roundTrip / (double) 1000000 });
+
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "computeStats", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_show_time_spent_in_delegate__FINE",
+ new Object[] { aCasReferenceId, (double) timeInService / (double) 1000000, aMessageContext.getEndpoint() });
+
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "computeStats", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_show_time_spent_in_comms__FINE",
+ new Object[] { aCasReferenceId, (double) totalTimeInComms / (double) 1000000, aMessageContext.getEndpoint() });
+
+ if ( aMessageContext.propertyExists(AsynchAEMessage.DelegateStats) )
+ {
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ CacheEntry entry = getController().getInProcessCache().getCacheEntryForCAS(aCasReferenceId);
+ DelegateStats stats;
+ if ( (stats = entry.getStat()) != null )
+ {
+ //DelegateStats delegateStats = (DelegateStats)
+ Object delegateStats = aMessageContext.getMessageObjectProperty(AsynchAEMessage.DelegateStats);
+
+ if ( delegateStats instanceof TimerStats )
+ {
+ //delegateStats.get(AsynchAEMessage.TimeToSerializeCAS);
+ /*
+ timerStats.put(AsynchAEMessage.TimeToSerializeCAS, entry.getTimeToSerializeCAS());
+ timerStats.put(AsynchAEMessage.TimeWaitingForCAS, entry.getTimeWaitingForCAS());
+ timerStats.put(AsynchAEMessage.TimeToDeserializeCAS, entry.getTimeToDeserializeCAS());
+ timerStats.put(AsynchAEMessage.TimeInService, timeInService);
+*/
+ //TimerStats timerStats = (TimerStats)delegateStats;
+
+ }
+ if (!stats.containsKey( aMessageContext.getEndpoint().getEndpoint()))
+ {
+ stats.put(aMessageContext.getEndpoint().getEndpoint(), delegateStats);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ private Endpoint lookupEndpoint(String anEndpointName, String aCasReferenceId)
+ {
+ return getController().getInProcessCache().getEndpoint(anEndpointName, aCasReferenceId);
+ }
+
+ private void cancelTimer(MessageContext aMessageContext, String aCasReferenceId, boolean removeEndpoint) throws AsynchAEException
+ {
+ if (aMessageContext != null && aMessageContext.getEndpoint() != null)
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "cancelTimer", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_cancel_timer__FINE",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint(), aCasReferenceId });
+
+ // Retrieve the endpoint from the cache using endpoint name
+ // and casRefereceId
+ if ( aCasReferenceId == null &&
+ aMessageContext.propertyExists(AsynchAEMessage.Command) &&
+ aMessageContext.getMessageIntProperty(AsynchAEMessage.Command) == AsynchAEMessage.CollectionProcessComplete
+ )
+ {
+ aCasReferenceId = ":CpC";
+ }
+ if ( aMessageContext != null && aMessageContext.getEndpoint() != null )
+ {
+ Endpoint endpoint = lookupEndpoint(aMessageContext.getEndpoint().getEndpoint(), aCasReferenceId);
+
+ if (endpoint != null)
+ {
+ // Received the response within timeout interval so
+ // cancel the running timer
+ endpoint.cancelTimer();
+ if ( removeEndpoint )
+ {
+ getController().getInProcessCache().removeEndpoint(aMessageContext.getEndpoint().getEndpoint(), aCasReferenceId);
+ }
+ }
+ else
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "cancelTimer", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_endpoint_not_found__INFO",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint(), aCasReferenceId });
+ }
+ }
+ }
+ }
+
+ private void cancelTimerAndProcess(MessageContext aMessageContext, String aCasReferenceId, CAS aCAS) throws AsynchAEException
+ {
+ computeStats(aMessageContext, aCasReferenceId);
+
+ cancelTimer(aMessageContext, aCasReferenceId, true);
+ super.invokeProcess(aCAS, aCasReferenceId, null, aMessageContext, null);
+
+ }
+
+ private boolean isMessageExpected(String aCasReferenceId, Endpoint anEndpointWithTimer)
+
+ {
+ if (getController().getInProcessCache().entryExists(aCasReferenceId) && anEndpointWithTimer.isWaitingForResponse())
+ {
+ return true;
+ }
+ return false;
+ }
+
+ private void handleUnexpectedMessage(String aCasReferenceId, Endpoint anEndpoint)
+ {
+ // Cas does not exist in the CAS Cache. This would be possible if the
+ // CAS has been dropped due to timeout and the delegate
+ // sends the response later. In asynch communication this scenario is
+ // possible. The service may not be up when the client sends
+ // the message. Messages accumulate in the service queue until the
+ // service becomes available. When this happens, the service
+ // will pickup messages from the queue, process them and send respones
+ // to an appropriate response queue. Most likely such
+ // respones should be thrown away. Well perhaps logged first.
+ ErrorContext errorContext = new ErrorContext();
+ errorContext.add(AsynchAEMessage.CasReference, aCasReferenceId);
+ errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
+ errorContext.add(AsynchAEMessage.Endpoint, anEndpoint);
+ AnalysisEngineController controller = getController();
+ controller.getErrorHandlerChain().handle(new ExpiredMessageException(), errorContext, controller);
+
+ }
+
+ private void handleProcessResponseWithXMI(MessageContext aMessageContext, String aDelegateKey)
+ {
+ CAS cas = null;
+ String casReferenceId = null;
+ Endpoint endpointWithTimer = null;
+ try
+ {
+ casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ endpointWithTimer = lookupEndpoint(aMessageContext.getEndpoint().getEndpoint(), casReferenceId);
+
+ if ( endpointWithTimer == null )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
+ "handleProcessResponseWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_invalid_endpoint__WARNING",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint(), casReferenceId});
+ throw new AsynchAEException("Unable to lookup endpoint:"+aMessageContext.getEndpoint().getEndpoint()+ " Cas Reference Id:"+casReferenceId);
+ }
+
+ // Check if this reply message is expected. A message is expected
+ // when it contains a reply for request
+ // that has not been already received nor the timeout for the
+ // request has been handled. This is done
+ // to catch duplicate replies for the same request. If a cache entry
+ // does not exist with a given CAS reference id
+ // it means that the initial request has been canceled and the CAS
+ // has either been processed or dropped. If a duplicate
+ // reply arrives containing CAS that is still being processed in
+ // this aggregate, the endpoint state will show that
+ // it is not waiting for a response.
+ if (!isMessageExpected(casReferenceId, endpointWithTimer))
+ {
+ handleUnexpectedMessage(casReferenceId, aMessageContext.getEndpoint());
+ return;
+ }
+ // Increment number of CASes processed by this delegate
+ if ( aDelegateKey != null)
+ {
+ ((AggregateAnalysisEngineController)getController()).
+ incrementCasProcessedByDelegate(aDelegateKey);
+ }
+
+ String xmi = aMessageContext.getStringMessage();
+
+ // Fetch entry from the cache for a given Cas Id. The entry contains a CAS that will be used during deserialization
+ CacheEntry cacheEntry = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId);
+ cas = cacheEntry.getCas();
+ int totalNumberOfParallelDelegatesProcessingCas = cacheEntry.getNumberOfParallelDelegates();
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessResponseWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_number_parallel_delegates_FINE",
+ new Object[] { totalNumberOfParallelDelegatesProcessingCas});
+
+ if (cas == null)
+ {
+ throw new AsynchAEException(Thread.currentThread().getName()+"-Cache Does not contain a CAS. Cas Reference Id::"+casReferenceId);
+ }
+ if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
+ "handleProcessResponseWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_rcvd_reply_FINEST",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint(), casReferenceId, xmi });
+ }
+
+ long t1 = System.nanoTime();
+
+ synchronized (monitor)
+ {
+ XmiSerializationSharedData deserSharedData;
+ if (totalNumberOfParallelDelegatesProcessingCas > 1 && cacheEntry.howManyDelegatesResponded() > 0)
+ {
+ // process secondary reply from a parallel step
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
+ "handleProcessResponseWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_delegate_responded_count_FINEST",
+ new Object[] { cacheEntry.howManyDelegatesResponded(), casReferenceId});
+
+ int highWaterMark = cacheEntry.getHighWaterMark();
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
+ "handleProcessResponseWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_high_water_mark_FINEST",
+ new Object[] { highWaterMark, casReferenceId });
+
+ deserSharedData = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId).getDeserSharedData();
+ UimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, highWaterMark);
+ }
+ else // general case, or first reply from a parallel step
+ {
+ // Processing the reply from a standard, non-parallel delegate
+ deserSharedData = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId).getDeserSharedData();
+ if (deserSharedData == null) {
+ deserSharedData = new XmiSerializationSharedData();
+ getController().getInProcessCache().getCacheEntryForCAS(casReferenceId).setXmiSerializationData(deserSharedData);
+ }
+ UimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, -1);
+ }
+ }
+
+
+ if ( cacheEntry != null && totalNumberOfParallelDelegatesProcessingCas > 1 )
+ {
+ synchronized( cacheEntry )
+ {
+ cacheEntry.incrementHowManyDelegatesResponded();
+ }
+ }
+
+ long timeToDeserializeCAS = System.nanoTime() - t1;
+
+ // Increment Total Deserialization Time for the remote delegate
+//((AggregateAnalysisEngineController)getController()).
+//incrementCasDeserializationTime(aDelegateKey, timeToDeserializeCAS);
+
+ if ( aMessageContext.propertyExists(AsynchAEMessage.TimeToDeserializeCAS))
+ {
+ long remoteDelegateTimeToDeserialize =
+ aMessageContext.getMessageLongProperty(AsynchAEMessage.TimeToDeserializeCAS);
+ ((AggregateAnalysisEngineController)getController()).
+ incrementCasDeserializationTime(aDelegateKey, remoteDelegateTimeToDeserialize);
+
+ }
+
+ if ( aMessageContext.propertyExists(AsynchAEMessage.TimeToSerializeCAS))
+ {
+ long remoteDelegateTimeToSerialize =
+ aMessageContext.getMessageLongProperty(AsynchAEMessage.TimeToSerializeCAS);
+ ((AggregateAnalysisEngineController)getController()).
+ incrementCasDeserializationTime(aDelegateKey, remoteDelegateTimeToSerialize);
+
+ }
+
+ getController().
+ getServicePerformance().
+ incrementCasDeserializationTime(timeToDeserializeCAS);
+
+
+
+
+ LongNumericStatistic statistic;
+ if ( (statistic = getController().getMonitor().getLongNumericStatistic("",Monitor.TotalDeserializeTime)) != null )
+ {
+ statistic.increment(timeToDeserializeCAS);
+ }
+
+ computeStats(aMessageContext, casReferenceId);
+
+ cancelTimer(aMessageContext, casReferenceId, true);
+
+ // Send CAS for processing when all delegates reply
+ // totalNumberOfParallelDelegatesProcessingCas indicates how many delegates are processing CAS in parallel. Default is 1, meaning only
+ // one delegate processes the CAS at the same. Otherwise, check if all delegates responded before passing CAS on to the Flow Controller.
+ // The idea is that all delegates processing one CAS concurrently must respond, before the CAS is allowed to move on to the next step.
+ // HowManyDelegatesResponded is incremented every time a parallel delegate sends response.
+ if (totalNumberOfParallelDelegatesProcessingCas == 1 || ( cacheEntry.howManyDelegatesResponded() == totalNumberOfParallelDelegatesProcessingCas) )
+ {
+ cacheEntry.resetDelegateResponded();
+ super.invokeProcess(cas, casReferenceId, null, aMessageContext, null);
+ }
+
+ }
+ catch ( Exception e)
+ {
+ e.printStackTrace();
+ ErrorContext errorContext = new ErrorContext();
+ errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
+ errorContext.add(AsynchAEMessage.CasReference, casReferenceId );
+ errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
+ getController().getErrorHandlerChain().handle(e, errorContext, getController());
+ }
+ finally
+ {
+ incrementDelegateProcessCount(aMessageContext);
+ }
+
+ }
+/*
+ private boolean deserialize(String xmi, String casReferenceId, CAS cas, Endpoint endpoint, XmiSerializationSharedData deserSharedData, boolean dflag, int highWaterMark )
+ {
+ try
+ {
+ UimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, dflag, -1);
+ }
+ catch( Exception e)
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
+ "deserialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_deserialize_error__WARNING",
+ new Object[] { getController().getName(), casReferenceId, endpoint.getEndpoint() });
+ try
+ {
+ getController().getOutputChannel().sendReply(e, casReferenceId, endpoint );
+ }
+ catch( Exception ee )
+ {
+ // Exception while sending reply to client. Handle this silently for now.
+ }
+ if ( cas != null )
+ {
+ getController().dropCAS(cas);
+ }
+ return false;
+ }
+ return true;
+
+ }
+*/
+ private void handleProcessResponseWithCASReference(MessageContext aMessageContext )
+ {
+ String casReferenceId = null;
+ CacheEntry cacheEntry = null;
+
+ try
+ {
+ casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ cacheEntry = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId);
+ CAS cas = cacheEntry.getCas();
+
+ //CAS cas = getController().getInProcessCache().getCasByReference(casReferenceId);
+ if (cas != null)
+ {
+ cancelTimerAndProcess(aMessageContext, casReferenceId, cas);
+ }
+ else
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handleProcessResponseWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_cas_not_in_cache__INFO",
+ new Object[] { getController().getName(), casReferenceId, aMessageContext.getEndpoint().getEndpoint() });
+ throw new AsynchAEException("CAS with Reference Id:" + casReferenceId + " Not Found in CasManager's CAS Cache");
+ }
+ }
+ catch ( Exception e)
+ {
+
+ e.printStackTrace();
+ ErrorContext errorContext = new ErrorContext();
+ errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
+ errorContext.add(AsynchAEMessage.CasReference, casReferenceId );
+ errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
+ getController().getErrorHandlerChain().handle(e, errorContext, getController());
+ }
+ finally
+ {
+ incrementDelegateProcessCount(aMessageContext);
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ try
+ {
+ String endpointName = aMessageContext.getEndpoint().getEndpoint();
+ String delegateKey =
+ ((AggregateAnalysisEngineController)getController()).lookUpDelegateKey(endpointName);
+ if ( delegateKey != null )
+ {
+ Endpoint endpoint =
+ ((AggregateAnalysisEngineController)getController()).lookUpEndpoint(delegateKey, false);
+
+ // Check if the multiplier aborted during processing of this input CAS
+ if ( endpoint != null && endpoint.isCasMultiplier() && cacheEntry.isAborted() )
+ {
+ if ( !getController().getInProcessCache().isEmpty() )
+ {
+ getController().getInProcessCache().registerCallbackWhenCacheEmpty(getController().getEventListener());
+ }
+ else
+ {
+ // Callback to notify that the cache is empty
+ getController().getEventListener().onCacheEmpty();
+ }
+ }
+
+ }
+ }
+ catch( Exception e)
+ {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ }
+ private void incrementDelegateProcessCount(MessageContext aMessageContext)
+ {
+ Endpoint endpoint = aMessageContext.getEndpoint();
+ if ( endpoint != null && getController() instanceof AggregateAnalysisEngineController)
+ {
+ try
+ {
+ String delegateKey = ((AggregateAnalysisEngineController)getController()).lookUpDelegateKey(endpoint.getEndpoint());
+ LongNumericStatistic stat = getController().getMonitor().getLongNumericStatistic(delegateKey, Monitor.ProcessCount);
+ stat.increment();
+ }
+ catch( Exception e) {}
+ }
+
+ }
+
+ private boolean isException( Object object )
+ {
+ return (object instanceof Exception || object instanceof Throwable );
+ }
+
+ private boolean isShutdownException( Object object )
+ {
+ return ( object instanceof Exception &&
+ object instanceof UimaEEServiceException &&
+ ((UimaEEServiceException)object).getCause() != null &&
+ ((UimaEEServiceException)object).getCause() instanceof ServiceShutdownException
+ );
+ }
+ private boolean ignoreException(Object object)
+ {
+ if ( object != null && isException(object) && !isShutdownException(object) )
+ {
+ return false;
+ }
+ return true;
+ }
+ private void handleProcessResponseWithException(MessageContext aMessageContext)
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessResponseWithException", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_handling_exception_from_delegate_FINE",
+ new Object[] { getController().getName(), aMessageContext.getEndpoint().getEndpoint() });
+
+ boolean isCpCError = false;
+ String casReferenceId = null;
+ try
+ {
+ Object object = aMessageContext.getObjectMessage();
+ if ( object == null )
+ {
+ // Could be a C++ exception. In this case the exception is just a String in the message cargo
+ if ( aMessageContext.getStringMessage() != null )
+ {
+ object = new UimaEEServiceException(aMessageContext.getStringMessage());
+ }
+ }
+ if ( ignoreException(object) )
+ {
+ return;
+ }
+
+ if ( getController() instanceof AggregateAnalysisEngineController &&
+ aMessageContext.propertyExists(AsynchAEMessage.Command ) &&
+ aMessageContext.getMessageIntProperty(AsynchAEMessage.Command) == AsynchAEMessage.CollectionProcessComplete
+ )
+ {
+ isCpCError = true;
+ ((AggregateAnalysisEngineController)getController()).
+ processCollectionCompleteReplyFromDelegate(aMessageContext.getEndpoint().getEndpoint(), false);
+ }
+ else
+ {
+ casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ cancelTimer(aMessageContext, casReferenceId, false);
+ }
+
+ if ( object != null && (object instanceof Exception || object instanceof Throwable ))
+ {
+
+
+ Exception remoteException = (Exception) object;
+ ErrorContext errorContext = new ErrorContext();
+ errorContext.add(AsynchAEMessage.Command, aMessageContext.getMessageIntProperty(AsynchAEMessage.Command));
+ errorContext.add(AsynchAEMessage.MessageType, aMessageContext.getMessageIntProperty(AsynchAEMessage.MessageType));
+ if ( !isCpCError )
+ {
+ errorContext.add(AsynchAEMessage.CasReference, casReferenceId);
+ }
+ errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
+ getController().getErrorHandlerChain().handle(remoteException, errorContext, getController());
+ }
+ }
+ catch ( Exception e)
+ {
+ e.printStackTrace();
+ ErrorContext errorContext = new ErrorContext();
+ errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
+ errorContext.add(AsynchAEMessage.CasReference, casReferenceId );
+ errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
+ getController().getErrorHandlerChain().handle(e, errorContext, getController());
+ }
+
+ }
+
+ private void handleCollectionProcessCompleteReply(MessageContext aMessageContext)
+ {
+ try
+ {
+ String delegateKey = ((Endpoint)aMessageContext.getEndpoint()).getEndpoint();
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ Object o;
+ if ( (o = aMessageContext.getMessageObjectProperty("Stats")) != null && o instanceof HashMap)
+ {
+ ((AggregateAnalysisEngineController) getController()).
+ saveStatsFromService(delegateKey, (HashMap)o );
+// saveStatsFromService(delegateKey, (HashMap)((HashMap)o).get(mapkey) );
+/*
+ Iterator it = ((HashMap)o).keySet().iterator();
+ while( it.hasNext() )
+ {
+ String mapkey = (String)it.next();
+ System.out.println("Controller:"+getController().getComponentName()+" HashMap Delegate Key:"+mapkey);
+ ((AggregateAnalysisEngineController) getController()).
+ saveStatsFromService(delegateKey, (HashMap)((HashMap)o).get(mapkey) );
+ }
+*/
+ }
+ ((AggregateAnalysisEngineController) getController())
+ .processCollectionCompleteReplyFromDelegate(delegateKey, true);
+ }
+ }
+ catch ( Exception e)
+ {
+ e.printStackTrace();
+ ErrorContext errorContext = new ErrorContext();
+ errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.CollectionProcessComplete);
+ errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
+ getController().getErrorHandlerChain().handle(e, errorContext, getController());
+ }
+
+ }
+ private void resetErrorCounts(String aDelegate)
+ {
+ getController().getMonitor().resetCountingStatistic(aDelegate, Monitor.ProcessErrorCount);
+ getController().getMonitor().resetCountingStatistic(aDelegate, Monitor.ProcessErrorRetryCount);
+ }
+ public synchronized void handle(Object anObjectToHandle) throws AsynchAEException
+ {
+ super.validate(anObjectToHandle);
+ MessageContext messageContext = (MessageContext) anObjectToHandle;
+
+ if (isHandlerForMessage(messageContext, AsynchAEMessage.Response, AsynchAEMessage.Process) ||
+ isHandlerForMessage(messageContext, AsynchAEMessage.Response, AsynchAEMessage.ACK ) ||
+ isHandlerForMessage(messageContext, AsynchAEMessage.Response, AsynchAEMessage.CollectionProcessComplete) )
+ {
+ int payload = messageContext.getMessageIntProperty(AsynchAEMessage.Payload);
+ int command = messageContext.getMessageIntProperty(AsynchAEMessage.Command);
+ String delegate = ((Endpoint)messageContext.getEndpoint()).getEndpoint();
+ String key = null;
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ key = ((AggregateAnalysisEngineController)getController()).lookUpDelegateKey(delegate);
+ }
+ if (AsynchAEMessage.CASRefID == payload)
+ {
+
+ handleProcessResponseWithCASReference(messageContext);
+ if ( key != null )
+ {
+ resetErrorCounts(key);
+ }
+ }
+ else if (AsynchAEMessage.XMIPayload == payload)
+ {
+ handleProcessResponseWithXMI(messageContext, key);
+ if ( key != null )
+ {
+ resetErrorCounts(key);
+ }
+ }
+ else if (AsynchAEMessage.Exception == payload)
+ {
+ handleProcessResponseWithException(messageContext);
+ }
+ else if (AsynchAEMessage.None == payload && AsynchAEMessage.CollectionProcessComplete == command)
+ {
+ handleCollectionProcessCompleteReply(messageContext);
+ }
+ else if (AsynchAEMessage.None == payload && AsynchAEMessage.ACK == command)
+ {
+ handleACK(messageContext);
+ }
+
+ else
+ {
+ throw new AsynchAEException("Invalid Payload. Expected XMI or CasReferenceId Instead Got::" + payload);
+ }
+
+
+ // Handled Request to Process with A given Payload
+ return;
+ }
+ // Not a Request nor Command. Delegate to the next handler in the chain
+ super.delegate(messageContext);
+
+ }
+ private void handleACK(MessageContext aMessageContext)
+ throws AsynchAEException
+ {
+ if ( getController() instanceof PrimitiveAnalysisEngineController )
+ {
+// ((PrimitiveAnalysisEngineController) getController()).throttleNext();
+ }
+ }
+
+private static final String copyright = com.ibm.uima.copyright.Copyright.COPYRIGHT;
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/AggregateServiceInfo.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/AggregateServiceInfo.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/AggregateServiceInfo.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/AggregateServiceInfo.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,10 @@
+package org.apache.uima.aae.jmx;
+
+public class AggregateServiceInfo
+extends ServiceInfo implements AggregateServiceInfoMBean
+{
+ private static final long serialVersionUID = 6076764724935117459L;
+ /**
+ *
+ */
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/AggregateServiceInfoMBean.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/AggregateServiceInfoMBean.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/AggregateServiceInfoMBean.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/AggregateServiceInfoMBean.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,7 @@
+package org.apache.uima.aae.jmx;
+
+import java.io.Serializable;
+
+public interface AggregateServiceInfoMBean extends ServiceInfoMBean
+{
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/JmxManagement.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/JmxManagement.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/JmxManagement.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/JmxManagement.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,62 @@
+package org.apache.uima.aae.jmx;
+
+import java.util.Map;
+
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+/**
+ *
+ * Jmx Management interface for registering and unregistering MBeans.
+ *
+ */
+public interface JmxManagement
+{
+ /**
+ * Initializes instnace of this interface using provided properties
+ *
+ * @param anInitMap - properties required during initialization
+ * @throws Exception
+ */
+ public void initialize( Map anInitMap ) throws Exception;
+ /**
+ * Returns instance of an MBeanServer
+ *
+ * @return
+ */
+ public MBeanServer getMBeanServer();
+
+ /**
+ * Returns configured Jmx Domain
+ * @return
+ */
+ public String getJmxDomain();
+
+ /**
+ * Sets the Jmx Domain
+ *
+ * @param aJmxDomain
+ */
+ public void setJmxDomain( String aJmxDomain );
+
+ /**
+ * Registers given MBean with MBeanServer
+ *
+ * @param anMBean - instance of MBean to register
+ * @param anMBeanName - name of the MBean
+ * @throws Exception
+ */
+ public void registerMBean( Object anMBean, ObjectName anMBeanName ) throws Exception;
+ /**
+ * Removed an MBean from the MBeanServer registry
+ *
+ * @param anMBeanName
+ */
+ public void unregisterMBean( ObjectName anMBeanName );
+
+ /**
+ * Performs cleanup when object goes out of scope
+ *
+ * @throws Exception
+ */
+ public void destroy() throws Exception;
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/JmxManager.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/JmxManager.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/JmxManager.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/JmxManager.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,212 @@
+package org.apache.uima.aae.jmx;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Method;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import org.apache.uima.UIMAFramework;
+import org.apache.uima.internal.util.JmxMBeanAgent;
+import org.apache.uima.util.Level;
+
+public class JmxManager implements JmxManagement
+{
+ private String jmxDomain = "";
+
+ public JmxManager( String aDomain )
+ {
+ jmxDomain = aDomain;
+ }
+ /**
+ * Removes all objects from the MBeanServer with t
+ *
+ * @param aDomain
+ */
+ public void unregisterDomainObjects(String aDomain)
+ {
+ if ( !isInitialized())
+ {
+ return;
+ }
+ try
+ {
+ Set set = getMBeansInDomain(aDomain);
+ Iterator it = set.iterator();
+ while (it.hasNext())
+ {
+ ObjectName on = (ObjectName) it.next();
+ try
+ {
+ unregisterMBean(on);
+ }
+ catch (Exception e)
+ {
+ } // Dont care if there is an exception
+ }
+ }
+ catch (Exception e)
+ {
+ UIMAFramework.getLogger().logrb(Level.WARNING, JmxMBeanAgent.class.getName(), "unregisterDomainObjects", LOG_RESOURCE_BUNDLE, "UIMA_JMX_failed_to_register_mbean__WARNING", e);
+ }
+
+ }
+
+ private Set getMBeansInDomain( String aDomain) throws Exception
+ {
+ ObjectName objName = new ObjectName(aDomain);
+ return ((MBeanServer)platformMBeanServer).queryNames(objName, null);
+ }
+
+ public String getJmxDomain()
+ {
+ return jmxDomain;
+ }
+
+ public void setJmxDomain(String aJmxDomain)
+ {
+ jmxDomain = aJmxDomain;
+ }
+
+ public MBeanServer getMBeanServer()
+ {
+ if ( !isInitialized())
+ {
+ return null;
+ }
+ return (MBeanServer) platformMBeanServer;
+ }
+
+
+ public void registerMBean(Object anMBeanToRegister, ObjectName aName ) throws Exception
+ {
+ if ( !isInitialized())
+ {
+ return;
+ }
+
+ try
+ {
+ if (((MBeanServer) platformMBeanServer).isRegistered(aName))
+ {
+ ((MBeanServer) platformMBeanServer).unregisterMBean(aName);
+ }
+ ((MBeanServer) platformMBeanServer).registerMBean(anMBeanToRegister, aName);
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ UIMAFramework.getLogger().logrb(Level.WARNING, JmxMBeanAgent.class.getName(), "registerMBean", LOG_RESOURCE_BUNDLE, "UIMA_JMX_failed_to_register_mbean__WARNING", e);
+ }
+ }
+
+ private boolean isInitialized()
+ {
+ if (!jmxAvailable || platformMBeanServer == null) // means we couldn't find the required classes and methods
+ {
+ System.out.println("No JMX Server");
+ UIMAFramework.getLogger().logrb(Level.CONFIG, JmxManager.class.getName(), "isInitialized", LOG_RESOURCE_BUNDLE, "UIMA_JMX_platform_mbean_server_not_available__CONFIG");
+ return false;
+ }
+ return true;
+ }
+
+ public void unregisterMBean(ObjectName anMBeanToUnregister)
+ {
+ if ( !isInitialized())
+ {
+ return;
+ }
+ try
+ {
+ if (((MBeanServer) platformMBeanServer).isRegistered(anMBeanToUnregister))
+ {
+ ((MBeanServer) platformMBeanServer).unregisterMBean(anMBeanToUnregister);
+ }
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ UIMAFramework.getLogger().logrb(Level.WARNING, JmxMBeanAgent.class.getName(), "registerMBean", LOG_RESOURCE_BUNDLE, "UIMA_JMX_failed_to_register_mbean__WARNING", e);
+ }
+
+ }
+ public void destroy() throws Exception
+ {
+ unregisterDomainObjects("org.apache.uima:type=ee.jms.services,*");
+ }
+
+ public void initialize(Map anInitMap) throws Exception
+ {
+ }
+
+ /** Class and Method handles for reflection */
+ private static Class mbeanServerClass;
+
+ private static Class objectNameClass;
+
+ private static Constructor objectNameConstructor;
+
+ private static Method isRegistered;
+
+ private static Method registerMBean;
+
+ private static Method unregisterMBean;
+
+ /**
+ * Set to true if we can find the required JMX classes and methods
+ */
+ private static boolean jmxAvailable;
+
+ /**
+ * The platform MBean server if one is available (Java 1.5 only)
+ */
+ private static Object platformMBeanServer;
+
+ /** Get class/method handles */
+ static
+ {
+ try
+ {
+ mbeanServerClass = Class.forName("javax.management.MBeanServer");
+ objectNameClass = Class.forName("javax.management.ObjectName");
+ objectNameConstructor = objectNameClass.getConstructor(new Class[]
+ { String.class });
+ isRegistered = mbeanServerClass.getMethod("isRegistered", new Class[]
+ { objectNameClass });
+ registerMBean = mbeanServerClass.getMethod("registerMBean", new Class[]
+ { Object.class, objectNameClass });
+ unregisterMBean = mbeanServerClass.getMethod("unregisterMBean", new Class[]
+ { objectNameClass });
+ jmxAvailable = true;
+ }
+ catch (Exception e)
+ {
+ // JMX not available
+ jmxAvailable = false;
+ }
+
+ // try to get platform MBean Server (Java 1.5 only)
+ try
+ {
+ Class managementFactory = Class.forName("java.lang.management.ManagementFactory");
+ Method getPlatformMBeanServer = managementFactory.getMethod("getPlatformMBeanServer", new Class[0]);
+ platformMBeanServer = getPlatformMBeanServer.invoke(null, (Object[])null);
+ if (platformMBeanServer != null)
+ {
+ System.out.println("Got MBeanServer Started");
+ }
+ }
+ catch (Exception e)
+ {
+ platformMBeanServer = null;
+ }
+ }
+
+ /**
+ * resource bundle for log messages
+ */
+ private static final String LOG_RESOURCE_BUNDLE = "org.apache.uima.impl.log_messages";
+
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ManagedObjectBase.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ManagedObjectBase.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ManagedObjectBase.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ManagedObjectBase.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,18 @@
+package org.apache.uima.aae.jmx;
+
+import javax.management.ObjectName;
+
+public class ManagedObjectBase
+{
+ private ObjectName objectName;
+
+ public ObjectName getObjectName()
+ {
+ return objectName;
+ }
+
+ public void setObjectName(ObjectName objectName)
+ {
+ this.objectName = objectName;
+ }
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/PrimitiveServiceInfo.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/PrimitiveServiceInfo.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/PrimitiveServiceInfo.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/PrimitiveServiceInfo.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,25 @@
+package org.apache.uima.aae.jmx;
+
+public class PrimitiveServiceInfo
+extends ServiceInfo implements PrimitiveServiceInfoMBean//, ServiceInfoMBean
+//extends ServiceInfo implements PrimitiveServiceInfoMBean//, ServiceInfoMBean
+{
+ /**
+ *
+ */
+ private static final long serialVersionUID = -6416623322329673083L;
+ /**
+ *
+ */
+
+ private int instanceCount = 0;
+
+ public int getAnalysisEngineInstanceCount()
+ {
+ return instanceCount;
+ }
+ public void setAnalysisEngineInstanceCount(int anAEInstanceCount)
+ {
+ instanceCount = anAEInstanceCount;
+ }
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/PrimitiveServiceInfoMBean.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/PrimitiveServiceInfoMBean.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/PrimitiveServiceInfoMBean.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/PrimitiveServiceInfoMBean.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,7 @@
+package org.apache.uima.aae.jmx;
+
+public interface PrimitiveServiceInfoMBean extends ServiceInfoMBean
+{
+ public int getAnalysisEngineInstanceCount();
+
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ServiceErrors.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ServiceErrors.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ServiceErrors.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ServiceErrors.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,68 @@
+package org.apache.uima.aae.jmx;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ServiceErrors implements ServiceErrorsMBean
+{
+ private static final long serialVersionUID = 1L;
+ private static final String label="Service Errors";
+ private static final int MAX_ERROR_COUNT = 5;
+
+ private long processErrors = 0;
+ private long metadataErrrors = 0;
+ private long cpcErrors = 0;
+ private String[] errors = new String[] {" ", " "," "," "," "};
+ private List throwables = new ArrayList();
+
+ public String getLabel()
+ {
+ return label;
+ }
+
+ public synchronized void addError( Throwable aThrowable)
+ {
+
+ }
+ public String[] getErrors()
+ {
+ return errors;
+ }
+
+ public long getProcessErrors()
+ {
+ return processErrors;
+ }
+
+ public synchronized void incrementProcessErrors()
+ {
+ processErrors++;
+ }
+
+ public long getMetadataErrors()
+ {
+ return metadataErrrors;
+ }
+ public synchronized void incrementMetadataErrors()
+ {
+ metadataErrrors++;
+ }
+
+ public long getCpcErrors()
+ {
+ return cpcErrors;
+ }
+
+ public synchronized void incrementCpcErrors()
+ {
+ cpcErrors++;
+ }
+
+ public void reset()
+ {
+ cpcErrors = 0;
+ metadataErrrors = 0;
+ processErrors = 0;
+ }
+
+}
Added: incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ServiceErrorsMBean.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ServiceErrorsMBean.java?rev=610003&view=auto
==============================================================================
--- incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ServiceErrorsMBean.java (added)
+++ incubator/uima/sandbox/trunk/uima-ee/uimaj-ee-core/src/main/java/org/apache/uima/aae/jmx/ServiceErrorsMBean.java Tue Jan 8 06:38:38 2008
@@ -0,0 +1,14 @@
+package org.apache.uima.aae.jmx;
+
+import java.io.Serializable;
+
+public interface ServiceErrorsMBean extends Serializable
+{
+ public String[] getErrors();
+ public long getProcessErrors();
+ public long getMetadataErrors();
+ public long getCpcErrors();
+
+ public void reset();
+
+}