How to Use Service Bus Topics/Subscriptions

This article describes how to use Service Bus topics and subscriptions from Ruby applications. 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.

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; instead they exchange messages via a topic, which acts as an intermediary.

TopicConcepts

In contrast with Service Bus queues, in which 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 subscription to a topic 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 enables you to filter or restrict which messages to a topic are received by which topic subscriptions.

Service Bus topics and subscriptions enable you to scale and process a very large number of messages across many users and applications.

Create a namespace

To begin using Service Bus topics and subscriptions in Azure, you must first create a service namespace. A namespace provides a scoping container for addressing Service Bus resources within your application.

To create a namespace:

  1. Log on to the Azure portal.
  2. In the left navigation pane of the portal, click New, then click Enterprise Integration, and then click Service Bus.
  3. In the Create namespace dialog, enter a namespace name. The system immediately checks to see if the name is available.
  4. After making sure the namespace name is available, choose the pricing tier (Basic, Standard, or Premium).
  5. In the Subscription field, choose an Azure subscription in which to create the namespace.
  6. In the Resource group field, choose an existing resource group in which the namespace will live, or create a new one.
  7. In Location, choose the country or region in which your namespace should be hosted.

    Create namespace

  8. Click the Create button. The system now creates your namespace and enables it. You might have to wait several minutes as the system provisions resources for your account.

Obtain the credentials

  1. In the list of namespaces, click the newly created namespace name.
  2. In the Service Bus namespace blade, click Shared access policies.
  3. In the Shared access policies blade, click RootManageSharedAccessKey.

    connection-info

  4. In the Policy: RootManageSharedAccessKey blade, click the copy button next to Connection string–primary key, to copy the connection string to your clipboard for later use.

    connection-string

Create a service namespace

To begin using Service Bus queues in Azure, you must first create a namespace. A namespace provides a scoping container for addressing Service Bus resources within your application.

To create a namespace:

  1. Log on to the Azure portal.
  2. In the left navigation pane of the portal, click New, then click Enterprise Integration, and then click Service Bus.
  3. In the Create namespace dialog, enter a namespace name. The system immediately checks to see if the name is available.
  4. After making sure the namespace name is available, choose the pricing tier (Basic, Standard, or Premium).
  5. In the Subscription field, choose an Azure subscription in which to create the namespace.
  6. In the Resource group field, choose an existing resource group in which the namespace will live, or create a new one.
  7. In Location, choose the country or region in which your namespace should be hosted.

    Create namespace

  8. Click Create. The system now creates your namespace and enables it. You might have to wait several minutes as the system provisions resources for your account.

Obtain the management credentials

  1. In the list of namespaces, click the newly created namespace name.
  2. In the namespace blade, click Shared access policies.
  3. In the Shared access policies blade, click RootManageSharedAccessKey.

    connection-info

  4. In the Policy: RootManageSharedAccessKey blade, click the copy button next to Connection string–primary key, to copy the connection string to your clipboard for later use. Paste this value into Notepad or some other temporary location.

    connection-string

  5. Repeat the previous step, copying and pasting the value of Primary key to a temporary location for later use.

Create a Ruby application

For instructions, see Create a Ruby Application on Azure.

Configure Your application to Use Service Bus

To use Service Bus, download and use the Azure Ruby package, which includes a set of convenience libraries that communicate with the storage REST services.

Use RubyGems to obtain the package

  1. Use a command-line interface such as PowerShell (Windows), Terminal (Mac), or Bash (Unix).
  2. Type "gem install azure" in the command window to install the gem and dependencies.

Import the package

Using your favorite text editor, add the following to the top of the Ruby file in which you intend to use storage:

require "azure"

Set up a Service Bus connection

The Azure module reads the environment variables AZURE_SERVICEBUS_NAMESPACE and AZURE_SERVICEBUS_ACCESS_KEY for information required to connect to your namespace. If these environment variables are not set, you must specify the namespace information before using Azure::ServiceBusService with the following code:

Azure.config.sb_namespace = "<your azure service bus namespace>"
Azure.config.sb_access_key = "<your azure service bus access key>"

Set the namespace value to the value you created rather than the entire URL. For example, use "yourexamplenamespace", not "yourexamplenamespace.servicebus.windows.net".

Create a topic

The Azure::ServiceBusService object enables you to work with topics. The following code creates an Azure::ServiceBusService object. To create a topic, use the create_topic() method. The following example creates a topic or prints out the errors if there are any.

azure_service_bus_service = Azure::ServiceBusService.new
begin
  topic = azure_service_bus_service.create_queue("test-topic")
rescue
  puts $!
end

You can also pass a Azure::ServiceBus::Topic object with additional options, which allow you to override default topic settings such as message time to live or maximum queue size. The following example shows setting the maximum queue size to 5GB and time to live to 1 minute:

topic = Azure::ServiceBus::Topic.new("test-topic")
topic.max_size_in_megabytes = 5120
topic.default_message_time_to_live = "PT1M"

topic = azure_service_bus_service.create_topic(topic)

Create subscriptions

Topic subscriptions are also created with the Azure::ServiceBusService object. Subscriptions are named and can have an optional filter that restricts the set of messages delivered to the subscription's virtual queue.

Subscriptions are persistent and will continue to exist until either they, or the topic they are associated with, are deleted. If your application contains logic to create a subscription, it should first check if the subscription already exists by using the getSubscription method.

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 "all-messages" and uses the default MatchAll filter.

subscription = azure_service_bus_service.create_subscription("test-topic", "all-messages")

Create subscriptions with filters

You can also define filters that enable you to specify which messages sent to a topic should show up within a specific subscription.

The most flexible type of filter supported by subscriptions is the Azure::ServiceBus::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 syntax.

You can add filters to a subscription by using the create_rule() method of the Azure::ServiceBusService object. This method enables you to add new filters to an existing subscription.

Since the default filter is applied automatically to all new subscriptions, you must first remove the default filter, or the MatchAll will override any other filters you may specify. You can remove the default rule by using the delete_rule() method on the Azure::ServiceBusService object.

The following example creates a subscription named "high-messages" with a Azure::ServiceBus::SqlFilter that only selects messages that have a custom message_number property greater than 3:

subscription = azure_service_bus_service.create_subscription("test-topic", "high-messages")
azure_service_bus_service.delete_rule("test-topic", "high-messages", "$Default")

rule = Azure::ServiceBus::Rule.new("high-messages-rule")
rule.topic = "test-topic"
rule.subscription = "high-messages"
rule.filter = Azure::ServiceBus::SqlFilter.new({
  :sql_expression => "message_number > 3" })
rule = azure_service_bus_service.create_rule(rule)

Similarly, the following example creates a subscription named "low-messages" with a Azure::ServiceBus::SqlFilter that only selects messages that have a message_number property less than or equal to 3:

subscription = azure_service_bus_service.create_subscription("test-topic", "low-messages")
azure_service_bus_service.delete_rule("test-topic", "low-messages", "$Default")

rule = Azure::ServiceBus::Rule.new("low-messages-rule")
rule.topic = "test-topic"
rule.subscription = "low-messages"
rule.filter = Azure::ServiceBus::SqlFilter.new({
  :sql_expression => "message_number <= 3" })
rule = azure_service_bus_service.create_rule(rule)

When a message is now sent to "test-topic", it is always be delivered to receivers subscribed to the "all-messages" topic subscription, and selectively delivered to receivers subscribed to the "high-messages" and "low-messages" topic subscriptions (depending upon the message content).

Send messages to a topic

To send a message to a Service Bus topic, your application must use the send_topic_message() method on the Azure::ServiceBusService object. Messages sent to Service Bus topics are instances of the Azure::ServiceBus::BrokeredMessage objects. Azure::ServiceBus::BrokeredMessage objects have a set of standard properties (such as label and time_to_live), a dictionary that is used to hold custom application-specific properties, and a body of string data. An application can set the body of the message by passing a string value to the send_topic_message() method and any required standard properties will be populated by default values.

The following example demonstrates how to send five test messages to "test-topic". Note that the message_number custom property value of each message varies on the iteration of the loop (this determines which subscription receives it):

5.times do |i|
  message = Azure::ServiceBus::BrokeredMessage.new("test message " + i,
    { :message_number => i })
  azure_service_bus_service.send_topic_message("test-topic", message)
end

Service Bus topics support a maximum message size of 256 KB in the Standard tier and 1 MB in the Premium tier. The header, which includes the standard and custom application properties, can have a maximum size of 64 KB. 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 topic size is defined at creation time, with an upper limit of 5 GB.

Receive messages from a subscription

Messages are received from a subscription using the receive_subscription_message() method on the Azure::ServiceBusService object. By default, messages are read(peak) and locked without deleting it from the subscription. You can read and delete the message from the subscription by setting the peek_lock option to false.

The default behavior makes the reading and deleting a two-stage operation, which also 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 delete_subscription_message() method and providing the message to be deleted as a parameter. The delete_subscription_message() method will mark the message as being consumed and remove it from the subscription.

If the :peek_lock parameter is set to false, reading and deleting the message becomes 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.

The following example demonstrates how messages can be received and processed using receive_subscription_message(). The example first receives and deletes a message from the "low-messages" subscription by using :peek_lock set to false, then it receives another message from the "high-messages" and then deletes the message using delete_subscription_message():

message = azure_service_bus_service.receive_subscription_message(
  "test-topic", "low-messages", { :peek_lock => false })
message = azure_service_bus_service.receive_subscription_message(
  "test-topic", "high-messages")
azure_service_bus_service.delete_subscription_message(message)

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 unlock_subscription_message() method on the Azure::ServiceBusService object. This causes 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 (for example, 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 delete_subscription_message() method is called, then the message is 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 logic is often achieved using the message_id property of the message, which will remain constant across delivery attempts.

Delete topics and subscriptions

Topics and subscriptions are persistent, and must be explicitly deleted either through the Azure portal or programmatically. The example below demonstrates how to delete the topic named "test-topic".

azure_service_bus_service.delete_topic("test-topic")

Deleting a topic also deletes any subscriptions that are registered with the topic. Subscriptions can also be deleted independently. The following code demonstrates how to delete the subscription named "high-messages" from the "test-topic" topic:

azure_service_bus_service.delete_subscription("test-topic", "high-messages")

Next steps

Now that you've learned the basics of Service Bus topics, follow these links to learn more.