Windows Azure Service Bus – Push based programming approach

Article
11/01/2011

While working with Azure Service Bus, either with the queue or with the topic/subscription, you need to explicitly write code to pull messages from the queue or from the subscription. A typical pattern looks like the following:

Code Snippet

// Create a SB queue
QueueDescription queue = CreateQueue();
// Create a factory and client to talk to this queue
MessagingFactory factory = MessagingFactory.Create(ServiceUri, TokenProvider);
QueueClient client = factory.CreateQueueClient(queue.Path);
// Send a message to the queue
string requestMessage = "Hello World!";
client.Send(new BrokeredMessage(requestMessage));
// Receive the message from the queue
BrokeredMessage responseQueueMessage = client.Receive();
string responseMessage = responseQueueMessage.GetBody<string>();

Note how you have to write code in line 13,14 to receive the messages from the queue. Programing with topic/subscription follow similar pattern in how you need to pull messages from the subscription client.

Now what if the queue is constantly receiving messages that you need to process or if you want to use a session full communication with the queue or if you want to do custom processing of messages based on some message content or if you want to carry out an async operation when you receive the message? You may be able to program these scenarios but the number of lines and the complexity of the code increases.

If you think about it – receiving a message and doing “something” with it is exactly what a WCF service does so wouldn’t it be cool if there is an ability to start up a WCF service which will listen for messages received by the queue/subscription and do whatever that you want to do with this message? This is essentially what I meant by the title – Push based programing approach. This means that you don’t have to worry about constantly pulling the messages from the queue – you define a service method (or methods as we will see) to do whatever you want to do when the message is received and the WCF service, like a humble servant, will do it for you.

Today we uploaded a Nuget package (containing source and sample) which introduce a set of extension methods on the QueueClient and SubscriptionClient which take care of starting up a WCF service instance and then we follow a convention based approach to route the queued messages to the appropriate service method which takes care of handling its processing.

The package walks you through a set of ten scenarios – first five of which explore Service Bus as it is today and the remaining five build on the push based programming model to show how easy it has become now to program against Service Bus.

Scenario 1 – We enumerate all the queues in your namesapce using NamespaceManager.GetQueues() method:

A typical output will look like the following:

Scenario 2 - We enumerate all the topics in your namespace using NamespaceManager.GetTopics() method:

A typical output will look like the following:

Scenario 3 - We show how you can send and receive messages to a queue using a QueueClient formed from a MessagingFactory:

Code Snippet

// Create a SB queue
QueueDescription queue = CreateQueue();
// Create a factory and client to talk to this queue
MessagingFactory factory = MessagingFactory.Create(ServiceUri, TokenProvider);
QueueClient client = factory.CreateQueueClient(queue.Path);
// Send a message to the queue
string requestMessage = "Hello World!";
client.Send(new BrokeredMessage(requestMessage));
Console.WriteLine("Message sent to the queue - '{0}' ", requestMessage);
// Receive the message from the queue
BrokeredMessage responseQueueMessage = client.Receive();
string responseMessage = responseQueueMessage.GetBody<string>();
Console.WriteLine("Message received from the queue - '{0}' ", responseMessage);
// Cleanup
client.Close();
factory.Close();

This example shows how you program against Service Bus queue right now. A typical output will look like the following:

Scenario 4 - We show how you can send messages to a topic and receive messages via a subscription client:

Code Snippet

string topicName = "news";
string subscriptionName = "mySubscription";
// Delete the topic if it already exists
if (NamespaceManager.TopicExists(topicName))
{
NamespaceManager.DeleteTopic(topicName);
}
// Create the topic
Console.WriteLine("Creating new topic - '{0}'", topicName);
TopicDescription topic = NamespaceManager.CreateTopic(topicName);
// Create a subscription
Console.WriteLine("Creating new subscription - '{0}'", subscriptionName);
SubscriptionDescription subs = NamespaceManager.CreateSubscription(topic.Path, subscriptionName);
// Create a factory to talk to the topic & subscription
MessagingFactory factory = MessagingFactory.Create(ServiceUri, TokenProvider);
// Create a topic client to talk to this topic
TopicClient topicClient = factory.CreateTopicClient(topic.Path);
// Send a message to the topic
string requestMessage = "Hello World!";
Console.WriteLine("Message sent to the topic - '{0}' ", requestMessage);
topicClient.Send(new BrokeredMessage(requestMessage));
// Create a subscription client to receive messages
SubscriptionClient subscriptionClient = factory.CreateSubscriptionClient(topic.Path, subs.Name);
// Receive the messages from the subscription
BrokeredMessage responseQueueMessage = subscriptionClient.Receive();
string responseMessage = responseQueueMessage.GetBody<string>();
Console.WriteLine("Message received from the subscription - '{0}' ", responseMessage);
// Cleanup
subscriptionClient.Close();
topicClient.Close();
factory.Close();

Again – this example shows how you program against Service Bus topic/subscription currently. A typical output will look like the following:

Scenario 5 - We show the WCF integration – how you can create a WCF service which will listen on a Service Bus endpoint and how you can relay messages using a WCF client to this WCF service via the Azure Service Bus.

Code Snippet

NetMessagingBinding binding = new NetMessagingBinding();
// Start up the Echo WCF service listening via Azure Service Bus
ServiceHost host = new ServiceHost(typeof(EchoService));
ServiceEndpoint echoServiceEndpoint
= host.AddServiceEndpoint(typeof(IEchoService), binding, QueueAddress);
echoServiceEndpoint.Behaviors.Add(new TransportClientEndpointBehavior(TokenProvider));
host.Open();
Console.WriteLine("WCF service up and running, listening on {0}", QueueAddress);
// Create a WCF channel factory and client to send message to the Azure Service Bus
ChannelFactory<IEchoService> factory = new ChannelFactory<IEchoService>
(binding, new EndpointAddress(QueueAddress));
factory.Endpoint.Behaviors.Add(new TransportClientEndpointBehavior(TokenProvider));
IEchoService proxy = factory.CreateChannel();
// Send messages to the service
string requestMessage = "Hello World";
proxy.Echo(requestMessage);
Console.WriteLine("Message sent to the queue - '{0}' ", requestMessage);
// clean up the client and service
factory.Close();
host.Close();

Nothing new here - a typical output will look like the following:

Scenario 6 – With scenario 6 onwards we start harnessing the power of the Service Bus push based programming library.

We start with a simple scenario applying this new approach to receive messages.

Code Snippet

// Create a ServiceBus queue.
QueueDescription queue = CreateQueue();
// Create a factory and client to talk to this queue
MessagingFactory factory = MessagingFactory.Create(ServiceUri, TokenProvider);
QueueClient client = factory.CreateQueueClient(queue.Path);
// Use the extension method to pass in a delegate
// which will handle all messages received by the queue
client.StartListener(message =>
Console.WriteLine("Message received from the queue - {0}", message.GetBody<string>()));
// Send a message to the queue
string helloMessage = "Hello World!";
client.Send(new BrokeredMessage(helloMessage));
Console.WriteLine("Message sent to the queue - '{0}' ", helloMessage);
// Send another message to the queue
string byeMessage = "Bye World!";
client.Send(new BrokeredMessage(byeMessage));
Console.WriteLine("Message sent to the queue - '{0}' ", byeMessage);
// clean up the client and service
client.StopListener();
client.Close();
factory.Close();

Note line 10 – here we are using the StartListener extension method on QueueClient which takes in a delegate which will handle the messages received by the queue henceforth. A typical output will look like the following:

Scenario 7 – We expand on the previous scenario by explicitly providing a service type which will be responsible for handling the incoming messages to the queue. We also make the scenario interesting by attaching an Action property to each BrokeredMessage we are sending to the queue. The idea is to route messages to different service methods based on this Action property of the incoming message.

Code Snippet

class MyActionOrientedEchoService
{
[OnMessage(Action = "EchoOnce")]
public void EchoOnce(BrokeredMessage msg)
{
Console.WriteLine("Message received by EchoOnce method");
string messageBody = msg.GetBody<string>();
Console.WriteLine(messageBody);
}
[OnMessage(Action = "EchoTwice")]
public void EchoTwice(BrokeredMessage msg)
{
Console.WriteLine("Message received by EchoTwice method");
string messageBody = msg.GetBody<string>();
Console.WriteLine(messageBody);
Console.WriteLine(messageBody);
}
}

Code Snippet

// Create a ServiceBus queue.
QueueDescription queue = CreateQueue();
// Create a factory and client to talk to this queue
MessagingFactory factory = MessagingFactory.Create(ServiceUri, TokenProvider);
QueueClient client = factory.CreateQueueClient(queue.Path);
// Use the extension method to start the WCF service
// which will handle the queued messages based on its Action property
client.StartListener(typeof(MyActionOrientedEchoService));
string requestMessage = "Hello World!";
// Send message to the queue - to the EchoOnce operation
BrokeredMessage message = new BrokeredMessage(requestMessage);
message.SetAction("EchoOnce");
Console.WriteLine("Message '{0}' sent with '{1}' action", requestMessage, "EchoOnce");
client.Send(message);
// Send message to the queue - to the EchoTwice operation
BrokeredMessage anotherMessage = new BrokeredMessage(requestMessage);
anotherMessage.SetAction("EchoTwice");
Console.WriteLine("Message '{0}' sent with '{1}' action", requestMessage, "EchoTwice");
client.Send(anotherMessage);
// Cleanup
client.StopListener();
client.Close();
factory.Close();

In line 10 – we are again using the StartListener extension method to start up the service of type ‘MyActionOrientedEchoService’. Note the following points about the service:

the service doesn’t need to implement any interface or be marked with a ServiceContract attribute
the service has two methods marked with the OnMessage attribute. This is a new attribute defined in the sample. All queued messages will be directed to a method decorated with this attribute.
The regular convention though, which is automatically understood, is to use a public method named OnReceivedMessage returning void and accepting a single parameter of type BrokeredMessage which will be used to dispatch the incoming messages. In the absence of such a method, the dispatching is done to a public method with return type void and which accepts a parameter of type BrokeredMessage only if there is exactly one such method. Of course – OnMessage attribute is used to override such conventions.
In this scenario – we want to do custom processing based on received messages and hence we define two service methods each with OnMessage attribute and using the Action parameter which will help in directing the message to the correct operation.

In line 15 and 21, we are using the SetAction extension method to add an ‘action’ property to the BrokeredMessage and this will be used to direct the message to the suitable service method.

A typical output will look like the following:

Scenario 8 – We integrate sessions in this scenario. We see how we can easily spawn up different service instances which will handle the incoming queued messages based on the SessionId we attach to the brokered message.

Code Snippet

class MySessionfulService
{
public void OnReceiveMessage(BrokeredMessage msg)
{
Console.WriteLine("Received Message - {0}, Received by service - {1}",
msg.GetBody<string>(), this.GetHashCode());
}
public void OnError(Exception e)
{
Console.WriteLine("Error occurred");
Console.WriteLine(e.Message);
}
}

Code Snippet

// Create a sessionful queue
QueueDescription queue = NamespaceManager.CreateQueue
(new QueueDescription(QueueName) { RequiresSession = true });
// Create a factory and client to talk to this queue
MessagingFactory factory = MessagingFactory.Create(ServiceUri, TokenProvider);
QueueClient client = factory.CreateQueueClient(queue.Path);
// Use the extension method to start up the WCF service
// which will process the sessionful queued messages received by the queue
client.StartListener(typeof(MySessionfulService), requiresSession: true);
string helloMessage = "Hello World!";
string byeMessage = "Bye World!";
// Start sending messages to the queue
// Session 1
BrokeredMessage message1 = new BrokeredMessage(helloMessage);
message1.SessionId = "1";
client.Send(message1);
BrokeredMessage message2 = new BrokeredMessage(byeMessage);
message2.SessionId = "1";
client.Send(message2);
// Session 2
BrokeredMessage message3 = new BrokeredMessage(helloMessage);
message3.SessionId = "2";
client.Send(message3);
BrokeredMessage message4 = new BrokeredMessage(byeMessage);
message4.SessionId = "2";
client.Send(message4);
// Cleanup
client.Close();
factory.Close();

In line 2, we are ensuring that the ServiceBus queue enables sessions and correspondingly in line 11, we are starting up the WCF service listener enabling sessions. Note that in the MySessionfulService implementation we are sticking to the convention of using the method with the signature – void OnReceivedMessage (BrokeredMessage msg) which will handle all incoming messages. Also note another convention based method void OnError(Exception ex) – this is used to handle all the errors arising out of handling the queued messages. Again – for handling errors - we will look for a method which conforms to this signature if an OnError method does not exist. Internally, we are simply registering an IErrorHandler that redirects Exceptions to the error handler. This means that anything that is bubbled up to the IErrorHandler by ServiceModel will make its way to the error handler

A typical output will look like the following:

Note that I can also add a service behavior specifying the InstanceContextMode of my WCF service if I so desire – by default it is PerSession but I can change it to PerCall or Single to influence how my service method processes the incoming messages.

Scenario 9 – We extend the example to show how you can use the same WCF based push approach to receive messages for a subscription from a topic.

Code Snippet

// Create the topic
Console.WriteLine("Creating new topic - '{0}'", topicName);
TopicDescription topic = NamespaceManager.CreateTopic(topicName);
// Create a subscription
Console.WriteLine("Creating new subscription - '{0}'", subscriptionName);
SubscriptionDescription subs = NamespaceManager.CreateSubscription(topic.Path, subscriptionName);
// Create a factory to talk to the topic & subscription
MessagingFactory factory = MessagingFactory.Create(ServiceUri, TokenProvider);
// Create a topic client to talk to this topic
TopicClient topicClient = factory.CreateTopicClient(topic.Path);
// Create a subscription client to retrive messages
SubscriptionClient subscriptionClient = factory.CreateSubscriptionClient(topic.Path, subs.Name);
// Start the WCF service which will receive the messages sent to the topic
subscriptionClient.StartListener(typeof(MyQueuedMessageHandlerService));
// Send a message to the topic
string requestMessage = "Hello World!";
Console.WriteLine("Message sent to the topic - '{0}' ", requestMessage);
topicClient.Send(new BrokeredMessage(requestMessage));
// Cleanup
subscriptionClient.StopListener();
subscriptionClient.Close();
topicClient.Close();
factory.Close();

All the overloads which exist on the QueueClient, as we saw above, are also available on the SubscriptionClient. A WCF service will be started up in a similar way to handle all messages received by this subscription.

A typical output will look like the following:

Scenario 10 – We show how you can use the same approach to carry out an async operation at the receipt of the queued message.

Code Snippet

class MyAsyncEchoService
{
public Task DoComplexTaskAsync(BrokeredMessage message)
{
Console.WriteLine("Async method called");
return Task.Factory.StartNew(() =>
{
// Do some complex task
Console.WriteLine("Starting complex task at {0}", DateTime.Now.ToString());
Thread.Sleep(5000);
Console.WriteLine("Completing complex task at {0}", DateTime.Now.ToString());
});
}
}

Code Snippet

// Create a ServiceBus queue.
QueueDescription queue = CreateQueue();
// Create a factory and client to talk to this queue
MessagingFactory factory = MessagingFactory.Create(ServiceUri, TokenProvider);
QueueClient client = factory.CreateQueueClient(queue.Path);
// Use the extension method to start the WCF service
// which will handle the queued messages based on its Action property
client.StartListener(typeof(MyAsyncEchoService));
// Send message to the queue
string requestMessage = "Hello World!";
BrokeredMessage message = new BrokeredMessage("Hello World!");
client.Send(message);
Console.WriteLine("Message sent to the queue - '{0}' ", requestMessage);
// Cleanup
client.StopListener();
client.Close();
factory.Close();

The service implementation is the only interesting part otherwise the sample is identical to what we have seen in earlier scenarios. Note how the method takes in a BrokeredMessage and returns a Task.

A typical output will look like the following:

Thank you!

Special thanks to Youssef who was instrumental in developing this library.

Windows Azure Service Bus – Push based programming approach

Additional resources