Skip to main content

Patterns in Windows Azure Service Bus - Message Splitter Pattern

In one of my post about Service Bus Topics from Windows Azure I told you that I will write about a post that describe how we can design an application that can route messages for a house which is controlled remotely. Before we can talk about this we need to see some design patterns that can be used in combination with Service Bus. I will write a series of post about this.
We will start with Splitter pattern. This pattern refers to the ability to have a collection of messages parts that form one message or an entity for us. This pattern gives as the ability to receive and process messages from the related messages separately. In this way all the messages that belong to that message part will be sent to the same consumer.
 
How we can use this pattern in Service Bus? Hmm, is pretty simple. We can use SessionId property of a message for this. In this way, when a client (consumer) start to receive a message with a given session id, we can specify to receive only messages with the given id. Service Bus guarantees that the messages from the given session will be received in the same order they were added.
When a client start to receive messages of a specific session, the session is automatically locked. Only that client will be able to receive messages with the given session id. This action is a transactional action. Because of this, if something will happen with the consumer (crash for example), all the messages for the specific session will be received, even if the client that consumed a part of the messages from that session crashed.
To be able to lock the messages from Service Bus to be consumed only by the client that started to receive messages with a given session id we need to the RequiresSession property of the QueueDescription or SubscriptionDescription. This is the only configuration that needs to be done on the Service Bus. On the consumer don’t forget to use the AcceptMessageSession() method to receive a reference to a session from ServiceBus. This method can be found under the abstract class MessageClientEntity that is implemented by all the crucial classes that are used to receive messages from Service Bus:
  • QueueClient
  • TopicClient
  • SubscriptionClient
  • MessageSender
  • MessageReceiver
  • MessagingFactory
Because of this we can write the same code that will be used when we use queues or topics. More information about how you can write code that can consume Service Bus Queues and Service Bus Topics can be found on the following link: www.vunvulearadu.blogspot.com/2012/08/service-bus-topic-how-we-can-migrate.html
In the next example I will get a reference to MessageSession from a queue, a topic and a MessageReceiver.
MessageReceiver
// Service Bus Queue
QueueClient queueClient =QueueClient.CreateFromConnectionString(
     myFooConnectionString, 
     "FooQueue");
MessageSession sessionFromQueue = queueClient.AcceptMessageSession();
// Service Bus Topic
SubscriptionClient subscriptionClient = SubscriptionClient.CreateFromConnectionString(
     CloudConfigurationManager.GetSetting(
          "ServiceBusConnectionString"),
          "myFooTopic",
          "seccondSubscription");
MessageSession sessionFromSubscriber = subscriptionClient.AcceptMessageSession();
// MessageReceiver
Uri serviceAddress = ServiceBusEnvironment.CreateServiceUri("sb", “myFooNamspace”, string.Empty);
MessagingFactory messagingFactory  = MessagingFactory.Create(serviceAddress, credentials);
MessageReceiver messageReceiver = messagingFactory.CreateMessageReceiver(“myFooQueueName”);
MessageSession sessionFromMessageReceiver = messageReceiver.AcceptMessageSession();
Don’t forget to activate the session support from the queue or topic of the Service Bus.
QueueDescription queueDescription = new QueueDescription("FooQueue");
queueDescription.MaxSizeInMegabytes = 5120;
queueDescription.DefaultMessageTimeToLive = new TimeSpan(0, 10, 30);
queueDescription.RequiresSession = true;
if (!namespaceManager.QueueExists("FooQueue"))
{
    namespaceManager.CreateQueue(queueDescription);
}

Once we have the reference to MessageSession we can consume messages with the same session id using Receive method of the MessageSession (as a hint: the base class of this class is MessageReceiver).
while ( true )
{
     BrokeredMessage message = session.Receive();
     ...
     message.Complete();
}
This pattern can be used when we the messages are a specific order that is important for the received. We want to be able to specify where the messages need to be sent. As an example we can imagine a hyper-market that sell a lot of things. Based on the type of the product we have different services that need to consume these messages and process this. For this case, each session can be a different product category. Another case when slitter pattern is very useful is for the case when we want to send content that is too big for a message. For this case we need to use a splitter.
Tomorrow we will continue with another pattern where Windows Azure Service Bus can be used.
Last edit: A list of all patterns that can be used with Windows Azure Service Bus, that were described by me LINK

Comments

  1. Interesting!
    There is a very good book on this subject:
    http://www.amazon.com/Enterprise-Integration-Patterns-Designing-Deploying/dp/0321200683

    ReplyDelete

Post a Comment

Popular posts from this blog

Windows Docker Containers can make WIN32 API calls, use COM and ASP.NET WebForms

After the last post , I received two interesting questions related to Docker and Windows. People were interested if we do Win32 API calls from a Docker container and if there is support for COM. WIN32 Support To test calls to WIN32 API, let’s try to populate SYSTEM_INFO class. [StructLayout(LayoutKind.Sequential)] public struct SYSTEM_INFO { public uint dwOemId; public uint dwPageSize; public uint lpMinimumApplicationAddress; public uint lpMaximumApplicationAddress; public uint dwActiveProcessorMask; public uint dwNumberOfProcessors; public uint dwProcessorType; public uint dwAllocationGranularity; public uint dwProcessorLevel; public uint dwProcessorRevision; } ... [DllImport("kernel32")] static extern void GetSystemInfo(ref SYSTEM_INFO pSI); ... SYSTEM_INFO pSI = new SYSTEM_INFO(

Azure AD and AWS Cognito side-by-side

In the last few weeks, I was involved in multiple opportunities on Microsoft Azure and Amazon, where we had to analyse AWS Cognito, Azure AD and other solutions that are available on the market. I decided to consolidate in one post all features and differences that I identified for both of them that we should need to take into account. Take into account that Azure AD is an identity and access management services well integrated with Microsoft stack. In comparison, AWS Cognito is just a user sign-up, sign-in and access control and nothing more. The focus is not on the main features, is more on small things that can make a difference when you want to decide where we want to store and manage our users.  This information might be useful in the future when we need to decide where we want to keep and manage our users.  Feature Azure AD (B2C, B2C) AWS Cognito Access token lifetime Default 1h – the value is configurable 1h – cannot be modified

What to do when you hit the throughput limits of Azure Storage (Blobs)

In this post we will talk about how we can detect when we hit a throughput limit of Azure Storage and what we can do in that moment. Context If we take a look on Scalability Targets of Azure Storage ( https://azure.microsoft.com/en-us/documentation/articles/storage-scalability-targets/ ) we will observe that the limits are prety high. But, based on our business logic we can end up at this limits. If you create a system that is hitted by a high number of device, you can hit easily the total number of requests rate that can be done on a Storage Account. This limits on Azure is 20.000 IOPS (entities or messages per second) where (and this is very important) the size of the request is 1KB. Normally, if you make a load tests where 20.000 clients will hit different blobs storages from the same Azure Storage Account, this limits can be reached. How we can detect this problem? From client, we can detect that this limits was reached based on the HTTP error code that is returned by HTTP