Azure Service Bus Queues with .NET Core Services

This article shows how to implement the ASP.NET Core API application to communicate with the ASP.NET Core MVC application using Azure Service Bus. The ASP.NET Core MVC application is used to push messages to Azure Service Bus Queue and the ASP.NET Core API is used to consume the messages from the queue for further processing (here to send email).

Azure Service Bus Queues are a cloud-based message queuing service that enables reliable communication and messaging between applications and services. They are a key component of Azure's messaging infrastructure and allow you to build distributed and decoupled systems that are scalable, reliable, and highly available.

One of the key benefits of Azure Service Bus Queues is their ability to store and forward messages, even if the receiving application is temporarily offline. This makes them ideal for building resilient, scalable systems that can handle bursts of traffic and recover gracefully from failures.

Source Code: https://github.com/nandkishor-yadav/azure-service-bus-queue

Setting up the Azure Service Bus Queue

Azure Service Bus is set up as described here.

To implement the messaging, a queue or a topic can be used. A queue is used as the messaging type in this example. Once the data has been received, it is removed from the queue. The applications are implemented as follows:

Implementing a Service Bus Queue

Microsoft.Azure.ServiceBus Nuget package is required to implement the Azure Service Bus clients. To run the example, create your own Azure Service Bus and set the connection strings in secrets or Azure Key Vault for the projects. In this post, I have hard coded the connection strings for simplicity.

The SendMessage method takes a PayloadForServiceBus type as a parameter and adds this to the message as a JSON payload.

using AzureBusServiceMVC.Models;
using Microsoft.Azure.ServiceBus;
using Microsoft.Extensions.Configuration;
using Newtonsoft.Json;
using System.Text;
using System.Threading.Tasks;

namespace AzureBusServiceMVC.Services
{
    public class ServiceBusSender
    {
        private readonly QueueClient _queueClient;
        private readonly IConfiguration _configuration;
        private const string QUEUE_NAME = "sendemailtest";

        public ServiceBusSender(IConfiguration configuration)
        {
            _configuration = configuration;
            _queueClient = new QueueClient("SET-YOUR-CONNECTION-STRING", QUEUE_NAME);
        }

        public async Task SendMessage(PayloadForServiceBus payload)
        {
            string data = JsonConvert.SerializeObject(payload);
            Message message = new Message(Encoding.UTF8.GetBytes(data));

            await _queueClient.SendAsync(message);
        }
    }
}

The ServiceBusSender is registered to the IOC of the ASP.NET Core MVC application in the Startup class.

        public void ConfigureServices(IServiceCollection services)
        {
            services.AddControllersWithViews();
            services.AddScoped<ServiceBusSender>();
        }

This service is used in the controller of the ASP.NET Core MVC application.

        [HttpPost]
        [ProducesResponseType(typeof(PayloadForServiceBus), StatusCodes.Status200OK)]
        [ProducesResponseType(typeof(PayloadForServiceBus),StatusCodes.Status409Conflict)]
        public async Task<IActionResult> Create([Required] PayloadForServiceBus request)
        {
            await _serviceBusSender.SendMessage(new PayloadForServiceBus
            {
                Email = request.Email,
                Message = request.Message
            });

            return RedirectToAction("Index", "Home");
        }

Consuming messages from the Queue and sending email

To receive the message from the Azure Service Bus, ServiceBusConsumer implements the IServiceBusConsumer interface. The connection string is hardcoded for simplicity. You may want to set it in project secrets or Azure Key Vault for production deployment.  The RegisterOnMessageHandlerAndReceiveMessages method adds the event handler for the messages and uses the ProcessMessagesAsync method to process these. The ProcessMessagesAsync method converts the message to an object and calls the IProcessData interface to complete the processing of the message.

using System.Threading.Tasks;

namespace AzureServiceBusDemo.Services
{
    public interface IServiceBusConsumer
    {
        void RegisterOnMessageHandlerAndReceiveMessages();

        Task CloseQueueAsync();
    }
}

using AzureServiceBusDemo.Models;
using Microsoft.Azure.ServiceBus;
using Newtonsoft.Json;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace AzureServiceBusDemo.Services
{
    public class ServiceBusConsumer : IServiceBusConsumer
    {

        private readonly QueueClient _queueClient;
        private const string QUEUE_NAME = "sendemailtest";
        private readonly IProcessData _processData;

        public ServiceBusConsumer(IProcessData processData)
        {
            _processData = processData;
            _queueClient = new QueueClient("SET-YOUR-CONNECTION-STRING", QUEUE_NAME);
        }

        public void RegisterOnMessageHandlerAndReceiveMessages()
        {
            var messageHandlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler)
            {
                MaxConcurrentCalls = 1,
                AutoComplete = false
            };

            _queueClient.RegisterMessageHandler(ProcessMessagesAsync, messageHandlerOptions);
        }

        private async Task ProcessMessagesAsync(Message message, CancellationToken token)
        {
            var myPayload = JsonConvert.DeserializeObject<PayloadForServiceBus>(Encoding.UTF8.GetString(message.Body));
            await _processData.Process(myPayload);
            await _queueClient.CompleteAsync(message.SystemProperties.LockToken);
        }

        private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
        {
            _ = exceptionReceivedEventArgs.ExceptionReceivedContext;

            return Task.CompletedTask;
        }

        public async Task CloseQueueAsync()
        {
            await _queueClient.CloseAsync();
        }
    }
}

Add the support for the Azure Service Bus in the Startup class.

 public void ConfigureServices(IServiceCollection services)
        {
            services.AddControllers();

            services.AddSingleton<IServiceBusConsumer, ServiceBusConsumer>();
            services.AddTransient<IProcessData, ProcessData>();
            services.AddTransient<IEmailSender, EmailSender>();
            services.Configure<EmailSenderOptions>(Configuration);
        }

public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }

            app.UseHttpsRedirection();

            app.UseRouting();

            app.UseAuthorization();

            app.UseEndpoints(endpoints =>
            {
                endpoints.MapControllers();
            });

            var bus = app.ApplicationServices.GetService<IServiceBusConsumer>();
            bus.RegisterOnMessageHandlerAndReceiveMessages();
        }

The ProcessData implements the IProcessData interface which receives the messages. In this post, I am sending an email with the message received from the Azure Service Bus Queue but you may do anything with the received message as per your requirements.

In addition to the basic send and receive functionality, Azure Service Bus Queues also support a number of other features, such as dead lettering, sessions, and transactions. You can use these features to build more advanced messaging scenarios and ensure the reliability and scalability of your applications.

In conclusion, Azure Service Bus Queues are a powerful and reliable messaging service that can be easily integrated into .NET Core applications. Whether you are building distributed systems, decoupling services, or just need a reliable way to communicate between applications, Azure Service Bus Queues can help you build scalable and resilient solutions.