DaoMqPump2/DaoMqPump2/Transport.cs

using System;
using System.Collections;
using System.Collections.Generic;
using System.IO;

using System.Diagnostics;

using IBM.WMQ;
using MySql.Data.MySqlClient;
using System.Globalization;

using DaoCommon;

namespace DaoMqPump2
{
    public class Transport
    {

        public static string SQL2MQ = "sql2mq";
        public static string MQ2SQL = "mq2sql";

        //private bool enabled;        

        TransportController controller;

        StatusData statusData = new StatusData();

        public Logfile logfile { get; private set; }

        public string name { get; private set; }
        public string direction { get; private set; }
        public string queueName { get; private set; }
        public string mq2sqlInsertQuery { get; private set; }
        public string sql2mqReadQuery { get; private set; }
        public string sql2mqUpdateQuery { get; private set; }

        //public bool lastrunOk { get; private set; }
        //public string lastErrorMessage { get; private set; }

        //public string lastOkTime { get; private set; }
        //public string lastErrorTime { get; private set; }
        //public string lastTransferTime { get; private set; }

        //public int counter { get; private set; }

        public StatusData TransportStatusData
        {
            get
            {
                return this.statusData;
            }
        }


        public bool Enabled
        {
            get {
                return statusData.transportEnabled;
            }
            set
            {
                statusData.transportEnabled = value;
                if (value == true)
                {
                    logfile.addSingleLogEntry("Transport enabled");
                }
                else
                {
                    logfile.addSingleLogEntry("Transport disabled");
                }
            }
        }


        public Transport(TransportController controller, string name, string direction, string queueName, string mq2sqlInsertQuery, string sql2mqReadQuery, string sql2mqUpdateQuery, bool enabled)
        {
            this.controller = controller;
            this.name = name;
            this.direction = direction;
            this.queueName = queueName;
            this.mq2sqlInsertQuery = mq2sqlInsertQuery;
            this.sql2mqReadQuery = sql2mqReadQuery;
            this.sql2mqUpdateQuery = sql2mqUpdateQuery;

            statusData.transportEnabled = enabled;


            statusData.lastrunOk = true;
            statusData.counter = 0;
            statusData.lastErrorMessage = statusData.lastOkTime = statusData.lastErrorTime = "";

            logfile = new Logfile(LogfileType.LogEvents,  name, controller.logDirectory);
            logfile.addSingleLogEntry("Starting ... ");
        }

        ~Transport()
        {
            logfile.addSingleLogEntry("Stopping ... ");
        }


        public void transportMessages()
        {
            if (statusData.transportEnabled == false)
                return;

            Console.WriteLine(name + " -> transportMessages() ");
            statusData.lastrunOk = true;

            int startCounter = statusData.counter;

            if (direction == SQL2MQ)
            {
                transportSql2Mq();
            }
            else
            {
                transportMq2Sql();
            }

            if (statusData.lastrunOk == true)
            {
                statusData.lastOkTime = Logfile.getNowString();

                if (statusData.counter != startCounter) 
                {
                    //Vi har transporteret beskeder - så gemmer vi lige transfer tidspunktet
                    statusData.lastTransferTime = Logfile.getNowString();
                }
            }
            else
            {
                logfile.addSingleLogEntry(statusData.lastErrorMessage);
                statusData.lastErrorTime = Logfile.getNowString();
            }
        }

        private void transportSql2Mq()
        {
            MQQueueManager mqMgr = null;
            MQQueue out_queue = null;

            string filename = Logfile.getLogFilename(LogfileType.LogTransactions, controller.logDirectory, name);
            using (StreamWriter translog = new StreamWriter(filename, true) )
            try
            {
                //MQ Options
                Hashtable connProps = MQHelper.getConnectionProperties(controller.mqHost, controller.mqChannel);
                int openOptions = MQC.MQOO_OUTPUT + MQC.MQOO_FAIL_IF_QUIESCING;

                //MySQL Options
                string mysqlString = buildMysqlConnString();

                //MQ objects i v6 implementerer ikke IDisposable og kan derfor ikke bruges med "using" statement
                mqMgr = new MQQueueManager(controller.mqQueueManager, connProps);//stage 1 connect to mq
                out_queue = mqMgr.AccessQueue(queueName, openOptions);
                using (MySqlConnection sqlReadConnection = new MySqlConnection(mysqlString)) //stage 2 connect to mysql
                using (MySqlConnection sqlWriteConnection = new MySqlConnection(mysqlString)) 
                {
                    sqlReadConnection.Open();
                    sqlWriteConnection.Open();

                    //stage 3 move messages
                    string readSql = "CALL " + sql2mqReadQuery + "()";
                    MySqlCommand readCmd = new MySqlCommand(readSql, sqlReadConnection);
                    MySqlDataReader dataReader = readCmd.ExecuteReader();
                    while (dataReader.Read())
                    {
                        int id = dataReader.GetInt32(0);
                        string msgString = dataReader.GetString(1);

                        MQPutMessageOptions pmo = new MQPutMessageOptions(); // accept the defaults,
                        // same as MQPMO_DEFAULT

                        MQMessage msg = new MQMessage();
                        msg.Format = MQC.MQFMT_STRING;
                        msg.CharacterSet = 1252;
                        msg.WriteString(msgString);

                        out_queue.Put(msg, pmo);

                        //now that the message has been put on queue mark it as such

                        string updateSql = "CALL " + sql2mqUpdateQuery + "(" + id + ")";
                        MySqlCommand updateCmd = new MySqlCommand(updateSql, sqlWriteConnection);
                        int numrows = updateCmd.ExecuteNonQuery();

                        translog.WriteLine(Logfile.getNowString() + " " + msgString);

                        if (numrows != 1)
                        {
                            break;
                        }
                        statusData.counter++;
                    }

                }
            }
            catch (Exception e)
            {
                statusData.lastrunOk = false;
                statusData.lastErrorMessage = name + ".transportMq2Sql error: " + e.GetType().FullName + " " + e.Message;
                Console.WriteLine(statusData.lastErrorMessage);
                Console.WriteLine(e.StackTrace);
                EventLog.WriteEntry("DaoMqPump2", statusData.lastErrorMessage, EventLogEntryType.Warning);
            }
            finally
            {
                MQHelper.closeQueueSafe(out_queue);
                MQHelper.closeQueueManagerSafe(mqMgr);


            }
        }

        private void transportMq2Sql()
        {
            int messageCount = 0;

            MQQueueManager mqMgr = null;
            MQQueue in_queue = null;
            string filename = Logfile.getLogFilename(LogfileType.LogTransactions, controller.logDirectory, name);
            using (StreamWriter translog = new StreamWriter(filename, true))
                try
                {
                    //MQ options
                    Hashtable connProps = MQHelper.getConnectionProperties(controller.mqHost, controller.mqChannel);
                    int openOptions = MQC.MQOO_INPUT_AS_Q_DEF + MQC.MQOO_FAIL_IF_QUIESCING;

                    //MySQL options
                    string mysqlString = buildMysqlConnString();

                    //MQ objects i v6 implementerer ikke IDisposable og kan derfor ikke bruges med "using" statement
                    mqMgr = new MQQueueManager(controller.mqQueueManager, connProps);//stage 1 connect to mq
                    in_queue = mqMgr.AccessQueue(queueName, openOptions);
                    using (MySqlConnection sqlConnection = new MySqlConnection(mysqlString)) //stage 2 connect to mysql
                    {

                        sqlConnection.Open();


                        //stage 3 move messages
                        bool isContinue = true;
                        while (isContinue)
                        {

                            MQMessage mqMsg = new MQMessage();
                            MQGetMessageOptions mqGetMsgOpts = new MQGetMessageOptions();

                            mqGetMsgOpts.Options = MQC.MQGMO_SYNCPOINT; //kræver en commit førend at den er fjernet fra køen

                            try
                            {
                                in_queue.Get(mqMsg, mqGetMsgOpts);
                                if (mqMsg.Format.CompareTo(MQC.MQFMT_STRING) == 0)
                                {
                                    string msgString = mqMsg.ReadString(mqMsg.MessageLength);
                                    //System.Console.WriteLine(msgString);


                                    // Hvis transaktionen starter med et ? er det ikke en gyldig transaktion
                                    // validér ligeledes at headeren er gyldig
                                    if ( Salt2Helper.validateSalt2Header(msgString) == false ) 
                                    {
                                        string discarded_filename = Logfile.getLogFilename(LogfileType.LogTransactions, controller.logDirectory, name);
                                        using (StreamWriter discardedlog = new StreamWriter(discarded_filename, true))
                                        {
                                            discardedlog.WriteLine(Logfile.getNowString() + " " + msgString);
                                        }
                                        mqMgr.Commit();//fjern den afviste transaktion fra køen
                                        statusData.discardedCounter++;
                                        continue; //gå frem til at tage næste transaktion fra køen
                                    }


                                    string sql = "CALL " + mq2sqlInsertQuery + "( '" + MySqlHelper.EscapeString(msgString) + "' )"; //opbygger en CALL somestoredprocedure('msgString'); sql streng

                                    MySqlCommand sqlcmd = new MySqlCommand(sql, sqlConnection);
                                    int numrows = sqlcmd.ExecuteNonQuery();

                                    if (numrows == 1)
                                    {
                                        translog.WriteLine(Logfile.getNowString() + " " + msgString);
                                        mqMgr.Commit();
                                        statusData.counter++;


                                        messageCount++;// increment per run message counter
                                        if (messageCount >= 10000) // if we have moved  10000 messages in this run - let it go and give the other transports a change
                                        {
                                            isContinue = false;
                                        }


                                    }
                                    else
                                    {
                                        mqMgr.Backout();
                                        isContinue = false;
                                    }

                                }
                                else
                                {
                                    System.Console.WriteLine("Non-text message");
                                }
                            }
                            catch (MQException mqe)
                            {
                                isContinue = false;

                                // report reason, if any
                                if (mqe.Reason == MQC.MQRC_NO_MSG_AVAILABLE)
                                {
                                    // special report for normal end
                                    System.Console.WriteLine("no more messages");
                                }
                                else
                                {
                                    // general report for other reasons
                                    System.Console.WriteLine("MQQueue::Get ended with " + mqe.Message); ;
                                    statusData.lastrunOk = false;
                                }

                            }


                        }

                    }

                }
                catch (Exception e)
                {
                    //Det vil være mest korrekt at Rollback/backout MQ Transaktionen her - for at være sikker på at Message'n fjernes fra køen
                    try
                    {
                        if (mqMgr != null)
                        {
                            mqMgr.Backout();
                        }
                    }
                    catch (Exception e2)
                    {
                        logfile.addSingleLogEntry("Error backing out from MQ Transaction: " + e2.Message);
                    }

                    statusData.lastrunOk = false;

                    statusData.lastErrorMessage = name + ".transportMq2Sql error: " + e.GetType().FullName + " " + e.Message;
                    Console.WriteLine(statusData.lastErrorMessage);
                    Console.WriteLine(e.StackTrace);
                    EventLog.WriteEntry("DaoMqPump2", statusData.lastErrorMessage, EventLogEntryType.Warning);
                }
                finally
                {

                    MQHelper.closeQueueSafe(in_queue);
                    MQHelper.closeQueueManagerSafe(mqMgr);

                }
        }

        private string buildMysqlConnString()
        {
            string connectionString = "";

            connectionString += "SERVER=" + controller.mysqlHost + ";";
            //connectionString += "DATABASE=" + controller.mysqlHost + ";";
            connectionString += "UID=" + controller.mysqlUser + ";";
            connectionString += "PASSWORD=" + controller.mysqlPassword + ";";
            connectionString += "Max Pool Size=20;";
            //connectionString += "ConnectionReset=true;";//ConnectionReset kræver muligvis at Default Database er angivet - det virker ihvertfald ikke uden

            return connectionString;
        }


    }
}
1	using System;
2	using System.Collections;
3	using System.Collections.Generic;
4	using System.IO;
5
6	using System.Diagnostics;
7
8	using IBM.WMQ;
9	using MySql.Data.MySqlClient;
10	using System.Globalization;
11
12	using DaoCommon;
13
14	namespace DaoMqPump2
15	{
16	public class Transport
17	{
18
19	public static string SQL2MQ = "sql2mq";
20	public static string MQ2SQL = "mq2sql";
21
22	//private bool enabled;
23
24	TransportController controller;
25
26	StatusData statusData = new StatusData();
27
28	public Logfile logfile { get; private set; }
29
30	public string name { get; private set; }
31	public string direction { get; private set; }
32	public string queueName { get; private set; }
33	public string mq2sqlInsertQuery { get; private set; }
34	public string sql2mqReadQuery { get; private set; }
35	public string sql2mqUpdateQuery { get; private set; }
36
37	//public bool lastrunOk { get; private set; }
38	//public string lastErrorMessage { get; private set; }
39
40	//public string lastOkTime { get; private set; }
41	//public string lastErrorTime { get; private set; }
42	//public string lastTransferTime { get; private set; }
43
44	//public int counter { get; private set; }
45
46	public StatusData TransportStatusData
47	{
48	get
49	{
50	return this.statusData;
51	}
52	}
53
54
55	public bool Enabled
56	{
57	get {
58	return statusData.transportEnabled;
59	}
60	set
61	{
62	statusData.transportEnabled = value;
63	if (value == true)
64	{
65	logfile.addSingleLogEntry("Transport enabled");
66	}
67	else
68	{
69	logfile.addSingleLogEntry("Transport disabled");
70	}
71	}
72	}
73
74
75
76
77
78	public Transport(TransportController controller, string name, string direction, string queueName, string mq2sqlInsertQuery, string sql2mqReadQuery, string sql2mqUpdateQuery, bool enabled)
79	{
80	this.controller = controller;
81	this.name = name;
82	this.direction = direction;
83	this.queueName = queueName;
84	this.mq2sqlInsertQuery = mq2sqlInsertQuery;
85	this.sql2mqReadQuery = sql2mqReadQuery;
86	this.sql2mqUpdateQuery = sql2mqUpdateQuery;
87
88	statusData.transportEnabled = enabled;
89
90
91	statusData.lastrunOk = true;
92	statusData.counter = 0;
93	statusData.lastErrorMessage = statusData.lastOkTime = statusData.lastErrorTime = "";
94
95	logfile = new Logfile(LogfileType.LogEvents, name, controller.logDirectory);
96	logfile.addSingleLogEntry("Starting ... ");
97	}
98
99	~Transport()
100	{
101	logfile.addSingleLogEntry("Stopping ... ");
102	}
103
104
105	public void transportMessages()
106	{
107	if (statusData.transportEnabled == false)
108	return;
109
110	Console.WriteLine(name + " -> transportMessages() ");
111	statusData.lastrunOk = true;
112
113	int startCounter = statusData.counter;
114
115	if (direction == SQL2MQ)
116	{
117	transportSql2Mq();
118	}
119	else
120	{
121	transportMq2Sql();
122	}
123
124	if (statusData.lastrunOk == true)
125	{
126	statusData.lastOkTime = Logfile.getNowString();
127
128	if (statusData.counter != startCounter)
129	{
130	//Vi har transporteret beskeder - så gemmer vi lige transfer tidspunktet
131	statusData.lastTransferTime = Logfile.getNowString();
132	}
133	}
134	else
135	{
136	logfile.addSingleLogEntry(statusData.lastErrorMessage);
137	statusData.lastErrorTime = Logfile.getNowString();
138	}
139	}
140
141	private void transportSql2Mq()
142	{
143	MQQueueManager mqMgr = null;
144	MQQueue out_queue = null;
145
146	string filename = Logfile.getLogFilename(LogfileType.LogTransactions, controller.logDirectory, name);
147	using (StreamWriter translog = new StreamWriter(filename, true) )
148	try
149	{
150	//MQ Options
151	Hashtable connProps = MQHelper.getConnectionProperties(controller.mqHost, controller.mqChannel);
152	int openOptions = MQC.MQOO_OUTPUT + MQC.MQOO_FAIL_IF_QUIESCING;
153
154	//MySQL Options
155	string mysqlString = buildMysqlConnString();
156
157	//MQ objects i v6 implementerer ikke IDisposable og kan derfor ikke bruges med "using" statement
158	mqMgr = new MQQueueManager(controller.mqQueueManager, connProps);//stage 1 connect to mq
159	out_queue = mqMgr.AccessQueue(queueName, openOptions);
160	using (MySqlConnection sqlReadConnection = new MySqlConnection(mysqlString)) //stage 2 connect to mysql
161	using (MySqlConnection sqlWriteConnection = new MySqlConnection(mysqlString))
162	{
163	sqlReadConnection.Open();
164	sqlWriteConnection.Open();
165
166	//stage 3 move messages
167	string readSql = "CALL " + sql2mqReadQuery + "()";
168	MySqlCommand readCmd = new MySqlCommand(readSql, sqlReadConnection);
169	MySqlDataReader dataReader = readCmd.ExecuteReader();
170	while (dataReader.Read())
171	{
172	int id = dataReader.GetInt32(0);
173	string msgString = dataReader.GetString(1);
174
175	MQPutMessageOptions pmo = new MQPutMessageOptions(); // accept the defaults,
176	// same as MQPMO_DEFAULT
177
178	MQMessage msg = new MQMessage();
179	msg.Format = MQC.MQFMT_STRING;
180	msg.CharacterSet = 1252;
181	msg.WriteString(msgString);
182
183	out_queue.Put(msg, pmo);
184
185	//now that the message has been put on queue mark it as such
186
187	string updateSql = "CALL " + sql2mqUpdateQuery + "(" + id + ")";
188	MySqlCommand updateCmd = new MySqlCommand(updateSql, sqlWriteConnection);
189	int numrows = updateCmd.ExecuteNonQuery();
190
191	translog.WriteLine(Logfile.getNowString() + " " + msgString);
192
193	if (numrows != 1)
194	{
195	break;
196	}
197	statusData.counter++;
198	}
199
200	}
201	}
202	catch (Exception e)
203	{
204	statusData.lastrunOk = false;
205	statusData.lastErrorMessage = name + ".transportMq2Sql error: " + e.GetType().FullName + " " + e.Message;
206	Console.WriteLine(statusData.lastErrorMessage);
207	Console.WriteLine(e.StackTrace);
208	EventLog.WriteEntry("DaoMqPump2", statusData.lastErrorMessage, EventLogEntryType.Warning);
209	}
210	finally
211	{
212	MQHelper.closeQueueSafe(out_queue);
213	MQHelper.closeQueueManagerSafe(mqMgr);
214
215
216	}
217	}
218
219	private void transportMq2Sql()
220	{
221	int messageCount = 0;
222
223	MQQueueManager mqMgr = null;
224	MQQueue in_queue = null;
225	string filename = Logfile.getLogFilename(LogfileType.LogTransactions, controller.logDirectory, name);
226	using (StreamWriter translog = new StreamWriter(filename, true))
227	try
228	{
229	//MQ options
230	Hashtable connProps = MQHelper.getConnectionProperties(controller.mqHost, controller.mqChannel);
231	int openOptions = MQC.MQOO_INPUT_AS_Q_DEF + MQC.MQOO_FAIL_IF_QUIESCING;
232
233	//MySQL options
234	string mysqlString = buildMysqlConnString();
235
236	//MQ objects i v6 implementerer ikke IDisposable og kan derfor ikke bruges med "using" statement
237	mqMgr = new MQQueueManager(controller.mqQueueManager, connProps);//stage 1 connect to mq
238	in_queue = mqMgr.AccessQueue(queueName, openOptions);
239	using (MySqlConnection sqlConnection = new MySqlConnection(mysqlString)) //stage 2 connect to mysql
240	{
241
242	sqlConnection.Open();
243
244
245	//stage 3 move messages
246	bool isContinue = true;
247	while (isContinue)
248	{
249
250	MQMessage mqMsg = new MQMessage();
251	MQGetMessageOptions mqGetMsgOpts = new MQGetMessageOptions();
252
253	mqGetMsgOpts.Options = MQC.MQGMO_SYNCPOINT; //kræver en commit førend at den er fjernet fra køen
254
255	try
256	{
257	in_queue.Get(mqMsg, mqGetMsgOpts);
258	if (mqMsg.Format.CompareTo(MQC.MQFMT_STRING) == 0)
259	{
260	string msgString = mqMsg.ReadString(mqMsg.MessageLength);
261	//System.Console.WriteLine(msgString);
262
263
264	// Hvis transaktionen starter med et ? er det ikke en gyldig transaktion
265	// validér ligeledes at headeren er gyldig
266	if ( Salt2Helper.validateSalt2Header(msgString) == false )
267	{
268	string discarded_filename = Logfile.getLogFilename(LogfileType.LogTransactions, controller.logDirectory, name);
269	using (StreamWriter discardedlog = new StreamWriter(discarded_filename, true))
270	{
271	discardedlog.WriteLine(Logfile.getNowString() + " " + msgString);
272	}
273	mqMgr.Commit();//fjern den afviste transaktion fra køen
274	statusData.discardedCounter++;
275	continue; //gå frem til at tage næste transaktion fra køen
276	}
277
278
279	string sql = "CALL " + mq2sqlInsertQuery + "( '" + MySqlHelper.EscapeString(msgString) + "' )"; //opbygger en CALL somestoredprocedure('msgString'); sql streng
280
281	MySqlCommand sqlcmd = new MySqlCommand(sql, sqlConnection);
282	int numrows = sqlcmd.ExecuteNonQuery();
283
284	if (numrows == 1)
285	{
286	translog.WriteLine(Logfile.getNowString() + " " + msgString);
287	mqMgr.Commit();
288	statusData.counter++;
289
290
291	messageCount++;// increment per run message counter
292	if (messageCount >= 10000) // if we have moved 10000 messages in this run - let it go and give the other transports a change
293	{
294	isContinue = false;
295	}
296
297
298
299	}
300	else
301	{
302	mqMgr.Backout();
303	isContinue = false;
304	}
305
306	}
307	else
308	{
309	System.Console.WriteLine("Non-text message");
310	}
311	}
312	catch (MQException mqe)
313	{
314	isContinue = false;
315
316	// report reason, if any
317	if (mqe.Reason == MQC.MQRC_NO_MSG_AVAILABLE)
318	{
319	// special report for normal end
320	System.Console.WriteLine("no more messages");
321	}
322	else
323	{
324	// general report for other reasons
325	System.Console.WriteLine("MQQueue::Get ended with " + mqe.Message); ;
326	statusData.lastrunOk = false;
327	}
328
329	}
330
331
332	}
333
334	}
335
336	}
337	catch (Exception e)
338	{
339	//Det vil være mest korrekt at Rollback/backout MQ Transaktionen her - for at være sikker på at Message'n fjernes fra køen
340	try
341	{
342	if (mqMgr != null)
343	{
344	mqMgr.Backout();
345	}
346	}
347	catch (Exception e2)
348	{
349	logfile.addSingleLogEntry("Error backing out from MQ Transaction: " + e2.Message);
350	}
351
352	statusData.lastrunOk = false;
353
354	statusData.lastErrorMessage = name + ".transportMq2Sql error: " + e.GetType().FullName + " " + e.Message;
355	Console.WriteLine(statusData.lastErrorMessage);
356	Console.WriteLine(e.StackTrace);
357	EventLog.WriteEntry("DaoMqPump2", statusData.lastErrorMessage, EventLogEntryType.Warning);
358	}
359	finally
360	{
361
362	MQHelper.closeQueueSafe(in_queue);
363	MQHelper.closeQueueManagerSafe(mqMgr);
364
365	}
366	}
367
368	private string buildMysqlConnString()
369	{
370	string connectionString = "";
371
372	connectionString += "SERVER=" + controller.mysqlHost + ";";
373	//connectionString += "DATABASE=" + controller.mysqlHost + ";";
374	connectionString += "UID=" + controller.mysqlUser + ";";
375	connectionString += "PASSWORD=" + controller.mysqlPassword + ";";
376	connectionString += "Max Pool Size=20;";
377	//connectionString += "ConnectionReset=true;";//ConnectionReset kræver muligvis at Default Database er angivet - det virker ihvertfald ikke uden
378
379	return connectionString;
380	}
381
382
383	}
384	}