Lieven Cardoen

 
using System;
using System.Messaging;
using System.Transactions;
 
namespace MSMQ
{
/// <summary> @Author: Lieven Cardoen
/// This little program tests some of the functionalities of MSMQ.
/// Sending messages synchronously and asynchronously are covered.
/// Transactions are also covered, internally by MSMQ and externally by DTC.
/// A next step would be to use some threads, but this was just some minor R&amp;D
/// that I had to do for our product Edumatic Exam.
/// Documentation:
/// http://msdn.microsoft.com/en-us/library/ms978425.aspx
/// http://msdn.microsoft.com/en-us/library/ms978430.aspx
/// </summary>
 
public class MsmqTryOut
{
protected MessageQueue MessageQueue;
private const string NonTransMessageQueueName = ".\\private$\\finishedSessionsNonTrans";
private const string IntTransMessageQueueName = ".\\private$\\finishedSessionsIntTrans";
private const string ExtTransMessageQueueName = ".\\private$\\finishedSessionsExtTrans";
private const string AckQueueName = ".\\private$\\Ack";
private const string OrdersQueueName = ".\\private$\\Orders";
 
/// <summary> 
/// Main entry point
/// </summary>
 
static void Main()
{
var program = new MsmqTryOut();
 
//NON TRANSACTIONAL QUEUE --&gt; SYNC AND ASYNC MESSAGES
program.SetUpQueue(NonTransMessageQueueName, false);
program.SendMessage(String.Format("Message sent at {0}", DateTime.Now)
, "Non-Transactional sync message (first message)", MessageQueueTransactionType.None);
program.SendMessage(String.Format("Message sent at {0}", DateTime.Now)
, "Non-Transactional async message (second message)", MessageQueueTransactionType.None);
/**
* Program would hang until a message is delivered (if no messages are in the queue
* and none are added anymore, program would hang forever).
* Here first message will be received
*/
program.ReceiveSyncMessage(MessageQueueTransactionType.None);
/**
* Program wouldn't hang forever. Handler would be triggered when new message in queue.
* Here second message will be received. If second message is received, the queue will be deleted
* in the async handler.
*/
program.ReceiveAsyncMessage();
 
//INTERNAL TRANSACTIONS (MSMQ)
program.SetUpQueue(IntTransMessageQueueName, true);
//Start the internal msmq transaction
var msgTx = new MessageQueueTransaction();
msgTx.Begin();
program.SendMessage(String.Format("Message sent at {0}", DateTime.Now), "Internal MSMQ Transactional sync Message", msgTx);
program.SendMessage(String.Format("Message sent at {0}", DateTime.Now), "Internal MSMQ Transactional sync Message", msgTx);
msgTx.Commit();
 
//Start new transaction
msgTx.Begin();
program.ReceiveSyncMessage(msgTx);
program.ReceiveSyncMessage(msgTx);
Console.WriteLine("----------------------------------------------------------------------------------");
Console.WriteLine("Dispose transaction");
msgTx.Dispose();
//Two messages should still be available because previous receives were disposed
program.ReceiveSyncMessage(msgTx);
program.ReceiveAsyncMessage(); //queue will be deleted in the async handler
 
//TRANSACTIONAL QUEUE WITH EXTERNAL TRANSACTIONS (DTC)
program.SetUpQueue(ExtTransMessageQueueName, true);
//If it's a transactional queue, then Single needs to be used if the message
//isn't send in a transaction
program.SendMessage(String.Format("Message sent at {0}", DateTime.Now), "DTC Transactional sync Message", MessageQueueTransactionType.Single);
program.SendMessage(String.Format("Message sent at {0}", DateTime.Now), "DTC Transactional sync Message", MessageQueueTransactionType.Single);
using (var scope = new TransactionScope())
{
//If a transaction is used and the queue is externally transactional (DTC),
//then Automatic needs to be used
program.ReceiveSyncMessage(MessageQueueTransactionType.Automatic);
program.ReceiveSyncMessage(MessageQueueTransactionType.Automatic);
//ReceiveAsync(); //!!! Apparently this won't be disposed. + error is thrown when async
//handler is triggered because the MessageQueue is already deleted.
Console.WriteLine("----------------------------------------------------------------------------------");
Console.WriteLine("Dispose transaction");
scope.Dispose();
}
//Two messages should still be available
program.ReceiveSyncMessage(MessageQueueTransactionType.Single);
program.ReceiveAsyncMessage(); //queue will be deleted in the async handler
 
//ACKNOWLEDGEMENTS
ReceiveAcknowledgementMessages();
DeleteQueue(OrdersQueueName);
DeleteQueue(AckQueueName);
 
//Pause the program
Console.ReadKey();
}
 
/// <summary> 
/// Tests acknowledgements
/// </summary>
 
private static void ReceiveAcknowledgementMessages()
{
Console.WriteLine("----------------------------------------------------------------------------------");
if (!MessageQueue.Exists(OrdersQueueName)) MessageQueue.Create(OrdersQueueName);
if(!MessageQueue.Exists(AckQueueName)) MessageQueue.Create(AckQueueName);
//Send Message
var msgQ = new MessageQueue(OrdersQueueName)
{
DefaultPropertiesToSend =
{
AcknowledgeType = AcknowledgeTypes.FullReachQueue |
AcknowledgeTypes.FullReceive,
AdministrationQueue = new MessageQueue(AckQueueName)
}
};
msgQ.Send("Sample");
 
//Receive Message
var msg = msgQ.Receive(new TimeSpan(0), MessageQueueTransactionType.Single);
if (msg != null) Console.WriteLine("Message Received {0}", msg.Id);
 
//Receive Acknowledgement Messages
var adminQ = new MessageQueue(AckQueueName)
{
MessageReadPropertyFilter = {CorrelationId = true}
};
do
{
try
{
msg = adminQ.Receive(new TimeSpan(0,0,5));
if (msg != null &amp;&amp; msg.MessageType == MessageType.Acknowledgment)
switch(msg.Acknowledgment)
{
case Acknowledgment.ReachQueue:
Console.WriteLine("Message Reached Queue {0}", msg.CorrelationId);
break;
case Acknowledgment.Receive:
Console.WriteLine("Message Received {0}", msg.CorrelationId);
break;
}
}
catch(Exception e)
{
Console.WriteLine(e.Message);
msg = null;
}
}
while (msg != null);
}
 
/// <summary> 
/// Get a message out of the queue synchronously
/// </summary
///Type of transaction to be used 
 
private void ReceiveSyncMessage(MessageQueueTransactionType messageQueueTransactionType)
{
Console.WriteLine("----------------------------------------------------------------------------------");
var message = MessageQueue.Receive(messageQueueTransactionType);
if (message != null)
{
message.Formatter = new XmlMessageFormatter(new[] {typeof (String)});
var text = message.Body as String;
var label = message.Label;
Console.WriteLine("Received sync message (type:" + messageQueueTransactionType + ")");
Console.WriteLine("Text:" + text);
Console.WriteLine("Label:" + label);
}
else
{
Console.WriteLine("ERROR: Message was null");
}
}
 
/// <summary> 
/// Get a message out of the queue synchronously
/// </summary>
///The MessageQueueTransaction to be used 
 
private void ReceiveSyncMessage(MessageQueueTransaction mgsTx)
{
Console.WriteLine("----------------------------------------------------------------------------------");
var message = MessageQueue.Receive(mgsTx);
if (message != null)
{
//We'll use the XmlMessageFormatter, but you could also use the binary formatter
message.Formatter = new XmlMessageFormatter(new[] { typeof(String) });
var text = message.Body as String;
var label = message.Label;
Console.WriteLine("Received sync message (type:" + mgsTx + ")");
Console.WriteLine("Text:" + text);
Console.WriteLine("Label:" + label);
}
else
{
Console.WriteLine("ERROR: Message was null");
}
}
 
/// <summary> 
/// Get a message out of the queue asynchronously.
/// If messages are very large and network is slow for instance, then it
/// take some time to get the message. If async is used, program will not
/// hang.
/// </summary>
 
private void ReceiveAsyncMessage()
{
MessageQueue.ReceiveCompleted += ReceiveCompletedHandler;
MessageQueue.BeginReceive();
}
 
 
/// <summary> 
/// Asynchronous message handler
/// </summary>
///The MessageQueue 
///Arguments (message can be accessed through the arguments 
 
private static void ReceiveCompletedHandler(Object sender, ReceiveCompletedEventArgs args)
{
Console.WriteLine("----------------------------------------------------------------------------------");
var message = args.Message;
//We'll use the XmlMessageFormatter, but you could also use the binary formatter
message.Formatter = new XmlMessageFormatter(new[] { typeof(String) });
var text = message.Body as String;
var label = message.Label;
Console.WriteLine("Received async message");
Console.WriteLine("Can be received after queue is deleted.");
Console.WriteLine("Text:" + text);
Console.WriteLine("Label:" + label);
var messageQeueue = sender as MessageQueue;
 
//We delete the queue when the async message is received
//This is weird, but it makes things simple
if(messageQeueue != null) DeleteQueue(messageQeueue.Path);
else Console.WriteLine("ERROR: messageQueue was null");
}
 
 
/// <summary> 
/// Sets up the queue. If queue doesn't exist, it is created.
/// </summary>
///Name of the queue 
///Boolean whether queue should be transactional or not 
 
private void SetUpQueue(string messageQueueName, bool transactional)
{
Console.WriteLine("----------------------------------------------------------------------------------");
Console.WriteLine("SetUpQueue(" + messageQueueName + Environment.NewLine + ", "
+ (transactional ? "Transactional" : "Non-Transactional") + ")");
if (!MessageQueue.Exists(messageQueueName))
{
try
{
MessageQueue = MessageQueue.Create(messageQueueName, transactional);
}
catch (Exception createException)
{
//Error could occur creating queue if the code does
//not have sufficient permissions to create queues.
throw new Exception("Error Creating Queue", createException);
}
}
else
{
try
{
//Warning: If existing queue has a transactional property that
//is not equal to the transaction argument of the funtion
//the queue should actually be recreated.
MessageQueue = new MessageQueue(messageQueueName);
}
catch(Exception getException)
{
throw new Exception("Error Getting Queue", getException);
}
}
//Gets or sets a value that indicates whether the message is guaranteed
//to be delivered in the event of a computer failure or network problem.
MessageQueue.DefaultPropertiesToSend.Recoverable = true;
}
 
/// <summary> 
/// Deletes a queue
/// </summary>
///Name of the queue 
 
private static void DeleteQueue(string messageQueueName)
{
Console.WriteLine("----------------------------------------------------------------------------------");
Console.WriteLine("DeleteQueue(" + messageQueueName + ")");
try
{
MessageQueue.Delete(messageQueueName);
}
catch (Exception e)
{
throw new Exception("Deletion Failed", e);
}
}
 
/// <summary> 
/// Sends a message to the queue
/// </summary>
///Text message 
///Label of the message 
///Type of transaction 
private void SendMessage(string textMessage, string label, MessageQueueTransactionType type)
{
Console.WriteLine("----------------------------------------------------------------------------------");
Console.WriteLine("SendMessage(" + textMessage + Environment.NewLine + ", " + label + ", " + type);
try
{
var message = new Message {Body = textMessage, Recoverable = true};
MessageQueue.Send(message, label, type);
}
catch (Exception e)
{
throw new Exception("Error sending message to Queueu", e);
}
}
 
/// <summary> 
/// Sends a message to the queue
/// </summary>
///Text message 
///Label of the message 
///The MessageQueueTransaction to be used 
 
private void SendMessage(string textMessage, string label, MessageQueueTransaction msgTx)
{
Console.WriteLine("----------------------------------------------------------------------------------");
Console.WriteLine("SendMessage(" + textMessage + Environment.NewLine + ", " + label + ", " + msgTx);
try
{
var message = new Message { Body = textMessage, Recoverable = true };
MessageQueue.Send(message, label, msgTx);
}
catch (Exception e)
{
throw new Exception("Error sending message to Queueu", e);
}
}
}
}