Chapter 30. Message Oriented Middleware - MSMQ

Chapter 30. Message Oriented Middleware - MSMQ

1. If the message queue is transactional and there is an ambient MessageQueueTransaction in thread local storage (put there via the use of Spring's MessageQueueTransactionManager or TransactionalMessageListenerContainer), the message will be sent transactionally using the MessageQueueTransaction object in thread local storage.

   	Note
   	This lets you group together multiple messaging operations within the same transaction without having to explicitly pass around the MessageQueueTransaction object.

2. f the message queue is transactional but there is no ambient MessageQueueTransaction, then a single message transaction is created on each messaging operation. (MessageQueueTransactionType = Single).

3. If there is an ambient System.Transactions transaction then that transaction will be used (MessageQueueTransactionType = Automatic).

4. If the queue is not transactional, then a non-transactional send (MessageQueueTransactionType = None) is used.

• NonTransactionalMessageListenerContainer - does not surround the receive operation with a transaction

• TransactionalMessageListenerContainer - surrounds the receive operation with local (non-DTC) based transaction(s).

• DistributedTxMessageListenerContainer - surrounds the receive operation with a distributed (DTC) transaction

This container performs a Receive operation on the MessageQueue without any transactional settings. As such messages will not be redelivered if an exception is thrown during message processing. Exceptions during message processing can be handled via an implementation of the interface IExceptionHandler. This can be set via the property ExceptionHandler on the listener. The IExceptionHandler interface is shown below

    public interface IExceptionHandler
    {
        void OnException(Exception exception, Message message);
    }

An example of configuring a NonTransactionalMessageListenerContainer with an IExceptionHandler is shown below

  <!-- Queue to receive from -->
  <object id='msmqTestQueue' type='Spring.Messaging.Support.MessageQueueFactoryObject, Spring.Messaging'>
    <property name='Path' value='.\Private$\testqueue'/>
    <property name='MessageReadPropertyFilterSetAll' value='true'/>
    <property name='ProductTemplate'>
      <object>
        <property name='Label' value='MyTestQueueLabel'/>
      </object>
    </property>
  </object>

  <!-- Queue to respond to -->
  <object id='msmqTestResponseQueue' type='Spring.Messaging.Support.MessageQueueFactoryObject, Spring.Messaging'>
    <property name='Path' value='.\Private$\testresponsequeue'/>
    <property name='MessageReadPropertyFilterSetAll' value='true'/>
    <property name='ProductTemplate'>
      <object>
        <property name='Label' value='MyTestResponseQueueLabel'/>
      </object>
    </property>
  </object>

  <!-- Listener container -->
  <object id="nonTransactionalMessageListenerContainer" type="Spring.Messaging.Listener.NonTransactionalMessageListenerContainer, Spring.Messaging">
    <property name="MessageQueueObjectName" value="msmqTestQueue"/>
    <property name="MaxConcurrentListeners" value="2"/>
    <property name="ListenerTimeLimit" value="20s"/>  <!-- 20 seconds -->
    <property name="MessageListener" ref="messageListenerAdapter"/>
    <property name="ExceptionHandler" ref="exceptionHandler"/>
  </object>

  <!-- Delegate to plain .NET object for message handling -->
  <object id="messageListenerAdapter" type="Spring.Messaging.Listener.MessageListenerAdapter, Spring.Messaging">
    <property name="DefaultResponseQueueName" value="msmqTestResponseQueue"/>
    <property name="HandlerObject" ref="simpleHandler"/>
  </object>

  <!-- Classes you need to write -->
  <object id="simpleHandler" type="MyNamespace.SimpleHandler, MyAssembly"/>

  <object id="exceptionHandler" type="MyNamespace.SimpleExceptionHandler, MyAssembly"/>

The SimpleHandler class would look something like this

public class SimpleHandler : ISimpleHandler
{
  public void HandleObject(string txt)
  {
     // perform message processing...
     Console.WriteLine("Received text: " + txt);
  }
}

This message listener container performs receive operations within the context of local transaction. This class requires an instance of Spring's IPlatformTransactionManager, either AdoPlatformTransactionManager, HibernateTransactionManager, or MessageQueueTransactionManager.

If you specify a MessageQueueTransactionManager then a MessageQueueTransaction will be started before receiving the message and used as part of the container's receive operation. As with other IPlatformTransactionManager implementation's, the transactional resources (in this case an instance of the MessageQueueTransaction class) is bound to thread local storage. MessageQueueTemplate will look in thread-local storage and use this 'ambient' transaction if found for its send and receive operations. The message listener is invoked and if no exception occurs, then the MessageQueueTransactionManager will commit the MessageQueueTransaction.

The message listener implementation can call into service layer classes that are made transactional using standard Spring declarative transactional techniques. In case of exceptions in the service layer, the database operation will be rolled back (nothing new here), and the TransactionalMessageListenerContainer will call it's IMessageTransactionExceptionHandler implementation to determine if the MessageQueueTransaction should commit (removing the message from the queue) or rollback (leaving the message on the queue for redelivery).

Note

The use of a transactional service layer in combination with a MessageQueueTransactionManager is a powerful combination that can be used to achieve "exactly one" transaction message processing with database operations. This requires a little extra programming effort and is a more efficient alternative than using distributed transactions which are commonly associated with this functionality since both the database and the message transaction commit or rollback together. The additional programming logic needed to achieve this is to keep track of the Message.Id that has been processed successfully within the transactional service layer. This is needed as there may be a system failure (e.g. power goes off) between the 'inner' database commit and the 'outer' messaging commit, resulting in message redelivery. The transactional service layer needs logic to detect if incoming message was processed successfully. It can do this by checking the database for an indication of successful processing, perhaps by recording the Message.Id itself in a status table. If the transactional service layer determines that the message has already been processed, it can throw a specific exception for this case. The container's exception handler will recognize this exception type and vote to commit (remove from the queue) the 'outer' messaging transaction. Spring provides an exception handler with this functionality, see SendToQueueExceptionHandler described below.

An example of configuring the TransactionalMessageListenerContainer using a MessageQueueTransactionManager is shown below

  <!-- Queue to receive from -->
  <object id='msmqTestQueue' type='Spring.Messaging.Support.MessageQueueFactoryObject, Spring.Messaging'>
    <property name='Path' value='.\Private$\testqueue'/>
    <property name='MessageReadPropertyFilterSetAll' value='true'/>
    <property name='ProductTemplate'>
      <object>
        <property name='Label' value='MyTestQueueLabel'/>
      </object>
    </property>
  </object>

  <!-- Queue to respond to -->
  <object id='msmqTestResponseQueue' type='Spring.Messaging.Support.MessageQueueFactoryObject, Spring.Messaging'>
    <property name='Path' value='.\Private$\testresponsequeue'/>
    <property name='MessageReadPropertyFilterSetAll' value='true'/>
    <property name='ProductTemplate'>
      <object>
        <property name='Label' value='MyTestResponseQueueLabel'/>
      </object>
    </property>
  </object>

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

  <!-- The transaction message listener container -->
  <object id="transactionalMessageListenerContainer" type="Spring.Messaging.Listener.TransactionalMessageListenerContainer, Spring.Messaging">
    <property name="MessageQueueObjectName" value="msmqTestQueue"/>
    <property name="PlatformTransactionManager" ref="messageQueueTransactionManager"/>
    <property name="MaxConcurrentListeners" value="5"/>
    <property name="ListenerTimeLimitIn" value="20s"/>
    <property name="MessageListener" ref="messageListenerAdapter"/>
    <property name="MessageTransactionExceptionHandler" ref="messageTransactionExceptionHandler"/>
  </object>

  <!-- Delegate to plain .NET object for message handling -->
  <object id="messageListenerAdapter" type="Spring.Messaging.Listener.MessageListenerAdapter, Spring.Messaging">
    <property name="DefaultResponseQueueName" value="msmqTestResponseQueue"/>
    <property name="HandlerObject" ref="simpleHandler"/>
  </object>

  <!-- Poison message handling -->
  <object id="messageTransactionExceptionHandler" type="Spring.Messaging.Listener.SendToQueueExceptionHandler, Spring.Messaging">
    <property name="MaxRetry" value="5"/>
    <property name="MessageQueueObjectName" value="testTxErrorQueue"/>
  </object>

  <!-- Classes you need to write -->
  <object id="simpleHandler" type="MyNamespace.SimpleHandler, MyAssembly"/>

If you specify either AdoPlatformTransactionManager or HibernateTransactionManager then a local database transaction will be started before the receiving the message. By default, the container will also start a local MessageQueueTransaction after the local database transaction has started, but before the receiving the message. This MessageQueueTransaction will be used to receive the message. By default the MessageQueueTransaction will be bound to thread local storage so that any MessageQueueTemplate send or receive operations will participate transparently in the same MessageQueueTransaction. If you do not want this behavior set the property ExposeContainerManagedMessageQueueTransaction to false.

In case of exceptions during IMessageListener processing when using either either AdoPlatformTransactionManager or HibernateTransactionManager the container's IMessageTransactionExceptionHandler will determine if the MessageQueueTransaction should commit (removing it from the queue) or rollback (placing it back on the queue for redelivery). The listener exception will always trigger a rollback in the 'outer' database transaction.

Poison message handing, that is, the endless redelivery of a message due to exceptions during processing, can be detected using implementations of the IMessageTransactionExceptionHandler. This interface is shown below

public interface IMessageTransactionExceptionHandler
{
    TransactionAction OnException(Exception exception, Message message,  MessageQueueTransaction messageQueueTransaction);
}

The return value is an enumeration with the values Commit and Rollback. A specific implementation is provided that will move the poison message to another queue after a maximum number of redelivery attempts. See SendToQueueExceptionHandler described below. You can set a specific implementation to by setting TransactionalMessageListenerContainer's property MessageTransactionExceptionHandler

The IMessageTransactionExceptionHandler implementation SendToQueueExceptionHandler keeps track of the Message's Id property in memory with a count of how many times an exception has occurred. If that count is greater than the handler's MaxRetry count it will be sent to another queue using the provided MessageQueueTransaction. The queue to send the message to is specified via the property MessageQueueObjectName.

This message listener container performs receive operations within the context of distributed transaction. A distributed transaction is started before a message is received. The receive operation participates in this transaction using by specifying MessageQueueTransactionType = Automatic. The transaction that is started is automatically promoted to two-phase-commit to avoid the default behavior of transaction promotion since the only reason to use this container is to use two different resource managers (messaging and database typically).

The commit and rollback semantics are simple, if the message listener does not throw an exception the transaction is committed, otherwise it is rolled back.

Exceptions in message listener processing are handled by implementations of the IDistributedTransactionExceptionHandler interface. This interface is shown below

    public interface IDistributedTransactionExceptionHandler
    {
        bool IsPoisonMessage(Message message);

        void HandlePoisonMessage(Message poisonMessage);        

        void OnException(Exception exception, Message message);
    }

the IsPoisonMessage method determines whether the incoming message is a poison message. This method is called before the IMessageListener is invoked. The container will call HandlePoisonMessage is IsPoisonMessage returns true and will then commit the distributed transaction (removing the message from the queue. Typical implementations of HandlePoisonMessage will move the poison message to another queue (under the same distributed transaction used to receive the message). The class SendToQueueDistributedTransactionExceptionHandler detects poison messages by tracking the Message Id property in memory with a count of how many times an exception has occurred. If that count is greater than the handler's MaxRetry count it will be sent to another queue. The queue to send the message to is specified via the property MessageQueueObjectName.

• XmlMessageConverter - uses a XmlMessageFormatter.

• BinaryMessageConverter - uses a BinaryMessageFormatter

• ActiveXMessageConverter - uses a ActiveXMessageFormatter

• XmlDocumentConverter - loads and saves an XmlDocument to the message BodyStream. This lets you manipulate directly the XML data independent of type serialization issues. This is quite useful if you use XPath expressions to pick out the relevant information to construct your business objects.

• RawBytesMessageConverter - directly write raw bytes to the message stream, compress

• CompressedMessageConverter - compresses the message payload

• EncryptedMessageConverter - encrypt the message (standard MSMQ encryptiong has several limitations)

• SoapMessageConverter - use soap formatting.

The MessageListenerAdapter allows methods of a class that does not implement the IMessageListener interface to be invoked upon message delivery. Lets call this class the 'message handler' class. To achieve this goal the MessageListenerAdapter implements the standard IMessageListener interface to receive a message and then delegates the processing to the message handler class. Since the message handler class does not contain methods that refer to MSMQ artifacts such as Message, the MessageListenerAdapter uses a IMessageConverter to bridge the MSMQ and 'plain object' worlds. As a reminder, the default XmlMessageConverter used in MessageQueueTemplate and the message listener containers converts from Message to string. Once the incoming message is converted to an object (string for example) a method with the name 'HandleMessage' is invoked via reflection passing in the string as an argument.

Using the default configuration of XmlMessageConverter in the message listeners, a simple string based message handler would look like this.

public class MyHandler
{

    public void HandleMessage(string text)
    {
      ...
    }
     
}

The next example has a similar method signature but the name of the handler method name has been changed to "DoWork", by setting the adapter's property DefaultHandlerMethod.

public interface IMyHandler
{
    void DoWork(string text);
}

If your IMessageConverter implementation will return multiple object types, overloading the handler method is perfectly acceptable, the most specific matching method will be used. A method with an object signature would be consider a 'catch-all' method of last resort.

public interface IMyHandler
{
   void DoWork(string text);
   void DoWork(OrderRequest orderRequest);
   void DoWork(InvoiceRequest invoiceRequest);
   void DoWork(object obj);
}

Another of the capabilities of the MessageListenerAdapter class is the ability to automatically send back a response Message if a handler method returns a non-void value. Any non-null value that is returned from the execution of the handler method will (in the default configuration) be converted to a string. The resulting string will then be sent to the ResponseQueue defined in the Message's ResponseQueue property of the original Message, or the DefaultResponseQueueName on the MessageListenerAdapter (if one has been configured) will be used. If not ResponseQueue is found then an Spring MessagingException will be thrown. Please note that this exception will not be swallowed and will propagate up the call stack.

Here is an example of Handler signatures that have various return types.

public interface IMyHandler
{
   string DoWork(string text);
   OrderResponse DoWork(OrderRequest orderRequest);
   InvoiceResponse DoWork(InvoiceRequest invoiceRequest);
   void DoWork(object obj);
}

The following configuration shows how to hook up the adapter to process incoming MSMQ messages using the default message converter.

  <!-- Delegate to plain .NET object for message handling -->
  <object id="messageListenerAdapter" type="Spring.Messaging.Listener.MessageListenerAdapter, Spring.Messaging">
    <property name="DefaultResponseQueueName" value="msmqTestResponseQueue"/>
    <property name="HandlerObject" ref="myHandler"/>
  </object>