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).

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 hardcoded 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 message 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.

Thank you for reading this article.