██FR█████ █INTELL███████████

Migrating to Brighter V10 with AWS SNS/SQS


This content originally appeared on DEV Community and was authored by Rafael Andrade

In previous articles, I covered Brighter integration with AWS SNS/SQS and Brighter V10 RC1. This guide focuses on migrating to Brighter V10, emphasizing AWS SNS/SQS configuration changes and breaking updates.

New Features in Brighter V10 for AWS SNS/SQS

Brighter V10 introduces significant enhancements for AWS SNS/SQS:

  • Direct SQS Support: Publish/consume messages from SQS without requiring SNS
  • FIFO Support: Full compatibility with SNS/SQS FIFO queues
  • LocalStack Integration: Improved support for local AWS emulation

Requirement

Brighter Recap

Before continuing about RabbitMQ configuration, let’s recap what we already know about Brighter.

Request (Command/Event)

Define messages using IRequest:

public class Greeting() : Event(Guid.NewGuid())
{
    public string Name { get; set; } = string.Empty;
}
  • Commands: Single-recipient operations (e.g., SendEmail).
  • Events: Broadcast notifications (e.g., OrderShipped).

Message Mapper (Optional)

Translates between Brighter messages and your app objects:

public class SqsFifoMapper : IAmAMessageMapperAsync<SqsFifoEvent>
{
    public Task<Message> MapToMessageAsync(SqsFifoEvent request, Publication publication,
        CancellationToken cancellationToken = new CancellationToken())
    {
        return Task.FromResult(new Message(new MessageHeader
            {
                MessageId = request.Id,
                Topic = publication.Topic!,
                PartitionKey = request.PartitionValue, // FIFO requirement
                MessageType = MessageType.MT_EVENT,
                TimeStamp = DateTimeOffset.UtcNow
            }, 
            new MessageBody(JsonSerializer.SerializeToUtf8Bytes(request, JsonSerialisationOptions.Options))));
    }

    public Task<SqsFifoEvent> MapToRequestAsync(Message message, CancellationToken cancellationToken = new CancellationToken())
    {
        return Task.FromResult(JsonSerializer.Deserialize<SqsFifoEvent>(message.Body.Bytes, JsonSerialisationOptions.Options)!);
    }

    public IRequestContext? Context { get; set; }
}

V10 Change: For async pipeline now require IAmAMessageMapperAsync. For FIFO SNS/SQS it’s necessary to implement a custom message mapper where it’ll set the partition key.

Request Handler

Processes incoming messages:

public class GreetingHandler(ILogger<GreetingHandler> logger) : RequestHandler<Greeting>
{
    public override Greeting Handle(Greeting command)
    {
        logger.LogInformation("Hello {Name}", command.Name);
        await processor.PostAsync(new Farewell { Name = command.Name }, cancellationToken: cancellationToken);
        return base.Handle(command);
    }
}

Configuring Brighter with AWS SNS/SQS

1. Connection Setup

Define AWS SNS connection details:

 var connection = new AWSMessagingGatewayConnection(new BasicAWSCredentials("test", "test"), 
            RegionEndpoint.USEast1,
            cfg => cfg.ServiceURL = "http://localhost:4566" // LocalStack
);

2. SQS Subscription

Subscribe to a SQS, in this sample I’ll show to how to connect the SNS to SQS, SQS to SQS and SNS to SQS FIFO:

 .AddServiceActivator(opt =>
{
    opt.Subscriptions = [
         // SNS → SQS
         new SqsSubscription<Greeting>(
             "greeting-subscription", // Optional
             "greeting-queue", // SQS queue name
             ChannelType.PubSub,  // Required for SNS
             "greeting.topic".ToValidSNSTopicName(), // SNS Topic Name
             bufferSize: 2, 
             messagePumpType: MessagePumpType.Proactor),

          // SQS → SQS (Point-to-Point)
         new SqsSubscription<Farewell>(
             new SubscriptionName("farawell-subscription"), // Optional
             new ChannelName("farewell.queue"), // SQS queue name
             ChannelType.PointToPoint, // Direct SQS
             new RoutingKey("farewell.queue".ToValidSQSQueueName()), // SNS Topic Name
             bufferSize: 2,
             messagePumpType: MessagePumpType.Proactor),

         // FIFO SNS → SQS
         new SqsSubscription<SnsFifoEvent>(
             new SubscriptionName("sns-sample-fifo-subscription"), // Optional
             new ChannelName("sns-sample-fifo".ToValidSQSQueueName(true)), // SQS queue name
             ChannelType.PubSub, 
             new RoutingKey("sns-sample-fifo".ToValidSNSTopicName(true)), // SNS Topic Name
             bufferSize: 2,
             messagePumpType: MessagePumpType.Proactor,
             topicAttributes: new SnsAttributes { Type = SqsType.Fifo }, // FIFO
             queueAttributes: new SqsAttributes(type: SqsType.Fifo)), // FIFO
    ];

    opt.DefaultChannelFactory = new ChannelFactory(connection);
})

3. SNS/SQS Producer Configuration

Publish events to a topic, because we want to produce to multi publication type (SNS and SQS) we will need to use the CombinedProducerRegistryFactory:

.UseExternalBus(opt =>
{
    opt.ProducerRegistry = new CombinedProducerRegistryFactory(
        // SNS Producers
        new SnsMessageProducerFactory(connection, [
            new SnsPublication<Greeting>
            {
                Topic = "greeting.topic".ToValidSNSTopicName(), 
                MakeChannels = OnMissingChannel.Create
            },
            new SnsPublication<SnsFifoEvent>
            {
                Topic = "sns-sample-fifo".ToValidSNSTopicName(true),
                MakeChannels = OnMissingChannel.Create,
                TopicAttributes = new SnsAttributes
                {
                    Type = SqsType.Fifo
                }
            }
       ]),

       // SQS Producers
       new SqsMessageProducerFactory(connection, [
           new SqsPublication<Farewell>
           {
               ChannelName = "farewell.queue".ToValidSQSQueueName(), 
               Topic = "farewell.queue".ToValidSQSQueueName(), 
               MakeChannels = OnMissingChannel.Create
            }
        ])
      ).Create();
});

Breaking Changes in Brighter V10

Brighter V10 introduces significant updates to AWS SNS/SQS integration. Below are the key breaking changes:

Message Mapper Overhaul

Default JSON Serialization :

In V9, message mappers were mandatory. In V10, JSON serialization is built-in unless custom logic is required.

Interface Split

  • IAmAMessageMapper (sync/reactor): For synchronous workflows.
  • IAmAMessageMapperAsync (async/proactor): For asynchronous workflows.

V10: Currently in V10 RC1, if you want to set a partition key, it’s necessary to implement the a custom mapper, I hope to be able to allow you to set it on the RequestContext before the final version is released

Changes on the IAmAMessageMapper 

// V10
IRequestContext? Context { get; set; }
Message MapToMessage(Greeting request, Publication publication);

// V9
Message MapToMessage(Greeting request);

Subscription

We had some changes on subscription.

Explicit Message Pump Types

The first one is before we had a field called runAsync or isAsync it was a boolean, to make everything clear we change it to messagePumpType and it’s the MessagePumpType(Reactor, Proactor, Unknown).

Property Renaming

On the AddServiceActivator where rename the ChannelFactory property to DefaultChannelFactory

SQS Subscription ChannelType

Now on AWS SqsSubscription Brighter have a channel type, it’s important to know if the subscription is to PubSub (SNS -> SQS) or PointToPoint (SQS -> SQS)

Queue and Topic attribute

Many property was moved from SqsSubscription and SnsPublication to SqsAttributes and SnsAttributes making the separation between them clear.

For FIFO, you must set the type as FIFO in both parameters queueAttributes and topicAttributes.

new SqsSubscription<SnsFifoEvent>(
   ...
   topicAttributes: new SnsAttributes { Type = SqsType.Fifo },
   queueAttributes: new SqsAttributes(type: SqsType.Fifo))

Publication

Here we also have some changes

Configuration Overhaul

Removed IAmAProducerRegistry. Use ExternalBusConfiguration to configure producers and outbox patterns:

// V10
.UseExternalBus(opt => { ... })

// V9
.UseExternalBus(new RmqProducerRegistryFactory(...))

Request Type Specification

The other change was about the Publication, you must setup an the request type or change the default IAmAPublicationFinder by UsePublicationFinder:

new SqsSubscription<Greeting>
{
    MakeChannels = OnMissingChannel.Create,
    Topic = new RoutingKey("greeting.topic"),
}

// or
new SqsSubscription
{
    MakeChannels = OnMissingChannel.Create,
    Topic = new RoutingKey("greeting.topic"),
    RequestType = typeof(Greeting)
}

SqsPublication

Now Brighter have a SqsPublication for SQS

Conclusion

Brighter V10 simplifies AWS integrations while adding critical features like FIFO support. Migrate by:

  1. Updating message mapper interfaces
  2. Configuring explicit ChannelType in subscriptions
  3. Using new publication models
  4. Leveraging built-in JSON serialization where possible

Full sample code: GitHub Repository


This content originally appeared on DEV Community and was authored by Rafael Andrade