 |
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.
Components of the Request/Reply Example
This example was developed using the Microsoft .NET Framework SDK and run on a Windows XP computer with MSMQ [MSMQ01] installed.
Request/Reply Example
This example consists of two main classes:
- Requestor — A Message Endpoint that sends a request message and waits to receive a reply message as a response.
- Replier — A Message Endpoint that waits to receive the request message; when it does, it responds by sending the reply message.
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:
- .\private$\RequestQueue — The MessageQueue the Requestor uses to send the request message to the Replier.
- .\private$\ReplyQueue — The MessageQueue the Replier uses to send the reply message to the Requestor.
- .\private$\InvalidQueue — The MessageQueue that the Requestor and Replier move a message to when they receive a message that they cannot interpret.
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-c3a881a34808\149
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-c3a881a34808\149
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-c3a881a34808\150
Correl. ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808\149
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-c3a881a34808\150
Correl. ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808\149
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.
Request/Reply Code
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.
- It uses the queue names to look up the queues, which are MessageQueues. The names are pathnames to MSMQ resources.
- It sets the reply queue's property filter so that when a message is read from the queue, all of the message's properties will be read as well. It also sets the queue's formatter to be an XmlMessageFormatter so that the message contents will be interpreted as strings.
One thing that the requestor needs to be able to do is send request messages. For that, it implements the Send() method.
- It creates a Message and sets its contents to "Hello world."
- It sets the message's ResponseQueue property to be the reply queue. This is a Return Address that will tell the replier how to send back the reply.
- It then sends the message to the queue.
- It then prints out the details of the message it just sent. This is done after the message is sent because the message ID is set by the messaging system and is not set until the message is actually sent.
The other thing the requestor needs to be able to do is receive reply messages. It implements the ReceiveSync() method for this purpose.
- It runs the queue's Receive() method to get the message, which synchronously blocks until a message is delivered to the queue and is read from the queue, so the requestor is a Polling Consumer. Because this receive is synchronous, the requestor's method is called ReceiveSync().
- The requestor gets the message's contents and prints out the message's details.
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:
- One difference is that the replier does not look up the reply queue. This is because the replier does not assume it will always send replies on that queue; rather, as we'll see it will let the request message tell it what queue to send the reply message on.
- Another difference is that replier is an Event-Driven Consumer, so it sets up a ReceiveCompletedEventHandler. When a message is delivered to the request queue, the messaging system will automatically call the specified method, OnReceiveCompleted.
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:
- The source is a MessageQueue, the request queue.
- The message itself is obtained by running the queue's EndReceive method. The replier then prints out the details about the message.
- Then the replier implements its part of the Return Address pattern. Remember that the requestor set the request message's response-queue property to specify the reply queue. The replier now gets that property's value and uses it reference the proper MessageQueue. The important part here is that the replier is not hard-coded to use a particular reply queue; it uses whatever reply queue each particular request message specifies.
- The replier then creates the reply message. In doing so, it implements the Correlation Identifier pattern by setting the relpy message's correlation-id property to the same value as the request message's message-id property.
- The replier then sends out the reply message and displays its details.
- If the message can be received but not successfully processed, and an Exception is thrown, the replier resends the message to the invalid message queue. In the process, it sets the new message's correlation id to the original message's message id.
- Once the replier has finished processing the message, it runs BeginReceive to start listening for the next message.
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.
Invalid Message Example
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-c3a881a34808\168
Correl. ID: 00000000-0000-0000-0000-000000000000\0
Reply to: .\private$\ReplyQueueType 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-c3a881a34808\168
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-c3a881a34808\169
Correl. ID: 8b0fc389-f21f-423b-9eaa-c3a881a34808\168
Reply to: FORMATNAME:DIRECT=OS:XYZ123\private$\ReplyQueueOne 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.
Conclusions
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.
Integration Pattern Language
