Amazon SQS/SNS Configuration

MassTransit combines Amazon SQS (Simple Queue Service) with SNS (Simple Notification Service) to provide both send and publish support.

Amazon SQS

Amazon Simple Queue Service (SQS) is a fully managed message queuing service that enables you to decouple and scale microservices, distributed systems, and serverless applications. SQS eliminates the complexity and overhead associated with managing and operating message oriented middleware, and empowers developers to focus on differentiating work.

With SQS, you can send, store, and receive messages between software components at any volume, without losing messages or requiring other services to be always available. SQS makes it simple and cost-effective to decouple and coordinate the components of a cloud application.

SQS offers two types of queues, Standard and FIFO (First-In-First-Out). Standard queues offer best-effort ordering, which ensures that messages are generally delivered in the order in which they are sent. FIFO queues guarantee that messages are processed exactly once, in the order that they are sent, and they are designed to prevent duplicates.

Amazon SNS

Amazon Simple Notification Service (SNS) is a fully managed messaging service that enables you to send messages to multiple subscribers or endpoints. SNS supports multiple protocols including HTTP, HTTPS, email, and Lambda, and it can be used to send push notifications to mobile devices, or to process messages asynchronously using AWS Lambda.

SNS allows you to send a message to a “topic” which is a logical access point and communication channel. Subscribers can then subscribe to that topic to receive the messages.

SNS also provides a feature called fan-out delivery, which enables messages to be delivered to multiple subscribers in parallel, this allows SNS to handle high-throughput and burst traffic, and can improve the overall performance of your application.

MassTransit uses SNS to route published messages to SQS queues.

Caution

Amazon SNS does not support polymorphic message dispatch

Configuring a receive endpoint will use the message topology to create and subscribe SNS topics to SQS queues so that published messages will be delivered to the receive endpoint queue.

Configure the Amazon SQS transport

In the example below, the Amazon SQS host settings are configured. The configuration includes:

namespace AmazonSqsConsoleListener;

using System.Threading.Tasks;
using MassTransit;
using Microsoft.Extensions.Hosting;

public class Program
{
    public static async Task Main(string[] args)
    {
        await Host.CreateDefaultBuilder(args)
            .ConfigureServices((hostContext, services) =>
            {
                services.AddMassTransit(x =>
                {
                    x.UsingAmazonSqs((context, cfg) =>
                    {
                        cfg.Host("us-east-2", h =>
                        {
                            h.AccessKey("your-iam-access-key");
                            h.SecretKey("your-iam-secret-key");
                        });
                    });
                });
            })
            .Build()
            .RunAsync();
    }
}

• The Amazon SQS host
  • Region name: us-east-2
  • Access key and secret key used to access the resources

Amazon SQS host settings

Property	Description
Region	The AWS Region
Scope	Will be used as a prefix for queue/topic name
AccessKey	Access Key
SecretKey	Access Secret

Configure the Amazon SQS transport options

All AWS SQS transport options can be configured using the .Host() method. The most commonly used settings can be configured via transport options.

services.AddOptions<AmazonSqsTransportOptions>()
    .Configure(options =>
    {
        // configure options manually, but usually bind them to a configuration section
    });

Configure a service scope

Because there is only ever one “SQS/SNS” per AWS account, it can be helpful to scope a service’s queues and topics. This will prefix all SQS queues and SNS topics with scope value.

services.AddMassTransit(x =>
{
    x.UsingAmazonSqs((context, cfg) =>
    {
        cfg.Host("us-east-2", h =>
        {
            h.AccessKey("your-iam-access-key");
            h.SecretKey("your-iam-secret-key");

            // specify a scope for all topics
            h.Scope("dev", true);
        });

        // additionally include the queues
        cfg.ConfigureEndpoints(context, new DefaultEndpointNameFormatter("dev-", false));
    });
});

Example IAM Policy

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "SqsAccess",
      "Effect": "Allow",
      "Action": [
        "sqs:SetQueueAttributes",
        "sqs:ReceiveMessage",
        "sqs:CreateQueue",
        "sqs:DeleteMessage",
        "sqs:SendMessage",
        "sqs:GetQueueUrl",
        "sqs:GetQueueAttributes",
        "sqs:ChangeMessageVisibility",
        "sqs:PurgeQueue",
        "sqs:DeleteQueue",
        "sqs:TagQueue"
      ],
      "Resource": "arn:aws:sqs:*:YOUR_ACCOUNT_ID:*"
    },
    {
      "Sid": "SnsAccess",
      "Effect": "Allow",
      "Action": [
        "sns:GetTopicAttributes",
        "sns:ListSubscriptionsByTopic",
        "sns:GetSubscriptionAttributes",
        "sns:SetSubscriptionAttributes",
        "sns:CreateTopic",
        "sns:Publish",
        "sns:Subscribe"
      ],
      "Resource": "arn:aws:sns:*:YOUR_ACCOUNT_ID:*"
    },
    {
      "Sid": "SnsListAccess",
      "Effect": "Allow",
      "Action": [
        "sns:ListTopics"
      ],
      "Resource": "*"
    }
  ]
}

Configure an SNS subscription

Any topic can be subscribed to a receive endpoint, as shown below. The topic attributes can also be configured, in case the topic needs to be created.

services.AddMassTransit(x =>
{
    x.UsingAmazonSqs((context, cfg) =>
    {
        cfg.Host("us-east-2", h =>
        {
            h.AccessKey("your-iam-access-key");
            h.SecretKey("your-iam-secret-key");
        });

        cfg.ReceiveEndpoint("input-queue", e =>
        {
            // disable the default topic binding
            e.ConfigureConsumeTopology = false;

            e.Subscribe("event-topic", s =>
            {
                // set topic attributes
                s.TopicAttributes["DisplayName"] = "Public Event Topic";
                s.TopicSubscriptionAttributes["some-subscription-attribute"] = "some-attribute-value";
                s.TopicTags.Add("environment", "development");
            });
        });
    });
});

Default roker Topology

With SQS/SNS, which supports topics and queues, messages are sent or published to SNS Topics and then routes those messages through subscriptions to the appropriate SQS Queues.

When the bus is started, MassTransit will create SNS Topics and SQS Queues for the receive endpoint.

The following messages are used in this example:

Here is the command contract for processing a file that was received:

namespace Acme;

public record ProcessFile
{
}

These are the event contracts for a consumer that receives files from a customer:

namespace Acme;

public record FileReceivedEvent
{
}

The consumers for these message contracts are shown below:

class ProcessFileConsumer :
    IConsumer<ProcessFile>
{
}

class FileReceivedConsumer :
    IConsumer<FileReceivedEvent>
{
}

class CustomerAuditConsumer :
    IConsumer<FileReceivedEvent>
{
}

Note

The broker topology can be customized using the topology API.

Send

These are the exchanges and queues for the example above when Sending a message:

Publish

These are the topics and queues for the example above when Publishing a message:

Fault

These are the exchanges and queues used when messages fail. The failing message gets forwarded to an _error queue by default. The following diagram shows which exchanges and queues are used when a message fails to be processed and is deadlettered for the example above.

Go to Exceptions to learn more on exception and faults

Retry messages

Faulted messages by default are forwarded to the corresponding *_error queue:

Messages can be inspected by:

1. Selecting the queue
2. Selecting Send and receive messages
3. In the Receive messages panel, select Poll for messages

A list of message appears and a message can be inspected by clicking it:

Configure the consumer queue its dead-letter queue:

1. Select a consumer queue, for example BillOrder
2. Select Edit in the Dead-letter queue panel
3. Enable Set this queue to receive undeliverable messages in the Dead-letter queue panel
4. Select the corresponding consumer queue (here arn:aws:sqs:***:***:BillOrder_error )
5. Select Save

The _error dead-letter needs to be configured to set a re-drive (return) queue;

1. Select a consumer _error queue, for example BillOrder_error
2. Select Edit in the Dead-letter queue panel
3. Enable the Redrive allow policy panel
4. Select By queue
5. Select the consumer queue, for example BillOrder (here arn:aws:sqs:***:***:BillOrder )
6. Select Save

The Start DLQ redrive button in the upper-right corner should now be enabled.

AmazonSQS isn’t aware that MassTransit forwarded the messages currently in the queue from another queue unless it actually forwarded the messages it self because the delivery count was exceeded. This requires the following steps to set a custom destination:

1. Select Start DLQ redrive
2. Select Redrive to a custom destination
3. Select the correct consumer queue, for example BillOrder
4. Scroll down and select 1 or more messages
5. Select DLQ redrive in the lower-right corner

Caution

This will re-drive all messages in the dead-letter queue, not just the polled messages.

Note

For advanced alternatives to managing failures see Exceptions - Managing Faults