Windows Azure Service Bus Splitter and Aggregator
- by Alan Smith
This article will cover basic implementations of the Splitter and Aggregator patterns using the Windows Azure Service Bus. The content will be included in the next release of the “Windows Azure Service Bus Developer Guide”, along with some other patterns I am working on.
I’ve taken the pattern descriptions from the book “Enterprise Integration Patterns” by Gregor Hohpe. I bought a copy of the book in 2004, and recently dusted it off when I started to look at implementing the patterns on the Windows Azure Service Bus. Gregor has also presented an session in 2011 “Enterprise Integration Patterns: Past, Present and Future” which is well worth a look.
I’ll be covering more patterns in the coming weeks, I’m currently working on Wire-Tap and Scatter-Gather. There will no doubt be a section on implementing these patterns in my “SOA, Connectivity and Integration using the Windows Azure Service Bus” course.
There are a number of scenarios where a message needs to be divided into a number of sub messages, and also where a number of sub messages need to be combined to form one message. The splitter and aggregator patterns provide a definition of how this can be achieved. This section will focus on the implementation of basic splitter and aggregator patens using the Windows Azure Service Bus direct programming model.
In BizTalk Server receive pipelines are typically used to implement the splitter patterns, with sequential convoy orchestrations often used to aggregate messages. In the current release of the Service Bus, there is no functionality in the direct programming model that implements these patterns, so it is up to the developer to implement them in the applications that send and receive messages.
Splitter
A message splitter takes a message and spits the message into a number of sub messages. As there are different scenarios for how a message can be split into sub messages, message splitters are implemented using different algorithms.
The Enterprise Integration Patterns book describes the splatter pattern as follows:
How can we process a message if it contains multiple elements, each of which may have to be processed in a different way?
Use a Splitter to break out the composite message into a series of individual messages, each containing data related to one item.
The Enterprise Integration Patterns website provides a description of the Splitter pattern here.
In some scenarios a batch message could be split into the sub messages that are contained in the batch. The splitting of a message could be based on the message type of sub-message, or the trading partner that the sub message is to be sent to.
Aggregator
An aggregator takes a stream or related messages and combines them together to form one message.
The Enterprise Integration Patterns book describes the aggregator pattern as follows:
How do we combine the results of individual, but related messages so that they can be processed as a whole?
Use a stateful filter, an Aggregator, to collect and store individual messages until a complete set of related messages has been received. Then, the Aggregator publishes a single message distilled from the individual messages.
The Enterprise Integration Patterns website provides a description of the Aggregator pattern here.
A common example of the need for an aggregator is in scenarios where a stream of messages needs to be combined into a daily batch to be sent to a legacy line-of-business application. The BizTalk Server EDI functionality provides support for batching messages in this way using a sequential convoy orchestration.
Scenario
The scenario for this implementation of the splitter and aggregator patterns is the sending and receiving of large messages using a Service Bus queue. In the current release, the Windows Azure Service Bus currently supports a maximum message size of 256 KB, with a maximum header size of 64 KB. This leaves a safe maximum body size of 192 KB.
The BrokeredMessage class will support messages larger than 256 KB; in fact the Size property is of type long, implying that very large messages may be supported at some point in the future. The 256 KB size restriction is set in the service bus components that are deployed in the Windows Azure data centers.
One of the ways of working around this size restriction is to split large messages into a sequence of smaller sub messages in the sending application, send them via a queue, and then reassemble them in the receiving application. This scenario will be used to demonstrate the pattern implementations.
Implementation
The splitter and aggregator will be used to provide functionality to send and receive large messages over the Windows Azure Service Bus. In order to make the implementations generic and reusable they will be implemented as a class library. The splitter will be implemented in the LargeMessageSender class and the aggregator in the LargeMessageReceiver class. A class diagram showing the two classes is shown below.
Implementing the Splitter
The splitter will take a large brokered message, and split the messages into a sequence of smaller sub-messages that can be transmitted over the service bus messaging entities. The LargeMessageSender class provides a Send method that takes a large brokered message as a parameter. The implementation of the class is shown below; console output has been added to provide details of the splitting operation.
public class LargeMessageSender
{
private static int SubMessageBodySize = 192 * 1024;
private QueueClient m_QueueClient;
public LargeMessageSender(QueueClient queueClient)
{
m_QueueClient = queueClient;
}
public void Send(BrokeredMessage message)
{
// Calculate the number of sub messages required.
long messageBodySize = message.Size;
int nrSubMessages = (int)(messageBodySize / SubMessageBodySize);
if (messageBodySize % SubMessageBodySize != 0)
{
nrSubMessages++;
}
// Create a unique session Id.
string sessionId = Guid.NewGuid().ToString();
Console.WriteLine("Message session Id: " + sessionId);
Console.Write("Sending {0} sub-messages", nrSubMessages);
Stream bodyStream = message.GetBody<Stream>();
for (int streamOffest = 0; streamOffest < messageBodySize;
streamOffest += SubMessageBodySize)
{
// Get the stream chunk from the large message
long arraySize = (messageBodySize - streamOffest) > SubMessageBodySize
? SubMessageBodySize : messageBodySize - streamOffest;
byte[] subMessageBytes = new byte[arraySize];
int result = bodyStream.Read(subMessageBytes, 0, (int)arraySize);
MemoryStream subMessageStream = new MemoryStream(subMessageBytes);
// Create a new message
BrokeredMessage subMessage = new BrokeredMessage(subMessageStream, true);
subMessage.SessionId = sessionId;
// Send the message
m_QueueClient.Send(subMessage);
Console.Write(".");
}
Console.WriteLine("Done!");
}}
The LargeMessageSender class is initialized with a QueueClient that is created by the sending application. When the large message is sent, the number of sub messages is calculated based on the size of the body of the large message. A unique session Id is created to allow the sub messages to be sent as a message session, this session Id will be used for correlation in the aggregator. A for loop in then used to create the sequence of sub messages by creating chunks of data from the stream of the large message. The sub messages are then sent to the queue using the QueueClient.
As sessions are used to correlate the messages, the queue used for message exchange must be created with the RequiresSession property set to true.
Implementing the Aggregator
The aggregator will receive the sub messages in the message session that was created by the splitter, and combine them to form a single, large message. The aggregator is implemented in the LargeMessageReceiver class, with a Receive method that returns a BrokeredMessage. The implementation of the class is shown below; console output has been added to provide details of the splitting operation.
public class LargeMessageReceiver
{
private QueueClient m_QueueClient;
public LargeMessageReceiver(QueueClient queueClient)
{
m_QueueClient = queueClient;
}
public BrokeredMessage Receive()
{
// Create a memory stream to store the large message body.
MemoryStream largeMessageStream = new MemoryStream();
// Accept a message session from the queue.
MessageSession session = m_QueueClient.AcceptMessageSession();
Console.WriteLine("Message session Id: " + session.SessionId);
Console.Write("Receiving sub messages");
while (true)
{
// Receive a sub message
BrokeredMessage subMessage = session.Receive(TimeSpan.FromSeconds(5));
if (subMessage != null)
{
// Copy the sub message body to the large message stream.
Stream subMessageStream = subMessage.GetBody<Stream>();
subMessageStream.CopyTo(largeMessageStream);
// Mark the message as complete.
subMessage.Complete();
Console.Write(".");
}
else
{
// The last message in the sequence is our completeness criteria.
Console.WriteLine("Done!");
break;
}
}
// Create an aggregated message from the large message stream.
BrokeredMessage largeMessage = new BrokeredMessage(largeMessageStream, true);
return largeMessage;
}
}
The LargeMessageReceiver initialized using a QueueClient that is created by the receiving application. The receive method creates a memory stream that will be used to aggregate the large message body. The AcceptMessageSession method on the QueueClient is then called, which will wait for the first message in a message session to become available on the queue. As the AcceptMessageSession can throw a timeout exception if no message is available on the queue after 60 seconds, a real-world implementation should handle this accordingly.
Once the message session as accepted, the sub messages in the session are received, and their message body streams copied to the memory stream. Once all the messages have been received, the memory stream is used to create a large message, that is then returned to the receiving application.
Testing the Implementation
The splitter and aggregator are tested by creating a message sender and message receiver application. The payload for the large message will be one of the webcast video files from http://www.cloudcasts.net/, the file size is 9,697 KB, well over the 256 KB threshold imposed by the Service Bus.
As the splitter and aggregator are implemented in a separate class library, the code used in the sender and receiver console is fairly basic. The implementation of the main method of the sending application is shown below.
static void Main(string[] args)
{
// Create a token provider with the relevant credentials.
TokenProvider credentials =
TokenProvider.CreateSharedSecretTokenProvider
(AccountDetails.Name, AccountDetails.Key);
// Create a URI for the serivce bus.
Uri serviceBusUri = ServiceBusEnvironment.CreateServiceUri
("sb", AccountDetails.Namespace, string.Empty);
// Create the MessagingFactory
MessagingFactory factory = MessagingFactory.Create(serviceBusUri, credentials);
// Use the MessagingFactory to create a queue client
QueueClient queueClient = factory.CreateQueueClient(AccountDetails.QueueName);
// Open the input file.
FileStream fileStream = new FileStream(AccountDetails.TestFile, FileMode.Open);
// Create a BrokeredMessage for the file.
BrokeredMessage largeMessage = new BrokeredMessage(fileStream, true);
Console.WriteLine("Sending: " + AccountDetails.TestFile);
Console.WriteLine("Message body size: " + largeMessage.Size);
Console.WriteLine();
// Send the message with a LargeMessageSender
LargeMessageSender sender = new LargeMessageSender(queueClient);
sender.Send(largeMessage);
// Close the messaging facory.
factory.Close();
}
The implementation of the main method of the receiving application is shown below.
static void Main(string[] args)
{
// Create a token provider with the relevant credentials.
TokenProvider credentials =
TokenProvider.CreateSharedSecretTokenProvider
(AccountDetails.Name, AccountDetails.Key);
// Create a URI for the serivce bus.
Uri serviceBusUri = ServiceBusEnvironment.CreateServiceUri
("sb", AccountDetails.Namespace, string.Empty);
// Create the MessagingFactory
MessagingFactory factory = MessagingFactory.Create(serviceBusUri, credentials);
// Use the MessagingFactory to create a queue client
QueueClient queueClient = factory.CreateQueueClient(AccountDetails.QueueName);
// Create a LargeMessageReceiver and receive the message.
LargeMessageReceiver receiver = new LargeMessageReceiver(queueClient);
BrokeredMessage largeMessage = receiver.Receive();
Console.WriteLine("Received message");
Console.WriteLine("Message body size: " + largeMessage.Size);
string testFile = AccountDetails.TestFile.Replace(@"\In\", @"\Out\");
Console.WriteLine("Saving file: " + testFile);
// Save the message body as a file.
Stream largeMessageStream = largeMessage.GetBody<Stream>();
largeMessageStream.Seek(0, SeekOrigin.Begin);
FileStream fileOut = new FileStream(testFile, FileMode.Create);
largeMessageStream.CopyTo(fileOut);
fileOut.Close();
Console.WriteLine("Done!");
}
In order to test the application, the sending application is executed, which will use the LargeMessageSender class to split the message and place it on the queue. The output of the sender console is shown below.
The console shows that the body size of the large message was 9,929,365 bytes, and the message was sent as a sequence of 51 sub messages.
When the receiving application is executed the results are shown below.
The console application shows that the aggregator has received the 51 messages from the message sequence that was creating in the sending application. The messages have been aggregated to form a massage with a body of 9,929,365 bytes, which is the same as the original large message. The message body is then saved as a file.
Improvements to the Implementation
The splitter and aggregator patterns in this implementation were created in order to show the usage of the patterns in a demo, which they do quite well. When implementing these patterns in a real-world scenario there are a number of improvements that could be made to the design.
Copying Message Header Properties
When sending a large message using these classes, it would be great if the message header properties in the message that was received were copied from the message that was sent. The sending application may well add information to the message context that will be required in the receiving application.
When the sub messages are created in the splitter, the header properties in the first message could be set to the values in the original large message. The aggregator could then used the values from this first sub message to set the properties in the message header of the large message during the aggregation process.
Using Asynchronous Methods
The current implementation uses the synchronous send and receive methods of the QueueClient class. It would be much more performant to use the asynchronous methods, however doing so may well affect the sequence in which the sub messages are enqueued, which would require the implementation of a resequencer in the aggregator to restore the correct message sequence.
Handling Exceptions
In order to keep the code readable no exception handling was added to the implementations. In a real-world scenario exceptions should be handled accordingly.
