Receiving Azure Service Bus Messages Using Queues

Introduction

Azure Service Bus is a message broker used by enterprises to send and receive messages or data payloads between applications. This approach is helpful when we want to avoid using APIs to send and receive data between applications. This approach is also very helpful for microservices communication when we want to exchange messages between multiple microservices. Queues are used for point-to-point communication. Message broker uses the FI-FO approach to send messages from source to destination. It assures at least one delivery of a message till certain attempts. Once the number of attempts exceeds at the receiver's end to acknowledge the messages, the messages are pushed into the dead letter queue. We will take a walkthrough of how to receive messages using queues.

The messages are received and sent using these three protocols:

AMQP - Advanced Messaging Queuing Protocol
SBMP - Service Bus Messaging Protocol
HTTP - HyperText Transfer Protocol

Setting messages is also very flexible in Azure Service Bus. Not every message or data payload is useful. We can do various operations on messages to fit our scenario. We can acknowledge them, abandon it, defer the message and move the message to alternate queue i.e. Dead letter queue that will not be received again by normal means.

Approaches

There are two approaches offered by Azure Service Bus SDK to receive messages.

Service Bus Receiver - It is used to simply receive the messages and perform operations on them. It is a vanilla implementation that does core operations.
Service Bus Processor - It is a wrapper around service bus receiver and works on an event based model to receive messages and perform operations on them. It contains a callback method to perform operations on received messages.

For more control, we must use a Service Bus Receiver.

Setup

Azure Service Bus Instance (Basic and above)
Visual Studio 2022
.NET 6

Provisioning Instance and Creating Messages

Open portal.azure.com
Search Service Bus
Click + Create
Fill the below details,
Click Next, Next and Create
Once created, Click Go to Resource.
Open Queues under Entities and on right side click ” + Queue”
Fill the following configuration
Click Create.
Open Queue.
Select Service Bus Explorer (preview) under Settings.
Select Send Tab
Create a JSON payload {"id":1,"message":"received"}
Click Send
Under Settings section select “Shared access policies”
Click on “ + Add ”
Write policy name and select the checkbox to listen and click Create.

Note: We must always follow the principle of least privilege when assigning any permissions
Select the policy name and copy Primary or Secondary Connection String.

Code Walkthrough

Create a .NET 6 Console Application.
Right click on the project.
Select Manage Nuget Packages..
Install Azure.Messaging.ServiceBus!

Using Service Bus Receiver:

using Azure.Messaging.ServiceBus;
//Configure ServiceBusClient
//In constructor, pass the connection string
await using var client = new ServiceBusClient("Endpoint=sb://learn-asb-cin-varuntutorial.servicebus.windows.net/;SharedAccessKeyName=sas-listen-policy;SharedAccessKey=<removed>;EntityPath=data-receive-queue");
//Call method CreateReceiver of client object and pass the queue name 
var receiver = client.CreateReceiver("data-receive-queue");
//Call method ReceiveMessageAsync from receiver object to receive 1 message at a time
var message = await receiver.ReceiveMessageAsync();
string body = message.Body.ToString();
Console.WriteLine(body);
//Acknowledge the message by marking it as complete
await receiver.CompleteMessageAsync(message);

Now, when we rerun the above code we will get no response and application will break as there is no exception handling because the message has been processed and we need to add a message again.

Using Service Bus Processor:

using Azure.Messaging.ServiceBus;
//Configure ServiceBusClient
//In constructor, pass the connection string
await using var client = new ServiceBusClient("Endpoint=sb://learn-asb-cin-varuntutorial.servicebus.windows.net/;SharedAccessKeyName=sas-listen-policy;SharedAccessKey=<Removed>;EntityPath=data-receive-queue");
//Create processor instance from client object, pass queue name and ServiceBisProcessorOptions instance
var processor = client.CreateProcessor("data-receive-queue", new ServiceBusProcessorOptions());
//Register ProcessMessageAsync event method
processor.ProcessMessageAsync += async (messageArgs) =>
{
    Console.WriteLine(messageArgs.Message?.Body.ToString());
    await messageArgs.CompleteMessageAsync(messageArgs.Message);
};
//Register ErrorAsync event method
processor.ProcessErrorAsync += async (messageArgs) =>
{
    Console.WriteLine(messageArgs.Exception.Message);
};
//Start the processor
await processor.StartProcessingAsync();
//In case of console application prevent application shutdown by adding keyboard interrupt
Console.ReadKey();

Now push messages to the service bus explorer and we will get messages in realtime.

Note: To Stop Processor and Dispose the connection use

await processor.StopProcessingAsync();
await processor.DisposeAsync();

For client

await client.DisposeAsync();

That’s it!! Thanks for reading! Hope you liked it. Please feel free to add your comments and suggestions.