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 PHP and use the Windows Azure SDK for PHP. 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.

Table of Contents

• What are Service Bus Topics and Subscriptions?
• Create a Service Namespace
• Obtain the Default Management Credentials for the Namespace
• Create a PHP application
• Get the Windows Azure Client Libraries
• Configure Your Application to Use Service Bus
• How to: Create a Topic
• How to: Create a Subscription
• How to: Send Messages to a Topic
• How to: Receive Messages from a Subscription
• How to: Handle Application Crashes and Unreadable Messages
• How to: Delete Topics and Subscriptions
• Next Steps

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:

1. Log on to the Windows Azure Management Portal.

2. In the left navigation pane of the Management Portal, click Service Bus.

3. In the lower pane of the Management Portal, click Create.
   

4. In the Add a new namespace dialog, enter a namespace name. The system immediately checks to see if the name is available.
   

5. 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).

   IMPORTANT: Pick the same region that you intend to choose for deploying your application. This will give you the best performance.

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

   

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.

1. In the left navigation pane, click the Service Bus node to display the list of available namespaces:
   

2. Select the namespace you just created from the list shown:
   

3. Click Access Key.
   

4. In the Connect to your namespace dialog, find the Default Issuer and Default Key entries. Make a note of these values, as you will use this information below to perform operations with the namespace.

Create a PHP application

The only requirement for creating a PHP application that accesses the Windows Azure Blob service is the referencing of classes in the Windows Azure SDK for PHP from within your code. You can use any development tools to create your application, including Notepad.

In this guide, you will use service features which can be called within a PHP application locally, or in code running within a Windows Azure web role, worker role, or web site.

Get the Windows Azure Client Libraries

Configure your application to use Service Bus

To use the Windows Azure Service Bus topic APIs, you need to:

1. Reference the autoloader file using the require_once statement, and
2. Reference any classes you might use.

The following example shows how to include the autoloader file and reference the ServiceBusService class.

require_once 'vendor\autoload.php';
use WindowsAzure\Common\ServicesBuilder;

In the examples below, the require_once statement will be shown always, but only the classes necessary for the example to execute will be referenced.

Setup a Windows Azure Service Bus connection

To instantiate a Windows Azure Service Bus client you must first have a valid connection string following this format:

Endpoint=[yourEndpoint];SharedSecretIssuer=[Default Issuer];SharedSecretValue=[Default Key]

Where the Endpoint is typically of the format https://[yourNamespace].servicebus.windows.net.

To create any Windows Azure service client you need to use the ServicesBuilder class. You can:

• pass the connection string directly to it or
• use the CloudConfigurationManager (CCM)to check multiple external sources for the connection string:
  • by default it comes with support for one external source - environmental variables
  • you can add new sources by extending the ConnectionStringSource class

For the examples outlined here, the connection string will be passed directly.

require_once 'vendor\autoload.php';

use WindowsAzure\Common\ServicesBuilder;

$connectionString = "Endpoint=[yourEndpoint];SharedSecretIssuer=[Default Issuer];SharedSecretValue=[Default Key]";

$serviceBusRestProxy = ServicesBuilder::getInstance()->createServiceBusService($connectionString);

How to: Create a topic

Management operations for Service Bus topics can be performed via the ServiceBusRestProxy class. A ServiceBusRestProxy object is constructed via the ServicesBuilder::createServiceBusService factory method with an appropriate connection string that encapsulates the token permissions to manage it.

The example below shows how to instantiate a ServiceBusRestProxy and call ServiceBusRestProxy->createTopic to create a topic named mytopic within a MySBNamespace service namespace:

require_once 'vendor\autoload.php';

use WindowsAzure\Common\ServicesBuilder;
use WindowsAzure\Common\ServiceException;
use WindowsAzure\ServiceBus\Models\TopicInfo;

// Create Service Bus REST proxy.
$serviceBusRestProxy = ServicesBuilder::getInstance()->createServiceBusService($connectionString);

try {       
    // Create topic.
    $topicInfo = new TopicInfo("mytopic");
    $serviceBusRestProxy->createTopic($topicInfo);
}
catch(ServiceException $e){
    // Handle exception based on error codes and messages.
    // Error codes and messages are here: 
    // http://msdn.microsoft.com/en-us/library/windowsazure/dd179357
    $code = $e->getCode();
    $error_message = $e->getMessage();
    echo $code.": ".$error_message."<br />";
}

How to: Create a subscription

Topic subscriptions are also created with the ServiceBusRestProxy->createSubscription method. 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 'mysubscription' and uses the default MatchAll filter.

require_once 'vendor\autoload.php';

use WindowsAzure\Common\ServicesBuilder;
use WindowsAzure\Common\ServiceException;
use WindowsAzure\ServiceBus\Models\SubscriptionInfo;

// Create Service Bus REST proxy.
$serviceBusRestProxy = ServicesBuilder::getInstance()->createServiceBusService($connectionString);

try {
    // Create subscription.
    $subscriptionInfo = new SubscriptionInfo("mysubscription");
    $serviceBusRestProxy->createSubscription("mytopic", $subscriptionInfo);
}
catch(ServiceException $e){
    // Handle exception based on error codes and messages.
    // Error codes and messages are here: 
    // http://msdn.microsoft.com/en-us/library/windowsazure/dd179357
    $code = $e->getCode();
    $error_message = $e->getMessage();
    echo $code.": ".$error_message."<br />";
}

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 information about SqlFilters, see SqlFilter.SqlExpression Property.

The example below creates a subscription named "HighMessages" with a SqlFilter that only selects messages that have a custom MessageNumber property greater than 3 (see How to: Send messages to a topic for information about adding custom properties to messages):

$subscriptionInfo = new SubscriptionInfo("HighMessages");
$serviceBusRestProxy->createSubscription("mytopic", $subscriptionInfo);

$serviceBusRestProxy->deleteRule("mytopic", "HighMessages", '$Default');

$ruleInfo = new RuleInfo("HighMessagesRule");
$ruleInfo->withSqlFilter("MessageNumber > 3");
$ruleResult = $serviceBusRestProxy->createRule("mytopic", "HighMessages", $ruleInfo);

Note that the code above requires the use of an additional namespace: WindowsAzure\ServiceBus\Models\SubscriptionInfo.

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:

$subscriptionInfo = new SubscriptionInfo("LowMessages");
$serviceBusRestProxy->createSubscription("mytopic", $subscriptionInfo);

$serviceBusRestProxy->deleteRule("mytopic", "LowMessages", '$Default');

$ruleInfo = new RuleInfo("LowMessagesRule");
$ruleInfo->withSqlFilter("MessageNumber <= 3");
$ruleResult = $serviceBusRestProxy->createRule("mytopic", "LowMessages", $ruleInfo);

When a message is now sent to the mytopic topic, it will always be delivered to receivers subscribed to the mysubscription subscription, and selectively delivered to receivers subscribed to the "HighMessages" and "LowMessages" 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 call the ServiceBusRestProxy->sendTopicMessage method. The code below demonstrates how to send a message to the mytopic topic we created above within the MySBNamespace service namespace.

require_once 'vendor\autoload.php';

use WindowsAzure\Common\ServicesBuilder;
use WindowsAzure\Common\ServiceException;
use WindowsAzure\ServiceBus\Models\BrokeredMessage;

// Create Service Bus REST proxy.
$serviceBusRestProxy = ServicesBuilder::getInstance()->createServiceBusService($connectionString);

try {
    // Create message.
    $message = new BrokeredMessage();
    $message->setBody("my message");

    // Send message.
    $serviceBusRestProxy->sendTopicMessage("mytopic", $message);
}
catch(ServiceException $e){
    // Handle exception based on error codes and messages.
    // Error codes and messages are here: 
    // http://msdn.microsoft.com/en-us/library/windowsazure/hh780775
    $code = $e->getCode();
    $error_message = $e->getMessage();
    echo $code.": ".$error_message."<br />";
}

Messages sent to Service Bus topics are instances of the BrokeredMessage class. BrokeredMessage objects have a set of standard properties and methods (such as getLabel, getTimeToLive, setLabel, and setTimeToLive), as well as properties that can be used to hold custom application-specific properties. The following example demonstrates how to send five test messages to the mytopic topic we created earlier. The setProperty method is used to add a custom property (MessageNumber) to each message. Note how the MessageNumber property value varies on each message (this can be used to determine which subscriptions receive it, as shown in the How to: Create a Subscription section above):

for($i = 0; $i < 5; $i++){
    // Create message.
    $message = new BrokeredMessage();
    $message->setBody("my message ".$i);

    // Set custom property.
    $message->setProperty("MessageNumber", $i);

    // Send message.
    $serviceBusRestProxy->sendTopicMessage("mytopic", $message);
}

Service Bus queues support a maximum message size of 256 KB (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 queue but there is a cap on the total size of the messages held by a queue. This upper limit on queue size is 5 GB.

How to: Receive messages from a subscription

The primary way to receive messages from a subscription is to use a ServiceBusRestProxy->receiveSubscriptionMessage method. Received messages can work in two different modes: ReceiveAndDelete (the default) 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, receiving a message 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 passing the received message to ServiceBusRestProxy->deleteMessage. When Service Bus sees the deleteMessage call, it will mark the message as being consumed and remove it from the queue.

The example below demonstrates how a message can be received and processed using PeekLock mode (not the default mode).

require_once 'vendor\autoload.php';

use WindowsAzure\Common\ServicesBuilder;
use WindowsAzure\Common\ServiceException;
use WindowsAzure\ServiceBus\Models\ReceiveMessageOptions;

// Create Service Bus REST proxy.
$serviceBusRestProxy = ServicesBuilder::getInstance()->createServiceBusService($connectionString);

try {
    // Set receive mode to PeekLock (default is ReceiveAndDelete)
    $options = new ReceiveMessageOptions();
    $options->setPeekLock();

    // Get message.
    $message = $serviceBusRestProxy->receiveSubscriptionMessage("mytopic", 
                                                                "mysubscription", 
                                                                $options);
    echo "Body: ".$message->getBody()."<br />";
    echo "MessageID: ".$message->getMessageId()."<br />";

    /*---------------------------
        Process message here.
    ----------------------------*/

    // Delete message. Not necessary if peek lock is not set.
    echo "Deleting message...<br />";
    $serviceBusRestProxy->deleteMessage($message);
}
catch(ServiceException $e){
    // Handle exception based on error codes and messages.
    // Error codes and messages are here:
    // http://msdn.microsoft.com/en-us/library/windowsazure/hh780735
    $code = $e->getCode();
    $error_message = $e->getMessage();
    echo $code.": ".$error_message."<br />";
}

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 unlockMessage method on the received message (instead of the deleteMessage method). This will cause Service Bus to unlock the message within the queue 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 queue, 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 deleteMessage 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 getMessageId method of the message, which will remain constant across delivery attempts.

How to Delete Topics and Subscriptions

To delete a topic or a subscription, use the ServiceBusRestProxy->deleteTopic or the ServiceBusRestProxy->deleteSubscripton methods respectively. Note that deleting a topic will also delete any subscriptions that are registered with the topic.

The following example shows how to delete a topic (mytopic) and its registered subscriptions.

require_once 'vendor\autoload.php';

use WindowsAzure\ServiceBus\ServiceBusService;
use WindowsAzure\ServiceBus\ServiceBusSettings;
use WindowsAzure\Common\ServiceException;

// Create Service Bus REST proxy.
$serviceBusRestProxy = ServicesBuilder::getInstance()->createServiceBusService($connectionString);

try {       
    // Delete topic.
    $serviceBusRestProxy->deleteTopic("mytopic");
}
catch(ServiceException $e){
    // Handle exception based on error codes and messages.
    // Error codes and messages are here: 
    // http://msdn.microsoft.com/en-us/library/windowsazure/dd179357
    $code = $e->getCode();
    $error_message = $e->getMessage();
    echo $code.": ".$error_message."<br />";
}

By using the deleteSubscription method, you can delete a subscription independently:

$serviceBusRestProxy->deleteSubscription("mytopic", "mysubscription");

Next steps

Now that you've learned the basics of Service Bus queues, see the MSDN topic Queues, Topics, and Subscriptions for more information.