Pipelines stage API - CProcess class

CProcess class

Pipelines v1.6

Description

The CProcess class is the base class of the CStage class.

It is through this class object that you access your stages’ core settings. The class comprises a Stage() function which returns the number of your stage, a Pipeline() function which returns the number of your pipeline, a Manager() function which provides access to the StageManager to which your stage belongs and two pure virtual functions, Initialise() and Go(). Initialise() represents the initialisation phase of your stage and Go() represents the runtime phase of your stage. In order to instantiate your CStage class object you must provide an implementation of both pure virtual functions and call each one, in turn and in that order from the required stage DLL interface function BeginStage().

Definition

The following class definition is an edited version; it details only those members which are accessible and of use to you. In addition; any examples that may be shown are only intended as a demonstration of how certain functions might be used, they do not represent best practice nor do they take into account the scope and de-allocation of class objects.

The Initialise() and Go() function are pure virtual functions and such they require a full implementation by your CStage class.

class CProcess

    public:

    virtual bool Initialise( void ) = 0;

    virtual DWORD Go( void ) = 0;

    DWORD Pipeline( void ) const;

    DWORD Stage( void ) const;

    CStageManager *Manager( void ) const;

};

Member variables

None.

Member functions

virtual bool CProcess::Initialise( void ) = 0;

●

Purpose

Initialise() is a required stage DLL function that must be called by your stages' BeginStage() function as part of the stage initialisation phase.

The function allows you to validate your stage command argument; to pre-allocate resources that your stage might require and to issue warning and quiescent error initialisation messages. A value or table that can be pre-computed should be determined in this function.

●

Parameters

●

void

●

Returns

●

bool

true which indicates that your stage command argument has parsed correctly and has successfully allocated all of its’ required resources, and is fit to proceed to its’ runtime phase, otherwise; false which indicates that a terminal condition has occurred and that the StageManager should issue a quiesce command against all the stage’s in your the pipeline; that the global stage initialisation phase of the pipeline to which your stage belongs, should terminate.

●

Messages

None

●

Usage

Your stage requires a DLL Initialise() function, and it is in this function that you define your stages’ minimum, maximum and required input and output stream connections, parse your stage command to validate that the argument is as required and if necessary issue warning and initialisation phase quiescent error messages. Every stage DLL must comprise this function as a minimum requirement. Initialise() allows you to pre-allocate and pre-compute any resources and values/tables which are required by your stage; everything that can be pre-computed ahead of your stages’ runtime phase should be determined in this function.

bool CMyStage::Initialise( void )

    // Access the manager.

    CStageManager *pManager = Manager();

    // Access my stage.

    CStage *pStage = pManager->GetStage( *this );

    pStage->MinInStreamsRequired( 1 );

    pStage->MaxOutStreamsAllowed( 2 );

    pStage->MinOutStreamsRequired( 1 );

    pStage->MaxOutStreamsAllowed( 1 );

    // Point to my command argument.

    const char *pszArgument = pStage->Argument();

    // My stage does not require a command argument; if there is one,

    // then it is unexpected.

    if( *pszArgument )

        // Tell the user the type of argument that I expect.

        pManager->StageMessage( *this, 33, _ERROR_, pszArgument );

        return( false );

    return( true );

virtual DWORD CProcess::Go( void ) = 0;

●

Purpose

Go() is a required stage DLL function that must be called by your stages' BeginStage() function as part of the stage runtime phase.

This function should be called immediately after your stage has returned a value of true from its’ Initialise() function (and provided that all the stages in your set of pipelines also return a value of true). Go() should comprise your stages’ main processing loop; reading records from its input streams and writing records to its output streams.

●

Parameters

●

void

●

Returns

●

DWORD

_RC_SUCCESS_ which indicates that your stage has completed its’ runtime processing phase successfully, otherwise; _RC_FAIL_ which indicates that a runtime error has occurred; a value which causes the StageManager to issue a global/pipeline-wide quiescent command; instructing all of the pipelines and stages in your stages’ set of pipelines to terminate.

●

Messages

None

●

Usage

Your stage requires a stage DLL Go() function, and it is in this function that your stage must either directly or indirectly perform its’ runtime processing; reading records from its input streams and writing records to its output streams. You may call any number of other functions from within the function; however, the value which you return from this function determines whether the pipeline or set of pipelines to which your stage belongs, runs to completion or is quiesced.

The following example reads records from its primary input stream; validates that the first four bytes of each input record contains the string “AF23” and if this is not true; the stage returns a value of _RC_FAIL_, which causes the stage to end and subsequently the StageManager to issue a global quiesce command which will cause your pipeline to terminate.

DWORD CMyStage::Go( void )

    // Access the manager.

    CStageManager *pManager = Manager();

    // Record buffer.

    CString szRecord;

    while( true )

        // Read a record from my primary input stream.

        if( !pManager->ReadRecord( *this, 0, &szRecord ) )

            // This is the end-of-file on my primary input stream.

            break;

        // If the first four bytes of the input record does not comprise

        // the string “AF23” then terminate the stage.

        CString szKeyWord = szRecord.Left( 4 );

        if( szKeyWord.CompareNoCase( szKeyWord, “AF23” ) )

            Return( _RC_FAIL_ );

//

        // TODO: operate on the record here.

//

...

        // Write the record to my primary output stream.

        if( !pManager->WriteRecord( *this, 0, &szRecord ) )

            // This is the end-of-file on my primary output stream.

            break;

    return( _RC_SUCCESS_ );

DWORD CProcess::Pipeline( void ) const;

●

Purpose

Use Pipeline() to get the number of your pipeline.

In a Pipelines script, or indeed in any set of pipelines, your stage may need to know the number of the pipeline to which it belongs; to determine its relative position and to make decisions about how it is to parse its’ stage command argument. There are a number of instances where you might need to know the number of your pipeline and these might include; validating the minimum and maximum number of required input and output stream connections and to generate uniquely identifiable filenames or values.

●

Parameters

●

void

●

Returns

●	DWORD The number of the pipeline to which your stage belongs; starting with pipeline number 0, then pipeline 1, then 2 and so on.

●

Messages

None

●

Usage

For example; to create a work filename which uniquely identifies your stage as the owner; you might use the following approach:

bool CMyStage::Initialise( void )

    // Access the manager.

    CStageManager *pManager = Manager();

    // Access my stage.

    CStage *pStage = pManager->GetStage( *this );

...

    // Create a unique filename.

    CString szTmpFile;

    szTmpFile.Format( "TEMP-%.5d%.3d%.3d.txt", ( DWORD )pManager->GetProcId(), Pipeline(), Stage() );

...

    return( true );

DWORD CProcess::Stage( void ) const;

●

Purpose

Use Stage() to get the number of your stage.

In a Pipelines script, or indeed in any set of pipelines, your stage may need to know its’ stage number; to determine its relative position and to make decisions about how it is to parse its’ stage command argument. There are a number of instances where you might need to know the number of your stage and these might include; validating the minimum and maximum number of required input and output stream connections and to generate uniquely identifiable filenames or values.

●

Parameters

●

void

●

Returns

●	DWORD The number of your stage; starting with stage number 0, then stage 1, then 2 and so on.

●

Messages

None

●

Usage

For example; your stage may require a command argument only when it is the first stage of the first pipeline.

bool CMyStage::Initialise( void )

    // Access the manager.

    CStageManager *pManager = Manager();

    // Access my stage.

    CStage *pStage = pManager->GetStage( *this );

...

    // Point to my command argument.

    const char *pszArgument = pStage->Argument();

    // If I am the first stage in the first pipeline, then I must have

    // a delimited string command argument.

    if( Pipeline() == 0 && Stage() == 0 && !( *pszArgument ) )

        // Tell the user the type of argument that I expect.

        pManager->StageMessage( *this, 19, _ERROR_, pszArgument );

        return( false );

...

    return( true );

CStageManager *CProcess::Manager( void ) const;

●

Purpose

Use Manager() to get a pointer to your StageManager.

The StageManager is responsible for the coordination and service of all your stages’ requirements, including; access to your input and output streams, reading records, writing records, testing stream connections, providing information on your pipelines’ runtime status and more. Manager() is your stages’ interface to the support functions which allow your stage to operate in a multi-pipeline, multi-stream configuration. Manager() returns a pointer to the CStageManager class object to which your stage belongs; the stage manager which services your stage.

●

Parameters

●

void

●

Returns

●	CStageManager * A pointer to the CStageManager to which your stage belongs.

●

Messages

None

●

Usage

Once you have a pointer to your CStageManager; you can access all of your stages’ settings and functions. The following pipeline example; reads a record from each of its connected input streams.

DWORD CMyStage::Go( void )

    // Access the manager.

    CStageManager *pManager = Manager();

    // Access my stage.

    CStage *pStage = pManager->GetStage( *this );

...

    // Loop through my input streams and read a record from each

    // connected stream.

    CString szRecord;

    DWORD dwMaxInStream = pStage->MaxInStream();

    for( DWORD dwInStream = 0; dwInStream < dwMaxInStream; ++dwInStream )

        CInStream *pInStream = pStage->GetInStream( dwInStream );

        if( pInStream->IsConnected() )

            // Read a record from the input stream.

            pManager->ReadRecord( *this, dwIndex, &szRecord ) )

...

...

    return( _RC_SUCCESS_ )