fintp_routingengine
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
RoutingEngine.h
Go to the documentation of this file.
1 /*
2 * FinTP - Financial Transactions Processing Application
3 * Copyright (C) 2013 Business Information Systems (Allevo) S.R.L.
4 *
5 * This program is free software: you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License as published by
7 * the Free Software Foundation, either version 3 of the License, or
8 * (at your option) any later version.
9 *
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
14 *
15 * You should have received a copy of the GNU General Public License
16 * along with this program. If not, see <http://www.gnu.org/licenses/>
17 * or contact Allevo at : 031281 Bucuresti, 23C Calea Vitan, Romania,
18 * phone +40212554577, office@allevo.ro <mailto:office@allevo.ro>, www.allevo.ro.
19 */
20 
21 #ifndef ROUTING_ENGINE_H
22 #define ROUTING_ENGINE_H
23 
24 #include <omp.h>
25 
26 #include "WorkItemPool.h"
27 #include "AbstractWatcher.h"
28 #include "AppSettings.h"
29 #include "InstrumentedObject.h"
30 #include "Transactions/AbstractStatePersistence.h"
31 #include "DB/DbWatcher.h"
32 
33 #include "RoutingMessage.h"
34 #include "RoutingExceptions.h"
35 #include "RoutingSchema.h"
36 #include "RoutingDbOp.h"
37 
38 #include "RoutingCOT.h"
39 #include "RoutingJobExecutor.h"
40 #include "RoutingKeyword.h"
41 
42 #define DEFAULT_QUEUE "_INTERNAL_DUMMY_QUEUE"
43 
44 class ExportedTestObject RoutingEngine : public InstrumentedObject
45 {
46 #if defined( TESTDLL_EXPORT ) || defined ( TESTDLL_IMPORT )
47  friend class RoutingStructuresTest;
48  friend class RoutingActionsTest;
49  friend class RoutingKeywordsTest;
50 #endif
51 
52  public:
53 
54  static RoutingEngine* TheRoutingEngine;
55 
56  // .ctor, destructor
57  RoutingEngine();
58  ~RoutingEngine();
59 
60  // public methods
61  void Start( bool startWatcher = true );
62  void Stop();
63 
64  // profile options
65  static void SetProfileMessageCount( const unsigned int value ) { m_ProfileMessageCount = value; }
66 
67  // callback from db watcher
68  static void NotificationCallback( const AbstractWatcher::NotificationObject* notification );
69  static void OnNewJob( const AbstractWatcher::NotificationObject* notification );
70 
71  static void IdleProcess( void );
72  static void PurgeHashes( void );
73  static void ArchiveIdle( void );
74 
75  static WorkItemPool< RoutingJob >& getJobPool() { return TheRoutingEngine->m_JobPool; }
76  static WorkItemPool< RoutingMessage >& getMessagePool() { return TheRoutingEngine->m_MessagePool; }
77  static WorkItemPool< RoutingJobExecutor >& getThreadExecPool() { return TheRoutingEngine->m_ThreadExecPool; }
78 
79  AppSettings GlobalSettings;
80 
81  /*
82  static string MapId( const string id, const string mappedId, const int autoUnmap );
83  static string RemapId( const string id, const string mappedId );
84  static string GetMappedId( const string id );
85  static string GetReverseMapId( const string id );
86  static bool UnmapId( const string id );*/
87 
88  static string getLiquiditiesSP() { return TheRoutingEngine->m_LiquiditiesSP; }
89  static string getUpdateDateXSLT() { return TheRoutingEngine->m_UpdateDateXSLT; }
90  static string getBatchXslt( const string& queue = DEFAULT_QUEUE );
91  static string getBatchSchema() { return TheRoutingEngine->m_BatchSchema; }
92  static string getBatchSchemaNamespace() { return TheRoutingEngine->m_BatchSchemaNamespace; }
93  static string getBulkReactivateQueue() { return TheRoutingEngine->m_BulkReactivateQueue; }
94 
95  pthread_t getCotMonitorThreadId() { return m_CotThreadId; }
96  pthread_t getRouterThreadId() { return m_RouterThreadId; }
97 
98  bool isRunning() const { return m_Running; }
99 
100  // read application settings at startup
101  void ReadApplicationSettings();
102 
103  // performs some checks/corrective actions to get a consistent state on startup
104  void RecoverLastSession() const;
105 
106  // backward compatibility - may have a negative impact on perfomance
107  static bool hasIBANForLiquidities() { return TheRoutingEngine->m_LiquiditiesHasIBAN; }
108  static bool hasIBANPLForLiquidities() { return TheRoutingEngine->m_LiquiditiesHasIBANPL; }
109  static bool hasCorrForLiquidities() { return TheRoutingEngine->m_LiquiditiesHasCorresps; }
110 
111  // config helpers ( reads and caches config settings )
112  static bool isDuplicateDetectionActive() { return TheRoutingEngine->m_DuplicateDetectionActive; }
113  static bool isIdleArchivingActive() { return TheRoutingEngine->m_IdleArchivingActive; }
114 
115 
116  static int getDuplicateDetectionTimeout() { return TheRoutingEngine->m_DuplicateDetectionTimeout; }
117  static bool shouldCheckDuplicates( const string& service ) { return m_DuplicateChecks[ service ]; }
118 
119  static RoutingSchema* getRoutingSchema();
120 
121  //TODO Move directly to RoutingKeyword class
122  static RoutingKeywordCollection getRoutingKeywords(){ return Keywords; }
123  static RoutingKeywordMappings* getRoutingMappings(){ return &KeywordMappings; }
124  static bool getRoutingIsoMessageType( const string& messageType )
125  {
126  RoutingIsoMessageTypes::const_iterator isoFinder = IsoMessageTypes.find( messageType );
127  if ( isoFinder == IsoMessageTypes.end() )
128  return false;
129 
130  return IsoMessageTypes[messageType];
131  }
132 
133  /*
134  static string EvaluateKeywordValue( const string& messageType, const string& value, const string& keyword, const string& field = "value" )
135  {
136  bool isIso = TheRoutingEngine->KeywordMappings->isIso( messageType );
137  return TheRoutingEngine->Keywords->Evaluate( value, keyword, field, isIso ).first;
138  }
139  static RoutingKeywordCollection* getKeywords() { return TheRoutingEngine->Keywords; }
140 
141  static string GetKeywordXPath( const string& messageType, const string& keyword )
142  {
143  return TheRoutingEngine->KeywordMappings->getXPath( messageType, keyword );
144  }
145 
146  static RoutingKeywordMappings* getKeywordMappings() { return TheRoutingEngine->KeywordMappings; }
147  */
148  static bool shouldTrackMessages(){ return TheRoutingEngine->m_TrackMessages; }
149 
150  private :
151 
152  // static members
153  static RoutingSchema DefaultSchema;
154 
155  static RoutingKeywordMappings KeywordMappings;
156  static RoutingIsoMessageTypes IsoMessageTypes;
157  static RoutingKeywordCollection Keywords;
158 
159  static void* ThreadPoolWatcher( void* data );
160  static void* RouterRoutine( void* data );
161  static void* COTMonitor( void* data );
162 
163  static unsigned int m_ProfileMessageCount;
164  static unsigned int m_CotDelay;
165  static int m_ParallelJobs;
166  static bool m_ShouldStop;
167  static RoutingCOTMarker m_ActiveCotMarker;
168  static RoutingCOTMarker m_PreviousCotMarker;
169  static pthread_mutex_t m_CotMarkersMutex;
170  //static pthread_rwlock_t m_CotMarkersRWLock;
171 
172  static string m_OwnCorrelationId;
173 
174  bool m_Running;
175 
176  static omp_lock_t parallelJobLock;
177 
178  string m_LiquiditiesSP;
179  string m_UpdateDateXSLT;
180  map< string, string > m_BatchXSLT;
181  string m_BatchSchema;
182  string m_BatchSchemaNamespace;
183  string m_BulkReactivateQueue;
184 
185  pthread_t m_CotThreadId, m_RouterThreadId, m_BMThreadId;
186  DbWatcher m_Watcher;
187 
188  WorkItemPool< RoutingMessage > m_MessagePool;
189  WorkItemPool< RoutingJob > m_JobPool;
190  WorkItemPool< RoutingJobExecutor > m_ThreadExecPool;
191 
192  static AbstractStatePersistence* m_IdFactory;
193 
194 #if defined( WIN32 ) && defined ( _DEBUG )
195  _CrtMemState m_NextMemoryState, m_OldMemoryState, m_FinishMemoryState;
196  bool m_OldStateAvailable;
197  static unsigned int m_Notifications;
198 #endif
199 
200  // backward compatibility
201  // database version prior to 4.1(db2) or 2.1(oracle) doesn't have IBAN in BusinessMessages
202  bool m_LiquiditiesHasIBAN;
203  bool m_LiquiditiesHasIBANPL;
204  bool m_LiquiditiesHasCorresps;
205  bool m_DuplicateDetectionActive;
206  bool m_IdleArchivingActive;
207 
208  int m_DuplicateDetectionTimeout;
209 
210  static map< string, bool > m_DuplicateChecks;
211 
212  static pthread_once_t SchemaKeysCreate;
213  static pthread_key_t SchemaKey;
214 
215  static void CreateKeys();
216  static void DeleteSchema( void* data );
217 
218  void LoadKeywords();
219  void LoadMappings();
220 
221  bool m_TrackMessages;
222 };
223 #endif