How to Use Service Bus Topics/Subscriptions
This guide will show you how to use Service Bus topics and subscriptions. The samples are written in C# and use the .NET API. The scenarios covered include creating topics and subscriptions, creating subscription filters, sending messages to a topic, receiving messages from a subscription, and deleting topics and subscriptions. For more information on topics and subscriptions, see the Next Steps section.
Table of Contents
What are Service Bus Topics and Subscriptions
Service Bus topics and subscriptions support a publish/subscribe messaging communication model. When using topics and subscriptions, components of a distributed application do not communicate directly with each other, they instead exchange messages via a topic, which acts as an intermediary.
In contrast to Service Bus queues, where each message is processed by a single consumer, topics and subscriptions provide a one-to-many form of communication, using a publish/subscribe pattern. It is possible to register multiple subscriptions to a topic. When a message is sent to a topic, it is then made available to each subscription to handle/process independently.
A topic subscription resembles a virtual queue that receives copies of the messages that were sent to the topic. You can optionally register filter rules for a topic on a per-subscription basis, which allows you to filter/restrict which messages to a topic are received by which topic subscriptions.
Service Bus topics and subscriptions enable you to scale to process a very large number of messages across a very large number of users and applications.
Create a Service Namespace
To begin using Service Bus topics and subscriptions in Windows Azure, you must first create a service namespace. A service namespace provides a scoping container for addressing Service Bus resources within your application.
To create a service namespace:
-
Log on to the Windows Azure Management Portal.
-
In the lower left navigation pane of the Management Portal, click Service Bus, Access Control & Caching.
-
In the upper left pane of the Management Portal, click the Service Bus node, and then click the New button.
-
In the Create a new Service Namespace dialog, enter a Namespace, and then to make sure that it is unique, click the Check Availability button.
-
After making sure the Namespace name is available, choose the country or region in which your namespace should be hosted (make sure you use the same Country/Region in which you are deploying your compute resources), and then click the Create Namespace button.
The namespace you created will then appear in the Management Portal and takes a moment to activate. Wait until the status is Active before moving on.
Obtain the Default Management Credentials for the Namespace
In order to perform management operations, such as creating a topic or subscription, on the new namespace, you need to obtain the management credentials for the namespace.
-
In the left navigation pane, click the Service Bus node to display the list of available namespaces:
-
Select the namespace you just created from the list shown:
-
The right-hand Properties pane will list the properties for the new namespace:
-
The Default Key is hidden. Click the View button to display the security credentials:
-
Make a note of the Default Issuer and the Default Key as you will use this information below to perform operations with the namespace.
Configure Your Application to Use Service Bus
When you create an application that uses Service Bus, you will need to add a reference to the Service Bus assembly and include the corresponding namespaces.
Add a Reference to the Service Bus Assembly
-
In Visual Studio's Solution Explorer, right-click References, and then click Add Reference.
-
In the Browse tab, go to C:\Program Files\Windows Azure SDK\v1.6\ServiceBus\ref\ and add a Microsoft.ServiceBus.dll reference.
Import the Service Bus Namespaces
Add the following to the top of any C# file where you want to use Service Bus topics and subscriptions:
using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;
You are now ready to write code against Service Bus.
How to Create a Security Token Provider
Service Bus uses a claims-based security model implemented using the Windows Azure Access Control Service (ACS). The TokenProvider class provides a security token provider with built-in factory methods. The code below creates a SharedSecretTokenProvider to hold the shared secret credentials and handle the acquisition of the appropriate tokens from the Access Control Service:
string issuer = "<obtained from portal>";
string key = "<obtained from portal>";
TokenProvider tP = TokenProvider.CreateSharedSecretTokenProvider(issuer, key);
Use the issuer and key values retrieved from the Management Portal as described in the previous section.
How to Create a Topic
Management operations for Service Bus topics and subscriptions can be performed via the NamespaceManager class. A NamespaceManager object is constructed with the base address of a Service Bus namespace and an appropriate token provider that has permissions to manage it. The base address of a Service Bus namespace is a URI of the form "sb://.servicebus.windows.net". The ServiceBusEnvironment class provides the CreateServiceUri helper method to assist the creation of these URIs.
The NamespaceManager class provides methods to create, enumerate, and delete topics. The example below shows how a NamespaceManager can be used to create a topic named "TestTopic" within a "HowToSample" service namespace:
string issuer = "<obtained from portal>";
string key = "<obtained from portal>";
TokenProvider tP = TokenProvider.CreateSharedSecretTokenProvider(issuer, key);
// Retrieve URI of our "HowToSample" service namespace (created via the portal)
Uri uri = ServiceBusEnvironment.CreateServiceUri("sb", "HowToSample", string.Empty);
// Create NamespaceManager for our "HowToSample" service namespace
NamespaceManager namespaceManager = new NamespaceManager(uri, tP);
// Create a new Topic named "TestQueue"
namespaceManager.CreateTopic("TestTopic");
There are overloads of the CreateTopic method that allow properties of the topic to be tuned, for example, to set the default time-to-live to be applied to messages sent to the topic. These settings are applied by using the TopicDescription class. The following example shows how to create a topic named "TestTopic" with a maximum size of 5 GB and a default message time-to-live of 1 minute.
string issuer = "<obtained from portal>";
string key = "<obtained from portal>";
TokenProvider tP = TokenProvider.CreateSharedSecretTokenProvider(issuer, key);
// Retrieve URI of our "HowToSample" service namespace (created via the portal)
Uri uri = ServiceBusEnvironment.CreateServiceUri("sb", "HowToSample", string.Empty);
// Create NamespaceManager for our "HowToSample" service namespace
NamespaceManager namespaceManager = new NamespaceManager(uri, tP);
// Configure Topic Settings
TopicDescription td = new TopicDescription("TestTopic");
td.MaxSizeInMegabytes = 5120;
td.DefaultMessageTimeToLive = new TimeSpan(0, 1, 0);
// Create a new Topic with custom settings
namespaceManager.CreateTopic(td);
Note: You can use the TopicExists method on NamespaceManager objects to check if a topic with a specified name already exists within a service namespace.
How to Create Subscriptions
Topic subscriptions are also created with the NamespaceManager class. Subscriptions are named and can have an optional filter that restricts the set of messages passed to the subscription's virtual queue.
Create a Subscription with the default (MatchAll) Filter
The MatchAll filter is the default filter that is used if no filter is specified when a new subscription is created. When the MatchAll filter is used, all messages published to the topic are placed in the subscription's virtual queue. The following example creates a subscription named "AllMessages" and uses the default MatchAll filter.
string issuer = "<obtained from portal>";
string key = "<obtained from portal>";
// TokenProvider and URI of our "HowToSample" service namespace
TokenProvider tP = TokenProvider.CreateSharedSecretTokenProvider(issuer, key);
Uri uri = ServiceBusEnvironment.CreateServiceUri("sb", "HowToSample", string.Empty);
// Create NamespaceManager for the "HowToSample" service namespace
NamespaceManager namespaceManager = new NamespaceManager(uri, tP);
// Create a new "AllMessages" subscription on our "TestTopic"
namespaceManager.CreateSubscription("TestTopic", "AllMessages");
Create Subscriptions with Filters
You can also setup filters that allow you to scope which messages sent to a topic should show up within a specific topic subscription.
The most flexible type of filter supported by subscriptions is the SqlFilter, which implements a subset of SQL92. SQL filters operate on the properties of the messages that are published to the topic. For more details about the expressions that can be used with a SQL filter, review the SqlFilter.SqlExpression syntax.
The example below creates a subscription named "HighMessages" with a SqlFilter that only selects messages that have a custom MessageNumber property greater than 3:
// Create a "HighMessages" filtered subscription
SqlFilter highMessages = new SqlFilter("MessageNumber > 3");
namespaceManager.CreateSubscription("TestTopic", "HighMessages", highMessages);
Similarly, the following example creates a subscription named "LowMessages" with a SqlFilter that only selects messages that have a MessageNumber property less than or equal to 3:
// Create a "LowMessages" filtered subscription
SqlFilter lowMessages = new SqlFilter("MessageNumber <= 3");
namespaceManager.CreateSubscription("TestTopic", "LowMessages", lowMessages);
When a message is now sent to the "TestTopic", it will always be delivered to receivers subscribed to the "AllMessages" topic subscription, and selectively delivered to receivers subscribed to the "HighMessages" and "LowMessages" topic subscriptions (depending upon the message content).
How to Send Messages to a Topic
To send a message to a Service Bus topic, your application will obtain a MessageSender object. Like NamespaceManager objects, this object is created from the base URI of the service namespace and the appropriate token provider.
The below code demonstrates how to retrieve a MessageSender object for the "TestTopic" topic we created above within our "HowToSample" service namespace:
string issuer = "<obtained from portal>";
string key = "<obtained from portal>";
// URI address and token for our "HowToSample" namespace
TokenProvider tP = TokenProvider.CreateSharedSecretTokenProvider(issuer, key);
Uri uri = ServiceBusEnvironment.CreateServiceUri("sb", "HowToSample", string.Empty);
// Retrieve MessageSender for the "TestTopic" within our "HowToSample" namespace
MessagingFactory factory = MessagingFactory.Create(uri, tP);
MessageSender testTopic = factory.CreateMessageSender("TestTopic");
Messages sent to Service Bus Topics are instances of the BrokeredMessage class. BrokeredMessage objects have a set of standard properties (such as Label and TimeToLive), a dictionary that is used to hold custom application specific properties, and a body of arbitrary application data. An application can set the body of the message by passing any serializable object into the constructor of the BrokeredMessage, and the appropriate DataContractSerializer will then be used to serialize the object. Alternatively, a System.IO.Stream can be provided.
The following example demonstrates how to send five test messages to the "TestTopic" MessageSender we obtained in the code snippet above. Note how the MessageNumber property value of each message varies on the iteration of the loop (this will determine which subscriptions receive it):
for (int i=0; i<5; i++)
{
// Create message, passing a string message for the body
BrokeredMessage message = new BrokeredMessage("Test message " + i);
// Set additional custom app-specific property
message.Properties["MessageNumber"] = i;
// Send message to the topic
testTopic.Send(message);
}
Service Bus topics support a maximum message size of 256 MB (the header, which includes the standard and custom application properties, can have a maximum size of 64 MB). There is no limit on the number of messages held in a topic but there is a cap on the total size of the messages held by a topic. This queue size is defined at creation time, with an upper limit of 5 GB.
How to Receive Messages from a Subscription
The simplest way to receive messages from a subscription is to use a MessageReceiver object. MessageReceiver objects can work in two different modes: ReceiveAndDelete and PeekLock.
When using the ReceiveAndDelete mode, receive is a single-shot operation - that is, when Service Bus receives a read request for a message in a subscription, it marks the message as being consumed and returns it to the application. ReceiveAndDelete mode is the simplest model and works best for scenarios in which an application can tolerate not processing a message in the event of a failure. To understand this, consider a scenario in which the consumer issues the receive request and then crashes before processing it. Because Service Bus will have marked the message as being consumed, then when the application restarts and begins consuming messages again, it will have missed the message that was consumed prior to the crash.
In PeekLock mode (which is the default mode), receive becomes a two stage operation which makes it possible to support applications that cannot tolerate missing messages. When Service Bus receives a request, it finds the next message to be consumed, locks it to prevent other consumers receiving it, and then returns it to the application. After the application finishes processing the message (or stores it reliably for future processing), it completes the second stage of the receive process by calling Complete on the received message. When Service Bus sees the Complete call, it will mark the message as being consumed and remove it from the subscription.
The example below demonstrates how messages can be received and processed using PeekLock mode (the default mode). The example below does an infinite loop and processes messages as they arrive to our "HighMessages" subscription. Note that the path to our "HighMessages" subscription is supplied in the form "<topic path>/subscriptions/<subscription name>".
string issuer = "<obtained from portal>";
string key = "<obtained from portal>";
// URI address and token for our "HowToSample" namespace
TokenProvider tP = TokenProvider.CreateSharedSecretTokenProvider(issuer, key);
Uri uri = ServiceBusEnvironment.CreateServiceUri("sb", "HowToSample", string.Empty);
// Retrieve MessageReceiver for the "HighMessages" subscription
MessagingFactory factory = MessagingFactory.Create(uri, tP);
MessageReceiver highMessages =
factory.CreateMessageReceiver("TestTopic/subscriptions/HighMessages");
// Continuously process messages received from the "HighMessages" subscription
while (true)
{
BrokeredMessage message = highMessages.Receive();
if (message != null)
{
try
{
Console.WriteLine("Body: " + message.GetBody<string>());
Console.WriteLine("MessageID: " + message.MessageId);
Console.WriteLine("MessageNumber: " + message.Properties["MessageNumber"]);
// Remove message from subscription
message.Complete();
}
catch (Exception)
{
// Indicate a problem, unlock message in subscription
message.Abandon();
}
}
}
How to Handle Application Crashes and Unreadable Messages
Service Bus provides functionality to help you gracefully recover from errors in your application or difficulties processing a message. If a receiver application is unable to process the message for some reason, then it can call the Abandon method on the received message (instead of the Complete method). This will cause Service Bus to unlock the message within the subscription and make it available to be received again, either by the same consuming application or by another consuming application.
There is also a timeout associated with a message locked within the subscription, and if the application fails to process the message before the lock timeout expires (e.g., if the application crashes), then Service Bus will unlock the message automatically and make it available to be received again.
In the event that the application crashes after processing the message but before the Complete request is issued, then the message will be redelivered to the application when it restarts. This is often called At Least Once Processing, that is, each message will be processed at least once but in certain situations the same message may be redelivered. If the scenario cannot tolerate duplicate processing, then application developers should add additional logic to their application to handle duplicate message delivery. This is often achieved using the MessageId property of the message, which will remain constant across delivery attempts.
How to Delete Topics and Subscriptions
The example below demonstrates how to delete the topic named TestTopic from the HowToSample service namespace:
string issuer = "<obtained from portal>";
string key = "<obtained from portal>";
// TokenProvider and URI of our "HowToSample" service namespace
TokenProvider tP = TokenProvider.CreateSharedSecretTokenProvider(issuer, key);
Uri uri = ServiceBusEnvironment.CreateServiceUri("sb", "HowToSample", string.Empty);
// Create NamespaceManager for the "HowToSample" service namespace
NamespaceManager namespaceManager = new NamespaceManager(uri, tP);
// Delete Topic
namespaceManager.DeleteTopic("TestTopic");
Deleting a topic will also delete any subscriptions that are registered with the topic. Subscriptions can also be deleted independently. The following code demonstrates how to delete a subscription named HighMessages from the TestTopic topic:
namespaceManager.DeleteSubscription("TestTopic", "HighMessages");
Next Steps
Now that you've learned the basics of Service Bus topics, follow these links to learn more.