Asynchronous Operations

I l @ ve RuBoard

The examples presented so far have assumed that receive and peek operations are synchronous. The applications have issued method calls that might block if the requested message queue is empty. One key strength of message queuing solutions is that they're designed for asynchronous operations, and blocking is a feature that should be designed out of any such systems. We'll examine some of the issues involved in asynchronous operations next .

Receiving Messages Asynchronously

The receive and peek operations of the MessageQueue class support the asynchronous programming model used elsewhere in the .NET Framework Class Library. The BeginPeek and BeginReceive methods start a receive or peek operation asynchronously, using a thread from the thread pool. In their simplest form, they raise PeekCompleted or ReceiveCompleted events as appropriate. An application can subscribe to these events to obtain the information retrieved from the message queue.

Before subscribing to the ReceiveCompleted event, you must define a method to act as a callback when the event is raised. This method must conform to the usual requirements of an event handler method: It should take an Object parameter that holds the sender of the object, as well as an EventArgs parameter that holds event-specific information. In the case of the ReceiveCompleted event, the EventArgs parameter should be a System.Messaging.ReceiveCompletedEventArgs object. The ReceiveCompletedEventArgs class exposes the message received through its Message property.

The MessageReceived method shown on the next page is an example that you can use to subscribe to the ReceiveCompleted event:

 privatevoidMessageReceived(System.Objectsender, ReceiveCompletedEventArgsargs) { //Extractthemessage MessagecakeMessage=args.get_Message(); //Specifytheformatter cakeMessage.set_Formatter(newBinaryMessageFormatter()); //Deserializethemessage CakeRequestcakeData=(CakeRequest)cakeMessage.get_Body(); //Processthemessagebody } 

To subscribe to the ReceiveCompleted event, you must set the ReceiveCompleted event property of the message queue to a ReceiveCompletedEvent ­Handler delegate that refers to the MessageReceived method. You can then invoke the BeginReceive method:

 MessageQueuecakeQueue=...; cakeQueue.add_ReceiveCompleted(newReceiveCompletedEventArgs(MessageReceived)); cakeQueue.BeginReceive(); 

Execution will continue after BeginReceive , but when a message has been received, the MessageReceived method will run (on its own thread). There is a corresponding event ( PeekCompleted ) and a delegate ( PeekCompletedEvent ­Args ) for performing an asynchronous peek.

Both the BeginReceive and BeginPeek methods are overloaded. You can optionally specify a TimeSpan . If a message does not become available in the intervening period, the waiting thread will terminate silently (without throwing an exception), and the event handler will not be invoked if a message subsequently arrives in the queue. Further overloads allow you to specify an AsyncCallback delegate rather than subscribing to an event, and a programmer-defined state object. In these cases, you should execute the EndReceive or EndPeek method to obtain the message retrieved from the queue. (If the corresponding BeginReceive or BeginPeek method timed out, you will receive an exception this time).

Note

Multithreaded peeking is limited to ordinary peek operations. If you want to call PeekById and PeekByCorrelationId asynchronously, you must create your own thread to do it.

Disconnected Queues

In the traveling salesperson scenario described at the start of this chapter, the application running on the salesperson's laptop computer submits messages to a queue. These messages are cached locally and then dispatched to the destination queue when the laptop computer is connected to the corporate network. Message Queue 3.0 supports this mode of operation, but there is one issue: Normally, when an application sends a message to a queue the underlying infrastructure must be able to locate the target queue. (Message Queue 3.0 takes the view that if the queue does not exist, it would be useful to throw some sort of exception rather than silently lose messages!) If the target queue is a private queue on another machine, the message queuing infrastructure will attempt to contact the destination machine, fail, and throw an exception almost immediately. If the target queue is a public queue, the message queuing infrastructure will instead try to contact the domain controller so it can query Active Directory for the details and routing information of the target queue. Again, if the laptop is disconnected from the network, this will fail and throw an exception (but not immediately ”the underlying mechanisms will allow a little time for the domain controller to respond before timing out).

Sending a message to a message queue when you're disconnected from the network therefore requires a little cunning. You can identify a message queue when you instantiate a MessageQueue object in a few ways. The techniques shown so far have specified the pathname of the message queue ( White ­Rabbit\\CakeQueue , for example), and these need to be resolved either by contacting the target machine in the case of a private message queue or through Active Directory if the message queue is public (as just described). An alternative is to use a format name . Format names do not require any further resolution, so if the named queue is not immediately accessible it doesn't matter.

Several types of format name are available, but the simplest and most portable simply specifies the GUID of the destination message queue with a PUBLIC or PRIVATE prefix. The GUID of a public message queue is assigned when the queue is created, and you can find it by using the Computer Management console and examining the ID property, as shown in Figure 12-8.

Figure 12-8. The properties of the CakeQueue message queue

Once you know the GUID of a public message queue, you can create a MessageQueue object that's capable of supporting a disconnected send operation:

 privatestaticStringserverQueuePath=  "FORMATNAME:PUBLIC=172B7EDD-5D47-4C72-80BC-61CABD9AA8AA"; MessageQueuecakeQueue=newMessageQueue(serverQueuePath); 

When a message is sent to this queue, if the laptop computer is connected to the network the message will be transmitted. If the laptop computer is disconnected, the message will be cached locally in an outgoing queue and transmitted automatically once a network connection is reestablished. Remember that if the TimeToReachQueue property of the message is specified, the message will start to age when it is sent and might expire while still cached locally if the period is too short. Also notice that if the public queue is dropped and re-created, its GUID will change, rendering the format name useless! In other words, once you've created the public queues needed for processing messages in your system, you should keep them ”don't drop them and rebuild them unless you're prepared to reset any disconnected clients that access them.

An alternative is to use a direct format name that comprises the address of the computer that holds the destination message queue and the name of the queue. The address can be specified in the form of an IP address, a machine name, or an HTTP URL. For example, the CakeQueue public queue on the computer called WhiteRabbit can be referenced as follows :

 privatestaticStringserverQueuePath=  "FORMATNAME:DIRECT=OS:WhiteRabbit\CakeQueue"; 

Then again, if you know the IP address of the destination computer, you can use it for referencing the queue. (Do not use this approach if the destination computer's address is obtained using DHCP because the address might change.)

 privatestaticStringserverQueuePath=  "FORMATNAME:DIRECT=TCP:192.168.1.2\CakeQueue"; 

A private message queue does not have a GUID, so a different type of format name must be used for disconnected operations. Again, several options are available, including the ability to use direct format names, as shown in this example:

 privatestaticStringresponseQueuePath=  "FORMATNAME:DIRECT=TCP:192.168.1.2\PRIVATE$\CakeResponse"; 

The simplest format comprises the GUID of the target machine and the number of the queue on that machine. The machine GUID is created when message queuing is installed on a computer, and you can find the GUID allocated to a computer by calling the static MessageQueue.GetMachineId method and passing the name of the machine as a string parameter. You can do this only while you're connected to the network, but once you've obtained the GUID for a computer, it will not change unless message queuing is reinstalled on that computer. The queue number is an eight-digit identifier.

Determining the queue number for a private queue is a convoluted process ”use Windows Explorer to navigate to the folder \WINDOWS\System32\MSMQ\Storage\Lqs, look through the names of the files to locate the storage file corresponding to your private message queue, and find the eight-digit prefix. For example, the storage file named 00000043.6ab7c4b8 is for private message queue number 00000043. (Storage files that comprise a GUID without a prefix are for public queues, and you can ignore those.) After establishing the GUID of the target computer and the number of the message queue, you can string them together and specify them as the format name. A typical format name for a private message queue looks like this:

FORMATNAME:PRIVATE=B4C092C0-29AD-43B5-8E73-0F7909F40114\00000003

Tip

You can also find the format name of a private message queue by creating a MessageQueue object using a pathname (such as .\private$\CakeResponse) while you're connected to the domain (so the Active Directory lookup will operate ) and then querying the FormatName property of the MessageQueue object. You can then replace the path in your code with the format name, recompile the program, and execute it while you're disconnected from the network.

Again, note that if a private message queue is deleted and rebuilt, its queue number will probably change, with all that this implies for client applications.

Some restrictions apply when you use format names: You cannot test for the existence of, create, or delete a queue that's specified using a format name, for example.

Requesting an Acknowledgment

When you send a message to a queue, you can arrange for information messages to be dispatched to another message queue whenever something significant happens to the original message, such as safe delivery to the destination queue or successful receipt by the server application. These messages are generated by the underlying message queuing infrastructure and are referred to as positive acknowledgments . You can also request that messages be generated if a failure is detected , such as a message expiring before it is delivered. These are examples of negative acknowledgments . If you want to see the different acknowledgments, positive and negative, that are available, you can look up the System.Messaging.AcknowledgeTypes enumeration in the Visual Studio .NET documentation.

To request an acknowledgement , you set the AdministrationQueue property of a message to a queue for receiving the acknowledgement messages. This queue is referred to as an administration queue , and it is often created specifically for holding acknowledgment messages, although any public or private queue can be used:

 privatestaticStringadminQueuePath= ".\private$\AdminQueue"; Messagemsg=...; //CreateoropentheAdminqueuetoholdtheacknowledgementmessages MessageQueueadminQueue; if(!MessageQueue.Exists(adminQueuePath)) { adminQueue=MessageQueue.Create(adminQueuePath); } //Ifthequeuealreadyexists,openit else { adminQueue=newMessageQueue(adminQueuePath); } msg.set_AdministrationQueue(adminQueue); 

You must also indicate which acknowledgements you want to receive by setting the AcknowledgeType property of the message. You can set this property to any of the values in the AcknowledgeTypes enumeration, and if you require acknowledgements for more than one event as the message passes through the network, you can combine values. The following code fragment requests acknowledgement when the message is delivered to the destination queue and when the server process has received the message:

 msg.set_AcknowledgeType(AcknowledgeTypes.FullReachQueueAcknowledgeTypes.FullReceive); 

When the message, msg , is sent to a queue, the various acknowledgment messages will appear in the administration queue specified by the message. An application can read the administration queue and track the message. Messages posted to the administration queue have no body; instead, they provide information about the message they refer to through properties. The two most important properties are Acknowledgment and CorrelationId . The Acknowledgment property indicates the acknowledgment event that generated the message and is one of the values in the System.Messaging.Acknowledgment enumeration (not to be confused with the AcknowledgeTypes enumeration). The value ReachQueue indicates that the message reached the destination queue, and the value Receive specifies that the message was successfully received. The CorrelationId contains the Id of the original message but is available only if you set the CorrelationId property of the MessageReaderPropertyFilter of the queue to true , as described earlier.

The following code fragment reads and processes a message from the administration queue to determine whether it is acknowledging delivery or receipt of the message:

 adminQueue.get_MessageReadPropertyFilter().set_CorrelationId(true); MessageadminMsg=adminQueue.Receive(); if(adminMsg.get_Acknowledgment().Equals(Acknowledgment.ReachQueue)) { Console.WriteLine("Message " +adminMsg.get_CorrelationId()+  " reacheditsdestinationqueue"); } if(adminMsg.get_Acknowledgment().Equals(Acknowledgment.Receive)) { Console.WriteLine("Message " +adminMsg.get_CorrelationId()+  " receivedbydestinationapplication"); } 

Note

Peeking at a message does not count as receiving it because the message is not actually removed from the queue. Therefore, a peek operation will not generate a Receive acknowledgment.

I l @ ve RuBoard