This is a simple example of how to use messaging, implemented in .NET [SysMsg] and C#. It shows how to implement Request-Reply, where a requestor application sends a request, a replier application receives the request and returns a reply, and the requestor receives the reply. It also shows how an invalid message will be rerouted to a special channel.
This example was developed using the Microsoft .NET Framework SDK and run on a Windows XP computer with MSMQ [MSMQ01] installed.
This example consists of two main classes:
The Requestor and the Replier will each run as a separate .NET program, which is what makes the communication distributed.
This example assumes that the messaging system has these three queues defined:
Here's how the example works. When the Requestor is started in a command-line window, it starts and prints output like this:
Sent request Time: 09:11:09.165342 Message ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808149円 Correl. ID: Reply to: .\private$\ReplyQueue Contents: Hello world.
What this shows is that the Requestor has sent a request message. Notice that this works even though the Replier isn't even running and therefore cannot receive the request.
When the Replier is started in another command-line window, it starts and prints output like this:
Received request Time: 09:11:09.375644 Message ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808149円 Correl. ID: <n/a> Reply to: FORMATNAME:DIRECT=OS:XYZ123\private$\ReplyQueue Contents: Hello world. Sent reply Time: 09:11:09.956480 Message ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808150円 Correl. ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808149円 Reply to: <n/a> Contents: Hello world.
This shows that the Replier received the request message and sent a reply message.
There are several items in this output that are interesting to notice. First, notice the request send and received timestamps; the request was received after it was sent (210302 ms later). Second, notice that the message ID is the same in both cases, because it's the same message. Third, notice that the contents, "Hello world," are the same, which is very good because this is the data being transmitted and it has got to be the same on both sides. (The request in this example is pretty lame. It is basically a Document Message; a real request would usually be a Command Message.) Forth, the queue named "jms/ReplyQueue" has been specified in the request message as the destination for the reply message (an example of the Return Address pattern).
Next, let's compare the output from receiving the request to that for sending the reply. First, notice the reply was not sent until after the request was received (580836 ms after). Second, the message ID for the reply is different from that for the request; this is because the request and reply messages are different, separate messages. Third, the contents of the request have been extracted and added to the reply. Forth, the reply-to destination is unspecified because no reply is expected (the reply does not use the Return Address pattern). Fifth, the reply's correlation ID is the same as the request's message ID (the reply does use the Correlation Identifier pattern).
Finally, back in the first window, the requester received the reply:
Received reply Time: 09:11:10.156467 Message ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808150円 Correl. ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808149円 Reply to: <n/a> Contents: Hello world.
This output contains several items of interest. The reply was received after it was sent (199987 ms). The message ID of the reply was the same when it was received as it was when it was sent, which proves that it is indeed the same message. The message contents received are the same as those sent. And the correlation ID tells the requestor which request this reply is for (the Correlation Identifier pattern).
Notice too that the requestor is designed to simply send a request, receive a reply, and exit. So having received the reply, the requestor is no longer running. The replier, on the other hand, doesn't know when it might receive a request, so it never stops running. To stop it, we go to its command shell window and press the return key, which causes the replier program to exit.
So this is the request/reply example. A request was prepared and sent by the requestor. The replier received the request and sent a reply. Then the requestor received the reply to its original request.
First, let's take a look at how the Requestor is implemented:
using System;
using System.Messaging;
public class Requestor
{
private MessageQueue requestQueue;
private MessageQueue replyQueue;
public Requestor(String requestQueueName, String replyQueueName)
{
requestQueue = new MessageQueue(requestQueueName);
replyQueue = new MessageQueue(replyQueueName);
replyQueue.MessageReadPropertyFilter.SetAll();
((XmlMessageFormatter)replyQueue.Formatter).TargetTypeNames = new string[]{"System.String,mscorlib"};
}
public void Send()
{
Message requestMessage = new Message();
requestMessage.Body = "Hello world.";
requestMessage.ResponseQueue = replyQueue;
requestQueue.Send(requestMessage);
Console.WriteLine("Sent request");
Console.WriteLine("\tTime: {0}", DateTime.Now.ToString("HH:mm:ss.ffffff"));
Console.WriteLine("\tMessage ID: {0}", requestMessage.Id);
Console.WriteLine("\tCorrel. ID: {0}", requestMessage.CorrelationId);
Console.WriteLine("\tReply to: {0}", requestMessage.ResponseQueue.Path);
Console.WriteLine("\tContents: {0}", requestMessage.Body.ToString());
}
public void ReceiveSync()
{
Message replyMessage = replyQueue.Receive();
Console.WriteLine("Received reply");
Console.WriteLine("\tTime: {0}", DateTime.Now.ToString("HH:mm:ss.ffffff"));
Console.WriteLine("\tMessage ID: {0}", replyMessage.Id);
Console.WriteLine("\tCorrel. ID: {0}", replyMessage.CorrelationId);
Console.WriteLine("\tReply to: {0}", "<n/a>");
Console.WriteLine("\tContents: {0}", replyMessage.Body.ToString());
}
}
An application that wants to send requests and recieve replies could use a requestor to do so. The application specifies the pathnames of two queues: the request queue and the reply queue. This is the information the requestor needs to initialize itself.
In the Requestor constructor, the requestor uses the queue names to connect to the messaging system.
One thing that the requestor needs to be able to do is send request messages. For that, it implements the Send() method.
The other thing the requestor needs to be able to do is receive reply messages. It implements the ReceiveSync() method for this purpose.
In this way, a requestor does everything necessary to send a request and receive a reply.
Next, let's take a look at how the Replier is implemented:
using System;
using System.Messaging;
class Replier {
private MessageQueue invalidQueue;
public Replier(String requestQueueName, String invalidQueueName)
{
MessageQueue requestQueue = new MessageQueue(requestQueueName);
invalidQueue = new MessageQueue(invalidQueueName);
requestQueue.MessageReadPropertyFilter.SetAll();
((XmlMessageFormatter)requestQueue.Formatter).TargetTypeNames = new string[]{"System.String,mscorlib"};
requestQueue.ReceiveCompleted += new ReceiveCompletedEventHandler(OnReceiveCompleted);
requestQueue.BeginReceive();
}
public void OnReceiveCompleted(Object source, ReceiveCompletedEventArgs asyncResult)
{
MessageQueue requestQueue = (MessageQueue)source;
Message requestMessage = requestQueue.EndReceive(asyncResult.AsyncResult);
try
{
Console.WriteLine("Received request");
Console.WriteLine("\tTime: {0}", DateTime.Now.ToString("HH:mm:ss.ffffff"));
Console.WriteLine("\tMessage ID: {0}", requestMessage.Id);
Console.WriteLine("\tCorrel. ID: {0}", "<n/a>");
Console.WriteLine("\tReply to: {0}", requestMessage.ResponseQueue.Path);
Console.WriteLine("\tContents: {0}", requestMessage.Body.ToString());
string contents = requestMessage.Body.ToString();
MessageQueue replyQueue = requestMessage.ResponseQueue;
Message replyMessage = new Message();
replyMessage.Body = contents;
replyMessage.CorrelationId = requestMessage.Id;
replyQueue.Send(replyMessage);
Console.WriteLine("Sent reply");
Console.WriteLine("\tTime: {0}", DateTime.Now.ToString("HH:mm:ss.ffffff"));
Console.WriteLine("\tMessage ID: {0}", replyMessage.Id);
Console.WriteLine("\tCorrel. ID: {0}", replyMessage.CorrelationId);
Console.WriteLine("\tReply to: {0}", "<n/a>");
Console.WriteLine("\tContents: {0}", replyMessage.Body.ToString());
}
catch ( Exception ) {
Console.WriteLine("Invalid message detected");
Console.WriteLine("\tType: {0}", requestMessage.BodyType);
Console.WriteLine("\tTime: {0}", DateTime.Now.ToString("HH:mm:ss.ffffff"));
Console.WriteLine("\tMessage ID: {0}", requestMessage.Id);
Console.WriteLine("\tCorrel. ID: {0}", "<n/a>");
Console.WriteLine("\tReply to: {0}", "<n/a>");
requestMessage.CorrelationId = requestMessage.Id;
invalidQueue.Send(requestMessage);
Console.WriteLine("Sent to invalid message queue");
Console.WriteLine("\tType: {0}", requestMessage.BodyType);
Console.WriteLine("\tTime: {0}", DateTime.Now.ToString("HH:mm:ss.ffffff"));
Console.WriteLine("\tMessage ID: {0}", requestMessage.Id);
Console.WriteLine("\tCorrel. ID: {0}", requestMessage.CorrelationId);
Console.WriteLine("\tReply to: {0}", requestMessage.ResponseQueue.Path);
}
requestQueue.BeginReceive();
}
}
Replier is what an application might use to receive a request and send a reply. The application specifies the pathnames of the request and invalid message queues. (It does not need to specify the name of the reply queue because, as we'll see, that will be provided by the message's Return Address.) This is the information the requestor needs to initialize itself.
The Replier constructor is pretty similar to the requestor's, but there are a couple of differences:
Once the replier has initialized itself to be a listener on the request queue, there's not much for it to do but wait for messages. Unlike the requestor, which has to explicitedly poll the reply queue for messages, the replier is event-driven and so does nothing until the messaging system calls its OnReceiveCompleted method with a new message. The message will be from the request queue because the constructor created the event handler on the request queue. Once OnReceiveCompleted is called, this is what it does to get the new message and processes it:
Thus a replier does everything necessary to receive a message (presumably a request) and send a reply. If it cannot reply to a message, it routes the message to the invalid message queue.
While we're at it, let's look at an example of the Invalid Message Channel pattern. Remember, one of the queues we need is one named "private$\InvalidMessages." This exists so that if an MSMQ client (a Message Endpoint) receives a message it cannot process, it can move the strange message to a special channel.
To demonstrate invalid message handling, we have designed an InvalidMessenger class. This object is specifically designed to send a message on the request channel whose format is incorrect. Like any channel, the request channel is a Datatype Channel, in that the request receivers expect the requests to be of a certain format. The invalid messenger simply sends a message of a different format; when the replier receives the message, it does not recognize the message's format, and so moves the message to the invalid message queue.
We'll run the Replier in one window and the Invalid Messenger in another window. When the invalid messenger sends its message, it displays output like this:
Sent request Type: 768 Time: 09:39:44.223729 Message ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808168円 Correl. ID: 00000000-0000-0000-0000-0000000000000円 Reply to: .\private$\ReplyQueue
Type 768 means that the format of the message contents is binary (whereas the replier is expecting the contents to be text/XML). The Replier recieves the invalid message and resends it to the invalid message queue:
Invalid message detected Type: 768 Time: 09:39:44.233744 Message ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808168円 Correl. ID: <n/a> Reply to: <n/a> Sent to invalid message queue Type: 768 Time: 09:39:44.233744 Message ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808169円 Correl. ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808168円 Reply to: FORMATNAME:DIRECT=OS:XYZ123\private$\ReplyQueue
One insight worth noting is that when the message is moved to the invalid message queue, it is actually being resent, so it gets a new message ID. Because of this, we apply the Correlation Identifier pattern; once the replier determines the message to be invalid, it copies the message's main ID to its correlation ID so as to preserve a record of the message's original ID.
The code that handles this invalid-message processing is in the Replier class shown earlier, in the OnReceiveCompleted method.
We've seen how to implement two classes, Requestor and Replier (Message Endpoints), that exchange a request and reply Messages using Request-Reply. The request message uses a Return Address to specify what queue to send the reply on. The reply messages uses a Correlation Identifier to specify which request this is a reply for. The Requestor implements a Polling Consumer to receive replies, whereas the Replier implements an Event-Driven Consumer to receive requests. The request and reply queues are Datatype Channels; when a consumer receives a message that is not of the right type, it reroutes the message to the Invalid Message Channel.
Enterprise Integration Patterns
The de-facto language for designing asynchronous, distributed systems. Over 100,000 copies sold.
The Software Architect Elevator
Rethink the role of architects as a connecting element across organizational layers. Acquire the technical, communication, and organizational skills to succeed in this new role.
Cloud Strategy
Make your cloud migration a success by translating high-level goals into conscious decisions with well-understood trade-offs.
Platform Strategy
Platforms can boost innovation through harmonization, but they aren't easy to build. Learn from over a decade of designing and rolling out IT platforms.