Endpoint.h

Go to the documentation of this file.

/*
* FinTP - Financial Transactions Processing Application
* Copyright (C) 2013 Business Information Systems (Allevo) S.R.L.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>
* or contact Allevo at : 031281 Bucuresti, 23C Calea Vitan, Romania,
* phone +40212554577, office@allevo.ro <mailto:office@allevo.ro>, www.allevo.ro.
*/

#ifndef QPIENDPOINT_H
#define QPIENDPOINT_H

#ifdef WIN32
        #ifdef CHECK_MEMLEAKS
                //#define _CRTDBG_MAP_ALLOC
                #include <stdlib.h>
                #include <crtdbg.h>
        #endif
#endif

#include "ConnectorMain.h"

#include "InstrumentedObject.h"
#include "FilterChain.h"
#include "Management/Management.h"
#include "Transactions/AbstractStatePersistence.h"
#include "BatchManager/BatchManager.h"
#include "AppSettings.h"
#include "TimeUtil.h"
#include "AbstractWatcher.h"
#include "AppExceptions.h"

class ExportedTestObject EndpointConfig
{
        private :
                EndpointConfig() {};

        public :
                
                enum ConfigDirection
                {
                        AppToWMQ,
                        WMQToApp, 
                        AppToMQ,
                        MQToApp,
                        Common
                };

                enum ConfigSettings
                {
                        // Common settings
                        BATCHMGRTYPE,
                        BATCHMGRXSLT,
                        BATCHMGRXPATH,
                        BATCHMGRTMPL,
                        RPLOPTIONS,
                        RECONSOURCE,
                        HASHXSLT,
                        REGULATIONS,
                        
                        // DB settings
                        DBPROVIDER,
                        DBNAME,
                        DBUSER,
                        DBPASS,
                        TABLENAME,
                        REPLIESTABLE,
                        DADOPTIONS,
                        DADFILE,
                    ACKDADFILE,
                        SPINSERT,
                        SPSELECT,
                        SPMARK,
                        SPMARKCOMMIT,
                        SPMARKABORT,
                        SPWATCHER,
                        UNCMTMAX,
                        NOTIFTYPE,
                    DATEFORMAT,
                    TIMESTAMPFORMAT,
                        DBTOXMLTRIMM,

                        // MQ settings
                        APPQUEUE,
                        RPLYQUEUE,
                        BAKQUEUE,
                        WMQQMGR,
                        MQURI,
                        WMQKEYREPOS,
                        WMQSSLCYPHERSPEC,
                        WMQPEERNAME,
                        WMQFMT,
                        ISSIGNED,
                        CERTIFICATEFILENAME,
                        CERTIFICATEPASSWD,
                        MQSERVERTYPE,

                        // File settings
                        FILESOURCEPATH, 
                        FILEDESTPATH,
                        FILETEMPDESTPATH,
                        FILEERRPATH,
                        FILEFILTER,
                        RPLDESTTPATH,
                        RPLFILEFILTER,
                        RPLFEEDBACK,
                        FILEXSLT,
                        STRICTSWIFTFMT,

                         RECONSFORMAT,

                         RPLSERVICENAME,

                         ISIDENABLED,

                         IDCERTIFICATEFILENAME,

                         IDCERTIFICATEPASSWD,

                         SSLCERTIFICATEFILENAME,
                        
                         SSLCERTIFICATEPASSWD,

                         SERVICENAME, 
                         PARAMFPATTERN,
                         PARAMFXSLT,
                         RENAMEPATTERN,
                         INSERTBLOB,
                         DBCFGNAME,
                         DBCFGUSER,
                         DBCFGPASS,
                         BLOBLOCATOR,
                         BLOBPATTERN,
                         TRACKMESSAGES,
                         LAUCERTIFICATE
                };

                static string getName( const ConfigDirection prefix, const ConfigSettings setting );
};

class ExportedTestObject Endpoint : public InstrumentedObject
{
        private :

                unsigned int m_MessageThrottling;

                string internalPrepare();
                void internalCommit( const string& correlationId );
                void internalAbort( const string& correlationId );
                void internalRollback( const string& correlationId );
                void internalProcess( const string& correlationId );    

                void fireManagementEvent( TransactionStatus::TransactionStatusEnum, const void* additionalData );

#if defined( WIN32 ) && defined ( CHECK_MEMLEAKS )
                _CrtMemState m_NextMemoryState, m_OldMemoryState;
                bool m_OldStateAvailable;
#endif

                bool m_FatalError;
                static AppSettings *m_GlobalSettings;

                //performance
                TimeUtil::TimeMarker m_LastReportTime;
                //unsigned long m_PerfAborted, m_PerfCommited, m_PerfTotal;

        protected : //methods

                // methods for controlling the endpoint 
                virtual void internalStart() = 0;
                virtual void internalStop() = 0;
        
                Endpoint();

                string getServiceName() const { return m_ServiceName; }
                void setCorrelationId( const string& correlationId );

                AppSettings& getGlobalSettings() const;
                bool haveGlobalSetting( const EndpointConfig::ConfigDirection prefix, const EndpointConfig::ConfigSettings setting ) const;
                string getGlobalSetting( const EndpointConfig::ConfigDirection prefix, const EndpointConfig::ConfigSettings setting, const string& defaultValue = "__NODEFAULT" ) const;

        protected : // members

                static void ( *m_ManagementCallback )( TransactionStatus::TransactionStatusEnum, void* additionalData );

                BatchManagerBase* m_BatchManager;
                AbstractStatePersistence* m_PersistenceFacility;
                
                unsigned int m_BackoutCount;
                unsigned int m_CurrentStage;
                unsigned int m_CrtBatchItem;

                string m_ServiceName;
                string m_ServiceThreadId;
                string m_CorrelationId, m_LastFailureCorrelationId, m_TransactionKey;

                pthread_t m_SelfThreadId;
                bool m_Running, m_LastOpSucceeded;

                AbstractWatcher::NotificationPool m_NotificationPool;
                
                NameValueCollection m_TransportHeaders;
                
                string m_XmlData;
                AppException m_TrackingData;
                                
                //indicate for Batch processing if last message or not
                bool m_IsLast; 
                bool m_TrackMessages;

                FilterChain* m_FilterChain;
        
        public:

                virtual ~Endpoint();
                                
                // methods for controlling the endpoint 
                static void* StartInNewThread( void* );
                void Start();
                void Stop();

                // called before start. allows the endpoint to prepare environment
                virtual void Init() = 0;

                // methods for transactional execution
                virtual string Prepare() = 0;

                virtual void Commit() = 0;

                virtual void Abort() = 0;
                
                virtual void Rollback() = 0;

                virtual void Process( const string& correlationId ) = 0;        

                // workflow methods
                bool PerformMessageLoop();
                bool PerformMessageLoop( bool inBatch );
                
                virtual bool moreMessages() const { return false; }

                virtual void trackMessage( const string& payload, const NameValueCollection& transportHeaders );
                                
                // management methods           
                static void setManagementCallback( void ( *callback )( TransactionStatus::TransactionStatusEnum, void* ) )
                {
                        m_ManagementCallback = callback;
                }

                // state persistence
                void setPersistenceFacility( AbstractStatePersistence* facility );
                AbstractStatePersistence* getPersistenceFacility();
                
                void setServiceName( const string& value ){ m_ServiceName = value; }
                void setServiceThreadId( const string& serviceThreadId ) { m_ServiceThreadId = serviceThreadId; }
                
                pthread_t getThreadId() { return m_SelfThreadId; }
                const pthread_t getThreadId() const { return m_SelfThreadId; }

                virtual pthread_t getWatcherThreadId() = 0;
                
                FilterChain* getFilterChain(){ return m_FilterChain; }

                BatchManagerBase* getBatchManager(){ return m_BatchManager; }

                static void setGlobalSettings( AppSettings* settings ) { m_GlobalSettings = settings; }
};

class EndpointFactory
{
        public :
                static Endpoint* CreateEndpoint( const string& serviceName, const string& type, bool fetcher );
};

#endif //QPIENDPOINT_H