BizTalk 2010 R2 CTP: Azure Service Bus Integration–Part 4 Sending Messages to Service Bus Topics

Back again with Part 4 in this series.  This time around we are going use BizTalk to send a message to an Azure Service Bus Topic using the new SB-Messaging Adapter.

What is the difference between a Service Bus Queue and Topic?

Conceptually they are very similar in the sense that they both provide a durable message store within Azure where Publishers can publish messages and Consumers can consume messages.  Queues store messages in a First In First Out (FIFO) manner and provide a competing consumer experience.  What this means is that if we have two queue clients that are polling the same queue then only one client will receive a copy of any given message.

Topics provide some additional features that really support a Publish/Subscribe (Pub/Sub) architecture.  Topics allow for multiple clients to subscribe to the same Topic through subscriptions.  Subscriptions also support SQL92 expressions and allow a consumer to filter messages out based upon BrokeredMessage Properties.

Scenario

In Part 3 I  discussed how our Work Order Management system can notify our Major Account System when an Estimated Time of Restore is available.  This allows Major Account Representatives the ability to reach out to customers to share the good/bad news about when their power will be restored.

We are going to build upon this scenario but instead of sending all messages to a Queue we are going to send it to a Azure Service Bus Topic instead.  Due to the , fictional, growth of our company there are now two distinct groups responsible for Major accounts.  One for the city of Edmonton and another for the city of Calgary. (Both of these cities exist within Alberta, Canada)  Each queue Subscription client will now subscribe to events for their city.  BizTalk will however just send messages to one Topic and let the Service Bus work out which message needs to be delivered to each client.

Modifying Client Application(s)

Calgary Application

• Once again we are going to leverage the work that we have done in previous posts(In this case Part 3).  The first thing we will do is rename our previous C# Console project from BrokeredMessageFromBizTalk to BrokeredMessageFromBizTalkCalgary.

• Next we will modify the namespace for the Program.cs file so that it reflects the name of the project (BrokeredMessageFromBizTalkCalgary).  Note we will leave the namespace of our EstimatedTimeToRestore.cs as is.
• Below is the entire code listing for our Calgary Client.  Within this code we will use our connectivity parameters to establish a connection.  Once we have a connection we will see if the Calgary subscription currently exists.  If it does not, we will create it.  We will also create it and add a SQLFilter.  This particular filter is interested in messages where the Address Brokered Message Property equals “Calgary”.

using System.IO;
using System.Runtime.Serialization;
using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;
using BrokeredMessageFromBizTalk;

namespace BrokeredMessageFromBizTalkCalgary
{
    class Receiver
    {
        const string TopicName = "<your_topic>";
        static string ServiceNamespace = "<your_namespace>";
        static string IssuerName = "<your_owner>";
        static string IssuerKey = "<your_key>";
        static string connectionString = String.Format("Endpoint=sb://{0}.servicebus.windows.net/;SharedSecretIssuer={1};SharedSecretValue={2}",
             ServiceNamespace, IssuerName, IssuerKey); 

        static void Main(string[] args)
        {
           
            TokenProvider tokenProvider = TokenProvider.CreateSharedSecretTokenProvider(
                Receiver.IssuerName, Receiver.IssuerKey);
            Uri uri = ServiceBusEnvironment.CreateServiceUri("sb", Receiver.ServiceNamespace, string.Empty);
            MessagingFactory messagingFactory = MessagingFactory.Create(uri, tokenProvider);

            // Create a "Calgary" SQLFilter
           SqlFilter calgaryFilter = new SqlFilter("Address = 'Calgary'");

            //Check to see if Calgary Subscription exists, if it does not then create it
            NamespaceManager namespaceManager = NamespaceManager.CreateFromConnectionString(Receiver.connectionString);
            if (!namespaceManager.SubscriptionExists(Receiver.TopicName, "Calgary"))
            {

                //Create Calgary Subscription with our calgaryFilter
                namespaceManager.CreateSubscription(Receiver.TopicName, "Calgary", calgaryFilter);

            }
            //Create Subscription Client using Peek/Lock Mode
            SubscriptionClient sc = messagingFactory.CreateSubscriptionClient(Receiver.TopicName , "Calgary", ReceiveMode.PeekLock);
            BrokeredMessage bm;
            while ((bm = sc.Receive(new TimeSpan(hours: 0, minutes: 0, seconds: 20))) != null)
            {
                var data = bm.GetBody<EstimatedTimeToRestore>(new DataContractSerializer(typeof(EstimatedTimeToRestore)));
                Console.WriteLine(String.Format("An estimated time of restore {0} has been received for {1}", data.RestoreTime, data.CustomerName));
                Console.WriteLine("Brokered Message Property Address has a value of {0}", bm.Properties["Address"]);
                //Remove message from Topic
                bm.Complete();
            }

        }
    }
}

Edmonton Application

• Within our existing Visual Studio solution we are going to add another C# Console Application called BrokeredMessageFromBizTalkEdmonton.

• Within this application we will create a reference to our “Calgary” project.  We need to do this so that we have access to the EstimatedTimeToRestore class. This is accomplished by including the following statement:
  • using BrokeredMessageFromBizTalk;
• Otherwise we can simply copy and past the “Calgary” Program.cs code and adapt it for our Edmonton scenario.The main areas that we need to focus on are creating a unique Edmonton Subscription and ensuring that our Edmonton filter matches Edmonton messages and not Calgary messages.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.IO;
using System.Runtime.Serialization;
using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;
using BrokeredMessageFromBizTalk;

namespace BrokeredMessageFromBizTalkEdmonton
{
    class Receiver
    {
        const string TopicName = "<your_topic>";
        static string ServiceNamespace = "<your_namespace>";
        static string IssuerName = "<your_owner>";
        static string IssuerKey = "<your_key>";
        static string connectionString = String.Format("Endpoint=sb://{0}.servicebus.windows.net/;SharedSecretIssuer={1};SharedSecretValue={2}",
             ServiceNamespace, IssuerName, IssuerKey);

        static void Main(string[] args)
        {

            TokenProvider tokenProvider = TokenProvider.CreateSharedSecretTokenProvider(
                Receiver.IssuerName, Receiver.IssuerKey);
            Uri uri = ServiceBusEnvironment.CreateServiceUri("sb", Receiver.ServiceNamespace, string.Empty);
            MessagingFactory messagingFactory = MessagingFactory.Create(uri, tokenProvider);

            // Create a "Edmonton" filtered subscription
            SqlFilter edmontonFilter = new SqlFilter("Address = 'Edmonton'");

            //Create NamespaceManager object and see if Edmonton Subscription exists for our Topic
            NamespaceManager namespaceManager = NamespaceManager.CreateFromConnectionString(Receiver.connectionString);
            if (!namespaceManager.SubscriptionExists(Receiver.TopicName, "Edmonton"))
            {
                //If the Subscription does not exist, create it
                namespaceManager.CreateSubscription(Receiver.TopicName, "Edmonton", edmontonFilter);

            }

            //Create Subscription client and use Peek/Lock mechanism for delivery
            SubscriptionClient sc = messagingFactory.CreateSubscriptionClient(Receiver.TopicName, "Edmonton", ReceiveMode.PeekLock);
            BrokeredMessage bm;
            while ((bm = sc.Receive(new TimeSpan(hours: 0, minutes: 0, seconds: 20))) != null)
            {
                var data = bm.GetBody<EstimatedTimeToRestore>(new DataContractSerializer(typeof(EstimatedTimeToRestore)));
                Console.WriteLine(String.Format("An estimated time of restore {0} has been received for {1}", data.RestoreTime, data.CustomerName));
                Console.WriteLine("Brokered Message Property Address has a value of {0}", bm.Properties["Address"]);
                //remove message from Topic
                bm.Complete();
            }

        }
    }
}

Creating Topic

Much like we did in the previous blog post where we created a Service Bus Queue, we can also create a Service Bus Topic from the http://windowsazure.com portal

• To create a new Topic simply click on the New Topic button.

• Provide a Topic Name.  For the purpose of this post I am using PowerRestoreTopic.  We can then leave the default settings as is.

Note: We can also create Topics via code by using the NamespaceManager class much like we did for Subscriptions.  I decided to use the portal just to demonstrate that we have a few options when creating Service Bus artifacts.

Modifying BizTalk Solution

Once again we are going to keep the BizTalk Solution changes to a minimum and will continue to use a Messaging Only scenario.

• The first change that we are going to make is to our BrokeredMessagePropertySchema by adding our Address field.  Note if you recall from our C# projects that we created SQLFilters based upon the Address property being equal to either Calgary or Edmonton(depending upon the application).  By adding this property to our PropertySchema we can now promote this value in our XSD.
• Next we will modify the ETRfile.xsd and promote the Address field.  We will then map this field to the property we just created in our PropertySchema.

• We can now deploy our BizTalk Application and bounce any related Host Instances.
• While inside the BizTalk Administration Console, we need to modify our SB-Messaging Send Port called SendEstimatedTimeToRestoreToAzure.  Previously we were using this Send Port to send messages to a Service Bus Queue.  We will now be sending messages to a Topic which uses the same convention within the URI.  Instead of including our Queue Name in the URI we need to specify the Topic Name.  In our scenario I am going to specify PowerRestoreTopic.

• Our Authentication tab remains unchanged. We will continue to use the same Issuer Name and Issuer Key as in our previous example.

• Finally, we need to click on the Properties tab so that we can enable our custom Brokered Message Properties. To enable these properties we need to provide the namespace for our Property Schema that we previously modified.  As you may recall, we added a property called Address that we will use to route messages within our Azure Topic Subscription.

Testing our Solution

• We are going to leverage the Receive Location that we created in Part 3 of this blog series.  Our solution is expecting messages that conform to the ETRfile.xsd schema.  Below, I have included two sample files.  One contains an Address that belongs in Edmonton and the other one has an Address that belongs in Calgary.  We will drop both of these files within the same receive location and they will be sent to the same PowerRestoreTopic.

• Once the messages have been sent we will launch our two client applications (Calgary and Edmonton).  The expectation is that the Calgary application will retrieve the Calgary Message and the Edmonton application will retrieve the Edmonton Message.
• As expected, our Edmonton client retrieved the Edmonton message and the Calgary client retrieved the Calgary message.

Conclusion

In this post we used existing BizTalk skills to send typed messages to an Azure Service Bus Topic.  We promoted a property within BizTalk that was converted to a Brokered Message Property that can be used within Subscription Filters in order to route messages between different consumers.

Once again the experience is very seamless. I really do like the .Net API for Service Bus.  I think it is intuitive and is very logical.  As simple as it is to use this API it is hard to imagine anything being much simpler. But it has been done. The BizTalk team has made interfacing with the Service Bus even easier than using the API.  From a BizTalk perspective the Service Bus is just another endpoint and requires some additional configuration.  I do find that this aligns with some of the early goals of BizTalk.