title: Azure Service Bus Topics and Subscriptions with Worker Role
body: <p>So I've recently gotten the need to use <code>Service Bus Topic and Subscriptions</code> and I've followed many articles and tutorials. I've been able to successfully implement Microsoft's <a href="https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dotnet-how-to-use-topics-subscriptions" rel="nofollow noreferrer">Get started with Service Bus topics</a> and also successfully used <code>Visual Studio 2017's</code> <code>Worker Role</code> template to access a database.</p>
  
  <p>However, I'm confused as to how to properly "combine" the two. While the <a href="https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dotnet-how-to-use-topics-subscriptions" rel="nofollow noreferrer">Get started with Service Bus topics</a> article shows how to create 2 apps, one to send and one to receive and then quits, the <code>Worker Role</code> template seems to loops endlessly with <code>await Task.Delay(10000);</code>.</p>
  
  <p>I'm not sure how to "mesh" the two properly. Essentially, I want my <code>Worker Role</code> to stay alive and listen for entries into it's subscription forever (or until it quits obviously).</p>
  
  <p>Any guidance would be great!</p>
  
  <p>P.S.: I've asked a related question concerning proper technology I should use for my case scenario at <a href="https://softwareengineering.stackexchange.com/questions/376601/cloud-service-or-service-fabric">StackExchange - Software Engineering</a> if you are interested.</p>
  
  <p><strong>Update #1 (2018/08/09)</strong></p>
  
  <p>Based on <a href="https://stackoverflow.com/a/51759928/286618">Arunprabhu's answer</a>, here is some code of how I'm sending a <code>Message</code> based on articles I've read and receiving using <code>Visual Studio 2017</code>'s <code>Worker Role with Service Bus Queue</code> template.</p>
  
  <p><em>Sending (based on <a href="https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dotnet-how-to-use-topics-subscriptions" rel="nofollow noreferrer">Get started with Service Bus topics</a>)</em></p>
  
  <pre><code>using System;
  using System.Text;
  using System.Threading.Tasks;
  using Microsoft.Azure.ServiceBus;
  
  namespace TopicsSender {
      internal static class Program {
          private const string ServiceBusConnectionString = "&lt;your_connection_string&gt;";
          private const string TopicName = "test-topic";
          private static ITopicClient _topicClient;
  
          private static void Main(string[] args) {
              MainAsync().GetAwaiter().GetResult();
          }
  
          private static async Task MainAsync() {
              const int numberOfMessages = 10;
              _topicClient = new TopicClient(ServiceBusConnectionString, TopicName);
  
              Console.WriteLine("======================================================");
              Console.WriteLine("Press ENTER key to exit after sending all the messages.");
              Console.WriteLine("======================================================");
  
              // Send messages.
              await SendMessagesAsync(numberOfMessages);
  
              Console.ReadKey();
  
              await _topicClient.CloseAsync();
          }
  
          private static async Task SendMessagesAsync(int numberOfMessagesToSend) {
              try {
                  for (var i = 0; i &lt; numberOfMessagesToSend; i++) {
                      // Create a new message to send to the topic
                      var messageBody = $"Message {i}";
                      var message = new Message(Encoding.UTF8.GetBytes(messageBody));
  
                      // Write the body of the message to the console
                      Console.WriteLine($"Sending message: {messageBody}");
  
                      // Send the message to the topic
                      await _topicClient.SendAsync(message);
                  }
              } catch (Exception exception) {
                  Console.WriteLine($"{DateTime.Now} :: Exception: {exception.Message}");
              }
          }
      }
  }
  </code></pre>
  
  <p><em>Receiving (based on <code>Worker Role with Service Bus Queue</code> template)</em></p>
  
  <pre><code>using System;
  using System.Diagnostics;
  using System.Net;
  using System.Threading;
  using Microsoft.ServiceBus.Messaging;
  using Microsoft.WindowsAzure.ServiceRuntime;
  
  namespace WorkerRoleWithSBQueue1 {
      public class WorkerRole : RoleEntryPoint {
          // The name of your queue
          private const string ServiceBusConnectionString = "&lt;your_connection_string&gt;";
          private const string TopicName = "test-topic";
          private const string SubscriptionName = "test-sub1";
  
          // QueueClient is thread-safe. Recommended that you cache 
          // rather than recreating it on every request
          private SubscriptionClient _client;
          private readonly ManualResetEvent _completedEvent = new ManualResetEvent(false);
  
          public override void Run() {
              Trace.WriteLine("Starting processing of messages");
  
              // Initiates the message pump and callback is invoked for each message that is received, calling close on the client will stop the pump.
              _client.OnMessage((receivedMessage) =&gt; {
                  try {
                      // Process the message
                      Trace.WriteLine("Processing Service Bus message: " + receivedMessage.SequenceNumber.ToString());
                      var message = receivedMessage.GetBody&lt;byte[]&gt;();
                      Trace.WriteLine($"Received message: SequenceNumber:{receivedMessage.SequenceNumber} Body:{message.ToString()}");
                  } catch (Exception e) {
                      // Handle any message processing specific exceptions here
                      Trace.Write(e.ToString());
                  }
              });
  
              _completedEvent.WaitOne();
          }
  
          public override bool OnStart() {
              // Set the maximum number of concurrent connections 
              ServicePointManager.DefaultConnectionLimit = 12;
  
              // Initialize the connection to Service Bus Queue
              _client = SubscriptionClient.CreateFromConnectionString(ServiceBusConnectionString, TopicName, SubscriptionName);
              return base.OnStart();
          }
  
          public override void OnStop() {
              // Close the connection to Service Bus Queue
              _client.Close();
              _completedEvent.Set();
              base.OnStop();
          }
      }
  }
  </code></pre>
  
  <p><strong>Update #2 (2018/08/10)</strong></p>
  
  <p>After a few suggestions from Arunprabhu and knowing I was using different libraries, below is my current solution with pieces taken from several sources. Is there anything I'm overlooking, adding that shouldering be there, etc? Currently getting an error that may be for another question or already answered so don't want to post it yet before further research.</p>
  
  <pre><code>using System;
  using System.Diagnostics;
  using System.Net;
  using System.Text;
  using System.Threading;
  using System.Threading.Tasks;
  using Microsoft.Azure.ServiceBus;
  using Microsoft.WindowsAzure.ServiceRuntime;
  
  namespace WorkerRoleWithSBQueue1 {
      public class WorkerRole : RoleEntryPoint {
          private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
          private readonly ManualResetEvent _runCompleteEvent = new ManualResetEvent(false);
  
          // The name of your queue
          private const string ServiceBusConnectionString = "&lt;your_connection_string&gt;";
          private const string TopicName = "test-topic";
          private const string SubscriptionName = "test-sub1";
  
          // _client is thread-safe. Recommended that you cache 
          // rather than recreating it on every request
          private SubscriptionClient _client;
  
          public override void Run() {
              Trace.WriteLine("Starting processing of messages");
  
              try {
                  this.RunAsync(this._cancellationTokenSource.Token).Wait();
              } catch (Exception e) {
                  Trace.WriteLine("Exception");
                  Trace.WriteLine(e.ToString());
              } finally {
                  Trace.WriteLine("Finally...");
                  this._runCompleteEvent.Set();
              }
          }
  
          public override bool OnStart() {
              // Set the maximum number of concurrent connections 
              ServicePointManager.DefaultConnectionLimit = 12;
  
              var result = base.OnStart();
  
              Trace.WriteLine("WorkerRole has been started");
  
              return result;
          }
  
          public override void OnStop() {
              // Close the connection to Service Bus Queue
              this._cancellationTokenSource.Cancel();
              this._runCompleteEvent.WaitOne();
  
              base.OnStop();
          }
  
          private async Task RunAsync(CancellationToken cancellationToken) {
              // Configure the client
              RegisterOnMessageHandlerAndReceiveMessages(ServiceBusConnectionString, TopicName, SubscriptionName);
  
              _runCompleteEvent.WaitOne();
  
              Trace.WriteLine("Closing");
              await _client.CloseAsync();
          }
  
          private void RegisterOnMessageHandlerAndReceiveMessages(string connectionString, string topicName, string subscriptionName) {
              _client = new SubscriptionClient(connectionString, topicName, subscriptionName);
  
              var messageHandlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler) {
                  // Maximum number of concurrent calls to the callback ProcessMessagesAsync(), set to 1 for simplicity.
                  // Set it according to how many messages the application wants to process in parallel.
                  MaxConcurrentCalls = 1,
  
                  // Indicates whether MessagePump should automatically complete the messages after returning from User Callback.
                  // False below indicates the Complete will be handled by the User Callback as in `ProcessMessagesAsync` below.
                  AutoComplete = false,
              };
  
              _client.RegisterMessageHandler(ProcessMessageAsync, messageHandlerOptions);
          }
  
          private async Task ProcessMessageAsync(Message message, CancellationToken token) {
              try {
                  // Process the message
                  Trace.WriteLine($"Received message: SequenceNumber:{message.SystemProperties.SequenceNumber} Body:{Encoding.UTF8.GetString(message.Body)}");
                  await _client.CompleteAsync(message.SystemProperties.LockToken);
              } catch (Exception e) {
                  // Handle any message processing specific exceptions here
                  Trace.Write(e.ToString());
                  await _client.AbandonAsync(message.SystemProperties.LockToken);
              }
          }
  
          private static Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs) {
              Console.WriteLine($"Message handler encountered an exception {exceptionReceivedEventArgs.Exception}.");
              var context = exceptionReceivedEventArgs.ExceptionReceivedContext;
              Console.WriteLine("Exception context for troubleshooting:");
              Console.WriteLine($"- Endpoint: {context.Endpoint}");
              Console.WriteLine($"- Entity Path: {context.EntityPath}");
              Console.WriteLine($"- Executing Action: {context.Action}");
              return Task.CompletedTask;
          }
      }
  }
  </code></pre>