The following walkthrough will show how topics and subscriptions can be used to implement a simple publish-subscribe messaging channel. The next walkthrough will build on this sample and explore the use of filters on subscriptions.
If you want to run through the sample create a new console application, set the Target Framework to .NET 4.0 and add the following assemblies.
· Microsoft.ServiceBus
· System.Runtime.Serialization
Creating the Data Contract
The application will send order messages through the publish-subscribe channel. The following data contract is used to represent an order. This is created as a separate class in the project.
|
[DataContract]
class Order
{
[DataMember]
public string Name { get; set; }
[DataMember]
public DateTime OrderDate { get; set; }
[DataMember]
public int Items { get; set; }
[DataMember]
public double Value { get; set; }
[DataMember]
public string Priority { get; set; }
[DataMember]
public string Region { get; set; }
[DataMember]
public bool HasLoyltyCard { get; set; }
}
|
Adding Account Details
An AccountDetails class is also added and the values set for the namespace and credentials.
|
class AccountDetails
{
public static string Namespace = "";
public static string Name = "";
public static string Key = "";
}
|
Creating Static NamespaceManager and MessagingFactory
The next step is to create and initialize a NamespaceManager and MessagingFactory as static variables. This allows the objects to be reused and saves the overhead of recreating them.
|
class Program
{
static NamespaceManager NamespaceMgr;
static MessagingFactory Factory;
static void Main(string[] args)
{
CreateManagerAndFactory();
}
static void CreateManagerAndFactory()
{
// Create a token provider with the relevant credentials.
TokenProvider credentials =
TokenProvider.CreateSharedSecretTokenProvider
(AccountDetails.Name, AccountDetails.Key);
// Create a URI for the serivce bus.
Uri serviceBusUri = ServiceBusEnvironment.CreateServiceUri
("sb", AccountDetails.Namespace, string.Empty);
// Create the NamespaceManager
NamespaceMgr = new NamespaceManager(serviceBusUri, credentials);
// Create the NamespaceManager
Factory = MessagingFactory.Create(serviceBusUri, credentials);
}
}
|
Creating a Topic and Adding Subscriptions
When this is done a topic and some can be created in the service bus namespace. The following code will delete the topic, and then re-create it. As subscriptions belong to a topic any subscriptions we add to the topic will be deleted with that topic. Three subscriptions are then added to the topic without specifying any filters.
Please not that at the time of writing Azure customers are not billed for using brokered messaging services. When the billing model is in place repeatedly deleting and recreating messaging entities could lead to a recorded consumption of a large number of entity hours.
|
static void CreateTopicsAndSubscriptions()
{
// If the topic exists, delete it.
if (NamespaceMgr.TopicExists("ordertopic"))
{
NamespaceMgr.DeleteTopic("ordertopic");
}
// Create the topic.
NamespaceMgr.CreateTopic("ordertopic");
// Create three subscriptions in the topic.
NamespaceMgr.CreateSubscription("ordertopic", "subscription1");
NamespaceMgr.CreateSubscription("ordertopic", "subscription2");
NamespaceMgr.CreateSubscription("ordertopic", "subscription3");
}
|
Implementing a SendOrder Method
As we will be sending several messages to the topic a static method is added to send message using a TopicClient.
|
static void SendOrder(TopicClient topicClient, Order order)
{
Console.ForegroundColor = ConsoleColor.Cyan;
Console.Write("Sending {0}...", order.Name);
// Create a message from the order.
BrokeredMessage orderMsg = new BrokeredMessage(order);
// Send the message.
topicClient.Send(orderMsg);
Console.WriteLine("Done!");
}
|
Receiving Messages form Subscriptions
As several subscriptions have been added to the topic it makes sense to add method to loop through all the subscriptions on a topic and receive all the messages from each subscription. As the messages will have already been sent to the topic we can use a short timeout for receiving them from the subscriptions.
|
private static void ReceiveFromSubscriptions (string topicPath)
{
Console.ForegroundColor = ConsoleColor.Magenta;
Console.WriteLine("Receiving from topic {0} subscriptions.", topicPath);
// Loop through the subscriptions in a topic.
foreach (SubscriptionDescription subDescription in
NamespaceMgr.GetSubscriptions(topicPath))
{
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine(" Receiving from subscription {0}...", subDescription.Name);
// Create a SubscriptionClient
SubscriptionClient subClient =
Factory.CreateSubscriptionClient(topicPath, subDescription.Name);
Console.ForegroundColor = ConsoleColor.Green;
// Receive all the massages form the subscription.
while (true)
{
// Receive any message with a one second timeout.
BrokeredMessage msg = subClient.Receive(TimeSpan.FromSeconds(1));
if (msg != null)
{
// Deserialize the message body to an order.
Order order = msg.GetBody<Order>();
Console.WriteLine(" Received {0}", order.Name);
// Mark the message as complete.
msg.Complete();
}
else
{
break;
}
}
// Close the SubscriptionClient.
subClient.Close();
}
Console.ResetColor();
}
|
Putting it All Together
We can now add code to the main method to create the topic and subscriptions, send three order messages to the topic, and then receive messages from all the subscriptions.
|
static void Main(string[] args)
{
CreateManagerAndFactory();
CreateTopicsAndSubscriptions();
// Create a TopicClient for ordertopic.
TopicClient orderTopicclient = Factory.CreateTopicClient("ordertopic");
// Send three orders.
Console.WriteLine("Sending orders...");
SendOrder(orderTopicclient, new Order() { Name = "Order 1" });
SendOrder(orderTopicclient, new Order() { Name = "Order 2" });
SendOrder(orderTopicclient, new Order() { Name = "Order 3" });
// Close the TopicClient.
orderTopicclient.Close();
// Receive all messages from the ordertopic subscriptions.
ReceiveFromSubscriptions("ordertopic");
// Close the MessagingFactory.
Factory.Close();
}
|
Testing the Application
When the application is tested the following output is displayed in the console.
A copy of each of the three order messages sent to the topic was received by the three subscriptions.
Exploring Default Filters
When the subscriptions were created in the, no filter was defined for them, and they subscribe to all messages that are sent to the topic. To understand why this happens we need to dig a little deeper into the rules on those subscriptions. The following method will loop through the subscriptions in a topic and display any rules that belong to the subscription.
|
private static void OutputSubscriptionsAndRules(string topicPath)
{
Console.ForegroundColor = ConsoleColor.Magenta;
Console.WriteLine("Subscriptions for topic {0}.", topicPath);
// Loop through the subscriptions in a topic.
foreach (SubscriptionDescription subDescription in
NamespaceMgr.GetSubscriptions(topicPath))
{
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine(" Subscription Name = {0}", subDescription.Name);
Console.ForegroundColor = ConsoleColor.Green;
// Loop through the rules in each subscription.
foreach (RuleDescription rule in NamespaceMgr.GetRules(topicPath,
subDescription.Name))
{
// Cast the rule to a SqlFilter and display the name and expression.
Console.WriteLine(" Rule: Name = {0} - SqlExpression = {1}", rule.Name,
(rule.Filter as SqlFilter).SqlExpression);
}
}
}
|
When this code is executed the results are as follows.
This shows that each subscription has been created with a default rule with a SQL filter expression of “1 =1”. The filters are actually instances of the TureFilter class, which derives from SqlFilter and sets the SQL expression to “1=1”. The FasleFilter class also derives from SqlFilter and sets the SQL expression to “1=0”. These filters can be used to cause a subscription to subscribe to all messages on the topic, or no messages.