Levaraging Azure Service Bus with C#

Fork me on GitHubAzure Service Bus
Microsoft Azure provides offers Azure Service Bus as a means of leveraging the Decoupled Middleware design pattern, among other things in your application. This post outlines a step-by-step guide to implementation, assuming that you have already established an Azure account, and have initialised an associated Service Bus.

Start with the Abstraction

This library abstracts the concept of a Service Bus to a level that is not restricted to MS Azure alone. Both ServiceBus and ServiceBusAdapter classes offer any Service Bus implementation the means to establish associated implementations in this library. Having said that, this library explicitly implements concrete classes that are specific to MS Azure Service Bus.

The MS Azure Service Bus

The MSAzureServiceBus class provides a succinct means of interfacing with an MS Azure Service Bus, consuming messages as they arrive. Upon initialisation, MSAzureServiceBus requires that a delegate be established that determines appropriate behaviour to invoke in the event of inbound new messages. Very simply, this functionality is exposed as follows:

Incoming Message-handling

public override event EventHandler<MessageReceivedEventArgs<BrokeredMessage>> MessageReceived;

private void OnMessageReceived(MessageReceivedEventArgs<BrokeredMessage> e) {
    var handler = MessageReceived;
    if (handler != null) handler(this, e);
}

Duplicate Message-handling

Similarly, behaviour applying to duplicate messages, that is, messages that have already been processed by MSAzureServiceBus, can also be established:

public override event EventHandler<MessageReceivedEventArgs<BrokeredMessage>> DuplicateMessageReceived;

private void OnDuplicateMessageReceived(MessageReceivedEventArgs<BrokeredMessage> e) {
    var handler = DuplicateMessageReceived;
    if (handler != null) handler(this, e);
}

Receiving Messages Explicitly

Bootstrapping delegates aside, MSAzureServiceBus provides a method designed to retrieve the next available message from the MS Service Bus. This method may be invoked on demand, or as part of a continuous loop, polling the MS Service Bus and consuming new messages immediately after they become available.

        protected override void ReceiveNextMessage(string publisherName, TimeSpan timeout, bool autoAcknowledge) {
            var message = serviceBusAdapter.ReceiveNextMessage(publisherName, timeout, autoAcknowledge);
            if (message == null) return;

            var isValidMessage = messageValidator.ValidateMessageId(message.MessageId);

            if (isValidMessage) {
                messageValidator.AddMessageIdToCache(message.MessageId);
                OnMessageReceived(new BrokeredMessageReceivedEventArgs(message));
            }
            else {
                OnMessageReceived(new BrokeredMessageReceivedEventArgs(message));
            }
        }

The MS Azure ServiceBus Adapter

The MSAzureServiceBusAdapter class is a Bridge that encapsulate the underlying mechanisms required to establish a connection to, send, and receive messages to and from MS Azure Service Bus. Let’s consider the functionality in that order:

Initialising a Connection

Firstly, we must establish a NamespaceManager based on an appropriate connection-string associated with out MS Azure Service Bus account:

            var connectionString = CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
            _namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);

Now we return a reference to a desired Topic, creating the Topic if it does not already exist:

                _topic = !_namespaceManager.TopicExists(topicName) ?
                _namespaceManager.CreateTopic(topicName) : _namespaceManager.GetTopic(topicName);

Lastly, we create a Subscription to the Topic, if one does not already exist:

                if (!_namespaceManager.SubscriptionExists(_topic.Path, subscriptionName))
                _namespaceManager.CreateSubscription(_topic.Path, subscriptionName);

The Complete Listing

        public override void Initialise(string topicName) {
            var connectionString = CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
            _namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);

            _topic = !_namespaceManager.TopicExists(topicName) ?
                _namespaceManager.CreateTopic(topicName) : _namespaceManager.GetTopic(topicName);

            if (!_namespaceManager.SubscriptionExists(_topic.Path, subscriptionName))
                _namespaceManager.CreateSubscription(_topic.Path, subscriptionName);

            _isInitialised = true;
        }

It’s worth noting that all methods pertaining to MSAzureServiceBusAdapter will implicitly invoke the Initialise method if a connection to MS Azure Service Bus has not already been established.

Sending Messages

This library offers the means to send messages in the form of BrokeredMessage objects to MS Azure Service Bus. Firstly, we must establish a connection, if one does not already exist:

if (!_isInitialised) Initialise(topicName);

Finally, initialise a SubscriptionClient, if one has not already been established, and simply send the message as-is, in BrokeredMessage-format:

            if (_topicClient == null)
                _topicClient = TopicClient.Create(topicName);
            _topicClient.Send(message);

The Complete Listing

        public override void SendMessage(string topicName, BrokeredMessage message) {
            if (!_isInitialised) Initialise(topicName);

            if (_topicClient == null)
                _topicClient = TopicClient.Create(topicName);
            _topicClient.Send(message);
        }

Receiving Messages

Receiving Messages
Messages are consumed from MS Azure Service Bus in a serial manner, one after the other. Once again, we must initially establish a connection, if one does not already exist:

            if (!_isInitialised)
                Initialise(topicName);

Next, we initialise a SubscriptionClient, if one has not already been established, and define a BrokeredMessage instance, the desired method return-type:

            if (_subscriptionClient == null)
                _subscriptionClient = SubscriptionClient.Create(topicName, subscriptionName);

            BrokeredMessage message = null;

Next, we return the next available message, or null, if there are no available messages:

                message = _subscriptionClient.Receive(timeout);
                if (message == null)
                    return null;

Note that this method defines an “autoAcknowledge” parameter. If true, we must explicitly acknowledge the consumption of the message:

                if (!autoAcknowledge) return message;
                message.Complete();

Finally, we return or abandon the message, depending on whether or not an Exception occurred:

            catch (Exception) {
                if (message != null) message.Abandon();
                throw;
            }
            return message;

The Complete Listing

        public override BrokeredMessage ReceiveNextMessage(string topicName, TimeSpan timeout, bool autoAcknowledge = false) {
            if (!_isInitialised)
                Initialise(topicName);

            if (_subscriptionClient == null)
                _subscriptionClient = SubscriptionClient.Create(topicName, subscriptionName);

            BrokeredMessage message = null;

            try {
                message = _subscriptionClient.Receive(timeout);
                if (message == null)
                    return null;
                if (!autoAcknowledge) return message;
                message.Complete();
            }
            catch (Exception) {
                if (message != null) message.Abandon();
                throw;
            }
            return message;
        }

A Practical Example

Let’s build a small Console Application to demonstrate the concept. Our application will interface with MS Azure Service Bus and continuously poll for messages until the application terminates:

            var serviceBus = new MSAzureServiceBus(new MSAzureServiceBusAdapter(), new MessageValidator());
            serviceBus.MessageReceived += serviceBus_MessageReceived;

            private static void serviceBus_MessageReceived(object sender, MessageReceivedEventArgs<BrokeredMessage> e) {
                Console.WriteLine(e.Message.MessageId);
            }

Message Validation

Notice the MessageValidator instance in the above code snippet. Let’s pause for a moment and consider the mechanics.

Messages contain message identifiers in GUID format. Our application retains an index that maps these identities. Incoming messages are validated by comparing the incoming message ID to those IDs stored within the index. If a match is found, the message is determined to be a duplicate, and appropriate action can be taken.

Here we can see that our inbound message IDs are stored in a simple HashSet of type String. Incidentally, we leverage a HashSet here to achieve what is known as constant complexity in terms of time. Essentially, the time taken to perform a lookup will remain constant (external factors such as garbage collection aside) regardless of HashSet size:

private readonly HashSet<string> _cache = new HashSet<string>();

public IEnumerable<string> Cache { get { return _cache; } }

Newly added messages are formatted to remove all hyphens, if any exist, so that the same standard is applied to message IDs, regardless of format:

        public void AddMessageIdToCache(string messageId) {
            _cache.Add(messageId.Replace('-', '\0'));
        }

        public bool ValidateMessageId(string messageId) {
            return _cache.Contains(messageId);
        }

Once initialised, the application will continuously poll MS Azure Service Bus until the return key is pressed:

            serviceBus.StartListening("TestTopic", new TimeSpan(0, 0, 1), true);
            Console.WriteLine("Listening to the Service Bus. Press any key to quit...");

            Console.ReadLine();
            serviceBus.StopListening();

            Console.WriteLine("Disconnecting...");

The Complete Listing

    internal class Program {
        private static void Main(string[] args) {
            var serviceBus = new MSAzureServiceBus(new MSAzureServiceBusAdapter(), new MessageValidator());
            serviceBus.MessageReceived += serviceBus_MessageReceived;

            serviceBus.StartListening("TestTopic", new TimeSpan(0, 0, 1), true);
            Console.WriteLine("Listening to the Service Bus. Press any key to quit...");

            Console.ReadLine();
            serviceBus.StopListening();

            Console.WriteLine("Disconnecting...");
        }

        private static void serviceBus_MessageReceived(object sender, MessageReceivedEventArgs<BrokeredMessage> e) {
            Console.WriteLine(e.Message.MessageId);
        }
    }

Simply add a new message to your MS Azure Service Bus instance. The application will consume the message and display the message ID on-screen.

Connect with me:

RSSGitHubTwitter
LinkedInYouTubeGoogle+

6 thoughts on “Levaraging Azure Service Bus with C#

  1. gotdaz

    Nice post.
    A couple of questions. I plan to use azure service bus within a library that others may consume via NuGet. My library will need to listen for particular messages sent on the service bus from another system. Once a message is sent, and my library acknowledges receipt – is there some way for the other system to get that acknowledgement – so it knows the message it sent was processed? My other question is, is it ok to use many short lived instances of MSAzureServiceBus so that I can ensure Start and Stop listening is always called – as opposed to a global / single instance which starts listening once (think static initialiser) but then relies on user of my library to call StopListening when application shuts down etc..

    Reply
    1. Paul Mooney Post author

      Thanks. There is no implicit way that the first system will receive an acknowledgement. I suggest that you create a second Topic, or other means of linking both systems, and explicitly send an acknowledgement message from the second system to the first, after the initial message is processed. Of course if you apply this solution, you then face the Byzantine Generals’ Problem: https://en.wikipedia.org/wiki/Byzantine_fault_tolerance
      Yes, it’s absolutely fine, and in fact recommended to invoke several instances of MSAzureServiceBus. This is the essence of Microservice design.

      Reply
      1. gotdaz

        Thanks Paul, that makes sense. I had a quick read of the BFT problem and you are right, that could well be problematic for me.

        The other main headache I am wrestling with is that, given that I would like my library to essentially act as a proxy for a transaction that is being processed within a remote process (hosted on Azure) – when the user signals to commit() the transaction via my library, I will need to implement some sort of 2 phase commit protocol, and ASB looks like one of the only communication methods I have between my library and the remote process in Azure (unless you count polling a web service) and I am not sure if ASB is going to be enough to implement such a protocol!

        Atleast I am learning a lot in the process 🙂

      2. Paul Mooney Post author

        It seems to me that the downstream system (the final system to which the data will end up) should be responsible for committing data and ensuring referential integrity, and that your library should simply facilitate the transition of data from point A to point B. Otherwise, All you can do is persist the first segment of data, wait for a reply (as per my previous comment) from the downstream system confirming receipt of the data, and then, assuming success, persist all remaining data segments. Again, this introduces the Byzantine General’s problem.

  2. gotdaz

    Ok – I think I follow you.

    Could you elaborate a bit more on where you see Byzantine General’s problem applying? My own understanding of where it applies in this situation is due to the fact that the client (upstream) always assumes that the transaction fails unless it gets a successful acknowledgment from the server (downstream) to say that it has been successfully committed. If the server commits the transaction, but cannot then send the acknowledgement to the client to say it has successfully committed, then the client could end up assuming that due to the missing acknowledgment, the transaction was not applied.. The same issue occurs with a missing acknowledgement in the result also of a failure to commit – except in that scenario the assumption by the client that it failed to commit would actually be ok!

    It was this precise reason that I was looking at the System.Transactions namespace to see if I could hook into the transaction 2 phase commit process on the server (downstream) in order to only allow the transaction on the server to commit – if a successful acknowledgment can be sent to the client (upstream) during the PREPARE phase of the 2 phase commit. But this whole line of thinking is starting the fee like a god awful hack so i am very hesitant to pursue it any further haha.

    Thanks for your help, I know this is somewhat beyond the scope of your article so if you don’t reply, I can forgive you.

    Reply
    1. Paul Mooney Post author

      Yes, essentially the problem boils down to how reliable the channels of communications are, and how much you can trust that downstream messages are received intact.
      There is arguably no complete solution to this problem. The concept of a blockchain has been introduced as a potential solution, however, in your case I might simply suggest implementing a more blunt approach. Simply apply a timeout, that when reached, causes System A to assume that the message did not reach System B. System A then re-issues the message, regardless of whether or not System B is still busy processing the initial message. System B, on receipt of the re-issued message, simply ignores it. Such a design requires that duplicate messages be flagged in some manner, and tracked by downstream systems.

      Reply

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out / Change )

Twitter picture

You are commenting using your Twitter account. Log Out / Change )

Facebook photo

You are commenting using your Facebook account. Log Out / Change )

Google+ photo

You are commenting using your Google+ account. Log Out / Change )

Connecting to %s