Creating Azure Service Bus Queues And Sending Messages To Queues

Service bus

The reason behind the use of service buses is to decouple applications and services. As we need to decouple applications and services, the data can be transferred between different applications and services using messages. The messages can be sent to the queue on the service bus.

Creating a service bus queue

Most of you might be aware of how to create a service bus queue in Azure, but let me review.

Way 1

Log in to the Azure portal and create a service bus, namespace. Once that is done, you can see the queues and topics on the left-hand side. Once we click on queue, we can click on Add and create a queue from the Azure portal.

Way 2

If we have service Bus Explorer, we can log in to the service Bus Explorer with the connection string of the service bus and right click on the queue, and create the queue. We can have the settings enabled for the queue.

Some of the useful settings are

Enable sessions

This allows a guarantee of processing the messages sequentially. This follows the pattern of FIFO.

Enabled dead lettering

This allows the messages to be in dead letters if the receiver fails to receive the messages in time. The message will not expire in a dead letter until and unless it is viewed.

Configuring the Service Bus queue in the development environment.

Install the Microsoft.Azure.ServiceBus NuGet package to the project where you are setting up the message queue.

First, let's create a class for configuring the QueueClient and have a generic method for sending the message to the queue.

using Azure.ServiceBus;
using Microsoft.Azure.ServiceBus;
using System;
using System.Collections.Generic;
using System.Text;
namespace AzureFunctionsUnitTesting.Service_Bus
{
    public class QueueSender : IQueueSender
    {
        static Dictionary<string, IQueueClient> _queueClients;
        public QueueSender(string connectionString, List<Queue> queues)
        {
            _queueClients = new Dictionary<string, IQueueClient>();
            if (!string.IsNullOrEmpty(connectionString))
            {
                foreach (Queue queue in queues)
                {
                    string queueId = queue.Id;
                    string queueName = queue.Name;
                    _queueClients.Add(queueId, new QueueClient(connectionString, queueName));
                }
            }
        }
        public async void SendMessageAsync(string queueName, string messageItem, string messageLabel = null, Dictionary<string, object> messageProperties = null)
        {
            try
            {
                Message message = BuildMessage(messageItem, messageLabel, messageProperties);
                await _queueClients[queueName].SendAsync(message);
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
        private Message BuildMessage(string messageItem, string messageLabel, Dictionary<string, object> messageProperties)
        {
            Message message = new Message(Encoding.UTF8.GetBytes(messageItem));
            message.UserProperties["TimeStampUTC"] = DateTime.UtcNow;
            message.Label = messageLabel;
            if (messageProperties != null)
            {
                foreach (var messageProperty in messageProperties)
                {
                    message.UserProperties[messageProperty.Key] = messageProperty.Value;
                }
            }
            return message;
        }
    }
}

using AzureFunctionsUnitTesting.Interfaces;
using AzureFunctionsUnitTesting.Service_Bus;
using Microsoft.Azure.ServiceBus;
namespace AzureFunctionsUnitTesting
{
    public class MessageQueueSender : IMessageQueueSender
    {
        private readonly IQueueSender _queueSender;
        public MessageQueueSender(IQueueSender queueSender)
        {
            _queueSender = queueSender;
        }
        public void SendMessageToQueue()
        {
            Message message = new Message();
            message.SessionId = "session1";
            string queueName = "testqueue";
            string content = "Hello Service Bus Queue";
            _queueSender.SendMessageAsync(queueName, content);
        }
    }
}

using AzureFunctionsUnitTesting.Service_Bus;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Moq;
using System.Collections.Generic;
namespace AzureFunctionsUnitTesting.Tests
{
    [TestClass]
    public class MessageQueueSenderTests
    {
        public readonly Mock<IQueueSender> _queueSender;
        public List<string> _queueMessages;
        public MessageQueueSenderTests()
        {
            _queueSender = new Mock<IQueueSender>();
            _queueMessages = new List<string>();
            _queueSender.Setup(x => x.SendMessageAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<Dictionary<string, object>>()))
                       .Callback<string, string, string, Dictionary<string, object>>(async (a, b, d, e) =>
                       {
                           _queueMessages.Add($"{a}");
                       });
        }
        [TestInitialize()]
        public virtual void Initialise() { }
        [TestMethod]
        public void SendMessageToQueue()
        {
            MessageQueueSender messageQueueSender = new MessageQueueSender(_queueSender.Object);
            messageQueueSender.SendMessageToQueue();
            // Check if message is delivered to queue
            Assert.AreEqual(1, _queueMessages.Count, "Message sending failed");
        }
    }
}