Using Service Bus Topics allows us to communicate with loosely coupled components or systems using publish/subscribe model. The difference between Azure Queues and Topics is that using Topics the message can be easily routed to the selected subscriber by using filters while using only the queues this is hard to achieve without adding extra logic to the subscriber's adapters or connectors.

When publishing a message we can choose to route it to all subscribers or to apply filters for each subscription resulting in each subscriber receiving messages that are addressed to him. With Service Bus topics we can easily scale distributed applications communicating with each other within or across multiple networks. 

 In this article I will show you how to build and test Service Bus topic on your local computer. In our example we will simulate sending messages from the web, mobile and service application to the Service Bus Topic. These messages will be then routed to relevant subscriptions based on defined filters we assigned for each of them. Subscription for messages from the web application will be using multiple auto scalable worker roles to process the business logic. Same will apply for service messages. If we don’t expect a lot of traffic coming from mobile application, we can then use single worker role.

Autoscaling worker roles can be performed using Enterprise Library 5.0 – Autoscaling Application Block (aka WASABi). This will ensure that appropriate number of worker roles will be automatically started when traffic increases and stopped if the traffic will ease.

Diagram below shows the high level architecture




In order to test Service Bus Topics locally, we need to install “Service Bus 1.0 for Windows Server” (runs on Win7 as well). After installation we also need to configure it in order to set-up a farm on the local machine. Setting up the farm will allow us to create Service Bus namespace that we will use to communicate with the Topics.


In order to establish connection to the Service Bus we will use following code

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

  //use this setting when deploying to Windows Azure
  <add key="Microsoft.ServiceBus.ConnectionString" value="Endpoint=sb://[your namespace].servicebus.windows.net;SharedSecretIssuer=owner;SharedSecretValue=[your secret]" />
 
 public static string getLocalServiceBusConnectionString()
 {
    var ServerFQDN = System.Net.Dns.GetHostEntry(string.Empty).HostName;
    var ServiceNamespace = "ServiceBusDefaultNamespace";
    var HttpPort = 9355;
    var TcpPort = 9354;
 
    var connBuilder = new ServiceBusConnectionStringBuilder();
    connBuilder.ManagementPort = HttpPort;
    connBuilder.RuntimePort = TcpPort;
    connBuilder.Endpoints.Add(new UriBuilder() { Scheme = "sb", Host = ServerFQDN, Path = ServiceNamespace }.Uri);
    connBuilder.StsEndpoints.Add(new UriBuilder() { Scheme = "https", Host = ServerFQDN, Port = HttpPort, Path = ServiceNamespace }.Uri);
 
    return connBuilder.ToString();
 }


Next, we need to create Topic, subscriptions and filters. In our case we want to simulate multiple subscribers receiving incoming messages from different sources; web, mobile and service application.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

 public void CreateServiceBusTopicAndSubscriptions(NamespaceManager namespaceManager)
 {
    #region Configure and create Service Bus Topic
    var serviceBusTestTopic = new TopicDescription(TopicName);
    serviceBusTestTopic.MaxSizeInMegabytes = 5120;
    serviceBusTestTopic.DefaultMessageTimeToLive = new TimeSpan(0, 1, 0);
 
    if (!namespaceManager.TopicExists(TopicName))
    {
        namespaceManager.CreateTopic(serviceBusTestTopic);
    }
    #endregion
 
    #region Create filters and subsctiptions
    //create filters
    var messagesFilter_Web = new SqlFilter("MessageOrigin = 'Web'");
    var messagesFilter_Mobile = new SqlFilter("MessageOrigin = 'Mobile'");
    var messagesFilter_Service = new SqlFilter("MessageOrigin = 'Service'");
 
    if (!namespaceManager.SubscriptionExists(TopicName, "WebMessages"))
    {
        namespaceManager.CreateSubscription(TopicName, "WebMessages", messagesFilter_Web);
    }
 
    if (!namespaceManager.SubscriptionExists(TopicName, "MobileMessages"))
    {
        namespaceManager.CreateSubscription(TopicName, "MobileMessages", messagesFilter_Mobile);
    }
 
    if (!namespaceManager.SubscriptionExists(TopicName, "WCfServiceMessages"))
    {
        namespaceManager.CreateSubscription(TopicName, "WCfServiceMessages", messagesFilter_Service);
    }
    #endregion
}

Finally, we want to simulate receiving messages. We will do it by starting 3 subscriptions and waiting for incoming messages separately for each of them. After an message arrives, we will display it's source type and message body.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

public override void Run()
{
 Parallel.ForEach(SubscriptionClients, currentSubscrtiption =>
 {
    while (!IsStopped)
    {
        #region Receive messages
        try
        {
            // Receive the message
            var receivedMessage = currentSubscrtiption.Receive();
 
            if (receivedMessage != null)
            {
                var messageFrom = receivedMessage.Properties["MessageOrigin"].ToString();
 
                switch (messageFrom)
                {
                    case "Web":
                        //send it to web processing logic
 
                        break;
                    case "Mobile":
                        //send it to mobile processing logic
 
                        break;
                    case "Service":
                        //send it to service processing logic
 
                        break;
                    default:
                        break;
                }
 
                // Process the message
                Trace.WriteLine(Environment.NewLine + "--------------------------" + Environment.NewLine);
                Trace.WriteLine(string.Format("{0} message content: {1}", messageFrom, receivedMessage.GetBody<string>()));
 
                receivedMessage.Complete();
            }
        }
        catch (MessagingException e)
        {
            if (!e.IsTransient)
            {
                Trace.WriteLine(e.Message);
                throw;
            }
 
            Thread.Sleep(10000);
        }
        catch (OperationCanceledException e)
        {
            if (!IsStopped)
            {
                Trace.WriteLine(e.Message);
                throw;
            }
        }
        #endregion
    }
});
}

The outcome of our test is as follows. Please note that we are differentiating messages by setting up message.Properties["MessageOrigin"] property string.

hits counter