AWS Message Processing Framework for .NET - Simplifying AWS Based Messaging Applications in .NET

The AWS Message Processing Framework for .NET is a wrapper over the AWS Messaging Services like Amazon SNS, SQS, and Event Bridge that helps reduce a significant amount of boilerplate code needed to interact with these AWS Services, further simplifying the development of messaging in .NET applications. In this article, we will explore this framework, and build sample applications that would use SQS, and SNS.

Why AWS Message Processing Framework for .NET?

From a previous article where we discussed working with Amazon SQS in .NET Applications, here is a code snippet meant for consuming messages from an SQS Queue.

var queueUrl = await GetQueueUrl(OrderCreatedEventQueueName);
var receiveRequest = new ReceiveMessageRequest()
{
    QueueUrl = queueUrl
};
while (!stoppingToken.IsCancellationRequested)
{
    var response = await _sqsClient.ReceiveMessageAsync(receiveRequest);
    if (response.Messages.Count > 0)
    {
        foreach (var message in response.Messages)
        {
            _logger.LogInformation("received message from queue");
            //perform some processing.
            //mock 2 seconds delay for processing
            Task.Delay(2000).Wait();
            await _sqsClient.DeleteMessageAsync(queueUrl, message.ReceiptHandle);
        }
    }
}

This usually includes the following steps,

1. Receiving the actual message.
2. Deserializing the message to it’s .NET type.
3. Processing.
4. Deleting the message from the Queue.

This involves a lot of boilerplate code, which AWS Message Processing Framework for .NET is built to reduce.

With AWS Message Processing Framework for .NET, you would just have to write the following to consume messages from an SQS Queue.

public class OrderCreatedEventHandler(ILogger<OrderCreatedEventHandler> logger) : IMessageHandler<OrderCreatedEvent>
{
    public Task<MessageProcessStatus> HandleAsync(MessageEnvelope<OrderCreatedEvent> messageEnvelope, CancellationToken token = default)
    {
        logger.LogInformation($"received order created message for order id: {messageEnvelope.Message.OrderId}");
        // do some processing
        return Task.FromResult(MessageProcessStatus.Success());
    }
}

The Framework would handle all the boilerplate codes internally, and lets you worry only about writing business logics for processing the received message, and nothing else.

We will build a .NET 8 Solution, which has 2 ASP.NET Core Web API project that will act as the Publisher and Consumer. We will try to use the AWS Message Processing Framework for communication between these services.

SQS and AWS Message Processing Framework

In this section, we will explore sending and receiving Message over an SQS Queue between 2 .NET 8 Applications.

I have named my Services as ServiceA and ServiceB. Install the following packages on both of the projects.

Install-Package AWS.Messaging

Note that this package is still in pre-release.

Publisher

First up, let’s open up AWS Management Console, and create a SQS Queue for us to test with.

I have named my queue as OrderQueue with minimal configuration. Keep the Queue URL handy, as we will be using it shortly.

The idea is to publish messages of type OrderCreatedEvent from ServiceA to ServiceB. For this, create a new .NET Class Library Project named contracts, and add in the following class.

public class OrderCreatedEvent : IEvent
{
    public Guid OrderId { get; }
    public Guid CustomerId { get; }
    public DateTime CreatedDate { get; }
    public OrderCreatedEvent(Guid orderId, Guid customerId)
    {
        this.OrderId = orderId;
        this.CustomerId = customerId;
        CreatedDate = DateTime.UtcNow;
    }
}
public interface IEvent
{
}

The only reason we have decided to put this contract to a separate project is because this has to be shared by both of our Publisher and the Consumer projects.

To register the framework with the DI Container of your ASP.NET Core Publisher Application, navigate to the Program.cs of ServiceA and add in the following

builder.Services.AddAWSMessageBus(builder =>
{
    builder.AddSQSPublisher<OrderCreatedEvent>("https://sqs.us-east-1.amazonaws.com/821175633958/OrderQueue", nameof(OrderCreatedEvent));
});

Here, we are registering that we will be publishing a message of type OrderCreatedEvent to the mentioned SQS URL, and with an identifier name of OrderCreatedEvent.

I have also added a sample endpoint on to ServiceA’s Program.cs file.

app.MapPost("/order", async (ISQSPublisher publisher) =>
{
    // call order creation service
    var orderId = Guid.NewGuid();
    var customerId = Guid.NewGuid();

    // create order created event
    var orderCreatedEvent = new OrderCreatedEvent(orderId, customerId);

    // publish the message
    await publisher.SendAsync(orderCreatedEvent);

    return Results.Created();
});

This is a simple POST method, that mocks an order creation endpoint, and internally creates an order ID and customer ID. Then we use these IDs to create a message of type OrderCreatedEvent. Note that we have also injected an instance of ISQSPublisher service into this endpoint. The message is published using this instance. This will push our message to the AWS SQS Queue, as mentioned in the URL.

Note that this framework allows you to use a generic interface IMessagePublisher, which can push the message to any supported AWS Service. Or, you can also go for a very specific service by using interfaces like ISQSPublisher, ISNSPublisher, or the IEventBridgePublisher.

If you want to customize even further, you can get access to SQSOptions while publishing the messages. Here’s how.

await publisher.SendAsync(orderCreatedEvent, new SQSOptions
{
    DelaySeconds = 10,
    QueueUrl = "custom-url-to-override",
    MessageGroupId = "group-id"
});

Here, you can mention SQS specific configurations, like Delay Time, Queue URL if you want to override the URL that you had set during the Service Registration, and even the Message Group ID if you are working with FIFO Queues.

Now that we have our Publisher configured, let’s use ServiceB to poll and consume the message from the SQS Queue.

Consumer

At ServiceB, ensure that you have installed the same AWS.Messaging package.

First, let set up the Message Handler class. Create a new folder named Handlers, and create a new class OrderCreatedEventHandler. As the name suggests, this would contain the message processing logic for message of type OrderCreatedEvent.

public class OrderCreatedEventHandler(ILogger<OrderCreatedEventHandler> logger) : IMessageHandler<OrderCreatedEvent>
{
    public Task<MessageProcessStatus> HandleAsync(MessageEnvelope<OrderCreatedEvent> messageEnvelope, CancellationToken token = default)
    {
        logger.LogInformation($"received order created message for order id: {messageEnvelope.Message.OrderId}");
        return Task.FromResult(MessageProcessStatus.Success());
    }
}

We will just have to implement the IMessageHandler<OrderCreatedEvent> interface’s HandleAsync method. The message coming from SQS would be wrapped under a MessageEnvelope ready for consumption.

I have kept the code simple for demonstration purposes. As soon as we receive the message from the SQS Queue, we will just log the incoming Order’s ID to the console.

You will have to return a MessageProcessStatus.Success() which ensures that the message is processed as expected, and is deleted from the SQS queue. If you return a MessageProcessStatus.Failed() response, the message stays in the queue, retries, or moved into a Dead Letter Queue as per your configuration while creating the SQS Queue.

Now, we need to register ServiceB as the consumer. Navigate to the Program.cs class and add the following.

builder.Services.AddAWSMessageBus(builder =>
{
    builder.AddSQSPoller("https://sqs.us-east-1.amazonaws.com/821175633958/OrderQueue");
    builder.AddMessageHandler<OrderCreatedEventHandler, OrderCreatedEvent>(nameof(OrderCreatedEvent));
});

This registers an SQS Queue URL that our application will poll for incoming messages. Then, at line #4, we will register the message handler that we have written in the previous step, which is the OrderCreatedEventHandler. So, anytime a message of type OrderCreatedEvent is polled, the OrderCreatedEventHandler handler would be invoked, and processed.

Further, you can customize the SQS Poller even more. For example,

builder.AddSQSPoller("https://sqs.us-east-1.amazonaws.com/821175633958/OrderQueue", options =>
{
    options.MaxNumberOfConcurrentMessages = 10;
    options.VisibilityTimeout = 60;
    options.WaitTimeSeconds = 10;
});

Note that these properties already have some default values set.

Here, VisibilityTimeout is one of the core features of SQS, where, as soon as a client picks up a message from the queue, the corresponding message would still stay in the queue, but would be hidden from other clients. For a default duration of 30 seconds other requests would not pick up this message. If the client #1 sends a success message back, the message is deleted from the queue, else it stays to be picked up by another client.

That’s it with the SQS Messaging. Let’s build and run both ServiceA and ServiceB in parallel. On ServiceA, I will invoke the /order POST endpoint, which should ideally send out a message to our SQS Queue, and in turn would be pulled into ServiceB that should throw some logs.

As you can see, our ServiceB is able to poll and process the messages sent out from ServiceA with a lot less boilerplate code, right!

SNS and AWS Message Processing Framework

Similarly, you can also use the ISNSPublisher to work easily with AWS SNS Topics. For instance, at your publisher you will have to register the topic by using the following.

builder.Services.AddAWSMessageBus(builder =>
{
    builder.AddSQSPublisher<OrderCreatedEvent>("queueUrl", nameof(OrderCreatedEvent));
    builder.AddSNSPublisher<SomeNotification>("topicUrl");
});

And, you can use the ISNSPublisher interface to publish your notifications.

app.MapPost("/notify", async (ISNSPublisher publisher) =>
{
    var notification = new SomeNotification();
    await publisher.PublishAsync(notification, new SNSOptions
    {
        Subject = "something",
        MessageGroupId = "id"
    });
    return Results.Created();
});

Here as well, you can specify additional options while publishing the notification to your topic, like Subject, Metadata, and MessageGroupId.

That’s a wrap for this article.

Do not forget to claim your Free 25$ AWS Credits by clicking on the link attached to the top of this article. This will help you explore AWS more!

The AWS.Messaging package would definitely make life easier for developers working with AWS Messaging Services in .NET Applications, giving an easier, and well-defined blueprint. You can find the entire source code here. Thanks!