ASYNCHRONOUS MESSAGING THROUGH MSMQ in C#
1. OVERVIEW
Message queues provide an asynchronous communications protocol. The sender and receiver of the message do not need to interact with the message queue at the same time. Messages placed onto the queue are stored until the recipient retrieves them. These message queuing systems typically provide enhanced resilience functionality to ensure that messages do not get "lost" in the event of a system failure. Microsoft Message Queuing or MSMQ is a Message Queue implementation developed by Microsoft and deployed in its Windows Server operating systems since Windows NT 4 and Windows 95.
MSMQ is essentially a messaging protocol that allows applications running on disparate servers to communicate in a failsafe manner. MSMQ enables communication across heterogeneous networks and between computers which may not always be connected. MSMQ is responsible for reliably delivering messages between applications inside and outside the enterprise. MSMQ ensures reliable delivery by placing messages that fail to reach their intended destination in a queue and then resending them once the destination is reachable. It also supports security and priority based messaging. Dead letter queues can be created for looking at messages which timed out or failed for other reasons. MSMQ also supports transactions. It permits multiple operations on multiple queues, with all of the operations wrapped in a single transaction, thus ensuring that either all or none of the operations will take effect.
2. A DATA MIGRATION SYSTEM INCORPORATING MSMQ AND HTTPLISTENER.
This case study deals with a data migration system which migrates data between application developed in java and dotnet. It a system to synchronise the data between the two systems. This example deals with a RoomType object (user business) created in 'Java' system to be migrated to the 'Dotnet' system. Once a room type is modified in one system, a message in standard XML format is sent to the other system. This case study deals with the processing of messages once it reaches the application developed in Dotnet.
The basic architecture of the data migration system is as shown below.Fig1 : Block Diagram of MSMQ Data Migration System
- Incoming message from system1(java).
- Acknowledgment for the message is send back.
- XML message is send to the parser.
- Parsed message is converted to entity(RoomType Entity in this scenario) and is send to MSMQ.
- MSMQ Listener keep on watching the queue.
- Once a message reaches MSMQ, it is being sent to the MSMQ Listener.
- Listener invokes methods in the Business layer to insert the RoomType entity into database.
- Response is being sent to listener from Business Layer.
3. TECHINAL OVERVIEW
The main classes that are used in this sample are
- System.Net.HttpListener
- System.Xml.XMLReader
- System.Messaging.MessageQueue
3.1. HTTP Listener
Using the HttpListener class, you can create a simple HTTP protocol listener that responds to HTTP requests. The listener is active for the lifetime of the HttpListener object .
To use HttpListener, create a new instance of the class using the HttpListener constructor and use the Prefixes property to gain access to the collection that holds the strings that specify which Uniform Resource Identifier (URI) prefixes the HttpListener should process.
A URI prefix string is composed of a scheme (http or https), a host, an optional port, and an optional path. An example of a complete prefix string is “http://localhost:8080/DataMigration/”. Prefixes must end in a forward slash (“/”). When a port is specified, the host element can be replaced with “*” to indicate that the HttpListener accepts requests sent to the port if the requested URI does not match any other prefix. Similarly, to specify that the HttpListener accepts all requests sent to a port, replace the host element with the “+” character, “https://+:8080”.
To begin listening for requests from clients, add the URI prefixes to the collection and call the Start method. HttpListener offers both synchronous and asynchronous models for processing client requests. Requests and their associated responses are accessed using the HttpListenerContext object returned by the GetContext method or its asynchronous counterparts, the BeginGetContext and EndGetContext methods.
In either model, incoming requests are accessed using the HttpListenerContext.Request property and are represented by HttpListenerRequest objects. Similarly, responses are accessed using the HttpListenerContext.Response property and are represented by HttpListenerResponse objects.
The following code example demonstrates using a HttpListener which listens at port 8080.
HttpListener listener = new HttpListener();
listener.Prefixes.Add("http://*:8080/");
listener.Start();
Console.WriteLine("Listening...");
while(true)//For continuous polling of http requests
{
// Note: The GetContext method blocks while waiting for a request.
HttpListenerContext context = listener.GetContext();
HttpListenerRequest request = context.Request;
Stream inputStream = request.InputStream;
/*******************
* XMLParser.ParseXML method which is explained in section 2.1 is invoked here. Data read from the Inputstream is the input to the ParseXML method. ParseXML method returns a RoomType entity. The RoomType entity is then pushed to the MSMQ.
********************/
// Obtain a response object.
HttpListenerResponse response = context.Response;
// Construct a response.
string responseString = " Response Message";
byte[] buffer = System.Text.Encoding.UTF8.GetBytes(responseString);
// Get a response stream and write the response to it.
response.ContentLength64 = buffer.Length;
System.IO.Stream output = response.OutputStream;
output.Write(buffer,0,buffer.Length);
output.Close();
}
listener.Stop();
Code Snippet 1: Http LISTENER
3.2. XML Parser
This section deals with converting the XML message received by the HTTP Listener into a Entity, say RoomType Entity. XMLReader and XMLTextReader classes are used to navigate through each of the nodes in the XML. The text and the attribute value read form the XML is assigned to the RoomType entity.
The logic behind the parsing is that each of the nodes are converted into a full element name.
ie. Consider the following XML:
Then each of the nodes become :
roomtypes/roomtype/@name, value = DELUXE
roomtypes/roomtype/@maxOccupancy , value=5
roomtypes/roomtype/@minOccupancy, value = 3
These sets of full element name and values are passed into a ParseMessage method which assigns the values to the entity as shown below(Code Snippet 4). The below code snippet converts the incoming request into a XMLReader object
XmlTextReader textReader = new XmlTextReader(new MemoryStream (System.Text.UTF8Encoding.UTF8.GetBytes(request)));
XMLReaderSettings settings = new XMLReaderSettings();
settings.ValidationType = ValidationType.None;
XMLReader reader= XMLReader.Create(textReader,settings);
Code Snippet 2: Converting incoming request to XmlReader object
while(reader.Read())
{
switch(reader.NodeType)
{
case XmlNodeType.Element:
/*
Business to format the node
*/
ParseMessage(formattedNode, value);
break;
case XmlNodeType.Text:
ParseMessage(formattedNode,value);
break;
}
}
Code Snippet 3: Navigating through Nodes in the XMLReader
public void ParseMessage(String elementname, string value)
{
switch( elementname )
{
case roomTypes/roomtype/@name:
roomType.Name= value;
break;
case roomTypes/roomtype/@maxoccupancy:
roomType.MaxOccupancy= value;
break;
.
.
}
}
Code Snippet 4: Assigning values into Entity(e.g: roomType Entity)
3.3. Queue Handler : Programming Message Queuing
Message Queuing 3.0 is a part of Windows Server 2003 and Windows XP, Windows 2000 comes with the Message Queuing 2.0, which did not have support for the HTTP protocol and multicast messages. You can install a message queue in Windows XP using Add or Remove Programs, a separate section within windows components where Message Queuing options can be selected. Within the message Queuing options, various components can be selected (refer Fig 2 and Fig 3).
Fig 2: Installing Message Queue
Fig 3: Message Queue Components
3.3.1. Creating a Message Queue
In C#, we can create Message Queues programmatically using the Create() method of MessageQueue class. With Create() method, the path of the new queue must be passed. The path consists of the hostname where the queue is located and the name of the queue
using (MessageQueue queue = MessageQueue.Create(@". \myqueue"))
{
queue.Label = "RoomType";
Console.WriteLine("Queue Created:");
}
Code Snippet 5: Creating a Queue
3.3.2. Sending a Message
By using the Send method of the MessageQueue class, you can send the message to the queue. The object passed as an argument of the Send() Method is serialized queue.
The message received by the HttpListener(3.1) is parsed by the XMLParser(3.2). The parsed message is converted to the RoomType Entity. Room Type entity is the input to the Send method.
try
{
if (!MessageQueue.Exists(@".\Private$\FirstQueue"))
MessageQueue.Create(@".\Private$\FirstQueue");
MessageQueue queue = new MessageQueue(@".\Private$FirstQueue");
queue.Formatter= new System.Messaging.XmlMessageFormatter(new string[]{"RoomType”});
queue.Send("RoomType ", objRoomType);
}
catch (MessageQueueException ex)
{
Console.WriteLine(ex.Message);
}
Code Snippet 6: Sending entity to queue
You can view the message in the Component Management admin tool as shown in figure 4. By opening the message and selecting the body tab of the dialog, you can see the message was formatted using XML.
Fig 4: Viewing MessageQueue in the Component Management admin tool
3.3.3. Receiving a Message : MSMQ Listener
This is the code that is to be used in the MSMQ Listener page load. This would start the MSMQ listener. BeginReceive, and EndReceive(IAsyncResult) methods provide ways to asynchronously read messages from the queue.
queue.Formatter = new XmlMessageFormatter(new Type[]{typeof(RoomType)});
queue.ReceiveCompleted += new ReceiveCompletedEventHandler(MessageArrived);
queue.BeginReceive();
Code Snippet 7: Starting a MSMQ Listener
public static void MessageArrived(object source,ReceiveCompletedEventArgs e)
{
MessageQueue queue = (MessageQueue)source;
Message message = queue.EndReceive(e.AsyncResult);
if(message.Label.Equals(“RoomType”)
{
RoomType objRoomType = (RoomType)message.Body;
/*
Business to process the object received from Queue
*/
}
queue.BeginReceive();
}
Code Snippet 8: Event handler for MSMQ Listener
