DCSIMG
Using AppFabric Azure Topics - Manu Cohen-Yashar's Blog

Manu Cohen-Yashar's Blog

Using AppFabric Azure Topics

Recently the AppFabric team released real pub-sub capabilities with the new queues and topics features.

Queues are simple. AppFabric provides a queue for durable messaging between senders and receivers. Topics are interesting because they allows to create different subscriptions so different clients can receive only the messages they need.

To demonstrate how to use topics I created two helper classes: Senders and Receiver that will send and receive messages using AppFabric Topics.

Sender
  1. /// <summary>
  2.     /// Send messages to AppFabric Topic
  3.     /// </summary>
  4.     /// <typeparam name="T"></typeparam>
  5.     public class ServiceBusSender<T> : IBrokeredMessageSender<T>
  6.     {
  7.         private string m_ServiceNamespace;
  8.         private string m_IssuerName;
  9.         private string m_IssuerKey;
  10.         private Uri m_serviceUri;
  11.         private TokenProvider m_tokenProvider;
  12.         private TopicDescription m_topicDescriptor;
  13.  
  14.         /// <summary>
  15.         /// Constructor
  16.         /// </summary>
  17.         /// <param name="topicName">The name of the topic to use</param>
  18.         public ServiceBusSender(string topicName)
  19.         {
  20.             m_ServiceNamespace = ServiceBusConfigurationAccessor.AppFabricServiceNamespace;
  21.             m_IssuerName = ServiceBusConfigurationAccessor.AppFabricIssuerName;
  22.             m_IssuerKey = ServiceBusConfigurationAccessor.AppFabricIssuerKey;
  23.  
  24.             m_tokenProvider = TokenProvider.CreateSharedSecretTokenProvider(m_IssuerName, m_IssuerKey);
  25.             m_serviceUri = ServiceBusEnvironment.CreateServiceUri("sb", m_ServiceNamespace, string.Empty);
  26.             NamespaceManager namespaceManager = new NamespaceManager(m_serviceUri, m_tokenProvider);
  27.  
  28.             if (!namespaceManager.TopicExists(topicName))
  29.                 m_topicDescriptor = namespaceManager.CreateTopic(topicName);
  30.             else
  31.                 m_topicDescriptor = namespaceManager.GetTopic(topicName);
  32.            
  33.         }      
  34.  
  35.         /// <summary>
  36.         /// Send a message to the topic
  37.         /// </summary>
  38.         /// <param name="messageContent"></param>
  39.         public void SendMessage(T messageContent)
  40.         {
  41.             Contract.Requires(messageContent != null);
  42.  
  43.             MessagingFactory factory = MessagingFactory.Create(m_serviceUri, m_tokenProvider);
  44.             TopicClient myTopicClient = factory.CreateTopicClient(m_topicDescriptor.Path);
  45.             BrokeredMessage message = new BrokeredMessage(messageContent);
  46.  
  47.             //All the message's properties are propagated so they can be used in a Sql Filter
  48.             foreach (var prop in typeof(T).GetProperties())
  49.             {
  50.                 message.Properties.Add(prop.Name, prop.GetValue(messageContent, null));
  51.             }
  52.  
  53.             myTopicClient.Send(message);
  54.             myTopicClient.Close();
  55.         }
  56.  
  57.     }

As you can see the sender creates the topic (if it does not exist already). A messaging factory to create a client that can send messages to the topic. For that we need to supply the credentials for our namespace when we create the factory. To send a message to the topic we have to wrap our data in a BrokeredMessage. One of the capabilities we gain when doing that is properties propagation. A propagated property can participate a sql-like rule for filtering messages (will be defined by the client's subscription). In this sender we propagate all the properties. In real life you might choose to propagate only a the properties that participate in filtering because propagation costs.

Receiver
  1. /// <summary>
  2.     /// Listen on an AppFabric service bus topic. One receiver can listen on multiple subscriptions
  3.     /// </summary>
  4.     /// <typeparam name="T"></typeparam>
  5.     public class ServiceBusReceiver<T> : IBrokeredMessageReceiver<T>
  6.     {
  7.         private string m_ServiceNamespace;
  8.         private string m_IssuerName;
  9.         private string m_IssuerKey;
  10.         private Dictionary<string, SubscriptionClient> m_subscriptions;
  11.         private MessagingFactory m_factory;
  12.         private Uri m_serviceUri;
  13.         private TokenProvider m_tokenProvider;
  14.  
  15.         /// <summary>
  16.         /// The name of the topic to use
  17.         /// </summary>
  18.         public string TopicName { get; set; }
  19.    
  20.    
  21.         /// <summary>
  22.         /// Constructor
  23.         /// </summary>
  24.         /// <param name="topicName">The name of the topic to listen to</param>
  25.         public ServiceBusReceiver(string topicName)
  26.         {
  27.             m_subscriptions = new Dictionary<string, SubscriptionClient>();
  28.             m_ServiceNamespace = ServiceBusConfigurationAccessor.AppFabricServiceNamespace;
  29.             m_IssuerName = ServiceBusConfigurationAccessor.AppFabricIssuerName;
  30.             m_IssuerKey = ServiceBusConfigurationAccessor.AppFabricIssuerKey;
  31.             m_tokenProvider = TokenProvider.CreateSharedSecretTokenProvider(m_IssuerName, m_IssuerKey);
  32.             m_serviceUri = ServiceBusEnvironment.CreateServiceUri("sb", m_ServiceNamespace, string.Empty);
  33.             m_factory = MessagingFactory.Create(m_serviceUri, m_tokenProvider);
  34.             TopicName = topicName;
  35.         }
  36.  
  37.         /// <summary>
  38.         /// Create a new subscription
  39.         /// </summary>
  40.         /// <param name="subscriptionName">The name of the subscription to create</param>
  41.         /// <param name="filter">filter to use for filtering messages</param>
  42.         public void InitSubscription(string subscriptionName, Filter filter=null)
  43.         {
  44.             Contract.Requires(!string.IsNullOrEmpty(subscriptionName));
  45.             Contract.Requires(!string.IsNullOrEmpty(TopicName), "Topic name must be provided. Use the receiver's TopicName property");
  46.                        
  47.             var namespaceManager = new NamespaceManager(m_serviceUri, m_tokenProvider);
  48.            
  49.             if (!namespaceManager.TopicExists(TopicName))           
  50.                 namespaceManager.CreateTopic(TopicName);
  51.  
  52.             if (!namespaceManager.SubscriptionExists(TopicName, subscriptionName))
  53.             {
  54.                 filter = (filter == null) ? new TrueFilter() : filter;                    
  55.                 namespaceManager.CreateSubscription(TopicName, subscriptionName, filter);               
  56.             }
  57.             
  58.             if (!m_subscriptions.ContainsKey(subscriptionName))
  59.                 m_subscriptions.Add(subscriptionName, m_factory.CreateSubscriptionClient(TopicName, subscriptionName, ReceiveMode.ReceiveAndDelete));
  60.         }
  61.  
  62.         /// <summary>
  63.         /// Delete a new subscription from the topic
  64.         /// </summary>
  65.         /// <param name="subscriptionName">The name of the subscription to delete</param>
  66.         public void DeleteSubscription(string subscriptionName)
  67.         {
  68.             Contract.Requires(!string.IsNullOrEmpty(subscriptionName));
  69.             Contract.Requires(!string.IsNullOrEmpty(TopicName), "Topic name must be provided. Use the receiver's TopicName property");
  70.  
  71.  
  72.             var namespaceManager = new NamespaceManager(m_serviceUri, m_tokenProvider);
  73.             if (namespaceManager.TopicExists(TopicName) && namespaceManager.SubscriptionExists(TopicName, subscriptionName))
  74.                 namespaceManager.DeleteSubscription(TopicName, subscriptionName);
  75.         }
  76.  
  77.  
  78.         /// <summary>
  79.         /// Start listening on a subscription
  80.         /// </summary>
  81.         /// <param name="subscriptionName">The name of the subscription to listen to</param>
  82.         /// <param name="processMessage">A function to trigger for each comming message</param>
  83.         public void ReceiveMessages(string subscriptionName, Action<T> processMessage)
  84.         {
  85.             Contract.Requires(!string.IsNullOrEmpty(subscriptionName));
  86.             Contract.Requires(m_subscriptions.ContainsKey(subscriptionName));
  87.  
  88.             try
  89.             {
  90.                 Task.Factory.StartNew(() =>
  91.                 {
  92.                     while (m_subscriptions.ContainsKey(subscriptionName))
  93.                     {
  94.                         BrokeredMessage message;
  95.                         while ((message = m_subscriptions[subscriptionName].Receive(TimeSpan.FromSeconds(5))) != null)
  96.                         {
  97.                             processMessage(message.GetBody<T>());
  98.                         }
  99.                     }
  100.                 });
  101.             }
  102.             catch (Exception ex)
  103.             {
  104.                 // Log this
  105.                 throw;
  106.             }
  107.             
  108.         }
  109.  
  110.  
  111.         /// <summary>
  112.         /// Stop Listening on a subscription and delete it from the current subscriptions list
  113.         /// </summary>
  114.         /// <param name="subscriptionName">The name of the subscription</param>
  115.         public void StopReceiveNessages(string subscriptionName)
  116.         {
  117.             if (!m_subscriptions.ContainsKey(subscriptionName))
  118.                 return;
  119.  
  120.             var subscriptionClient = m_subscriptions[subscriptionName];
  121.  
  122.             if ((subscriptionClient != null) && (!subscriptionClient.IsClosed))
  123.                 subscriptionClient.Close();
  124.  
  125.             subscriptionClient = null;
  126.             m_subscriptions.Remove(subscriptionName);
  127.         }
  128.  
  129.     
  130.     }

The receiver can handle a collection of subscriptions. (Use InitSubscription to create one).
After a subscription is created call ReceiveMessages and provide a method to run for each message received. The receiver will constantly pool the topic and wait for a message as long as the subscription exist. When a message arrives it will extract the data object out of the brokered message at trigger the method you provided on the data.

The following test demonstrate how to use the sender and receiver.

Code Snippet
  1. [TestMethod]
  2.         public void ServiceBusSenderAndReceiverTest()
  3.         {
  4.             var sender = new ServiceBusSender<TestMessage>("testTopic");
  5.             var receiver = new ServiceBusReceiver<TestMessage>("testTopic");
  6.             var finished = new AutoResetEvent(false);            
  7.  
  8.             receiver.InitSubscription("testSubscription");
  9.             receiver.ReceiveMessages("testSubscription",
  10.                 (msg) => { Assert.AreEqual(39, msg.Age); Assert.AreEqual("manu", msg.Name); finished.Set(); });
  11.             sender.SendMessage(new TestMessage() { Name = "manu", Age = 39 });               
  12.  
  13.             finished.WaitOne();
  14.             receiver.DeleteSubscription("testSubscription");
  15.         }


Enjoy

Manu

Posted: Nov 17 2011, 03:54 PM by Manu Cohen-Yashar | with no comments
תגים:,

Comments

No Comments

Leave a Comment

(required) 

(required) 

(optional)

(required) 


Enter the numbers above: