Package weka.knowledgeflow

Class StepManagerImpl

  • All Implemented Interfaces:
    StepManager
    public class StepManagerImpl
extends java.lang.Object
implements StepManager
    Concrete implementation of the StepManager interface. Has a number of methods, beyond those aimed at Step implementations, that are useful for applications that manipulate Steps and their connections.
    Version:
    $Revision: $
    Author:
    Mark Hall (mhall{[at]}pentaho{[dot]}com)

    • Constructor Detail

      • StepManagerImpl

        public StepManagerImpl​(Step step)
        Constructor
        Parameters:
        step - the Step to manage

    • Method Detail

      • getName

        public java.lang.String getName()
        Get the name of the Step being managed
        Specified by:
        getName in interface StepManager
        Returns:
        the name of the Step being managed

      • getManagedStep

        public Step getManagedStep()
        Get the step managed by this manager
        Specified by:
        getManagedStep in interface StepManager
        Returns:
        the step managed by this manager

      • setManagedStep

        public void setManagedStep​(Step step)
        Set the step managed by this manager
        Parameters:
        step - the step to manage

      • setStepIsResourceIntensive

        public void setStepIsResourceIntensive​(boolean resourceIntensive)
        Set whether the managed step is resource (cpu/memory) intensive or not
        Specified by:
        setStepIsResourceIntensive in interface StepManager
        Parameters:
        resourceIntensive - true if the managed step is resource intensive

      • stepIsResourceIntensive

        public boolean stepIsResourceIntensive()
        Get whether the managed step is resource (cpu/memory) intensive or not
        Specified by:
        stepIsResourceIntensive in interface StepManager
        Returns:
        true if the step is resource intensive

      • setStepMustRunSingleThreaded

        public void setStepMustRunSingleThreaded​(boolean mustRunSingleThreaded)
        Set whether the managed step must run single-threaded. I.e. in an executor service with one worker thread, thus effectively preventing more than one copy of the step from executing at any one point in time
        Specified by:
        setStepMustRunSingleThreaded in interface StepManager
        Parameters:
        mustRunSingleThreaded - true if the managed step must run single-threaded

      • getStepMustRunSingleThreaded

        public boolean getStepMustRunSingleThreaded()
        Get whether the managed step must run single-threaded. I.e. in an executor service with one worker thread, thus effectively preventing more than one copy of the step from executing at any one point in time
        Specified by:
        getStepMustRunSingleThreaded in interface StepManager
        Returns:
        true if the managed step must run single-threaded

      • getStepVisual

        public StepVisual getStepVisual()
        Get the step visual in use (if running in a visual environment)
        Returns:
        the step visual in use

      • setStepVisual

        public void setStepVisual​(StepVisual visual)
        Set the step visual to use when running in a graphical environment
        Parameters:
        visual - the step visual to use

      • setStepProperty

        public void setStepProperty​(java.lang.String name,
                            java.lang.Object value)
        Set a property for this step
        Parameters:
        name - the name of the property
        value - the value of the property

      • getStepProperty

        public java.lang.Object getStepProperty​(java.lang.String name)
        Get a named property for this step.
        Parameters:
        name - the name of the property to get
        Returns:
        the value of the property or null if the property is not set

      • getSettings

        public Settings getSettings()
        Get the current knowledge flow settings
        Specified by:
        getSettings in interface StepManager
        Returns:
        the current knowledge flow settings
        Throws:
        java.lang.IllegalStateException - if there is no execution environment available

      • setLoggingLevel

        public void setLoggingLevel​(LoggingLevel newLevel)
        Set the logging level to use
        Parameters:
        newLevel - the level to use

      • getLog

        public Logger getLog()
        Get the log to use
        Specified by:
        getLog in interface StepManager
        Returns:
        the log in use or null if no log has been set

      • setLog

        public void setLog​(Logger log)
        Set the log to use
        Parameters:
        log - the log to use

      • isStepBusy

        public boolean isStepBusy()
        Returns true if, at the current time, the managed step is busy with processing
        Specified by:
        isStepBusy in interface StepManager
        Returns:
        true if the managed step is busy with processing

      • isStopRequested

        public boolean isStopRequested()
        Return true if a stop has been requested by the runtime environment
        Specified by:
        isStopRequested in interface StepManager
        Returns:
        true if a stop has been requested

      • isStepFinished

        public boolean isStepFinished()
        Return true if the current step is finished.
        Specified by:
        isStepFinished in interface StepManager
        Returns:
        true if the current step is finished

      • setStopRequested

        public void setStopRequested​(boolean stopRequested)
        Set the status of the stop requested flag
        Parameters:
        stopRequested - true if a stop has been requested

      • processing

        public void processing()
        Started processing. Sets the busy flag to true.
        Specified by:
        processing in interface StepManager

      • finished

        public void finished()
        Finished all processing. Sets the busy flag to false and prints a finished message to the status area of the log.
        Specified by:
        finished in interface StepManager

      • interrupted

        public void interrupted()
        Finished processing due to a stop being requested. Sets the busy flag to false.
        Specified by:
        interrupted in interface StepManager

      • isStreamFinished

        public boolean isStreamFinished​(Data data)
        Returns true if this data object marks the end of an incremental stream. Note - does not check that the data object is actually an incremental one of some sort! Just checks to see if the CON_AUX_DATA_INCREMENTAL_STREAM_END flag is set to true or not;
        Specified by:
        isStreamFinished in interface StepManager
        Parameters:
        data - the data element to check
        Returns:
        true if the data element is flagged as end of stream

      • throughputUpdateStart

        public void throughputUpdateStart()
        Clients can use this to record a start point for streaming throughput measuring
        Specified by:
        throughputUpdateStart in interface StepManager

      • throughputUpdateEnd

        public void throughputUpdateEnd()
        Clients can use this to record a stop point for streaming throughput measuring
        Specified by:
        throughputUpdateEnd in interface StepManager

      • throughputFinished

        public void throughputFinished​(Data... data)
                        throws WekaException
        Clients can use this to indicate that throughput measuring is finished (i.e. the stream being processed has ended). Final throughput information is printed to the log and status
        Specified by:
        throughputFinished in interface StepManager
        Parameters:
        data - one or more Data events (with appropriate connection type set) to pass on to downstream connected steps. These are used to carry any final data and to inform the downstream step(s) that the stream has ended
        Throws:
        WekaException - if a problem occurs

      • disconnectStepWithConnection

        public void disconnectStepWithConnection​(Step toDisconnect,
                                         java.lang.String connType)
        Disconnect the supplied step under the associated connection type from both the incoming and outgoing connections for the step managed by this manager. Does nothing if this step does not have any connections to the supplied step, or does not have connections to the supplied step of the required type.
        Parameters:
        toDisconnect - the step to disconnect
        connType - the connection type to disconnect

      • disconnectStep

        public void disconnectStep​(Step toDisconnect)
        Remove the supplied step from connections (both incoming and outgoing of all types) for the step managed by this manager. Does nothing if the this step does not have any connections to the supplied step
        Parameters:
        toDisconnect - the step to disconnect

      • clearAllConnections

        public void clearAllConnections()
        Clear all connections to/from the step managed by this manager. Also makes sure that all directly connected upstream and downstream steps remove their respective outgoing and incoming connections to this step

      • addIncomingConnection

        public void addIncomingConnection​(java.lang.String connectionName,
                                  StepManagerImpl step)
        Add an incoming connection (comprising of the type of connection and associated step component) to this step of the specified type
        Parameters:
        connectionName - the name of the type of connection to add
        step - the source step component that is connecting with given connection type

      • removeIncomingConnection

        public void removeIncomingConnection​(java.lang.String connectionName,
                                     StepManagerImpl step)
        Remove an incoming connection to this step of the specified type
        Parameters:
        connectionName - the name of the type of connection to remove
        step - the source step component associated with the given connection type

      • addOutgoingConnection

        public boolean addOutgoingConnection​(java.lang.String connectionName,
                                     StepManagerImpl step)
        Add an outgoing connection (comprising of the type of connection and associated target step) to this step of the specified type. Connection is only made if the target step will accept the connection type at this time
        Parameters:
        connectionName - the name of the type of connection to add
        step - the target step component that is receiving the given connection type it can't accept the connection at the present time
        Returns:
        true if the connection was successful

      • addOutgoingConnection

        public boolean addOutgoingConnection​(java.lang.String connectionName,
                                     StepManagerImpl step,
                                     boolean force)
        Add an outgoing connection (comprising of the type of connection and associated target step) to this step of the specified type. Connection is only made if the target step will accept the connection type at this time
        Parameters:
        connectionName - the name of the type of connection to add
        step - the target step component that is receiving the given connection type
        force - whether to force the connection, even if the target step says it can't accept the connection at the present time
        Returns:
        true if the connection was successful

      • removeOutgoingConnection

        public void removeOutgoingConnection​(java.lang.String connectionName,
                                     StepManagerImpl step)
        Remove an outgoing connection from this step of the specified type
        Parameters:
        connectionName - the name of the type of connection to remove
        step - the target step component associated with the given connection type

      • getIncomingConnectedStepsOfConnectionType

        public java.util.List<StepManager> getIncomingConnectedStepsOfConnectionType​(java.lang.String connectionName)
        Get a list of steps providing incoming connections of the specified type
        Specified by:
        getIncomingConnectedStepsOfConnectionType in interface StepManager
        Parameters:
        connectionName - the type of connection being received by this step
        Returns:
        a list of connected steps

      • getOutgoingConnectedStepsOfConnectionType

        public java.util.List<StepManager> getOutgoingConnectedStepsOfConnectionType​(java.lang.String connectionName)
        Description copied from interface: StepManager
        Get a list of downstream steps connected to this step with the given connection type.
        Specified by:
        getOutgoingConnectedStepsOfConnectionType in interface StepManager
        Parameters:
        connectionName - the name of the outgoing connection
        Returns:
        a list of downstream steps connected to this one with the named connection type

      • getIncomingConnectedStepWithName

        public StepManager getIncomingConnectedStepWithName​(java.lang.String stepName)
        Get a named step connected to this step with an incoming connection
        Specified by:
        getIncomingConnectedStepWithName in interface StepManager
        Parameters:
        stepName - the name of the step to look for
        Returns:
        the connected step

      • getOutgoingConnectedStepWithName

        public StepManager getOutgoingConnectedStepWithName​(java.lang.String stepName)
        Get a named step connected to this step with an outgoing connection
        Specified by:
        getOutgoingConnectedStepWithName in interface StepManager
        Parameters:
        stepName - the name of the step to look for
        Returns:
        the connected step

      • getOutgoingConnections

        public java.util.Map<java.lang.String,​java.util.List<StepManager>> getOutgoingConnections()
        Get the map of downstream (outgoing connections) connected steps
        Specified by:
        getOutgoingConnections in interface StepManager
        Returns:
        the map of downstream connected steps

      • getIncomingConnections

        public java.util.Map<java.lang.String,​java.util.List<StepManager>> getIncomingConnections()
        Get the man of upstream (incoming connections) connected steps
        Specified by:
        getIncomingConnections in interface StepManager
        Returns:
        the map of upstream connected steps

      • addStepOutputListener

        public void addStepOutputListener​(StepOutputListener listener,
                                  java.lang.String outputConnectionName)
        Register non-step third party to receive data from the managed step for the specified outgoing connection type. Output listeners are not serialized into the JSON flow when flows are saved.
        Parameters:
        listener - the output listener to register
        outputConnectionName - the name of the connection type

      • removeStepOutputListener

        public void removeStepOutputListener​(StepOutputListener listener,
                                     java.lang.String outputConnectionName)
        De-register non-step third party from receiving data from the managed step
        Parameters:
        listener - the output listener to de-register
        outputConnectionName - the name of the connection type the listener is registered against

      • clearAllStepOutputListeners

        public void clearAllStepOutputListeners()
        Clear all registered StepOutputListeners

      • clearStepOutputListeners

        public void clearStepOutputListeners​(java.lang.String outputConnectionName)
        Clear all the StepOutputListeners that are registered to receive the supplied connection type.
        Parameters:
        outputConnectionName - type of the connection to clear the listeners for

      • outputData

        public void outputData​(java.lang.String outgoingConnectionName,
                       Data data)
                throws WekaException
        Output a Data object to all downstream connected Steps that are connected with the supplied connection name. Sets the connection type on the supplied Data object to the supplied connection name. Also notifies any registered StepOutputListeners.
        Specified by:
        outputData in interface StepManager
        Parameters:
        outgoingConnectionName - the type of the outgoing connection to send data to
        data - a single Data object to send
        Throws:
        WekaException

      • outputData

        public void outputData​(Data... data)
                throws WekaException
        Output one or more Data objects to all relevant steps. Populates the source in each Data object for the client, HOWEVER, the client must have populated the connection type in each Data object to be output so that the StepManager knows which connected steps to send the data to. Also notifies any registered StepOutputListeners. Note that the downstream step(s)' processIncoming() method is called in a separate thread for batch connections. Furthermore, if multiple Data objects are supplied via the varargs argument, and a target step will receive more than one of the Data objects, then they will be passed on to the step in question sequentially within the same thread of execution.
        Specified by:
        outputData in interface StepManager
        Parameters:
        data - one or more Data objects to be sent
        Throws:
        WekaException - if a problem occurs

      • outputData

        public void outputData​(java.lang.String outgoingConnectionName,
                       java.lang.String stepName,
                       Data data)
                throws WekaException
        Outputs the supplied Data object to the named Step. Does nothing if the named step is not connected immediately downstream of this Step. Sets the supplied connection name on the Data object. Also notifies any StepOutputListeners.
        Specified by:
        outputData in interface StepManager
        Parameters:
        outgoingConnectionName - the name of the outgoing connection
        stepName - the name of the step to send the data to
        data - the data to send
        Throws:
        WekaException

      • getStepOutgoingConnectionTypes

        public java.util.List<java.lang.String> getStepOutgoingConnectionTypes()
        Used by the rendering routine in LayoutPanel to ensure that connections downstream from a deleted connection get rendered in grey rather than red.
        Returns:
        a list of outgoing connection types that the managed step can produce (adjusted to take into account any upstream broken connections)

      • numIncomingConnections

        public int numIncomingConnections()
        Get the number of incoming connections to the managed step
        Specified by:
        numIncomingConnections in interface StepManager
        Returns:
        the number of incoming connections

      • numIncomingConnectionsOfType

        public int numIncomingConnectionsOfType​(java.lang.String connectionName)
        Get the number of incoming connections to the managed step of a given type
        Specified by:
        numIncomingConnectionsOfType in interface StepManager
        Parameters:
        connectionName - the name of the connection type
        Returns:
        the number of incoming connections of this type

      • numOutgoingConnections

        public int numOutgoingConnections()
        Get the number of outgoing connections from the managed step
        Specified by:
        numOutgoingConnections in interface StepManager
        Returns:
        the number of incoming connections

      • numOutgoingConnectionsOfType

        public int numOutgoingConnectionsOfType​(java.lang.String connectionName)
        Get the number of outgoing connections from the managed step of a given type
        Specified by:
        numOutgoingConnectionsOfType in interface StepManager
        Parameters:
        connectionName - the name of the connection type
        Returns:
        the number of outgoing connections of this type

      • getIncomingStructureForConnectionType

        public Instances getIncomingStructureForConnectionType​(java.lang.String connectionName)
                                                throws WekaException
        Attempt to get the incoming structure (as a header-only set of instances) for the named incoming connection type. Assumes that there is only one incoming connection of the named type. If there are zero, or more than one, then null is returned
        Specified by:
        getIncomingStructureForConnectionType in interface StepManager
        Parameters:
        connectionName - the name of the incoming connection to get the structure for
        Returns:
        the structure as a header-only set of instances or null if there are zero or more than one upstream connected steps producing the named connection, or if the upstream step can't tell us the structure, or if the upstream step can't represent the structure of the connection type as a set of instances.
        Throws:
        WekaException - if a problem occurs

      • getIncomingStructureForConnectionType

        public Instances getIncomingStructureForConnectionType​(java.lang.String connectionName,
                                                       Environment env)
                                                throws WekaException
        Attempt to retrieve the structure (as a header-only set of instances) for the named incoming connection type. Assumes that there is only one step connected with the supplied incoming connection type.
        Specified by:
        getIncomingStructureForConnectionType in interface StepManager
        Parameters:
        connectionName - the type of the incoming connection to get the structure for
        env - the Environment to use
        Returns:
        the structure of the data for the specified incoming connection, or null if the structure can't be determined (or represented as an Instances object)
        Throws:
        WekaException - if a problem occurs

      • getIncomingStructureFromStep

        public Instances getIncomingStructureFromStep​(StepManager sourceStep,
                                              java.lang.String connectionName)
                                       throws WekaException
        Attempt to get the incoming structure (as a header-only set of instances) from the given managed step for the given connection type.
        Specified by:
        getIncomingStructureFromStep in interface StepManager
        Parameters:
        sourceStep - the step manager managing the source step
        connectionName - the name of the connection to attempt to get the structure for
        Returns:
        the structure as a header-only set of instances, or null if the source step can't determine this at present or if it can't be represented as a set of instances.
        Throws:
        WekaException - if a problem occurs

      • logLow

        public void logLow​(java.lang.String message)
        Log a message at the low logging level
        Specified by:
        logLow in interface StepManager
        Parameters:
        message - the message to log

      • logBasic

        public void logBasic​(java.lang.String message)
        Log a message at the basic logging level
        Specified by:
        logBasic in interface StepManager
        Parameters:
        message - the message to log

      • logDetailed

        public void logDetailed​(java.lang.String message)
        Log a message at the detailed logging level
        Specified by:
        logDetailed in interface StepManager
        Parameters:
        message - the message to log

      • logDebug

        public void logDebug​(java.lang.String message)
        Log a message at the debugging logging level
        Specified by:
        logDebug in interface StepManager
        Parameters:
        message - the message to log

      • logWarning

        public void logWarning​(java.lang.String message)
        Log a warning message
        Specified by:
        logWarning in interface StepManager
        Parameters:
        message - the message to log

      • logError

        public void logError​(java.lang.String message,
                     java.lang.Throwable cause)
        Log an error
        Specified by:
        logError in interface StepManager
        Parameters:
        message - the message to log
        cause - the optional Throwable to log

      • statusMessage

        public void statusMessage​(java.lang.String message)
        Output a status message to the status area of the log
        Specified by:
        statusMessage in interface StepManager
        Parameters:
        message - the message to output

      • log

        public void log​(java.lang.String message,
                LoggingLevel level)
        Log a message at the supplied logging level
        Specified by:
        log in interface StepManager
        Parameters:
        message - the message to write
        level - the level for the message

      • environmentSubstitute

        public java.lang.String environmentSubstitute​(java.lang.String source)
        Substitute the values of environment variables in the given string
        Specified by:
        environmentSubstitute in interface StepManager
        Parameters:
        source - the source string to substitute in
        Returns:
        the source string with all known environment variables resolved

      • getInfoStep

        public Step getInfoStep​(java.lang.Class stepClass)
                 throws WekaException
        Returns a reference to the step being managed if it has one or more outgoing CON_INFO connections and the managed step is of the supplied class
        Specified by:
        getInfoStep in interface StepManager
        Parameters:
        stepClass - the expected class of the step
        Returns:
        the step being managed if outgoing CON_INFO connections are present and the step is of the supplied class
        Throws:
        WekaException - if there are no outgoing CON_INFO connections or the managed step is the wrong type

      • getInfoStep

        public Step getInfoStep()
                 throws WekaException
        Returns a reference to the step being managed if it has one or more outgoing CON_INFO connections.
        Specified by:
        getInfoStep in interface StepManager
        Returns:
        the step being managed if outgoing CON_INFO connections are present
        Throws:
        WekaException - if there are no outgoing CON_INFO connections

      • findStepInFlow

        public StepManager findStepInFlow​(java.lang.String stepNameToFind)
        Finds a named step in the current flow. Returns null if the named step is not present in the flow
        Specified by:
        findStepInFlow in interface StepManager
        Parameters:
        stepNameToFind - the name of the step to find
        Returns:
        the StepManager of the named step, or null if the step does not exist in the current flow.

      • stepStatusMessagePrefix

        public java.lang.String stepStatusMessagePrefix()
        Gets a prefix for the step managed by this manager. Used to uniquely identify steps in the status area of the log
        Returns:
        a unique prefix for the step managed by this manager