Chapter 43. MSMQ QuickStart

The MSMQ quick start application demonstrates how to use asynchronous messaging to implement a system for purchasing a stock. Is follows the same basic approach as in the NMS QuickStart but is adapted as need for use with MSMQ. Please read the introduction in that chapter to get an overview of the system.

When there is direct overlap in functionality between the MSMQ and NMS quickstart a reference to the appropriate section in the NMS QuickStart documentation is given.

To communicate between th client and server a pair of queues will be used. Messages sent from the client to the server will use the transactional queue named .\Private$\request.txqueue. Messages sent from the server to the client will use the transactional queue .\Private$\response.joe.txqueue. The queue for messages that cannot be processed, so called 'poison messages' will be sent to the queue .\Private$\dead.queue. You can create these queues using the computer management administration console. Private queues are used to simplify the application setup requirements.

Since MSMQ does not natively support the publish-subscribe messaging style as in other messaging systems, Apache MQ, IBM Websphere MQ, TIBCO EMS, the market data information is sent on the same queue as the responses from the server to the client for trade requests..

The gateway interfaces are the same as those described in the NMS QuickStart here.

TradeRequest and TradeResponse messages are defined using XML Schema and classes are generated from that schema. This is the same approach as described in more details in the NMS QuickStart here.

An important difference in the types of message data formats supported 'out-of-the-box' with Apache, IBM, TIBCO as compared to Microsoft MSMQ is the latter support sending a hashtable data structure. As a result, the hashtable that was used to send market data information from the server to the client was changed to be of type System.String in the MSMQ example.

The message handlers are the same as used in the NMS QuickStart here, aside from the change of the hashtable data structure to a string. This is an important benefit of enforcing a separation between the messaging specific classes and the business processing layer.

The message converter used is Spring.Messaging.Support.Converters.XmlMessageConverter. It is configured by specifying the data types that will be send and received. Here is a configuration example for types generated from the XML Schema and a plain string.

<object id="xmlMessageConverter" type="Spring.Messaging.Support.Converters.XmlMessageConverter, Spring.Messaging">
  <property name="TargetTypes">
    <list>
      <value>Spring.MsmqQuickStart.Common.Data.TradeRequest, Spring.MsmqQuickStart.Common</value>
      <value>Spring.MsmqQuickStart.Common.Data.TradeResponse, Spring.MsmqQuickStart.Common</value>
      <value>System.String, mscorlib</value>        
    </list>
  </property>
</object>

The implementations of the gateway interfaces inherit from Spring's helper class MessageQueueGatewaySupport in order to get easy access to a MessageQueueTemplate for sending. The implementation of the IStockService interface is shown below

public class MsmqStockServiceGateway : MessageQueueGatewaySupport, IStockService
{
    private Random random = new Random();

    private string defaultResponseQueueObjectName;

    public string DefaultResponseQueueObjectName
    {
        set { defaultResponseQueueObjectName = value; }
    }

    public void Send(TradeRequest tradeRequest)
    {
        MessageQueueTemplate.ConvertAndSend(tradeRequest, delegate(Message message)
                                                              {
                                                                  message.ResponseQueue = GetResponseQueue();
                                                                  message.AppSpecific = random.Next();
                                                                  return message;
                                                              });
    }
   
    private MessageQueue GetResponseQueue()
    {
        return MessageQueueFactory.CreateMessageQueue(defaultResponseQueueObjectName);
    }
   
}

The Send method is using MessageQueueTemplate's ConvertAndSend(object obj, MessagePostProcessorDelegate messagePostProcessorDelegate) method. The anonymous delegate allows you to modify the message properties, such as ResponseQueue and AppSpecific after the message has been converted from an object but before it has been sent. The use of an anonymous delegate allows makes it very easy to apply any post processing logic to the converted message.

The configuration for MsmqStockServiceGateway and all its dependencies is shown below, highlighting important dependency links.

<object name="stockServiceGateway" type="Spring.MsmqQuickStart.Client.Gateways.MsmqStockServiceGateway, Spring.MsmqQuickStart.Client">
  <property name="MessageQueueTemplate" ref="messageQueueTemplate"/>    
  <property name="DefaultResponseQueueObjectName" value="responseTxQueue"/>
</object>

<object id="messageQueueTemplate" type="Spring.Messaging.Core.MessageQueueTemplate, Spring.Messaging">
  <property name="DefaultMessageQueueObjectName" value="requestTxQueue"/>
  <property name="MessageConverterObjectName" value="xmlMessageConverter"/>
</object>

<object id="xmlMessageConverter" type="Spring.Messaging.Support.Converters.XmlMessageConverter, Spring.Messaging">
  <property name="TargetTypes">
    <list>
      <value>Spring.MsmqQuickStart.Common.Data.TradeRequest, Spring.MsmqQuickStart.Common</value>
      <value>Spring.MsmqQuickStart.Common.Data.TradeResponse, Spring.MsmqQuickStart.Common</value>
      <value>System.String, mscorlib</value>        
    </list>
  </property>
</object>

<object id="requestTxQueue" type="Spring.Messaging.Support.MessageQueueFactoryObject, Spring.Messaging">
  <property name="Path" value=".\Private$\request.txqueue"/>
  <property name="MessageReadPropertyFilterSetAll" value="true"/>
</object>

<object id="responseTxQueue" type="Spring.Messaging.Support.MessageQueueFactoryObject, Spring.Messaging">
  <property name="Path" value=".\Private$\response.joe.txqueue"/>
  <property name="MessageReadPropertyFilterSetAll" value="true"/>
</object>

Since the client also needs to listen to incoming messages on the responseTxQueue, a TransactionalMessageListenerContainer is configured. The configuration for the message listener container and all its dependencies is shown below, highlighting important dependency links.

<!-- MSMQ Transaction Manager -->
<object id="messageQueueTransactionManager" type="Spring.Messaging.Core.MessageQueueTransactionManager, Spring.Messaging"/>

<!-- Message Listener Container that uses MSMQ transactional for receives -->
<object id="transactionalMessageListenerContainer" type="Spring.Messaging.Listener.TransactionalMessageListenerContainer, Spring.Messaging">
  <property name="MessageQueueObjectName" value="responseTxQueue"/>
  <property name="PlatformTransactionManager" ref="messageQueueTransactionManager"/>
  <property name="MessageListener" ref="messageListenerAdapter"/>
  <property name="MessageTransactionExceptionHandler" ref="sendToQueueExceptionHandler"/>

</object>

<!-- Delegate to plain .NET object for message handling -->
<object id="messageListenerAdapter" type="Spring.Messaging.Listener.MessageListenerAdapter, Spring.Messaging">
  <property name="HandlerObject" ref="stockAppHandler"/>
  <property name="DefaultHandlerMethod" value="Handle"/>
  <property name="MessageConverterObjectName" value="xmlMessageConverter"/>
</object>

<object id="sendToQueueExceptionHandler" type="Spring.Messaging.Listener.SendToQueueExceptionHandler, Spring.Messaging">
  <property name="MessageQueueObjectName" value="deadTxQueue"/>
</object>

<object id="deadTxQueue" type="Spring.Messaging.Support.MessageQueueFactoryObject, Spring.Messaging">
  <property name="Path" value=".\Private$\dead.queue"/>
  <property name="MessageReadPropertyFilterSetAll" value="true"/>
</object>

A similar configuration is used on the server to configure the class Spring.MsmqQuickStart.Server.Gateways.MarketDataServiceGateway that implements the IMarketDataService interface and a TransactionalMessageListenerContainer to process messages on the requestTxQueue. You can increase the number of processing thread in the TransactionalMessageListenerContainer by setting the property MaxConcurrentListeners, the default value is 1.

To run both the client and server make sure that you select 'Multiple Startup Projects' within VS.NET. The GUI has a button to make a hard coded trade request and show confirmation in a text box. A text area is used to display the market data. There is a 'Get Portfolio' button that is not implemented at the moment. A picture of the GUI after it has been running for a while and trade has been sent and responded to is shown below.