Class NonThreadedProcessingUnit
java.lang.Object
org.apache.uima.collection.impl.cpm.engine.NonThreadedProcessingUnit
The Class NonThreadedProcessingUnit.
-
Field Summary
FieldsModifier and TypeFieldDescriptionprotected Object[]
The artifact.private CAS[]
The cas cache.protected CAS[]
The cas list.protected CPECasPool
The cas pool.protected CAS
The conversion cas.protected CAS[]
The conversion cas array.protected CpeConfiguration
The cpe configuration.protected CPMEngine
The cpm.protected CasConverter
The m converter.protected boolean
The notify listeners.protected long
The num to process.protected BoundedWorkQueue
The output queue.protected LinkedList
The process containers.protected ProcessTrace
The processing unit process trace.protected boolean
The relase CAS.protected ArrayList
The status cb L.protected String
The thread id.int
The thread state.protected UimaTimer
The timer.protected BoundedWorkQueue
The work queue. -
Constructor Summary
ConstructorsConstructorDescriptionInstantiates a new non threaded processing unit.NonThreadedProcessingUnit
(CPMEngine acpm, BoundedWorkQueue aInputQueue, BoundedWorkQueue aOutputQueue) Initialize the PU. -
Method Summary
Modifier and TypeMethodDescriptionvoid
Plugs in Listener object used for notifications.protected boolean
analyze
(Object[] aCasObjectList, ProcessTrace pTrTemp) Analyze.void
cleanup()
Null out fields of this object.private void
Clear cas cache.private boolean
containerDisabled
(ProcessingContainer aContainer) Container disabled.private void
convertCasDataToCasObject
(int casIndex, String aContainerName, Object[] aCasObjectList) Convert cas data to cas object.void
disableCasProcessor
(int aCasProcessorIndex) Disable a CASProcessor in the processing pipeline.void
disableCasProcessor
(String aCasProcessorName) Alternative method to disable Cas Processor.private void
doEndOfBatch
(ProcessingContainer aContainer, CasProcessor aProcessor, ProcessTrace aProcessTr, int howManyCases) Do end of batch.protected void
doNotifyListeners
(Object aCas, boolean isCasObject, EntityProcessStatus aEntityProcStatus) Notifies all configured listeners.private void
doReleaseCasProcessor
(ProcessingContainer aContainer, CasProcessor aCasProcessor) Do release cas processor.void
enableCasProcessor
(String aCasProcessorName) Enables Cas Processor with a given name.private boolean
filterOutTheCAS
(ProcessingContainer aContainer, boolean isCasObject, Object[] aCasObjectList) Filter out the CAS.protected long
Returns the size of the CAS object.Returns list of listeners used by this PU for callbacks.private boolean
handleErrors
(Throwable e, ProcessingContainer aContainer, CasProcessor aProcessor, ProcessTrace aProcessTrace, Object[] aCasObjectList, boolean isCasObject) Main routine that handles errors occuring in the processing loop.private void
handleServiceException
(ProcessingContainer aContainer, CasProcessor aProcessor, ProcessTrace aProcessTr, Exception ex) Handle service exception.private void
handleSkipCasProcessor
(ProcessingContainer aContainer, Object[] aCasObjectList, boolean isLastCP) Handle skip cas processor.private void
invokeCasDataCasProcessor
(ProcessingContainer container, CasProcessor processor, Object[] aCasObjectList, ProcessTrace pTrTemp, boolean isCasObject, boolean retry) Invoke cas data cas processor.private void
invokeCasObjectCasProcessor
(ProcessingContainer container, CasProcessor processor, Object[] aCasObjectList, ProcessTrace pTrTemp, boolean isCasObject) Invoke cas object cas processor.protected boolean
isProcessorReady
(int aStatus) Check if the CASProcessor status is available for processing.protected void
notifyListeners
(Object aCas, boolean isCasObject, EntityProcessStatus aEntityProcStatus) Notifies Listeners of the fact that the pipeline has finished processing the current set Cas'es.private boolean
pauseContainer
(ProcessingContainer aContainer, Exception aException, String aThreadId) Determines if the thread should be paused.private void
postAnalysis
(Object[] aCasObjectList, boolean isCasObject, Object[] casObjects, ProcessTrace aProcessTr, boolean doneAlready) Post analysis.private void
releaseCases
(Object aCasList, boolean lastProcessor, String aName) Conditionally, releases CASes back to the CAS pool.void
Removes given listener from the list of listeners.void
setCasPool
(CPECasPool aPool) Sets the cas pool.void
setContainers
(LinkedList processorList) Plugs in a list of Cas Processor containers.void
setCPMEngine
(CPMEngine acpm) Alternative method of providing the reference to the component managing the lifecycle of the CPE.void
setInputQueue
(BoundedWorkQueue aInputQueue) Alternative method of providing a queue from which this PU will read bundle of Cas.void
setNotifyListeners
(boolean aDoNotify) Set a flag indicating if notifications should be made via configured Listeners.void
setOutputQueue
(BoundedWorkQueue aOutputQueue) Alternative method of providing a queue where this PU will deposit results of analysis.void
setProcessingUnitProcessTrace
(ProcessTrace aProcessingUnitProcessTrace) Plugs in ProcessTrace object used to collect statistics.void
setReleaseCASFlag
(boolean aFlag) Sets the release CAS flag.void
setUimaTimer
(UimaTimer aTimer) Plugs in custom timer used by the PU for getting time.void
stopCasProcessors
(boolean kill) Stops all Cas Processors that are part of this PU.
-
Field Details
-
threadState
public int threadStateThe thread state. -
casPool
The cas pool. -
relaseCAS
protected boolean relaseCASThe relase CAS. -
cpm
The cpm. -
workQueue
The work queue. -
outputQueue
The output queue. -
mConverter
The m converter. -
processingUnitProcessTrace
The processing unit process trace. -
processContainers
The process containers. -
numToProcess
protected long numToProcessThe num to process. -
casList
The cas list. -
statusCbL
The status cb L. -
notifyListeners
protected boolean notifyListenersThe notify listeners. -
conversionCas
The conversion cas. -
artifact
The artifact. -
conversionCasArray
The conversion cas array. -
timer
The timer. -
threadId
The thread id. -
cpeConfiguration
The cpe configuration. -
casCache
The cas cache.
-
-
Constructor Details
-
NonThreadedProcessingUnit
public NonThreadedProcessingUnit(CPMEngine acpm, BoundedWorkQueue aInputQueue, BoundedWorkQueue aOutputQueue) Initialize the PU.- Parameters:
acpm
- - component managing life cycle of the CPEaInputQueue
- - queue to read fromaOutputQueue
- - queue to write to
-
NonThreadedProcessingUnit
Instantiates a new non threaded processing unit.- Parameters:
acpm
- the acpm
-
-
Method Details
-
setInputQueue
Alternative method of providing a queue from which this PU will read bundle of Cas.- Parameters:
aInputQueue
- - read queue
-
setOutputQueue
Alternative method of providing a queue where this PU will deposit results of analysis.- Parameters:
aOutputQueue
- - queue to write to
-
setCPMEngine
Alternative method of providing the reference to the component managing the lifecycle of the CPE.- Parameters:
acpm
- - reference to the contrlling engine
-
cleanup
public void cleanup()Null out fields of this object. Call this only when this object is no longer needed. -
setNotifyListeners
public void setNotifyListeners(boolean aDoNotify) Set a flag indicating if notifications should be made via configured Listeners.- Parameters:
aDoNotify
- - true if notification is required, false otherwise
-
addStatusCallbackListener
Plugs in Listener object used for notifications.- Parameters:
aListener
- -BaseStatusCallbackListener
instance
-
getCallbackListeners
Returns list of listeners used by this PU for callbacks.- Returns:
- - lif of
BaseStatusCallbackListener
instances
-
removeStatusCallbackListener
Removes given listener from the list of listeners.- Parameters:
aListener
- - object to remove from the list
-
setProcessingUnitProcessTrace
Plugs in ProcessTrace object used to collect statistics.- Parameters:
aProcessingUnitProcessTrace
- - object to compile stats
-
setUimaTimer
Plugs in custom timer used by the PU for getting time.- Parameters:
aTimer
- - custom timer to use
-
setContainers
Plugs in a list of Cas Processor containers. During processing Cas Processors in this list are called sequentially. Each Cas Processor is contained in the container that is managing errors, counts and totals, and restarts.- Parameters:
processorList
- CASProcessor to be added to the processing pipeline
-
disableCasProcessor
public void disableCasProcessor(int aCasProcessorIndex) Disable a CASProcessor in the processing pipeline. Locate it by provided index. The disabled Cas Processor remains in the Processing Pipeline, however it is not used furing processing.- Parameters:
aCasProcessorIndex
- - location in the pipeline of the Cas Processor to delete
-
disableCasProcessor
Alternative method to disable Cas Processor. Uses a name to locate it.- Parameters:
aCasProcessorName
- - a name of the Cas Processor to disable
-
enableCasProcessor
Enables Cas Processor with a given name. Enabled Cas Processor will immediately begin to receive bundles of Cas.- Parameters:
aCasProcessorName
- - name of the Cas Processor to enable
-
analyze
Analyze.- Parameters:
aCasObjectList
- the a cas object listpTrTemp
- the tr temp- Returns:
- true, if successful
- Throws:
Exception
- the exception
-
setReleaseCASFlag
public void setReleaseCASFlag(boolean aFlag) Sets the release CAS flag.- Parameters:
aFlag
- the new release CAS flag
-
setCasPool
Sets the cas pool.- Parameters:
aPool
- the new cas pool
-
postAnalysis
private void postAnalysis(Object[] aCasObjectList, boolean isCasObject, Object[] casObjects, ProcessTrace aProcessTr, boolean doneAlready) throws Exception Post analysis.- Parameters:
aCasObjectList
- the a cas object listisCasObject
- the is cas objectcasObjects
- the cas objectsaProcessTr
- the a process trdoneAlready
- the done already- Throws:
Exception
- -
-
doReleaseCasProcessor
Do release cas processor.- Parameters:
aContainer
- the a containeraCasProcessor
- the a cas processor
-
doEndOfBatch
private void doEndOfBatch(ProcessingContainer aContainer, CasProcessor aProcessor, ProcessTrace aProcessTr, int howManyCases) Do end of batch.- Parameters:
aContainer
- the a containeraProcessor
- the a processoraProcessTr
- the a process trhowManyCases
- the how many cases
-
handleErrors
private boolean handleErrors(Throwable e, ProcessingContainer aContainer, CasProcessor aProcessor, ProcessTrace aProcessTrace, Object[] aCasObjectList, boolean isCasObject) throws Exception Main routine that handles errors occuring in the processing loop.- Parameters:
e
- - exception in the main processing loopaContainer
- - current container of the Cas ProcessoraProcessor
- - current Cas ProcessoraProcessTrace
- - an object containing stats for this procesing loopaCasObjectList
- - list of CASes being analyzedisCasObject
- - determines type of CAS in the aCasObjectList ( CasData or CasObject)- Returns:
- boolean
- Throws:
Exception
- -
-
invokeCasObjectCasProcessor
private void invokeCasObjectCasProcessor(ProcessingContainer container, CasProcessor processor, Object[] aCasObjectList, ProcessTrace pTrTemp, boolean isCasObject) throws Exception Invoke cas object cas processor.- Parameters:
container
- the containerprocessor
- the processoraCasObjectList
- the a cas object listpTrTemp
- the tr tempisCasObject
- the is cas object- Throws:
Exception
- -
-
convertCasDataToCasObject
private void convertCasDataToCasObject(int casIndex, String aContainerName, Object[] aCasObjectList) throws Exception Convert cas data to cas object.- Parameters:
casIndex
- the cas indexaContainerName
- the a container nameaCasObjectList
- the a cas object list- Throws:
Exception
- -
-
invokeCasDataCasProcessor
private void invokeCasDataCasProcessor(ProcessingContainer container, CasProcessor processor, Object[] aCasObjectList, ProcessTrace pTrTemp, boolean isCasObject, boolean retry) throws Exception Invoke cas data cas processor.- Parameters:
container
- the containerprocessor
- the processoraCasObjectList
- the a cas object listpTrTemp
- the tr tempisCasObject
- the is cas objectretry
- the retry- Throws:
Exception
- -
-
containerDisabled
Container disabled.- Parameters:
aContainer
- the a container- Returns:
- true, if successful
-
isProcessorReady
protected boolean isProcessorReady(int aStatus) Check if the CASProcessor status is available for processing.- Parameters:
aStatus
- the a status- Returns:
- true, if is processor ready
-
filterOutTheCAS
private boolean filterOutTheCAS(ProcessingContainer aContainer, boolean isCasObject, Object[] aCasObjectList) Filter out the CAS.- Parameters:
aContainer
- the a containerisCasObject
- the is cas objectaCasObjectList
- the a cas object list- Returns:
- true, if successful
-
notifyListeners
protected void notifyListeners(Object aCas, boolean isCasObject, EntityProcessStatus aEntityProcStatus) Notifies Listeners of the fact that the pipeline has finished processing the current set Cas'es.- Parameters:
aCas
- - object containing an array of OR a single instance of CasisCasObject
- - true if instance of Cas is of type Cas, false otherwiseaEntityProcStatus
- - status object that may contain exceptions and trace
-
doNotifyListeners
protected void doNotifyListeners(Object aCas, boolean isCasObject, EntityProcessStatus aEntityProcStatus) Notifies all configured listeners. Makes sure that appropriate type of Cas is sent to the listener. Convertions take place to ensure compatibility.- Parameters:
aCas
- - Cas to pass to listenerisCasObject
- - true is Cas is of type CASaEntityProcStatus
- - status object containing exceptions and trace info
-
clearCasCache
private void clearCasCache()Clear cas cache. -
pauseContainer
private boolean pauseContainer(ProcessingContainer aContainer, Exception aException, String aThreadId) Determines if the thread should be paused. Pausing container effectively pauses ALL Cas Processors that are managed by the container. The pause is needed when there are multiple pipelines shareing a common service. If this service dies (Socket Down), only one thread should initiate service restart. While the service is being restarted no invocations on the service should be done. Containers will be resumed on successfull service restart.- Parameters:
aContainer
- - a container that manages the current Cas Processor.aException
- the a exceptionaThreadId
- - id of the current thread- Returns:
- true, if successful
-
handleServiceException
private void handleServiceException(ProcessingContainer aContainer, CasProcessor aProcessor, ProcessTrace aProcessTr, Exception ex) throws Exception Handle service exception.- Parameters:
aContainer
- the a containeraProcessor
- the a processoraProcessTr
- the a process trex
- the ex- Throws:
Exception
- -
-
handleSkipCasProcessor
private void handleSkipCasProcessor(ProcessingContainer aContainer, Object[] aCasObjectList, boolean isLastCP) throws Exception Handle skip cas processor.- Parameters:
aContainer
- the a containeraCasObjectList
- the a cas object listisLastCP
- the is last CP- Throws:
Exception
- -
-
getBytes
Returns the size of the CAS object. Currently only CASData is supported.- Parameters:
aCas
- - Cas to get the size for- Returns:
- the size of the CAS object. Currently only CASData is supported.
-
releaseCases
Conditionally, releases CASes back to the CAS pool. The release only occurs if the Cas Processor is the last in the processing chain.- Parameters:
aCasList
- - list of CASes to releaselastProcessor
- - determines if the release takes placeaName
- the a name
-
stopCasProcessors
public void stopCasProcessors(boolean kill) Stops all Cas Processors that are part of this PU.- Parameters:
kill
- - true if CPE has been stopped before finishing processing during external stop
-