Class BaseCPMImpl

  • All Implemented Interfaces:
    java.lang.Runnable, BaseCPM
    Direct Known Subclasses:
    CPMImpl

    public class BaseCPMImpl
    extends java.lang.Object
    implements BaseCPM, java.lang.Runnable
    Main thread that launches CPE and manages it. An application interacts with the running CPE via this object. Through an API, an application may start, pause, resume, and stop a CPE.
    • Field Detail

      • defaultProcessTrace

        private boolean defaultProcessTrace
        The default process trace.
      • cpEngine

        private CPMEngine cpEngine
        The cp engine.
      • checkpoint

        private Checkpoint checkpoint
        The checkpoint.
      • checkpointData

        private CheckpointData checkpointData
        The checkpoint data.
      • num2Process

        private long num2Process
        The num 2 process.
      • killed

        private boolean killed
        The killed.
      • completed

        private boolean completed
        The completed.
      • cpeFactory

        private CPEFactory cpeFactory
        The cpe factory.
      • useJediiReport

        private boolean useJediiReport
        The use jedii report.
      • mEventTypeMap

        private java.util.Map mEventTypeMap
        The m event type map.
      • cpmExecutorService

        public CPMExecutorService cpmExecutorService
        The CPE executor service.
    • Constructor Detail

      • BaseCPMImpl

        public BaseCPMImpl​(CpeDescription aDescriptor)
                    throws java.lang.Exception
        Instantiates and initializes CPE Factory with a given CPE Descriptor and defaults.
        Parameters:
        aDescriptor - - parsed CPE descriptor
        Throws:
        java.lang.Exception - -
      • BaseCPMImpl

        public BaseCPMImpl​(CpeDescription aDescriptor,
                           ResourceManager aResourceManager,
                           boolean aDefaultProcessTrace,
                           java.util.Properties aProps)
                    throws java.lang.Exception
        Instantiates and initializes CPE Factory responsible for creating individual components that are part of the processing pipeline.
        Parameters:
        aDescriptor - - parsed CPE descriptor
        aResourceManager - - ResourceManager instance to be used by the CPE
        aDefaultProcessTrace - - ProcessTrace instance to capture events and stats
        aProps - the a props
        Throws:
        java.lang.Exception - -
      • BaseCPMImpl

        public BaseCPMImpl​(java.lang.Boolean mode,
                           java.lang.String aDescriptor,
                           ResourceManager aResourceManager)
                    throws java.lang.Exception
        Parses CPE descriptor.
        Parameters:
        mode - - indicates if the CPM should use a static descriptor or one provided
        aDescriptor - - provided descriptor path
        aResourceManager - ResourceManager to be used by CPM
        Throws:
        java.lang.Exception - -
    • Method Detail

      • setPerformanceTuningSettings

        public void setPerformanceTuningSettings​(java.util.Properties aPerformanceTuningSettings)
        Plugs in custom perfomance tunning parameters.
        Parameters:
        aPerformanceTuningSettings - the new performance tuning settings
      • setProcessControllerAdapter

        public void setProcessControllerAdapter​(ProcessControllerAdapter aPca)
        Plugs in a given ProcessControllerAdapter. The CPM uses this adapter to request Cas Processor restarts and shutdown.
        Parameters:
        aPca - - instance of the ProcessControllerAdapter
      • setJediiReport

        public void setJediiReport​(boolean aUseJediiReport)
        Sets Jedii-style reporting resources and sets the global flag to indicate what report-style to use at the end of processing. Jedii-style reporting shows a summary for this run. The CPM default report shows more detail information.
        Parameters:
        aUseJediiReport - the new jedii report
      • init

        public void init​(boolean aDummyCasProcessor,
                         java.util.Properties aProps)
                  throws java.lang.Exception
        Instantiates and initializes a CPE.
        Parameters:
        aDummyCasProcessor - -
        aProps - the a props
        Throws:
        java.lang.Exception - -
      • getCPEConfig

        public CpeConfiguration getCPEConfig()
                                      throws java.lang.Exception
        Returns CPEConfig object holding current CPE configuration.
        Returns:
        CPEConfig instance
        Throws:
        java.lang.Exception - -
      • getCasProcessors

        public CasProcessor[] getCasProcessors()
        Description copied from interface: BaseCPM
        Gets the CasProcessorss assigned to this CPM, in the order in which they will be called by the CPM.
        Specified by:
        getCasProcessors in interface BaseCPM
        Returns:
        an array of CasProcessors
      • addCasProcessor

        public void addCasProcessor​(CasProcessor aCasProcessor,
                                    int aIndex)
                             throws ResourceConfigurationException
        Description copied from interface: BaseCPM
        Adds a CasProcessor to this CPM's list of consumers. The new CasProcessor will be added at the specified index.
        Specified by:
        addCasProcessor in interface BaseCPM
        Parameters:
        aCasProcessor - the CasProcessor to add
        aIndex - the index at which to add the CasProcessor
        Throws:
        ResourceConfigurationException - if this CPM is currently processing
      • removeCasProcessor

        public void removeCasProcessor​(CasProcessor aCasProcessor)
        Description copied from interface: BaseCPM
        Removes a CasProcessor to this CPM's list of consumers.
        Specified by:
        removeCasProcessor in interface BaseCPM
        Parameters:
        aCasProcessor - the CasProcessor to remove
      • disableCasProcessor

        public void disableCasProcessor​(java.lang.String aCasProcessorName)
        Description copied from interface: BaseCPM
        Disables a CasProcessor in this CPM's list of CasProcessors.
        Specified by:
        disableCasProcessor in interface BaseCPM
        Parameters:
        aCasProcessorName - the name of the CasProcessor to disable
      • enableCasProcessor

        public void enableCasProcessor​(java.lang.String aCasProcessorName)
        Enable cas processor.
        Parameters:
        aCasProcessorName - the a cas processor name
      • isSerialProcessingRequired

        public boolean isSerialProcessingRequired()
        Description copied from interface: BaseCPM
        Gets whether this CPM is required to process the collection's elements serially (as opposed to performing parallelization). Note that a value of false does not guarantee that parallelization is performed; this is left up to the CPM implementation.
        Specified by:
        isSerialProcessingRequired in interface BaseCPM
        Returns:
        true if and only if serial processing is required
      • setSerialProcessingRequired

        public void setSerialProcessingRequired​(boolean aRequired)
        Description copied from interface: BaseCPM
        Sets whether this CPM is required to process the collection's elements serially (as opposed to performing parallelization). If this method is not called, the default is false. Note that a value of false does not guarantee that parallelization is performed; this is left up to the CPM implementation.
        Specified by:
        setSerialProcessingRequired in interface BaseCPM
        Parameters:
        aRequired - true if and only if serial processing is required
      • isPauseOnException

        public boolean isPauseOnException()
        Description copied from interface: BaseCPM
        Gets whether this CPM will automatically pause processing if an exception occurs. If processing is paused it can be resumed by calling the BaseCPM.resume(boolean) method.
        Specified by:
        isPauseOnException in interface BaseCPM
        Returns:
        true if and only if this CPM will pause on exception
      • setPauseOnException

        public void setPauseOnException​(boolean aPause)
        Description copied from interface: BaseCPM
        Sets whether this CPM will automatically pause processing if an exception occurs. If processing is paused it can be resumed by calling the BaseCPM.resume(boolean) method.
        Specified by:
        setPauseOnException in interface BaseCPM
        Parameters:
        aPause - true if and only if this CPM should pause on exception
      • run

        public void run()
        Specified by:
        run in interface java.lang.Runnable
      • finalizeIt

        public void finalizeIt()
        Called to cleanup CPE on shutdown.
      • setCollectionReader

        public void setCollectionReader​(BaseCollectionReader aCollectionReader)
        Sets the Collection Reader for this CPE.
        Specified by:
        setCollectionReader in interface BaseCPM
        Parameters:
        aCollectionReader - the collection reader
      • isProcessing

        public boolean isProcessing()
        Description copied from interface: BaseCPM
        Determines whether this CPM is currently processing. This means that a processing request has been submitted and has not yet completed or been BaseCPM.stop()ped. If processing is paused, this method will still return true.
        Specified by:
        isProcessing in interface BaseCPM
        Returns:
        true if and only if this CPM is currently processing.
      • pause

        public void pause()
        Description copied from interface: BaseCPM
        Pauses processing. Processing can later be resumed by calling the BaseCPM.resume(boolean) method.
        Specified by:
        pause in interface BaseCPM
      • isPaused

        public boolean isPaused()
        Description copied from interface: BaseCPM
        Determines whether this CPM's processing is currently paused.
        Specified by:
        isPaused in interface BaseCPM
        Returns:
        true if and only if this CPM's processing is currently paused.
      • resume

        public void resume​(boolean aRetryFailed)
        Description copied from interface: BaseCPM
        Resumes processing that has been paused.
        Specified by:
        resume in interface BaseCPM
        Parameters:
        aRetryFailed - if processing was paused because an exception occurred (see BaseCPM.setPauseOnException(boolean)), setting a value of true for this parameter will cause the failed entity to be retried. A value of false (the default) will cause processing to continue with the next entity after the failure.
      • resume

        public void resume()
        Description copied from interface: BaseCPM
        Resumes processing that has been paused.
        Specified by:
        resume in interface BaseCPM
      • kill

        public void kill()
        Kills the CPM hard. CASes in transit are not processed.
      • stop

        public void stop()
        Description copied from interface: BaseCPM
        Stops processing.
        Specified by:
        stop in interface BaseCPM
      • asynchStop

        public void asynchStop()
        Asynch stop.
      • decodeStatus

        private java.lang.String decodeStatus​(int aStatus)
        Decode status.
        Parameters:
        aStatus - the a status
        Returns:
        the string
      • copyComponentEvents

        private void copyComponentEvents​(java.lang.String aEvType,
                                         java.util.List aList,
                                         ProcessTrace aPTr)
                                  throws java.io.IOException
        Copy component events.
        Parameters:
        aEvType - the a ev type
        aList - the a list
        aPTr - the a P tr
        Throws:
        java.io.IOException - Signals that an I/O exception has occurred.
      • displayStats

        public void displayStats​(ProcessTrace aProcessTrace,
                                 int aNumDocsProcessed)
        Helper method to display stats and totals.
        Parameters:
        aProcessTrace - - trace containing stats
        aNumDocsProcessed - - number of entities processed so far
      • buildEventTree

        public void buildEventTree​(ProcessTraceEvent aEvent,
                                   int aTotalTime)
        Helper method to help build the CPM report.
        Parameters:
        aEvent - the a event
        aTotalTime - the a total time
      • getPerformanceReport

        public ProcessTrace getPerformanceReport()
        Returns PerformanceReport for the CPM. This report contains a snapshot of the CPM state.
        Specified by:
        getPerformanceReport in interface BaseCPM
        Returns:
        the performance report
      • createDefaultProcessTrace

        private void createDefaultProcessTrace​(CasProcessor[] aProcessors,
                                               ProcessTrace srcProcTr,
                                               ProcessTrace aProcessTrace)
        Creates the default process trace.
        Parameters:
        aProcessors - the a processors
        srcProcTr - the src proc tr
        aProcessTrace - the a process trace
      • getProgress

        public Progress[] getProgress()
        Returns current CPE progress. How many entities processed and bytes processed.
        Specified by:
        getProgress in interface BaseCPM
        Returns:
        the progress
      • getDescriptor

        public java.lang.String getDescriptor​(java.util.List aList)
                                       throws ResourceConfigurationException
        Returns a CPE descriptor as a String.
        Parameters:
        aList - - list of components
        Returns:
        - descriptor populated with a given components
        Throws:
        ResourceConfigurationException - the resource configuration exception
      • getSynchPoint

        public SynchPoint getSynchPoint()
        Returns a SynchPoint object initialized by the Collection Reader if the Collection Reader implements RecoverableCollectionReader. The synchpoint object contains the current snapshot that includes the last document processed.
        Returns:
        - instance of SynchPoint if the Collection Reader is recoverable, null otherwise