fintp_transport
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
WMqHelper.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 MQHELPER_H
22 #define MQHELPER_H
23 
24 #include <imqi.hpp>
25 #include <cmqbc.h>
26 #include <cmqcfc.h>
27 #include <string>
28 
29 #include "../TransportHelper.h"
30 
31 using namespace std;
32 
33 namespace FinTP
34 {
35  class ExportedTransportObject WMqHelper : public TransportHelper
36  {
37  private :
38 
39  // WMQ objects
40  ImqQueueManager m_QueueManager;
41  ImqQueue m_Queue, m_BackupQueue, m_BackoutQueue;
42  ImqChannel m_Channel;
43 
44  long m_QueueManagerCCSID, m_QueueManagerEncoding;
45  bool m_ConversionNeeded;
46  long m_QueueManagerPlatform;
47 
48  long m_OpenOption;
49  string m_ConnectionString;
50 
51  long m_MessageType;
52 
53  long m_DataOffset;
54 
55  // reply/report options
56  long m_ReplyOptions;
57  string m_ReplyQueueManager;
58  string m_MessageFormat;
59 
60  bool m_UseSyncpoint;
61 
62  int m_QueueManagerOpenRefCount;
63 
64  void setOpenQueueOptions( const long& openOptions );
65  void setChannel();
66  TransportHelper::TRANSPORT_MESSAGE_TYPE ToTransportMessageType( long messageType );
67  unsigned int StripJMSHeader( ImqMessage& msg, string& msgFormat ) const;
68  unsigned int InjectJMSHeader( ImqMessage& msg );
69  long getLastDataOffset() const { return m_DataOffset; }
70  long ParseReplyOptions( const TransportReplyOptions& options);
71 
72  public :
73  WMqHelper();
74  ~WMqHelper();
75 
76  /*
77  Function: connect
78  Connects to the queue manager
79 
80  This function will not reconnect to the queue manager unless the connection was broken/closed or the queue manager name has changed.
81 
82  Parameters:
83 
84  queueManagerName - Queue manager name that the helper will connect to
85  channelDefinition - WMQ Channel definition ( when running as client ) which is of the form "channel-name/transport-type/connection-name(port)"
86  force - Force reconnect even if already connected. It will first close the existing connection.
87  */
88  void connect( const string& queueManagerName = "", const string& channelDefinition = "", bool force = false );
89  void connect( const string& queueManagerName, const string& channelDefinition, const string& keyRepository , const string& sslCypherSpec , const string& sslPeerName , bool force = false );
90 
91  void openQueue( const string& queueName = "" );
92  void openBackupQueue();
93  void openBackoutQueue( string queueName );
94 
95  void closeQueue();
96  void closeBackupQueue();
97  void closeBackoutQueue();
98 
99  void disconnect();
100 
104  long getOne( unsigned char* buffer, size_t maxSize, bool syncpoint = true );
105  long getOne( ManagedBuffer* buffer, bool syncpoint = true, bool keepJMSHeader = true );
109  long getOne( unsigned char* buffer, size_t maxSize, ImqGetMessageOptions& gmo );
110  long getOne( unsigned char* buffer, size_t maxSize, ImqGetMessageOptions& gmo, ImqMessage& msg );
111  long getOne( ManagedBuffer* buffer, ImqGetMessageOptions& gmo );
112  long getOne( ManagedBuffer* buffer, ImqGetMessageOptions& gmo, ImqMessage& msg, bool getForClean = false, bool keepJMSHeader = true );
113 
117  void putOne( unsigned char* buffer, size_t bufferSize, bool syncpoint = true );
121  void putOne( unsigned char* buffer, size_t bufferSize, ImqPutMessageOptions& pmo );
122  void putOne( unsigned char* buffer, size_t bufferSize, ImqPutMessageOptions& pmo, ImqMessage& msg );
123  void putOne( ManagedBuffer* buffer, ImqPutMessageOptions& pmo, ImqMessage& msg );
124 
125  void putToDeadLetterQueue( ImqMessage& msg );
126  void clearMessages();
127 
128  // syncpoint = true, messagetype = MQMT_REQUEST,
129  // setReplyToQueueManagerName, setReplyToQueueName, setReport
130 
134  long getGroupMessage( ManagedBuffer* groupMessageBuffer, const string& groupId, bool& isCleaningUp );
135  void putGroupMessage( ManagedBuffer* buffer,const string& batchId, long messageSequence, bool isLast );
136 
137  void putSAAmessage( const TransportReplyOptions& replyOptions, const string& replyQueue, ManagedBuffer* buffer, const string& batchId, long messageSequence, bool isLast );
138 
142  void putOneRequest( unsigned char* buffer, size_t bufferSize, const string& rtqName, const string& rtqmName, TransportReplyOptions& replyOptions );
143  void putOneReply( unsigned char* buffer, size_t bufferSize = 0, long feedback = 0, TRANSPORT_MESSAGE_TYPE replyType = TMT_REPLY );
147  void putOneReply( ManagedBuffer* buffer, long feedback );
148 
152  long peek( const string& queue = "", bool first = true );
153 
154  string getApplicationName() const { return m_ApplicationName; }
155  void setApplicationName( const string& applicationName );
156 
157  void setMessageFormat( const string& format );
158  string getLastMessageFormat() const;
159 
160  TRANSPORT_MESSAGE_TYPE getLastMessageType() { return ToTransportMessageType( m_MessageType ); }
161 
162  TransportReplyOptions getLastReplyOptions() const ;
163  string getLastReplyQueueManager() const { return m_ReplyQueueManager; }
164 
165  void clearSSLOptions();
166  bool commit();
167  bool rollback();
168 
169  long getQueueDepth( const string& queueName = "");
170  long getOpenOutputCount( string queueName = "" );
171  long getOpenInputCount( string queueName = "" );
172 
173  static void setSyncPointParticipation( ImqPutMessageOptions& pmo, unsigned char value );
174  static void setSyncPointParticipation( ImqGetMessageOptions& gmo, unsigned char value );
175 
176  static ImqBinary createBinary( void* value );
177 
178  // XA resource coordination
179  void beginXATransaction();
180  };
181 }
182 
183 #endif