public void process(CAS aCAS, String aCasReferenceId, Endpoint anEndpoint) {
if (stopped) {
return;
}
CasStateEntry parentCasStateEntry = null;
try {
parentCasStateEntry = getLocalCache().lookupEntry(aCasReferenceId);
} catch (Exception e) {
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
"process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAEE_exception__WARNING", new Object[] { e });
}
return;
}
long totalProcessTime = 0; // stored total time spent producing ALL CASes
boolean inputCASReturned = false;
boolean processingFailed = false;
// This is a primitive controller. No more processing is to be done on the Cas. Mark the
// destination as final and return CAS in reply.
anEndpoint.setFinal(true);
AnalysisEngine ae = null;
try {
// Checkout an instance of AE from the pool
ae = aeInstancePool.checkout();
// Get input CAS entry from the InProcess cache
long time = super.getCpuTime();
CasIterator casIterator = ae.processAndOutputNewCASes(aCAS);
// Store how long it took to call processAndOutputNewCASes()
totalProcessTime = (super.getCpuTime() - time);
long sequence = 1;
long hasNextTime = 0; // stores time in hasNext()
long getNextTime = 0; // stores time in next();
boolean moreCASesToProcess = true;
boolean casAbortedDueToExternalRequest = false;
while (moreCASesToProcess) {
long timeToProcessCAS = 0; // stores time in hasNext() and next() for each CAS
hasNextTime = super.getCpuTime();
if (!casIterator.hasNext()) {
moreCASesToProcess = false;
// Measure how long it took to call hasNext()
timeToProcessCAS = (super.getCpuTime() - hasNextTime);
totalProcessTime += timeToProcessCAS;
break; // from while
}
// Measure how long it took to call hasNext()
timeToProcessCAS = (super.getCpuTime() - hasNextTime);
getNextTime = super.getCpuTime();
CAS casProduced = casIterator.next();
// Add how long it took to call next()
timeToProcessCAS += (super.getCpuTime() - getNextTime);
// Add time to call hasNext() and next() to the running total
totalProcessTime += timeToProcessCAS;
casAbortedDueToExternalRequest = abortGeneratingCASes(aCasReferenceId);
// If the service is stopped or aborted, stop generating new CASes and just return the input
// CAS
if (stopped || casAbortedDueToExternalRequest) {
if (getInProcessCache() != null && getInProcessCache().getSize() > 0
&& getInProcessCache().entryExists(aCasReferenceId)) {
try {
// Set a flag on the input CAS to indicate that the processing was aborted
getInProcessCache().getCacheEntryForCAS(aCasReferenceId).setAborted(true);
} catch (Exception e) {
// An exception be be thrown here if the service is being stopped.
// The top level controller may have already cleaned up the cache
// and the getCacheEntryForCAS() will throw an exception. Ignore it
// here, we are shutting down.
} finally {
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
// We are terminating the iterator here, release the internal CAS lock
// so that we can release the CAS. This approach may need to be changed
// as there may potentially be a problem with a Class Loader.
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
((CASImpl) aCAS).enableReset(true);
try {
// We are either stopping the service or aborting input CAS due to explicit STOP
// request
// from a client. If a new CAS was produced, release it back to the pool.
if (casProduced != null) {
casProduced.release();
}
} catch (Exception e) {
System.out.println("Controller:" + getComponentName()
+ " Attempt to release CAS Failed");
}
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
UIMAFramework.getLogger(CLASS_NAME).logrb(
Level.INFO,
getClass().getName(),
"process",
UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAEE_stopped_producing_new_cases__INFO",
new Object[] { Thread.currentThread().getId(), getComponentName(),
aCasReferenceId });
}
System.out.println(">>>> Cas Multiplier:" + getComponentName()
+ " Stopped Generating CASes from Input CAS:" + aCasReferenceId);
}
}
if (casAbortedDueToExternalRequest) {
// The controller was told to stop generating new CASes. Just return the input CAS to
// the
// client
// throw new ResourceProcessException(new
// InterruptedException("Cas Multiplier:"+getComponentName()+" Aborted CAS:"+aCasReferenceId));
break; // break out of the cas producing loop and return an input CAS to the client
} else {
// The controller is stopping
return;
}
}
OutOfTypeSystemData otsd = getInProcessCache().getOutOfTypeSystemData(aCasReferenceId);
MessageContext mContext = getInProcessCache()
.getMessageAccessorByReference(aCasReferenceId);
CacheEntry newEntry = getInProcessCache().register(casProduced, mContext, otsd);
// if this Cas Multiplier is not Top Level service, add new Cas Id to the private
// cache of the parent aggregate controller. The Aggregate needs to know about
// all CASes it has in play that were generated from the input CAS.
CasStateEntry childCasStateEntry = null;
if (!isTopLevelComponent()) {
newEntry.setNewCas(true, parentController.getComponentName());
// Create CAS state entry in the aggregate's local cache
childCasStateEntry = parentController.getLocalCache().createCasStateEntry(
newEntry.getCasReferenceId());
// Fetch the parent CAS state entry from the aggregate's local cache. We need to increment
// number of child CASes associated with it.
parentCasStateEntry = parentController.getLocalCache().lookupEntry(aCasReferenceId);
} else {
childCasStateEntry = getLocalCache().createCasStateEntry(newEntry.getCasReferenceId());
}
// Associate parent CAS (input CAS) with the new CAS.
childCasStateEntry.setInputCasReferenceId(aCasReferenceId);
// Increment number of child CASes generated from the input CAS
parentCasStateEntry.incrementSubordinateCasInPlayCount();
// Associate input CAS with the new CAS
newEntry.setInputCasReferenceId(aCasReferenceId);