in uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/NonThreadedProcessingUnit.java [378:592]
protected boolean analyze(Object[] aCasObjectList, ProcessTrace pTrTemp) throws Exception // throws
// ResourceProcessException,
// IOException,
// CollectionException,
// AbortCPMException
{
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_start_analysis__FINEST",
new Object[] { Thread.currentThread().getName() });
}
// String lastDocId = "";
CasProcessor processor = null;
// This is used to hold an index of the current CasObject
boolean doneAlready = false;
// If there are no CASes in the list, return false since there is nothing else to do
if (aCasObjectList == null || aCasObjectList[0] == null) {
if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.SEVERE, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_invalid_cas_reference__SEVERE",
new Object[] { Thread.currentThread().getName() });
}
return false;
}
Object[] casObjects = null;
// Determine if the Cas'es contained in the CasList are of type CAS. Samples the first CAS in
// the list.
// The list contains CASes of the same type ( either CasData or CAS ). Mixed model not
// supported.
boolean isCasObject = aCasObjectList[0] instanceof CAS;
// String docid = "";
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_entering_pipeline__FINEST",
new Object[] { Thread.currentThread().getName() });
}
ProcessingContainer container = null;
String containerName = "";
// *******************************************
// ** P R O C E S S I N G P I P E L I N E **
// *******************************************
// Send Cas Object through the processing pipeline.
for (int i = 0; processContainers != null && i < processContainers.size(); i++) {
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_retrieve_container__FINEST",
new Object[] { Thread.currentThread().getName(), String.valueOf(i) });
}
// Retrieve the container. Container manages one or more instances of CAS Processor
container = (ProcessingContainer) processContainers.get(i);
// container can be disabled in multi-processing pipeline configurations. The container is
// disabled
// when one of the processing threads is in the process of restarting/reconnecting to a shared
// fenced service. Shared, meaning that all processing pipelines use the same service for
// invocations.
// Container must be disabled to prevent concurrent restarts.
if (containerDisabled(container) || filterOutTheCAS(container, isCasObject, aCasObjectList)) {
continue;
}
containerName = container.getName();
// Flag controlling do-while loop that facilitates retries. Retries are defined in the
// CasProcessor configuration.
boolean retry = false;
do // Retry
{
try {
if (System.getProperty("SHOW_MEMORY") != null) {
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_show_memory__FINEST",
new Object[] { Thread.currentThread().getName(),
String.valueOf(Runtime.getRuntime().totalMemory() / 1024),
String.valueOf(Runtime.getRuntime().freeMemory() / 1024) });
}
}
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_checkout_cp_from_container__FINEST",
new Object[] { Thread.currentThread().getName(), containerName });
}
threadState = 2004;
// Get the CasProcessor from the pool managed by the container
processor = container.getCasProcessor();
if (processor == null) {
if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.SEVERE,
this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_checkout_null_cp_from_container__SEVERE",
new Object[] { Thread.currentThread().getName(), containerName });
}
throw new ResourceProcessException(CpmLocalizedMessage.getLocalizedMessage(
CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_EXP_invalid_component_reference__WARNING",
new Object[] { Thread.currentThread().getName(), "CasProcessor", "NULL" }),
null);
}
// Check to see if the CasProcessor is available for processing
// The CasProcessor may have been disabled due to excessive errors and error policy
// defined
// in the CPE descriptor.
if (!isProcessorReady(container.getStatus())) {
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_container_not_ready__FINEST",
new Object[] { Thread.currentThread().getName(), containerName });
}
if (container.getStatus() == Constants.CAS_PROCESSOR_KILLED) {
container.releaseCasProcessor(processor);
// Another thread has initiated CPM Abort. That Thread has already notified
// the application of the Abort. Here we just return as the CPM has been
// killed most likely due to excessive errors.
return false;
}
// Skip any CasProcessor that is not ready to process
break;
}
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_checkedout_cp_from_container__FINEST",
new Object[] { Thread.currentThread().getName(), containerName,
processor.getClass().getName() });
}
// ************************* P E R F O R M A N A L Y S I S *************************
if (processor instanceof CasDataProcessor) {
invokeCasDataCasProcessor(container, processor, aCasObjectList, pTrTemp, isCasObject,
retry);
isCasObject = false;
} else if (processor instanceof CasObjectProcessor) {
invokeCasObjectCasProcessor(container, processor, aCasObjectList, pTrTemp, isCasObject);
isCasObject = true;
}
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_analysis_successfull__FINEST",
new Object[] { Thread.currentThread().getName(), containerName,
processor.getClass().getName() });
}
retry = false;
// On successfull processing reset the restart counter. Restart counter determines how
// many times to restart Cas Processor on the same CAS
// Do this conditionally. If the CAS is to be dropped on Exception this restart counter
// scope extends to the entire collection not just one CAS
if (!cpm.dropCasOnException()) {
container.resetRestartCount();
}
} catch (Exception e) {
retry = handleErrors(e, container, processor, pTrTemp, aCasObjectList, isCasObject);
if (cpm.dropCasOnException()) {
retry = false; // override
return false; // Dont pass the CAS to the CasConsumer. CAS has been dropped
}
} finally {
if (!retry) {
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_end_of_batch__FINEST",
new Object[] { Thread.currentThread().getName(), containerName,
processor.getClass().getName() });
}
if (isProcessorReady(container.getStatus())) {
// Let the container take action if the end-of-batch marker has been reached.
// End-of-batch marker is defined in the cpm configuration for every CasProcessor.
// This marker is defined in the <checkpoint> section of the CasProcessor Definition
// and corresponds to the attribute "batch". If end-of-batch marker is reached the
// container
// invokes batchProcessComplete() on the CasProcessor
doEndOfBatch(container, processor, pTrTemp, aCasObjectList.length);
}
} else {
container.incrementRetryCount(1);
}
// Release current Cas Processor before continuing with the next Cas Processor in the
// pipeline
if (processor != null) {
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_release_cp__FINEST",
new Object[] { Thread.currentThread().getName(), containerName,
processor.getClass().getName(), String.valueOf(casCache == null) });
}
doReleaseCasProcessor(container, processor);
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_ok_release_cp__FINEST",
new Object[] { Thread.currentThread().getName(), containerName,
processor.getClass().getName(), String.valueOf(casCache == null) });
}
processor = null;
}
}
} while (retry);
} // end of: For All CasProcessors
postAnalysis(aCasObjectList, isCasObject, casObjects, pTrTemp, doneAlready);
casObjects = null;
return true;
}