Forem: Rafael Andrade

Trupe: Implementando Supervisor

Rafael Andrade — Sat, 28 Mar 2026 09:00:00 +0000

Na minha saga para implementar um Actor System, cobri até agora o uso básico de Actor. Agora, vou abordar o supervisor.

Supervisor

Caso você não saiba, no Model de Atores, um supervisor é uma peça fundamental para a tolerância a falhas. O supervisor monitora os filhos (atores) e reage quando um deles exibe um comportamento inadequado.

Um Supervisor é um tipo especial de ator, focado apenas em monitorar e realizar alguma ação quando os filhos apresentam mau comportamento.

Mau comportamento

No Model de Atores, um mau comportamento ocorre quando um ator sofre uma exceção durante a execução de uma mensagem.

Quando um ator lança uma exceção durante o processamento de uma mensagem, temos duas estratégias principais:

One For One (Um para Um): Reinicia apenas o filho que falhou.
All for One (Todos para Um): Reinicia todos os filhos, caso um deles falhe.

E o que fazer quando o reinício não funciona e ele continua falhando? Você pode parar o filho ou escalar o erro.

Reinício (Restart)

Outra questão é o que "reiniciar" significa no Model de Atores. No Model de Atores e no .NET, isso significa matar/descartar (dispose) aquela instância do ator e sua mailbox, mas manter a IActorReference. Se o ator for um supervisor, vamos parar e matar todos os filhos, tornando a IActorReference desses filhos inválida.

Tipos de supervisor

Antes de mergulhar no código, deixe-me mostrar quais são os supervisores que irei implementar:

Supervisor: Um supervisor preemptivo; não permite a adição de mais filhos após a inicialização.
DynamicSupervisor: Um supervisor dinâmico; permite criar filhos após a inicialização, mas suporta apenas a estratégia One For One.
PartitionSupervisor: Um supervisor baseado em partição que cria um pool fixo de workers idênticos e roteia as mensagens usando particionamento baseado em hash.

Implementação

Base (Foundation)

Antes de implementar os supervisores, precisamos de alguns blocos de construção: a estratégia de supervisão (Strategy), a ação de falha (FailureAction) e a política de reinício (RestartPolicy).

public enum Strategy
{
    OneForOne,
    AllForOne,
}

public enum FailureAction
{
    Restart,
    Stop,
    Escalate,
    Resume,
}

Além da estratégia, cada ator filho possui uma RestartPolicy que define se ele deve ser reiniciado:

public enum RestartPolicy
{
    Permanent, // Sempre reinicia, independentemente do motivo do término
    Transient, // Reinicia apenas se terminar de forma anormal
    Temporary, // Nunca reinicia
}

Agora, vamos primeiro criar uma interface chamada ISupervisor:

/// <summary>
/// Representa um ator supervisor que gerencia atores filhos.
/// </summary>
public interface ISupervisor : IActor
{
    /// <summary>
    /// Obtém a coleção de referências de atores filhos gerenciados por este supervisor.
    /// </summary>
    IEnumerable<IActorReference> Children { get; }
}

Também precisamos de uma forma de descrever como um filho deve ser criado. Para isso, temos o IChildSpecification:

public interface IChildSpecification
{
    Type ActorType { get; }
    IMailbox Mailbox { get; set; }
    RestartPolicy RestartPolicy { get; set; }
}

E a implementação padrão:

public record ChildSpecification : IChildSpecification
{
    public ChildSpecification(Type actorType)
    {
        ActorType = actorType;
    }

    public Type ActorType { get; }
    public IMailbox Mailbox { get; set; } = new ChannelMailbox();
    public RestartPolicy RestartPolicy { get; set; } = RestartPolicy.Permanent;
}

Cada ator supervisionado é rastreadu através de uma classe de metadados Child, que armazena a instância do ator, seu processo, mailbox, referência, política de reinício e contadores:

public class Child(
    IActor actor,
    IMailbox mailbox,
    ActorProcess process,
    LocalActorReference reference,
    RestartPolicy restartPolicy,
    Type actorType)
{
    public IActor Actor { get; set; } = actor;
    public ActorProcess Process { get; set; } = process;
    public IMailbox Mailbox { get; } = mailbox;
    public LocalActorReference Reference { get; } = reference;
    public RestartPolicy RestartPolicy { get; } = restartPolicy;
    public Type ActorType { get; } = actorType;
    public int RestartCount { get; set; } = 0;
    public DateTimeOffset LastRestartTime { get; set; } = DateTimeOffset.MinValue;
    public Dictionary<string, object> Metadata { get; } = [];
    public bool IsSupervisor => Actor is ISupervisor;
}

Supervisor

O Supervisor base é uma classe abstrata que implementa ISupervisor. Ele é preemptivo — os filhos são definidos durante a inicialização e não podem ser adicionados posteriormente.

public abstract partial class Supervisor(IActorFactory actorFactory, ILogger logger)
    : Actor,
        ISupervisor,
        IHandleActorMessage<AddActor>,
        IHandleActorMessage<ActorFailed>,
        IHandleActorMessage<ActorTerminated>,
        IAsyncDisposable
{

Ele expõe propriedades virtuais para que as subclasses possam customizar o comportamento:

    protected virtual Strategy Strategy => Strategy.OneForOne;
    protected virtual int MaxRestarts => 3;
    protected virtual TimeSpan RestartWindow => TimeSpan.FromSeconds(5);

Durante a inicialização, ele chama o método abstrato OnInitializeAsync, onde as subclasses definem seus filhos:

    public sealed override async ValueTask InitializeAsync(
        CancellationToken cancellationToken = default)
    {
        await OnInitializeAsync(cancellationToken);
        _initialized = true;
    }

    protected abstract ValueTask OnInitializeAsync(
        CancellationToken cancellationToken = default);

Quando um filho falha, o supervisor determina o que fazer através do GetFailureAction. Por padrão, ele reinicia o filho, a menos que o limite de reinícios seja excedido, caso em que ele escala a falha:

    protected virtual FailureAction GetFailureAction(Child child, Exception exception)
    {
        if (child.RestartCount >= MaxRestarts)
        {
            return FailureAction.Escalate;
        }

        return FailureAction.Restart;
    }

Quando a ação é Restart, a estratégia define quem será reiniciado:

    protected virtual async Task ApplyRestartAsync(Child child)
    {
        child.RestartCount++;
        child.LastRestartTime = DateTimeOffset.UtcNow;

        if (Strategy == Strategy.OneForOne)
        {
            await ResetActorAsync(child);
        }
        else if (Strategy == Strategy.AllForOne)
        {
            await Task.WhenAll(Children.Select(ResetActorAsync));
        }
    }

DynamicSupervisor

O DynamicSupervisor estende o Supervisor para permitir a adição e remoção de filhos em tempo de execução. Ele é selado com a estratégia OneForOne, pois reiniciar todos os atores quando um falha não faz sentido quando os atores são criados de forma independente.

public abstract class DynamicSupervisor(IActorFactory actorFactory, ILogger logger)
    : Supervisor(actorFactory, logger),
        IHandleActorMessage<RemoveChild>
{
    protected sealed override Strategy Strategy => Strategy.OneForOne;

RootSupervisor

O RootSupervisor é uma implementação concreta do Supervisor que serve como o supervisor de nível superior no sistema de atores. Ele implementa a interface de marcação IRootSupervisor e é configurado via RootSupervisorOptions.

Diferente do Supervisor base, o RootSupervisor sempre reinicia os filhos que falham — ele nunca escala, já que não existe um supervisor pai acima dele.

PartitionSupervisor

O PartitionSupervisor<TActor> cria um pool fixo de atores workers idênticos e roteia mensagens para eles usando particionamento baseado em hash. O número de workers é, por padrão, Environment.ProcessorCount.

As mensagens são roteadas para um worker usando uma chave de partição. A chave sofre o hash e é mapeada para um índice de worker:

    protected virtual IActorReference GetActorReference<TKey>(TKey key)
        where TKey : notnull
    {
        var hash = Math.Abs(GetHashcode(key));
        return Children[hash % Children.Count].Reference;
    }

Isso garante que mensagens com a mesma chave sempre vão para o mesmo worker, o que é útil para cenários como afinidade de sessão ou particionamento consistente de dados.

Conclusão

Com o Supervisor, DynamicSupervisor, PartitionSupervisor e o RootSupervisor como ponto de entrada de nível superior, agora temos uma base sólida para tolerância a falhas em nosso sistema de atores. Cada variante de supervisor aborda um caso de uso diferente.

Se você estiver interessado nos detalhes profundos da implementação, sinta-se à vontade para explorar o código-fonte completo no GitHub: lillo42/trupe.

Trupe: Implementing Supervisor

Rafael Andrade — Fri, 27 Mar 2026 09:00:00 +0000

In my saga to implement an Actor System, I've covered so far the basic usage of Actor. Now, I'm going to cover supervisor.

Supervisor

In case you don't know, in Actor Model, a supervisor is a fundamental piece for fault tolerance. Supervisor will monitor children (actors) and react when one of them exhibits bad behaviour.

A Supervisor is a special type of actor, focused only on monitoring and doing some action when children exhibit bad behaviour.

Bad behaviour

In Actor Model, bad behaviour is when an actor has an exception during message execution.

When an actor throws an exception during message processing we have 2 main strategies.

One For One: It'll restart only the failed child.
All for One: It'll restart all children, if one failed.

Now what to do when a restart doesn't work, and it's still failing? You can stop the child or scale it.

Restart

Another question is what restart means in Actor Model. In Actor Model and .NET it means killing/disposing that actor instance, the mailbox, but keeping the IActorReference. If the actor is a supervisor, we are going to stop and kill all children, making IActorReference for those children invalid.

Type of supervisor

Before deep diving into the code, let me show what are the supervisors that I'll implement

Supervisor: a preemptive supervisor, doesn't allow adding more children after startup
DynamicSupervisor: a dynamic supervisor, allows creating children after startup, but it only supports One For One strategy
PartitionSupervisor: a partition-based supervisor that creates a fixed pool of identical workers and routes messages using hash-based partitioning

Implementation

Foundation

Before implementing the supervisors, we need some building blocks: the supervision Strategy, the FailureAction, and the RestartPolicy.

public enum Strategy
{
    OneForOne,
    AllForOne,
}

public enum FailureAction
{
    Restart,
    Stop,
    Escalate,
    Resume,
}

Besides the strategy, each child actor has a RestartPolicy that defines whether it should be restarted:

public enum RestartPolicy
{
    Permanent,  // Always restart regardless of termination reason
    Transient,  // Restart only if it terminates abnormally
    Temporary,  // Never restart
}

Now let's first create an interface called ISupervisor

/// <summary>
/// Represents a supervisor actor that manages child actors.
/// </summary>
public interface ISupervisor : IActor
{
    /// <summary>
    /// Gets the collection of child actor references managed by this supervisor.
    /// </summary>
    IEnumerable<IActorReference> Children { get; }
}

We also need a way to describe how a child should be created. For that we have IChildSpecification:

public interface IChildSpecification
{
    Type ActorType { get; }
    IMailbox Mailbox { get; set; }
    RestartPolicy RestartPolicy { get; set; }
}

And the default implementation:

public record ChildSpecification : IChildSpecification
{
    public ChildSpecification(Type actorType)
    {
        ActorType = actorType;
    }

    public Type ActorType { get; }
    public IMailbox Mailbox { get; set; } = new ChannelMailbox();
    public RestartPolicy RestartPolicy { get; set; } = RestartPolicy.Permanent;
}

Each supervised actor is tracked via a Child metadata class that holds the actor instance, its process, mailbox, reference, restart policy, and restart counters:

public class Child(
    IActor actor,
    IMailbox mailbox,
    ActorProcess process,
    LocalActorReference reference,
    RestartPolicy restartPolicy,
    Type actorType)
{
    public IActor Actor { get; set; } = actor;
    public ActorProcess Process { get; set; } = process;
    public IMailbox Mailbox { get; } = mailbox;
    public LocalActorReference Reference { get; } = reference;
    public RestartPolicy RestartPolicy { get; } = restartPolicy;
    public Type ActorType { get; } = actorType;
    public int RestartCount { get; set; } = 0;
    public DateTimeOffset LastRestartTime { get; set; } = DateTimeOffset.MinValue;
    public Dictionary<string, object> Metadata { get; } = [];
    public bool IsSupervisor => Actor is ISupervisor;
}

Supervisor

The base Supervisor is an abstract class that implements ISupervisor. It is preemptive — children are defined during initialization and cannot be added afterwards.

public abstract partial class Supervisor(IActorFactory actorFactory, ILogger logger)
    : Actor,
        ISupervisor,
        IHandleActorMessage<AddActor>,
        IHandleActorMessage<ActorFailed>,
        IHandleActorMessage<ActorTerminated>,
        IAsyncDisposable
{

It exposes virtual properties so subclasses can customize the behaviour:

    protected virtual Strategy Strategy => Strategy.OneForOne;
    protected virtual int MaxRestarts => 3;
    protected virtual TimeSpan RestartWindow => TimeSpan.FromSeconds(5);

During initialization, it calls the abstract OnInitializeAsync where subclasses define their children:

    public sealed override async ValueTask InitializeAsync(
        CancellationToken cancellationToken = default)
    {
        await OnInitializeAsync(cancellationToken);
        _initialized = true;
    }

    protected abstract ValueTask OnInitializeAsync(
        CancellationToken cancellationToken = default);

When a child fails, the supervisor determines what to do via GetFailureAction. By default, it restarts the child unless the restart limit is exceeded, in which case it escalates:

    protected virtual FailureAction GetFailureAction(Child child, Exception exception)
    {
        if (child.RestartCount >= MaxRestarts)
        {
            return FailureAction.Escalate;
        }

        return FailureAction.Restart;
    }

When the action is Restart, the strategy determines who gets restarted:

    protected virtual async Task ApplyRestartAsync(Child child)
    {
        child.RestartCount++;
        child.LastRestartTime = DateTimeOffset.UtcNow;

        if (Strategy == Strategy.OneForOne)
        {
            await ResetActorAsync(child);
        }
        else if (Strategy == Strategy.AllForOne)
        {
            await Task.WhenAll(Children.Select(ResetActorAsync));
        }
    }

When the supervisor itself is restarted, all children are stopped and disposed:

    public override async ValueTask BeforeRestartAsync(
        CancellationToken cancellationToken = default)
    {
        foreach (var metadata in Children)
        {
            await StopActorAsync(metadata);
            await DisposeObjectAsync(metadata.Actor);

            metadata.Actor = null!;
            metadata.Process = null!;
            metadata.Metadata.Clear();
        }

        Children = [];
    }

DynamicSupervisor

The DynamicSupervisor extends Supervisor to allow adding and removing children at runtime. It is sealed to the OneForOne strategy because restarting all actors when one fails doesn't make sense when actors are created independently.

public abstract class DynamicSupervisor(IActorFactory actorFactory, ILogger logger)
    : Supervisor(actorFactory, logger),
        IHandleActorMessage<RemoveChild>
{
    protected sealed override Strategy Strategy => Strategy.OneForOne;

Children are added dynamically by sending an AddActor command to the supervisor's own mailbox:

    protected override IActorReference AddChild(IChildSpecification specification)
    {
        var actorRef = new LocalActorReference(specification.Mailbox);
        Context.Self.Tell(new AddActor(specification, actorRef));
        return actorRef;
    }

And removed via RemoveChild:

    public async ValueTask HandleAsync(
        RemoveChild message,
        CancellationToken cancellationToken = default)
    {
        var child = Children.FirstOrDefault(x => x.Actor == message.Actor);

        if (child != null)
        {
            Children = Children.Remove(child);

            await StopActorAsync(child);
            await DisposeObjectAsync(child.Actor);
            await child.Process.DisposeAsync();

            child.Actor = null!;
            child.Process = null!;
        }
    }

The DynamicSupervisor also respects RestartPolicy during failures and terminations. Temporary actors are removed after failure, and non-permanent actors are removed after termination.

RootSupervisor

The RootSupervisor is a concrete implementation of Supervisor that serves as the top-level supervisor in the actor system. It implements the IRootSupervisor marker interface and is configured via RootSupervisorOptions, which holds a list of IChildSpecification children.

public class RootSupervisor(
    IOptions<RootSupervisorOptions> options,
    IActorFactory actorFactory,
    ILogger<RootSupervisor> logger
) : Supervisor(actorFactory, logger), IRootSupervisor

During initialization, it adds all children defined in the options:

    protected override ValueTask OnInitializeAsync(CancellationToken cancellationToken = default)
    {
        foreach (var child in options.Value.Children)
        {
            AddChild(child);
        }
        return new ValueTask();
    }

Unlike the base Supervisor, the RootSupervisor always restarts failed children — it never escalates, since there is no parent supervisor above it:

    protected override FailureAction GetFailureAction(Child child, Exception exception)
    {
        return FailureAction.Restart;
    }

PartitionSupervisor

The PartitionSupervisor<TActor> creates a fixed pool of identical worker actors and routes messages to them using hash-based partitioning. The number of workers defaults to Environment.ProcessorCount.

public abstract partial class PartitionSupervisor<TActor>(
    IActorFactory actorFactory, ILogger logger, int workers)
    : Actor,
        ISupervisor,
        IHandleActorMessage<ActorFailed>,
        IHandleActorMessage<ActorTerminated>,
        IAsyncDisposable
{

It exposes similar configuration properties:

    protected virtual int Workers { get; } = workers;
    protected virtual Strategy Strategy => Strategy.OneForOne;
    protected virtual RestartPolicy DefaultRestartPolicy => RestartPolicy.Permanent;
    protected virtual int MaxRestarts => 3;
    protected virtual TimeSpan RestartWindow => TimeSpan.FromSeconds(5);

During initialization, it creates all workers upfront:

    public override async ValueTask InitializeAsync(
        CancellationToken cancellationToken = default)
    {
        for (var i = 0; i < Workers; i++)
        {
            CreateActor(
                new ChildSpecification(typeof(TActor))
                {
                    RestartPolicy = DefaultRestartPolicy,
                    Mailbox = CreateMailbox(),
                });
        }

        await OnInitializeAsync(cancellationToken);
        _initialized = true;
    }

Messages are routed to a worker using a partition key. The key is hashed and mapped to a worker index:

    protected virtual IActorReference GetActorReference<TKey>(TKey key)
        where TKey : notnull
    {
        var hash = Math.Abs(GetHashcode(key));
        return Children[hash % Children.Count].Reference;
    }

This ensures that messages with the same key always go to the same worker, which is useful for scenarios like session affinity or consistent data partitioning.

Conclusion

With the Supervisor, DynamicSupervisor, PartitionSupervisor, and the RootSupervisor as the top-level entry point, we now have a solid foundation for fault tolerance in our actor system. Each supervisor variant addresses a different use case — from static child sets, to dynamic actor management, to hash-based worker partitioning.

If you're interested in the deep details of the implementation, feel free to explore the full source code on GitHub: lillo42/trupe.

Usando DynamoDB com Brighter V10

Rafael Andrade — Fri, 20 Mar 2026 09:00:00 +0000

O Brighter possui suporte nativo para o DynamoDB por meio de provedores dedicados de Inbox, Outbox e bloqueio distribuído (distributed locking). Neste artigo, explorarei como implementar os padrões Inbox e Outbox usando o DynamoDB para criar serviços .NET resilientes e com consistência eventual.

Visão Geral do Projeto

Construiremos um serviço em .NET 8+ que consome e produz mensagens de um tópico Kafka, processa-as usando o Brighter e utiliza o DynamoDB como armazenamento persistente para o Inbox e o Outbox. Isso garante que todas as mensagens sejam processadas exatamente uma vez e que nenhuma mensagem seja perdida, mesmo diante de falhas. Também configuraremos um bloqueio distribuído para evitar a publicação de mensagens duplicadas quando a aplicação for executada em um ambiente multi-nó.

Pacotes Necessários

Escolha entre dois estilos de configuração:

Pacotes "Pure Brighter" (AWS SDK v4 recomendado)

<PackageReference Include="Paramore.Brighter.Inbox.DynamoDB.V4" Version="10.3.0" />
<PackageReference Include="Paramore.Brighter.Outbox.DynamoDB.V4" Version="10.3.0" />
<PackageReference Include="Paramore.Brighter.Locking.DynamoDB.V4" Version="10.3.0" />
<PackageReference Include="Paramore.Brighter.MessagingGateway.Kafka" Version="10.3.0" />
<PackageReference Include="Paramore.Brighter.ServiceActivator.Extensions.Hosting" Version="10.3.0" />
<PackageReference Include="Paramore.Brighter.Extensions.DependencyInjection" Version="10.3.0" />

Wrappers "Fluent Brighter"

<PackageReference Include="Fluent.Brighter.Kafka" Version="1.8.1" />
<PackageReference Include="Fluent.Brighter.AWS.V4" Version="1.8.1" />
<PackageReference Include="Paramore.Brighter.ServiceActivator.Extensions.Hosting" Version="10.3.0" />

Uma Breve Recapitulação do Brighter

Antes de mergulharmos na configuração, vamos revisar rapidamente os conceitos fundamentais do Brighter.

Request: Command e Event

O primeiro conceito-chave é a interface IRequest, normalmente implementada como um Command ou um Event. Esses objetos são usados pelo IAmACommandProcessor do Brighter para:

Send: espera exatamente um handler.
Publish: espera zero ou mais handlers.
Post: envia para um message broker externo.

Commands representam uma ação que você deseja executar, como CreateOrder. Events representam algo que já aconteceu, como OrderCreated.

public class CreateOrder() : Command(Id.Random());
public class OrderCreated() : Event(Id.Random());

Message Mapper

Ao postar ou consumir mensagens de um broker externo, você precisa de um mapeador de mensagens. Este componente mapeia seu objeto IRequest para (e de) uma mensagem do Brighter (que inclui o cabeçalho e o corpo para o broker).

O Brighter v10+ usa por padrão um mapeador JSON, mas você pode fornecer o seu próprio implementando IAmAMessageMapper ou IAmAMessageMapperAsync.

Request Handler

O RequestHandler (ou RequestHandlerAsync) contém sua lógica de negócio para um IRequest específico. O modelo de pipeline do Brighter permite encadear handlers e adicionar atributos para preocupações transversais (cross-cutting concerns), como retentativas, logs e o padrão Inbox.

public class GreetingHandler : RequestHandler<Greeting>
{
    public override Greeting Handle(Greeting @event)
    {
        Console.WriteLine("Olá, {0}", @event.Name);
        return base.Handle(@event);
    }
}

Configurando o Brighter com Kafka

Agora, a configuração principal. Precisamos dizer ao Brighter como se conectar ao Kafka. Abaixo, mostramos as abordagens "Fluent" e "Pure".

Usando Fluent Brighter

services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddFluentBrighter(opt => opt
        .UsingKafka(kf => kf
            .SetConnection(c => c
                .SetName("sample")
                .SetBootstrapServers("localhost:9092")
                .SetSecurityProtocol(SecurityProtocol.Plaintext)
                .SetSaslMechanisms(SaslMechanism.Plain))
            .UseSubscriptions(s => s
                .AddSubscription<OrderPlaced>(sb => sb
                    .SetTopic("order-placed-topic")
                    .SetConsumerGroupId("order-placed-topic-1")
                    .SetRequeueCount(3)
                    .CreateInfrastructureIfMissing()
                    .UseReactorMode())
                .AddSubscription<OrderPaid>(sb => sb
                    .SetTopic("order-paid-topic")
                    .SetConsumerGroupId("order-paid-topic-1")
                    .CreateInfrastructureIfMissing()
                    .UseReactorMode()))
            .UsePublications(p => p
                .AddPublication<OrderPaid>(kp => kp
                    .SetTopic("order-paid-topic")
                    .CreateTopicIfMissing())
                .AddPublication<OrderPlaced>(kp => kp
                    .SetTopic("order-placed-topic")
                    .CreateTopicIfMissing()))));

Usando Pure Brighter

var connection = new KafkaMessagingGatewayConfiguration
{
    Name = "sample",
    BootStrapServers = ["localhost:9092"],
    SecurityProtocol = SecurityProtocol.Plaintext,
    SaslMechanisms = SaslMechanism.Plain,
};

services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddConsumers(opt =>
    {
        opt.Subscriptions =
        [
            new KafkaSubscription<OrderPlaced>(
                new SubscriptionName("subscription-orderplaced"),
                new ChannelName("order-placed-queue"),
                new RoutingKey("order-placed-topic"),
                makeChannels: OnMissingChannel.Create,
                messagePumpType: MessagePumpType.Reactor,
                groupId: "test-1"),
            new KafkaSubscription<OrderPaid>(
                new SubscriptionName("subscription-orderpaid"),
                new ChannelName("order-paid-queue"),
                new RoutingKey("order-paid-topic"),
                makeChannels: OnMissingChannel.Create,
                messagePumpType: MessagePumpType.Reactor,
                groupId: "test-2"),
        ];

        opt.DefaultChannelFactory = new ChannelFactory(new KafkaMessageConsumerFactory(connection));
    })
    .AutoFromAssemblies()
    .AddProducers(opt =>
    {
        opt.ProducerRegistry = new KafkaProducerRegistryFactory(
            connection,
            [
                new KafkaPublication<OrderPaid>
                {
                    MakeChannels = OnMissingChannel.Create,
                    Topic = new RoutingKey("order-paid-topic"),
                },
                new KafkaPublication<OrderPlaced>
                {
                    MakeChannels = OnMissingChannel.Create,
                    Topic = new RoutingKey("order-placed-topic"),
                }
            ]).Create();
    });

Configuração do DynamoDB

Agora configuramos o Inbox, Outbox e o bloqueio distribuído usando o DynamoDB.

Usando Fluent Brighter

services
    .AddFluentBrighter(opt => opt
        .UsingAws(pg => pg
            .SetConnection(conn => conn
                .SetCredentials(new BasicAWSCredentials("test", "test"))
                .SetRegion(RegionEndpoint.USEast1)
                .SetClientConfigAction(cfg => cfg.ServiceURL = "http://localhost:4566"))  // LocalStack
            .UseDynamoDbOutbox("outbox")
            .UseDynamoDbInbox("inbox")
            .UseDynamoDbDistributedLock(c => c
                .SetLeaseholderGroupId("some-group")
                .SetTableName("locking")))
        // ... a configuração do Kafka continua aqui
    );

Usando Pure Brighter

Para a abordagem pura, você precisa configurar o cliente DynamoDB e fornecer as instâncias dos provedores.

// Configurar cliente DynamoDB (exemplo usando LocalStack)
var dynamoDbConfig = new AmazonDynamoDBConfig
{
    RegionEndpoint = RegionEndpoint.USEast1,
    ServiceURL = "http://localhost:4566"
};
var dynamoDbClient = new AmazonDynamoDBClient(new BasicAWSCredentials("test", "test"), dynamoDbConfig);

// Criar os provedores de outbox, inbox e locking
var outbox = new DynamoDbOutbox(dynamoDbClient, new DynamoDbConfiguration("outbox"));

var inbox = new DynamoDbInbox(dynamoDbClient, new DynamoDbInboxConfiguration
{
    TableName = "inbox"
});

var lockingProvider = new DynamoDbLockingProvider(dynamoDbClient, new DynamoDbLockingProviderOptions("locking", "some-group"));

// Registrar na infraestrutura do Brighter
services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddConsumers(opt =>
    {
        opt.InboxConfiguration = new InboxConfiguration(inbox);
        // ... outras configs de consumidor (subscriptions, etc.)
    })
    .AddProducers(opt =>
    {
        opt.Outbox = outbox;
        opt.DistributedLock = lockingProvider;
        opt.ConnectionProvider = typeof(DynamoDbUnitOfWork);  // Se estiver usando transações
        // ... config de registro de produtores
    });

Nota: As tabelas do DynamoDB (outbox, inbox, locking) devem existir previamente. O Brighter fornece helpers para criá-las se necessário, mas em produção você deve gerenciar a criação das tabelas separadamente.

Esquemas de Tabela do DynamoDB

Abaixo estão os esquemas que o Brighter espera. Você pode criá-los manualmente via Console AWS, CloudFormation, Terraform, ou usar o DynamoDbTableFactory do Brighter.

Tabela Outbox

Configuração	Valor
Nome padrão da tabela	`brighter_outbox`
Modo de faturamento	PAY_PER_REQUEST (recomendado) ou PROVISIONED

Chave Primária:

Chave	Atributo	Tipo
Chave de Partição (HASH)	`MessageId`	String (S)

Tabela Inbox

Configuração	Valor
Nome padrão da tabela	`brighter_inbox`
Modo de faturamento	PAY_PER_REQUEST

Chave Primária (composta):

Chave	Atributo	Tipo
Chave de Partição (HASH)	`CommandId`	String (S)
Chave de Classificação (RANGE)	`ContextKey`	String (S)

A chave composta (CommandId + ContextKey) permite que o Inbox rastreie se um handler específico já processou uma mensagem específica, suportando o consumo idempotente.

Exemplo: Mensagens e Request Handlers

Request Handlers

O CreateNewOrderHandler recebe um comando CreateNewOrder (via Send) e deposita dois eventos no Outbox. O processo de varredura (sweeper) em segundo plano os publicará posteriormente no Kafka.

public class CreateNewOrderHandler : RequestHandler<CreateNewOrder>
{
    private readonly IAmACommandProcessor _commandProcessor;
    private readonly ILogger<CreateNewOrderHandler> _logger;

    public CreateNewOrderHandler(IAmACommandProcessor commandProcessor, ILogger<CreateNewOrderHandler> logger)
    {
        _commandProcessor = commandProcessor;
        _logger = logger;
    }

    public override CreateNewOrder Handle(CreateNewOrder command)
    {
        var id = Uuid.NewAsString();
        _logger.LogInformation("Criando novo pedido: {OrderId}", id);

        _commandProcessor.DepositPost(new OrderPlaced { OrderId = id, Value = command.Value });
        _commandProcessor.DepositPost(new OrderPaid { OrderId = id });

        return base.Handle(command);
    }
}

Como os padrões funcionam juntos

Padrão Inbox – Quando o OrderPlacedHandler falha, a mensagem não é marcada como processada no Inbox. Quando os offsets do Kafka são resetados, apenas mensagens genuinamente não processadas são reexecutadas, evitando duplicatas.
Padrão Outbox – O DepositPost armazena as mensagens no Outbox do DynamoDB antes de tentar publicar. Um processo "sweeper" publica-as no Kafka, garantindo que nenhuma mensagem seja perdida mesmo se o Kafka estiver temporariamente indisponível.
Bloqueio Distribuído – Em ambientes multi-nó, o sweeper usa o bloqueio distribuído para garantir que apenas uma instância publique as mensagens por vez, evitando publicações duplicadas.

Conclusão

Os padrões Inbox e Outbox são essenciais para construir sistemas distribuídos resilientes. O Paramore.Brighter, combinado com sua integração com DynamoDB e o gateway de mensagens Kafka, oferece uma implementação limpa e robusta para aplicações .NET.

Essa configuração garante que seus serviços lidem com falhas graciosamente, mantendo a consistência de dados em todo o seu sistema distribuído.

Using DynamoDB with Brighter V10

Rafael Andrade — Fri, 20 Mar 2026 09:00:00 +0000

Brighter has native support for DynamoDB via dedicated inbox, outbox, and distributed locking providers. In this article, I'll explore how to implement the Inbox and Outbox patterns using DynamoDB to build resilient, eventually consistent .NET services.

Project Overview

We'll build a .NET 8+ service that consumes and produces messages from a Kafka topic, processes them using Brighter, and uses DynamoDB as the persistent store for the inbox and outbox. This guarantees that all messages are processed exactly once and that no messages are lost, even in the face of failures. We also need to configure a distributed lock to avoid publishing duplicate messages when the application runs in a multi‑node environment.

Required Packages

Choose between two setup styles:

Pure Brighter Packages (AWS SDK v4 recommended)

<PackageReference Include="Paramore.Brighter.Inbox.DynamoDB.V4" Version="10.3.0" />
<PackageReference Include="Paramore.Brighter.Outbox.DynamoDB.V4" Version="10.3.0" />
<PackageReference Include="Paramore.Brighter.Locking.DynamoDB.V4" Version="10.3.0" />
<PackageReference Include="Paramore.Brighter.MessagingGateway.Kafka" Version="10.3.0" />
<PackageReference Include="Paramore.Brighter.ServiceActivator.Extensions.Hosting" Version="10.3.0" />
<PackageReference Include="Paramore.Brighter.Extensions.DependencyInjection" Version="10.3.0" />

Fluent Brighter Wrappers

<PackageReference Include="Fluent.Brighter.Kafka" Version="1.8.1" />
<PackageReference Include="Fluent.Brighter.AWS.V4" Version="1.8.1" />
<PackageReference Include="Paramore.Brighter.ServiceActivator.Extensions.Hosting" Version="10.3.0" />

A Quick Brighter Recap

Before we dive into the configuration, let's quickly recap the core Brighter concepts.

Request: Command and Event

The first key concept is the IRequest interface, typically implemented as a Command or an Event. These objects are used by Brighter’s IAmACommandProcessor to:

Send: expects exactly one handler.
Publish: expects zero or more handlers.
Post: sends to an external message broker.

Commands represent an action you want to execute, like CreateOrder. Events represent something that has already happened, like OrderCreated.

public class CreateOrder() : Command(Id.Random());

public class OrderCreated() : Event(Id.Random());

Message Mapper

When you post or consume messages from an external broker, you need a message mapper. This component maps your IRequest object to and from a Brighter Message (which includes the header and body for the broker).

Brighter v10+ defaults to a JSON message mapper, but you can provide your own by implementing IAmAMessageMapper or IAmAMessageMapperAsync.

Request Handler

The RequestHandler (or RequestHandlerAsync) contains your business logic for a specific IRequest. Brighter’s pipeline model allows you to chain handlers and add attributes for cross‑cutting concerns like retries, logging, and the inbox pattern.

public class GreetingHandler : RequestHandler<Greeting>
{
    public override Greeting Handle(Greeting @event)
    {
        Console.WriteLine("Hello, {0}", @event.Name);
        return base.Handle(@event);
    }
}

Configuring Brighter with Kafka

Now for the main setup. We need to tell Brighter how to connect to Kafka.

Both the “pure” and “fluent” approaches are shown below

Using Fluent Brighter

services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddFluentBrighter(opt => opt
        .UsingKafka(kf => kf
            .SetConnection(c => c
                .SetName("sample")
                .SetBootstrapServers("localhost:9092")
                .SetSecurityProtocol(SecurityProtocol.Plaintext)
                .SetSaslMechanisms(SaslMechanism.Plain))
            .UseSubscriptions(s => s
                .AddSubscription<OrderPlaced>(sb => sb
                    .SetTopic("order-placed-topic")
                    .SetConsumerGroupId("order-placed-topic-1")
                    .SetRequeueCount(3)
                    .CreateInfrastructureIfMissing()
                    .UseReactorMode())
                .AddSubscription<OrderPaid>(sb => sb
                    .SetTopic("order-paid-topic")
                    .SetConsumerGroupId("order-paid-topic-1")
                    .CreateInfrastructureIfMissing()
                    .UseReactorMode()))
            .UsePublications(p => p
                .AddPublication<OrderPaid>(kp => kp
                    .SetTopic("order-paid-topic")
                    .CreateTopicIfMissing())
                .AddPublication<OrderPlaced>(kp => kp
                    .SetTopic("order-placed-topic")
                    .CreateTopicIfMissing()))));

Using Pure Brighter

var connection = new KafkaMessagingGatewayConfiguration
{
    Name = "sample",
    BootStrapServers = ["localhost:9092"],
    SecurityProtocol = SecurityProtocol.Plaintext,
    SaslMechanisms = SaslMechanism.Plain,
};

services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddConsumers(opt =>
    {
        opt.Subscriptions =
        [
            new KafkaSubscription<OrderPlaced>(
                new SubscriptionName("subscription-orderplaced"),
                new ChannelName("order-placed-queue"),
                new RoutingKey("order-placed-topic"),
                makeChannels: OnMissingChannel.Create,
                messagePumpType: MessagePumpType.Reactor,
                groupId: "test-1"),
            new KafkaSubscription<OrderPaid>(
                new SubscriptionName("subscription-orderpaid"),
                new ChannelName("order-paid-queue"),
                new RoutingKey("order-paid-topic"),
                makeChannels: OnMissingChannel.Create,
                messagePumpType: MessagePumpType.Reactor,
                groupId: "test-2"),
        ];

        opt.DefaultChannelFactory = new ChannelFactory(new KafkaMessageConsumerFactory(connection));
    })
    .AutoFromAssemblies()
    .AddProducers(opt =>
    {
        opt.ProducerRegistry = new KafkaProducerRegistryFactory(
            connection,
            [
                new KafkaPublication<OrderPaid>
                {
                    MakeChannels = OnMissingChannel.Create,
                    Topic = new RoutingKey("order-paid-topic"),
                },
                new KafkaPublication<OrderPlaced>
                {
                    MakeChannels = OnMissingChannel.Create,
                    Topic = new RoutingKey("order-placed-topic"),
                }
            ]).Create();
    });

DynamoDB Configuration

Now we configure the inbox, outbox, and distributed lock using DynamoDB.

Using Fluent Brighter

services
    .AddFluentBrighter(opt => opt
        .UsingAws(pg => pg
            .SetConnection(conn => conn
                .SetCredentials(new BasicAWSCredentials("test", "test"))
                .SetRegion(RegionEndpoint.USEast1)
                .SetClientConfigAction(cfg => cfg.ServiceURL = "http://localhost:4566"))  // LocalStack
            .UseDynamoDbOutbox("outbox")
            .UseDynamoDbInbox("inbox")
            .UseDynamoDbDistributedLock(c => c
                .SetLeaseholderGroupId("some-group")
                .SetTableName("locking")))
        // ... Kafka configuration continues here
    );

Using Pure Brighter

For the pure approach, you need to set up the DynamoDB client and provide the outbox, inbox, and lock provider instances.

// Configure DynamoDB client (example using LocalStack)
var dynamoDbConfig = new AmazonDynamoDBConfig
{
    RegionEndpoint = RegionEndpoint.USEast1,
    ServiceURL = "http://localhost:4566"
};
var dynamoDbClient = new AmazonDynamoDBClient(new BasicAWSCredentials("test", "test"), dynamoDbConfig);

// Create the outbox, inbox, and locking providers
var outbox = new DynamoDbOutbox(dynamoDbClient, new DynamoDbConfiguration("outbox"));

var inbox = new DynamoDbInbox(dynamoDbClient, new DynamoDbInboxConfiguration
{
    TableName = "inbox"
});

var lockingProvider = new DynamoDbLockingProvider(dynamoDbClient, new DynamoDbLockingProviderOptions("locking", "some-group"));

// Register them with the Brighter infrastructure
services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddConsumers(opt =>
    {
        opt.InboxConfiguration = new InboxConfiguration(inbox);
        // ... other consumer config (subscriptions, etc.)
    })
    .AddProducers(opt =>
    {
        opt.Outbox = outbox;
        opt.DistributedLock = lockingProvider;
        opt.ConnectionProvider = typeof(DynamoDbUnitOfWork);  // if using transactions
        // ... producer registry config
    });

Note: The DynamoDB tables (outbox, inbox, locking) must exist beforehand. Brighter provides helpers to create them if needed, but for production you should manage table creation separately.

DynamoDB Table Schemas

Below are the DynamoDB table schemas Brighter expects. You can create them manually via the AWS Console, CloudFormation, Terraform, or use Brighter's built-in DynamoDbTableFactory to generate the CreateTableRequest from the entity types.

Outbox Table

Setting	Value
Default table name	`brighter_outbox`
Billing mode	PAY_PER_REQUEST (recommended) or PROVISIONED

Primary Key:

Key	Attribute	Type
Partition key (HASH)	`MessageId`	String (S)

Attributes:

Attribute	Type	Description
`MessageId`	S	Unique message identifier
`Body`	B	Message body (binary)
`ContentType`	S	MIME content type
`CorrelationId`	S	Correlation identifier
`CreatedTime`	N	Creation time in ticks
`OutstandingCreatedTime`	N	Creation time in ticks; set to `null` once dispatched
`DeliveryTime`	N	Delivery time in ticks; `null` until dispatched
`DeliveredAt`	S	Delivery date (`yyyy-MM-dd`)
`ExpiresAt`	N	TTL in ticks (optional)
`TopicShard`	S	`{Topic}_{ShardNumber}`
`Topic`	S	Routing key / topic name
`PartitionKey`	S	Broker partition key
`ReplyTo`	S	Reply-to channel
`MessageType`	S	`MT_COMMAND`, `MT_EVENT`, etc.
`HeaderBag`	S	JSON dictionary of extra headers
`CharacterEncoding`	S	Body encoding (e.g. `UTF8`)
`CreatedAt`	S	ISO 8601 timestamp
`HandledCount`	N	Retry counter
`DelayedMilliseconds`	N	Delayed delivery offset
`Type`	S	CloudEvents type
`Source`	S	CloudEvents source URI
`Subject`	S	CloudEvents subject
`SpecVersion`	S	CloudEvents spec version
`DataSchema`	S	CloudEvents data schema URI
`DataRef`	S	Claim-check data reference
`TraceParent`	S	W3C Trace Parent
`TraceState`	S	W3C Trace State
`Baggage`	S	W3C Baggage
`JobId`	S	Workflow instance identifier
`WorkflowId`	S	Workflow identity

Global Secondary Indexes (GSIs):

Index Name	Partition Key (HASH)	Sort Key (RANGE)	Purpose
`Outstanding`	`TopicShard` (S)	`OutstandingCreatedTime` (N)	Query outstanding messages per topic shard
`OutstandingAllTopics`	`OutstandingCreatedTime` (N)	`MessageId` (S)	Scan all outstanding messages across topics
`Delivered`	`TopicShard` (S)	`DeliveryTime` (N)	Query delivered messages per topic shard
`DeliveredAllTopics`	`DeliveryTime` (N)	`MessageId` (S)	Scan all delivered messages across topics

The outbox uses sharding (TopicShard = "{Topic}_{ShardNumber}") to avoid hot partitions on active topics. The default number of shards is 3 and can be configured via DynamoDbConfiguration.NumberOfShards.

Inbox Table

Setting	Value
Default table name	`brighter_inbox`
Billing mode	PAY_PER_REQUEST (recommended) or PROVISIONED

Primary Key (composite):

Key	Attribute	Type
Partition key (HASH)	`CommandId`	String (S)
Sort key (RANGE)	`ContextKey`	String (S)

Attributes:

Attribute	Type	Description
`CommandId`	S	The request/command identifier
`ContextKey`	S	Handler context key (enables the same message ID to be handled by different handlers)
`CommandType`	S	The .NET type name of the command
`CommandBody`	S	Serialized command as JSON
`TimeStamp`	S	When the command was stored
`Time`	S	Timestamp in ticks (as string)

The composite key (CommandId + ContextKey) allows the inbox to track whether a specific handler has already processed a specific message, supporting idempotent consumption.

Lock Table

Setting	Value
Default table name	`brighter_distributed_lock`
Billing mode	PAY_PER_REQUEST (recommended) or PROVISIONED

Primary Key:

Key	Attribute	Type
Partition key (HASH)	`ResourceId`	String (S)

Attributes:

Attribute	Type	Description
`ResourceId`	S	`{LeaseholderGroupId}_{Resource}` — identifies the locked resource scoped to a group
`LeaseExpiry`	N	Unix timestamp in milliseconds when the lease expires
`LockId`	S	Unique identifier of the lock holder

Locking uses a conditional PutItem with attribute_not_exists(LockId) OR LeaseExpiry <= :now to atomically acquire the lock. Expired leases are automatically superseded. The default lease validity is 1 minute, configurable via DynamoDbLockingProviderOptions.LeaseValidity.

Creating the Tables with AWS CLI

For quick local development (e.g. with LocalStack), you can create the tables using the AWS CLI:

# Outbox table
aws dynamodb create-table \
  --table-name outbox \
  --attribute-definitions \
    AttributeDefinition={AttributeName=MessageId,AttributeType=S} \
    AttributeDefinition={AttributeName=TopicShard,AttributeType=S} \
    AttributeDefinition={AttributeName=OutstandingCreatedTime,AttributeType=N} \
    AttributeDefinition={AttributeName=DeliveryTime,AttributeType=N} \
  --key-schema AttributeDefinition={AttributeName=MessageId,KeyType=HASH} \
  --global-secondary-indexes \
    '[{"IndexName":"Outstanding","KeySchema":[{"AttributeName":"TopicShard","KeyType":"HASH"},{"AttributeName":"OutstandingCreatedTime","KeyType":"RANGE"}],"Projection":{"ProjectionType":"ALL"}},
      {"IndexName":"OutstandingAllTopics","KeySchema":[{"AttributeName":"OutstandingCreatedTime","KeyType":"HASH"},{"AttributeName":"MessageId","KeyType":"RANGE"}],"Projection":{"ProjectionType":"ALL"}},
      {"IndexName":"Delivered","KeySchema":[{"AttributeName":"TopicShard","KeyType":"HASH"},{"AttributeName":"DeliveryTime","KeyType":"RANGE"}],"Projection":{"ProjectionType":"ALL"}},
      {"IndexName":"DeliveredAllTopics","KeySchema":[{"AttributeName":"DeliveryTime","KeyType":"HASH"},{"AttributeName":"MessageId","KeyType":"RANGE"}],"Projection":{"ProjectionType":"ALL"}}]' \
  --billing-mode PAY_PER_REQUEST \
  --endpoint-url http://localhost:4566

# Inbox table
aws dynamodb create-table \
  --table-name inbox \
  --attribute-definitions \
    AttributeDefinition={AttributeName=CommandId,AttributeType=S} \
    AttributeDefinition={AttributeName=ContextKey,AttributeType=S} \
  --key-schema \
    AttributeDefinition={AttributeName=CommandId,KeyType=HASH} \
    AttributeDefinition={AttributeName=ContextKey,KeyType=RANGE} \
  --billing-mode PAY_PER_REQUEST \
  --endpoint-url http://localhost:4566

# Lock table
aws dynamodb create-table \
  --table-name locking \
  --attribute-definitions \
    AttributeDefinition={AttributeName=ResourceId,AttributeType=S} \
  --key-schema AttributeDefinition={AttributeName=ResourceId,KeyType=HASH} \
  --billing-mode PAY_PER_REQUEST \
  --endpoint-url http://localhost:4566

Example: Message and Request Handlers

Let's define the messages and handlers for our order service.

Messages

public class CreateNewOrder() : Command(Id.Random())
{
    public decimal Value { get; set; }
}

public class OrderPlaced() : Event(Id.Random())
{
    public string OrderId { get; set; } = string.Empty;
    public decimal Value { get; set; }
}

public class OrderPaid() : Event(Id.Random())
{
    public string OrderId { get; set; } = string.Empty;
}

Request Handlers

CreateNewOrderHandler receives a CreateNewOrder command (via Send) and deposits two events into the outbox. The background sweeper will later publish them to Kafka.

public class CreateNewOrderHandler : RequestHandler<CreateNewOrder>
{
    private readonly IAmACommandProcessor _commandProcessor;
    private readonly ILogger<CreateNewOrderHandler> _logger;

    public CreateNewOrderHandler(IAmACommandProcessor commandProcessor, ILogger<CreateNewOrderHandler> logger)
    {
        _commandProcessor = commandProcessor;
        _logger = logger;
    }

    public override CreateNewOrder Handle(CreateNewOrder command)
    {
        var id = Uuid.NewAsString();
        _logger.LogInformation("Creating new order: {OrderId}", id);

        _commandProcessor.DepositPost(new OrderPlaced { OrderId = id, Value = command.Value });
        _commandProcessor.DepositPost(new OrderPaid { OrderId = id });

        return base.Handle(command);
    }
}

OrderPaidHandler consumes the OrderPaid event from Kafka and logs it.

public class OrderPaidHandler : RequestHandler<OrderPaid>
{
    private readonly ILogger<OrderPaidHandler> _logger;

    public OrderPaidHandler(ILogger<OrderPaidHandler> logger) => _logger = logger;

    public override OrderPaid Handle(OrderPaid command)
    {
        _logger.LogInformation("{OrderId} paid", command.OrderId);
        return base.Handle(command);
    }
}

OrderPlacedHandler simulates a failure for orders with a value divisible by 3. The [UseResiliencePipeline] attribute adds retry logic (e.g., 3 retries using a policy named "kafka-policy").

public class OrderPlacedHandler : RequestHandler<OrderPlaced>
{
    private readonly ILogger<OrderPlacedHandler> _logger;

    public OrderPlacedHandler(ILogger<OrderPlacedHandler> logger) => _logger = logger;

    [UseResiliencePipeline("kafka-policy", 1)]
    public override OrderPlaced Handle(OrderPlaced command)
    {
        _logger.LogInformation("{OrderId} placed with value {OrderValue}", command.OrderId, command.Value);
        if (command.Value % 3 == 0)
        {
            _logger.LogError("Simulating error for {OrderId} with value {OrderValue}", command.OrderId, command.Value);
            throw new InvalidOperationException("Simulated error");
        }
        return base.Handle(command);
    }
}

How the Patterns Work Together

Inbox Pattern – When OrderPlacedHandler fails, the message is not marked as processed in the inbox. When Kafka offsets are reset, only genuinely unprocessed messages are reprocessed, preventing duplicates.
Outbox Pattern – DepositPost stores messages in the DynamoDB outbox before attempting to publish. A background sweeper process then publishes them to Kafka, ensuring no messages are lost even if Kafka is temporarily unavailable.
Distributed Lock – In a multi‑node environment, the sweeper uses the distributed lock to ensure that only one instance publishes messages at a time, avoiding duplicate publications.

Conclusion

The inbox and outbox patterns are essential for building resilient distributed systems. Paramore.Brighter, combined with its DynamoDB integration and Kafka messaging gateway, provides a clean, robust implementation for .NET applications.

This setup guarantees that your services can handle failures gracefully while maintaining data consistency across your distributed system.

Implementing Tell and Ask in Trupe — Actor Model library in .NET

Rafael Andrade — Fri, 23 Jan 2026 09:00:00 +0000

In the previous article I introduced Trupe, a lightweight actor model library for .NET. Since then, I’ve made several design improvements to enhance clarity, performance, and developer experience.

In this article, I’ll walk you through the latest modifications and implement two essential messaging patterns: Tell and Ask.

What's Changed in Trupe?

I'm going to starting talking about the change I've done in the library

Actor Interface

In the initial design, I debated between IActor and IActor<TMessage>. I wanted to avoid forcing a base class on developers. After some reflection, I’ve refined the naming to better reflect intent.

IActor: This remains the marker interface that identifies a class as an Actor.
IHandleActorMessage<TMessage>: Renamed from IActor<TMessage>. This makes it explicit that the interface is responsible for handling a specific message type.
HandleAsync: Renamed from ReceiveAsync. "Handle" more accurately describes the processing action.

The updated interfaces look like this:

public interface IActor
{
    IActorContext Context { get; }

    ValueTask HandleAsync(object? message, CancellationToken cancellationToken = default);
}

public interface IHandleActorMessage<TMessage>
{
    ValueTask HandleAsync(TMessage message, CancellationToken cancellationToken = default);
}

Mailbox Simplification

I have modernized the IMailbox interface. Instead of a traditional Dequeue method, the mailbox now implements IAsyncEnumerable<IMessage>. This simplifies the consumption loop in ActorProcess by allowing us to use the elegant await foreach pattern.

public interface IMailbox : IAsyncEnumerable<IMessage>
{
    ValueTask EnqueueAsync(IMessage message, CancellationToken cancellationToken = default);
}

Actor Process Improvements

Task Execution Strategy

Previously, I started the actor loop using TaskCreationOptions.LongRunning. However, research highlighted that this forces the creation of a new thread outside the ThreadPool. For a library like this, that is not ideal; we want to leverage the ThreadPool's efficiency and avoid thread exhaustion. I also want to avoid capturing the SynchronizationContext unnecessarily.

Lifecycle Management

I’ve refined the start/stop API so only the supervisor controls the actor’s lifecycle. This reduces complexity and eliminates redundant sync/async APIs.

public class ActorProcess(IActor actor, IMailbox mailbox)
{
    private CancellationTokenSource? _cts;
    private Task? _executing;

    public void Start()
    {
        if (_executing != null)
        {
            return;
        }

        _cts = new CancellationTokenSource();
        _executing = Task.Run(() => RunAsync(_cts.Token));
    }


    public async Task StopAsync()
    {
        if (_cts == null || _executing == null)
        {
            return;
        }

        await _cts.CancelAsync();
        try
        {
            await _executing;
        }
        catch (OperationCanceledException)
        {
            // Ignore cancellation exceptions during shutdown
        }

        _cts.Dispose();

        _cts = null;
        _executing = null;
    }
}

Dropping Sigil for Simplicity and AOT Support

To simplify the library and support Native AOT, I removed the Sigil dependency (which relies on dynamic IL generation). Instead, I am now using C#'s reflection capabilities (MakeGenericMethod and CreateDelegate) to dispatch messages to strongly-typed handlers.

Here is the revised RunAsync method utilizing a concurrent dictionary to cache delegates:

public class ActorProcess(IActor actor, IMailbox mailbox)
{
    private static readonly ConcurrentDictionary<
        Type,
        Func<IActor, IMessage, ValueTask>
    > _typedCallHandle = new();

    private async Task RunAsync(CancellationToken cancellationToken)
    {
        await foreach (var message in mailbox.WithCancellation(cancellationToken))
        {
            if (RuntimeFeature.IsDynamicCodeSupported)
            {
                var callHandle = _typedCallHandle.GetOrAdd(
                    message.Payload.GetType(),
                    CreateCallHandleDelegate
                );

                await callHandle(actor, message);
            }
            else
            {
                await actor.HandleAsync(message.Payload, message.CancellationToken);
            }
        }
    }

    private static async ValueTask CallHandle<TMessage>(IActor actor, IMessage message)
    {
        if (actor is IHandleActorMessage<TMessage> handle)
        {
            await handle.HandleAsync((TMessage)message.Payload, message.CancellationToken);
        }
        else
        {
            await actor.HandleAsync(message.Payload, message.CancellationToken);
        }
    }

    private static readonly MethodInfo s_callHandleMethodInfo = typeof(ActorProcess).GetMethod(
        nameof(CallHandle),
        BindingFlags.Static | BindingFlags.NonPublic
    )!;

    private static Func<IActor, IMessage, ValueTask> CreateCallHandleDelegate(Type messageType)
    {
        var typed = s_callHandleMethodInfo.MakeGenericMethod(messageType);
        return typed.CreateDelegate<Func<IActor, IMessage, ValueTask>>();
    }
}

This approach ensures efficient typed message handling when dynamic code is supported, while falling back to a generic handler for AOT scenarios.

Implementing Tell and Ask

Now, let’s implement two core messaging primitives in actor systems:

Tell: A "Fire-and-forget" mechanism. The sender pushes a message and continues immediately without waiting for a result.
Ask: A "Request-Response" mechanism. The sender sends a message and waits (asynchronously) for a reply.

The IActorReference Interface

First, we define an interface for actor references that supports both patterns:

public interface IActorReference
{
    TResponse Ask<TRequest, TResponse>(TRequest request, TimeSpan? timeout = null)
        where TRequest : notnull;

    ValueTask<TResponse> AskAsync<TRequest, TResponse>(TRequest request, CancellationToken cancellationToken = default)
        where TRequest : notnull;

    void Tell<TMessage>(TMessage message, TimeSpan? timeout = null)
        where TMessage : notnull;

    ValueTask TellAsync<TMessage>(TMessage message, CancellationToken cancellationToken = default)
        where TMessage : notnull;
}

Tell – Fire and Forget

Tell is straightforward—it sends a message without waiting for a response.

public class LocalActorReference
{
    public void Tell<TMessage>(TMessage message, TimeSpan? timeout = null)
        where TMessage : notnull
    {
        var cts = new CancellationTokenSource();
        if (timeout.HasValue)
        {
            cts.CancelAfter(timeout.Value);
        }

        try
        {
            var task = TellAsync(message, cts.Token);
            if (task.IsCompletedSuccessfully)
            {
                return;
            }

            task.AsTask().GetAwaiter().GetResult();
        }
        catch (OperationCanceledException ex)
        {
            throw new TimeoutException(
                $"Tell operation timed out after {timeout?.TotalMilliseconds ?? 0} ms.",
                ex
            );
        }
    }

    public async ValueTask TellAsync<TMessage>(TMessage message, CancellationToken cancellationToken = default)
        where TMessage : notnull
    {
        await mailbox.EnqueueAsync(
            new LocalTellMessage(message, CancellationToken.None),
            cancellationToken
        );
    }
}

Ask – Request-Response

Ask is more complex. It requires a temporary mechanism to capture the response. In Trupe, the LocalAskMessage (not shown here) acts as a TaskCompletionSource that the Actor process can complete when processing finishes.

public class LocalActorReference
{
    public TResponse Ask<TRequest, TResponse>(TRequest request, TimeSpan? timeout = null)
        where TRequest : notnull
    {
        var cts = new CancellationTokenSource();
        if (timeout.HasValue)
        {
            cts.CancelAfter(timeout.Value);
        }

        try
        {
            var result = AskAsync<TRequest, TResponse>(request, cts.Token);
            if (result.IsCompletedSuccessfully)
            {
                return result.Result;
            }

            return result.AsTask().GetAwaiter().GetResult();
        }
        catch (OperationCanceledException ex)
        {
            throw new TimeoutException(
                $"Ask operation timed out after {timeout?.TotalMilliseconds ?? 0} ms.",
                ex
            );
        }
    }

    public async ValueTask<TResponse> AskAsync<TRequest, TResponse>(TRequest request, CancellationToken cancellationToken = default)
        where TRequest : notnull
    {
        var message = new LocalAskMessage(request, cancellationToken);

        await mailbox.EnqueueAsync(message, cancellationToken);

        var response = await message.AsTask();

        if (response is TResponse val)
        {
            return val;
        }

        throw new InvalidCastException(
            $"Cannot cast response of type {response?.GetType().FullName ?? "null"} to {typeof(TResponse).FullName}."
        );
    }
}

Closing the Loop: Updating ActorProcess

Finally, we need to update the RunAsync loop in ActorProcess to handle the response.

Trupe uses a context-based approach for responses. The actor handles the message and sets the result on Context.Response. The process loop then grabs this result and completes the Ask promise.

public class ActorProcess(IActor actor, IMailbox mailbox)
{
    private async Task RunAsync(CancellationToken cancellationToken)
    {
        await foreach (var message in mailbox.WithCancellation(cancellationToken))
        {
            actor.Context.Response = null;
            if (RuntimeFeature.IsDynamicCodeSupported)
            {
                var callHandle = _typedCallHandle.GetOrAdd(
                    message.Payload.GetType(),
                    CreateCallHandleDelegate
                );

                await callHandle(actor, message);
            }
            else
            {
                await actor.HandleAsync(message.Payload, message.CancellationToken);
            }

            if (message is IAskMessage askMessage)
            {
                askMessage.SetResult(actor.Context.Response);
            }

            actor.Context.Response = null;
        }
    }
}

Conclusion

With these changes, Trupe is becoming a robust foundation for building Actor systems in .NET.

Clarity: Renamed interfaces make the code self-documenting.
Simplicity: Moving to IAsyncEnumerable and removing Sigil drastically reduces maintenance overhead.
Compatibility: We are now ready for Native AOT.
Functionality: The Tell and Ask implementations provide the essential communication channels actors need.

In the next article, I will dive into Supervision and how we handle faults in Trupe.

Implementando Tell e Ask no Trupe — Biblioteca de Modelo de Atores em .NET

Rafael Andrade — Fri, 23 Jan 2026 09:00:00 +0000

No artigo anterior, apresentei o Trupe, uma biblioteca leve de modelo de atores para .NET. Desde então, fiz várias melhorias de design para aprimorar a clareza, o desempenho e a experiência do desenvolvedor.

Neste artigo, vou guiá-lo pelas modificações mais recentes e implementar dois padrões de mensagens essenciais: Tell e Ask.

O que mudou no Trupe?

Vou começar falando sobre as mudanças que realizei na biblioteca.

Interface do Ator

No design inicial, debati entre IActor e IActor<TMessage>. Eu queria evitar forçar uma classe base para os desenvolvedores. Após alguma reflexão, refinei a nomenclatura para refletir melhor a intenção.

IActor: Esta permanece como a interface marcadora (marker interface) que identifica uma classe como um Ator.
IHandleActorMessage<TMessage>: Renomeada de IActor<TMessage>. Isso torna explícito que a interface é responsável por lidar com um tipo específico de mensagem.
HandleAsync: Renomeado de ReceiveAsync. "Handle" (lidar/tratar) descreve com mais precisão a ação de processamento.

As interfaces atualizadas ficaram assim:

public interface IActor
{
    IActorContext Context { get; }

    ValueTask HandleAsync(object? message, CancellationToken cancellationToken = default);
}

public interface IHandleActorMessage<TMessage>
{
    ValueTask HandleAsync(TMessage message, CancellationToken cancellationToken = default);
}

Simplificação da Mailbox

Modernizei a interface IMailbox. Em vez de um método tradicional de Dequeue, a mailbox agora implementa IAsyncEnumerable<IMessage>. Isso simplifica o loop de consumo no ActorProcess, permitindo usarmos o elegante padrão await foreach.

public interface IMailbox : IAsyncEnumerable<IMessage>
{
    ValueTask EnqueueAsync(IMessage message, CancellationToken cancellationToken = default);
}

Melhorias no Processo do Ator

Estratégia de Execução de Tasks

Anteriormente, eu iniciava o loop do ator usando TaskCreationOptions.LongRunning. No entanto, pesquisas destacaram que isso força a criação de uma nova thread fora do ThreadPool. Para uma biblioteca como esta, isso não é ideal; queremos aproveitar a eficiência do ThreadPool e evitar a exaustão de threads. Também quero evitar capturar o SynchronizationContext desnecessariamente.

Gerenciamento de Ciclo de Vida

Refinei a API de start/stop para que apenas o supervisor controle o ciclo de vida do ator. Isso reduz a complexidade e elimina APIs redundantes de sincronização/assincronia.

public class ActorProcess(IActor actor, IMailbox mailbox)
{
    private CancellationTokenSource? _cts;
    private Task? _executing;

    public void Start()
    {
        if (_executing != null)
        {
            return;
        }

        _cts = new CancellationTokenSource();
        _executing = Task.Run(() => RunAsync(_cts.Token));
    }


    public async Task StopAsync()
    {
        if (_cts == null || _executing == null)
        {
            return;
        }

        await _cts.CancelAsync();
        try
        {
            await _executing;
        }
        catch (OperationCanceledException)
        {
            // Ignora exceções de cancelamento durante o desligamento
        }

        _cts.Dispose();

        _cts = null;
        _executing = null;
    }
}

Removendo Sigil por Simplicidade e Suporte a AOT

Para simplificar a biblioteca e suportar Native AOT, removi a dependência do Sigil (que depende de geração dinâmica de IL). Em vez disso, agora estou usando as capacidades de reflexão do C# (MakeGenericMethod e CreateDelegate) para despachar mensagens para manipuladores fortemente tipados.

Aqui está o método RunAsync revisado utilizando um dicionário concorrente para fazer cache dos delegates:

public class ActorProcess(IActor actor, IMailbox mailbox)
{
    private static readonly ConcurrentDictionary<
        Type,
        Func<IActor, IMessage, ValueTask>
    > _typedCallHandle = new();

    private async Task RunAsync(CancellationToken cancellationToken)
    {
        await foreach (var message in mailbox.WithCancellation(cancellationToken))
        {
            if (RuntimeFeature.IsDynamicCodeSupported)
            {
                var callHandle = _typedCallHandle.GetOrAdd(
                    message.Payload.GetType(),
                    CreateCallHandleDelegate
                );

                await callHandle(actor, message);
            }
            else
            {
                await actor.HandleAsync(message.Payload, message.CancellationToken);
            }
        }
    }

    private static async ValueTask CallHandle<TMessage>(IActor actor, IMessage message)
    {
        if (actor is IHandleActorMessage<TMessage> handle)
        {
            await handle.HandleAsync((TMessage)message.Payload, message.CancellationToken);
        }
        else
        {
            await actor.HandleAsync(message.Payload, message.CancellationToken);
        }
    }

    private static readonly MethodInfo s_callHandleMethodInfo = typeof(ActorProcess).GetMethod(
        nameof(CallHandle),
        BindingFlags.Static | BindingFlags.NonPublic
    )!;

    private static Func<IActor, IMessage, ValueTask> CreateCallHandleDelegate(Type messageType)
    {
        var typed = s_callHandleMethodInfo.MakeGenericMethod(messageType);
        return typed.CreateDelegate<Func<IActor, IMessage, ValueTask>>();
    }
}

Essa abordagem garante um manuseio eficiente de mensagens tipadas quando código dinâmico é suportado, enquanto recorre a um manipulador genérico para cenários AOT.

Implementando Tell e Ask

Agora, vamos implementar dois primitivos de mensagens centrais em sistemas de atores:

Tell: Um mecanismo "Fire-and-forget" (dispare e esqueça). O remetente envia uma mensagem e continua imediatamente sem esperar por um resultado.
Ask: Um mecanismo "Request-Response" (solicitação-resposta). O remetente envia uma mensagem e aguarda (assincronamente) por uma resposta.

A Interface IActorReference

Primeiro, definimos uma interface para referências de atores que suporta ambos os padrões:

public interface IActorReference
{
    TResponse Ask<TRequest, TResponse>(TRequest request, TimeSpan? timeout = null)
        where TRequest : notnull;

    ValueTask<TResponse> AskAsync<TRequest, TResponse>(TRequest request, CancellationToken cancellationToken = default)
        where TRequest : notnull;

    void Tell<TMessage>(TMessage message, TimeSpan? timeout = null)
        where TMessage : notnull;

    ValueTask TellAsync<TMessage>(TMessage message, CancellationToken cancellationToken = default)
        where TMessage : notnull;
}

Tell – Fire and Forget

Tell é direto — ele envia uma mensagem sem esperar por uma resposta.

public class LocalActorReference
{
    public void Tell<TMessage>(TMessage message, TimeSpan? timeout = null)
        where TMessage : notnull
    {
        var cts = new CancellationTokenSource();
        if (timeout.HasValue)
        {
            cts.CancelAfter(timeout.Value);
        }

        try
        {
            var task = TellAsync(message, cts.Token);
            if (task.IsCompletedSuccessfully)
            {
                return;
            }

            task.AsTask().GetAwaiter().GetResult();
        }
        catch (OperationCanceledException ex)
        {
            throw new TimeoutException(
                $"Tell operation timed out after {timeout?.TotalMilliseconds ?? 0} ms.",
                ex
            );
        }
    }

    public async ValueTask TellAsync<TMessage>(TMessage message, CancellationToken cancellationToken = default)
        where TMessage : notnull
    {
        await mailbox.EnqueueAsync(
            new LocalTellMessage(message, CancellationToken.None),
            cancellationToken
        );
    }
}

Ask – Request-Response

Ask é mais complexo. Requer um mecanismo temporário para capturar a resposta. No Trupe, a LocalAskMessage (não mostrada aqui) atua como um TaskCompletionSource que o processo do Ator pode completar quando o processamento terminar.

public class LocalActorReference
{
    public TResponse Ask<TRequest, TResponse>(TRequest request, TimeSpan? timeout = null)
        where TRequest : notnull
    {
        var cts = new CancellationTokenSource();
        if (timeout.HasValue)
        {
            cts.CancelAfter(timeout.Value);
        }

        try
        {
            var result = AskAsync<TRequest, TResponse>(request, cts.Token);
            if (result.IsCompletedSuccessfully)
            {
                return result.Result;
            }

            return result.AsTask().GetAwaiter().GetResult();
        }
        catch (OperationCanceledException ex)
        {
            throw new TimeoutException(
                $"Ask operation timed out after {timeout?.TotalMilliseconds ?? 0} ms.",
                ex
            );
        }
    }

    public async ValueTask<TResponse> AskAsync<TRequest, TResponse>(TRequest request, CancellationToken cancellationToken = default)
        where TRequest : notnull
    {
        var message = new LocalAskMessage(request, cancellationToken);

        await mailbox.EnqueueAsync(message, cancellationToken);

        var response = await message.AsTask();

        if (response is TResponse val)
        {
            return val;
        }

        throw new InvalidCastException(
            $"Cannot cast response of type {response?.GetType().FullName ?? "null"} to {typeof(TResponse).FullName}."
        );
    }
}

Fechando o Ciclo: Atualizando o ActorProcess

Finalmente, precisamos atualizar o loop RunAsync no ActorProcess para lidar com a resposta.

O Trupe usa uma abordagem baseada em contexto para respostas. O ator processa a mensagem e define o resultado em Context.Response. O loop do processo então pega esse resultado e completa a promise do Ask.

public class ActorProcess(IActor actor, IMailbox mailbox)
{
    private async Task RunAsync(CancellationToken cancellationToken)
    {
        await foreach (var message in mailbox.WithCancellation(cancellationToken))
        {
            actor.Context.Response = null;
            if (RuntimeFeature.IsDynamicCodeSupported)
            {
                var callHandle = _typedCallHandle.GetOrAdd(
                    message.Payload.GetType(),
                    CreateCallHandleDelegate
                );

                await callHandle(actor, message);
            }
            else
            {
                await actor.HandleAsync(message.Payload, message.CancellationToken);
            }

            if (message is IAskMessage askMessage)
            {
                askMessage.SetResult(actor.Context.Response);
            }

            actor.Context.Response = null;
        }
    }
}

Conclusão

Com essas mudanças, o Trupe está se tornando uma base robusta para construir sistemas de Atores em .NET.

Clareza: Interfaces renomeadas tornam o código autoexplicativo.
Simplicidade: Mover para IAsyncEnumerable e remover o Sigil reduz drasticamente a sobrecarga de manutenção.
Compatibilidade: Agora estamos prontos para Native AOT.
Funcionalidade: As implementações de Tell e Ask fornecem os canais de comunicação essenciais que os atores precisam.

No próximo artigo, mergulharei em Supervisão e como lidamos com falhas no Trupe.

Entendendo o Pipeline do Brighter

Rafael Andrade — Fri, 28 Nov 2025 09:00:00 +0000

O Brighter adota uma abordagem distinta em comparação a muitos outros frameworks, priorizando a clareza explícita em seu pipeline de tratamento de requisições. Em vez de depender de convenções ocultas ou configurações complexas, você define explicitamente o comportamento do pipeline usando atributos, garantindo controle total sobre a ordem de execução.

O Modelo da Boneca Russa (Matryoshka)

O pipeline do Brighter é arquitetado com base no Modelo da Boneca Russa. Imagine um conjunto de bonecas encaixadas: cada boneca contém uma menor em seu interior.

No Brighter, cada componente de middleware (uma "boneca") envolve o próximo na cadeia. Um método handler fica no centro. Quando uma requisição é processada, ela atravessa cada camada de middleware antes de chegar ao handler, e depois retorna passando novamente por cada camada na ordem inversa. Esse design implementa perfeitamente o Padrão Pipe and Filter.

Um recurso essencial desse modelo é a capacidade de interromper o pipeline (short-circuit). Qualquer middleware pode parar o processamento e retornar um resultado imediatamente, impedindo que a requisição chegue às camadas internas (incluindo o próprio handler).

Essa arquitetura permite que qualquer etapa do pipeline:

Execute ações antes da requisição chegar ao handler.
Encaminhe a requisição para a próxima etapa.
Execute ações após o retorno do handler interno.
Interrompa o pipeline retornando antecipadamente (cancelando o processamento).

Observação: Embora este guia use exemplos de handlers síncronos, os mesmos conceitos se aplicam a handlers assíncronos. Basta usar os atributos e classes base correspondentes (ex: RequestHandlerAsync<T>).

Adicionando um Pipeline ao Seu Handler

No Brighter, middlewares são adicionados via Atributos do C#. Você deve decorar explicitamente o método Handle do seu Request Handler.

A ordem de execução é controlada pelo parâmetro step:

Números menores executam primeiro na entrada e por último na saída.

public class SampleHandler : RequestHandler<SomeMessage>
{
    // O 'step' define a ordem. 
    // Step 1 executa antes do Step 2.
    [RequestLogging(step: 0)]
    public virtual TRequest Handle(TRequest command)
    {
        ...
    }
}

Implementando Middleware "Global"

O Brighter não oferece uma API fluente (ex: services.AddGlobalMiddleware(...)) para injetar lógica globalmente em todas as requisições. Essa é uma escolha deliberada de design para manter o pipeline explícito no nível do handler.

Porém, para aplicar lógica a todos os handlers ("middleware global"), use herança: crie um Handler Base com os atributos desejados. Todos os handlers que herdarem dessa classe adotarão o pipeline definido.

Exemplo: Criando um Handler Base com Políticas Globais

// 1. Crie um handler base com middleware comum
public abstract class MyGlobalPolicyHandler<T> : RequestHandler<T> where T : class, IRequest
{
    [RequestLogging(step: 0)]
    [FallbackPolicy(backstop: true, circuitBreaker: false, step: 1)]
    public override T Handle(T command)
    {
        // Este método propaga o comando pelo pipeline até o handler real.
        return base.Handle(command);
    }
}

// 2. Herde do handler base nas suas implementações
public class SampleHandler : MyGlobalPolicyHandler<SomeCommand>
{
    [UseResiliencePipeline("MyPolicy", step: 2)]
    public override SomeCommand Handle(SomeCommand command)
    {
        // Sua lógica de negócio principal aqui.
        Console.WriteLine("Handling SomeCommand");
        return base.Handle(command);
    }
}

Ordem de execução para SampleHandler:

RequestLogging → FallbackPolicy → UseResiliencePipeline → SampleHandler.Handle.

Middleware Oferecido pelo Brighter

O Brighter inclui middlewares prontos para uso:

RequestLoggingAttribute / RequestLoggingAsyncAttribute → Registra a entrada/saída do pipeline, incluindo tempo de execução e detalhes da requisição.
FallbackPolicyAttribute / FallbackPolicyAsyncAttribute → Define uma ação de fallback se a execução falhar. Chama o método Fallback do handler.
TimeoutPolicyAttribute → (Obsoleto) Define tempo máximo de execução. Prefira UseResiliencePipeline.
UsePolicyAttribute / UsePolicyAsyncAttribute → (Obsoleto) Usa políticas do Polly. Prefira UseResiliencePipeline.
UseResiliencePipelineAsyncAttribute / UseResiliencePipelineAttribute → Aplica pipelines de resiliência do Polly (retry, circuit-breaker, timeouts).
MonitorAttribute / MonitorAsyncAttribute → Permite monitoramento de handlers.
FeatureSwitchAttribute / FeatureSwitchAsyncAttribute → Habilita/desabilita dinamicamente um handler com base em um feature flag.

Implementando Seu Próprio Middleware

Para criar middleware personalizado, você precisa de duas classes por modo de execução (síncrono/assíncrono):

O Atributo: Decora o método do handler.
O Handler: Contém a lógica do middleware.

Exemplo de Implementação:

// 1. Classe do Atributo
[AttributeUsage(AttributeTargets.Method)]
public class MyFeatureFlagAttribute : RequestHandlerAttribute
{
    public string FeatureName { get; }

    public MyFeatureFlagAttribute(string featureName, int step) : base(step, HandlerTiming.Before)
    {
        FeatureName = featureName;
    }

    public override object[] InitializerParams()
    {
        return new object[] { FeatureName };
    }

    public override Type GetHandlerType()
    {
        // Indica qual classe handler usar para este atributo.
        return typeof(MyFeatureFlagHandler<>);
    }
}

// 2. Classe do Handler
public class MyFeatureFlagHandler<TRequest> : RequestHandler<TRequest> where TRequest : class, IRequest
{
    private string _featureName;
    private IFeatureManager _featureManager; // Gerenciador de features hipotético

    public override void InitializeFromAttributeParams(params object[] initializerList)
    {
        // Recebe parâmetros do método InitializerParams()
        _featureName = (string)initializerList[0];
        _featureManager = ...; // Normalmente injetado via DI (Injeção de Dependência)
    }

    public override TRequest Handle(TRequest command)
    {
        // Verifica se a funcionalidade está habilitada
        if (!_featureManager.IsEnabled(_featureName))
        {
            return command;
        }

        // Se habilitada, continua para o próximo handler no pipeline.
        return base.Handle(command);
    }
}

// 3. Usando o Middleware Personalizado
public class DiscountHandler : RequestHandler<ApplyDiscountCommand>
{
    [MyFeatureFlag("AdvancedDiscounts", step: 1)]
    public override ApplyDiscountCommand Handle(ApplyDiscountCommand command)
    {
        // Este código só executa se a funcionalidade "AdvancedDiscounts" estiver habilitada.
        return base.Handle(command);
    }
}

Conclusão

O pipeline explícito e baseado em atributos do Brighter oferece clareza e controle incomparáveis sobre as preocupações transversais da sua aplicação. Ao utilizar o Modelo da Boneca Russa, você constrói pipelines de handlers robustos, bem ordenados e fáceis de manter. Seja com políticas embutidas ou middlewares personalizados, o processo permanece consistente e transparente.

Referência

Documentação Oficial: Building a Pipeline of Request Handlers

Understanding the Brighter Pipeline

Rafael Andrade — Fri, 28 Nov 2025 09:00:00 +0000

Brighter takes a distinct approach compared to many other frameworks by prioritizing explicitness in its request handling pipeline. Instead of relying on hidden conventions or complex configuration, you explicitly define the behavior of your pipeline using attributes, giving you full control over the execution order.

The Russian Doll Model

The Brighter pipeline is architectured around the Russian Doll (Matryoshka) Model. Imagine a set of nested dolls: each doll contains a smaller one inside it.

In Brighter, each middleware component (a "doll") wraps the next one in the chain. A handler method is at the very center. When a request is handled, it passes through each middleware layer before reaching the handler, and then back out through each layer. This design perfectly implements the Pipe and Filter Pattern.

A key feature of this model is the ability to short-circuit the pipeline. Any middleware can stop the processing and return a result immediately, preventing the request from reaching the inner layers, including the handler itself.

This architecture allows any step in the pipeline to:

Perform work before the request reaches the handler.
Pass the request to the next step.
Perform work after the inner handler returns.
Short-circuit the pipeline by returning early (stopping the processing).

Note: While this guide uses synchronous handler examples, the same concepts apply to asynchronous handlers. Simply use the corresponding async attributes and handler base classes (e.g., RequestHandlerAsync).

Adding a Pipeline to Your Handler

In Brighter, adding middleware is done via C# Attributes. You must explicitly add the attribute to the Handle method of your Request Handler.

You control the order of execution using the step parameter. Lower numbers execute first (on the way in) and last (on the way out).

public class SampleHandler : RequestHandler<SomeMessage>
{
    // The step determines the order. 
    // Step 1 runs before Step 2.
    [RequestLogging(step: 0)]
    public virtual TRequest Handle(TRequest command)
    {
        ...
    }
}

Implementing "Global" Middleware

Brighter does not provide a fluent API (e.g., services.AddGlobalMiddleware(...)) to inject logic into every request globally. This is a deliberate design choice to keep the pipeline explicit at the handler level.

However, if you want "Global" middleware (logic that applies to all your handlers), you can achieve this through inheritance. By creating a Base Handler with the desired attributes, all inheriting handlers will adopt that pipeline.

Example: Creating a Base Handler with Global Policies

// 1. Create a base handler with common middleware
public abstract class MyGlobalPolicyHandler<T> : RequestHandler<T> where T : class, IRequest
{
    [RequestLogging(step: 0)]
    [FallbackPolicy(backstop: true, circuitBreaker: false, step: 1)]
    public override T Handle(T command)
    {
        // This call propagates the command down the pipeline to the actual handler.
        return base.Handle(command);
    }
}

// 2. Inherit from the base handler in your implementations
public class SampleHandler : MyGlobalPolicyHandler<SomeCommand>
{
    [UseResiliencePipeline("MyPolicy", step: 2)]
    public override SomeCommand Handle(SomeCommand command)
    {
        // Your core business logic here.
        Console.WriteLine("Handling SomeCommand");
        return base.Handle(command);
    }
}

The execution order for SampleHandler will be: RequestLogging -> FallbackPolicy -> UseResiliencePipeline -> SampleHandler.Handle.

Middleware Provided by Brighter

Brighter provides several useful middleware attributes out of the box:

RequestLoggingAttribute / RequestLoggingAsyncAttribute -> Logs the entry and exit of the handler pipeline, including timing and request details.
FallbackPolicyAttribute / FallbackPolicyAsyncAttribute -> Provides a fallback action if the handler execution fails. Calls a Fallback method on the handler.
TimeoutPolicyAttribute -> (Deprecated) Sets a maximum timeout for the handler. Prefer UseResiliencePipeline.
UsePolicyAttribute / UsePolicyAsyncAttribute -> (Deprecated) Wraps the handler in a Polly policy. Prefer UseResiliencePipeline.
UseResiliencePipelineAsyncAttribute / UseResiliencePipelineAttribute -> Wraps the handler in a Polly resilience pipeline for advanced retry, circuit-breaker, and timeout strategies.
MonitorAttribute / MonitorAsyncAttribute -> Allow you to monitor your handlers
FeatureSwitchAttribute / FeatureSwitchAsyncAttribute -> Dynamically enables or disables a handler based on a feature switch.

Implementing Your Own Middleware

To implement custom middleware, you generally need two classes per execution mode (sync or async):

The Attribute: Used to decorate the handler method.
The Handler: Contains the actual middleware logic.

Example Implementation:

// 1. The Attribute Class
[AttributeUsage(AttributeTargets.Method)]
public class MyFeatureFlagAttribute : RequestHandlerAttribute
{
    public string FeatureName { get; }

    public MyFeatureFlagAttribute(string featureName, int step) : base(step, HandlerTiming.Before)
    {
        FeatureName = featureName;
    }

    public override object[] InitializerParams()
    {
        return new object[] { FeatureName };
    }

    public override Type GetHandlerType()
    {
        // This tells Brighter which handler class to use for this attribute.
        return typeof(MyFeatureFlagHandler<>);
    }
}

// 2. The Handler Class
public class MyFeatureFlagHandler<TRequest> : RequestHandler<TRequest> where TRequest : class, IRequest
{
    private string _featureName;
    private IFeatureManager _featureManager; // Hypothetical feature manager

    public override void InitializeFromAttributeParams(params object[] initializerList)
    {
        // Receives the parameters from InitializerParams()
        _featureName = (string)initializerList[0];
        _featureManager = ...; // Would typically be injected via DI
    }

    public override TRequest Handle(TRequest command)
    {
        // Check if the feature is enabled
        if (!_featureManager.IsEnabled(_featureName))
        {
            return command;
        }

        // If it is enabled, continue to the next handler in the pipeline.
        return base.Handle(command);
    }
}

// 3. Using the Custom Middleware
public class DiscountHandler : RequestHandler<ApplyDiscountCommand>
{
    [MyFeatureFlag("AdvancedDiscounts", step: 1)]
    public override ApplyDiscountCommand Handle(ApplyDiscountCommand command)
    {
        // This will only execute if the "AdvancedDiscounts" feature is enabled.
        return base.Handle(command);
    }
}

Conclusion

Brighter's explicit, attribute-based pipeline offers unparalleled clarity and control over your application's cross-cutting concerns. By leveraging the Russian Doll model, you can build robust, well-ordered handler pipelines that are easy to reason about and maintain. Whether using built-in policies or creating your own custom middleware, the process remains consistent and transparent.

Reference

Official Documentation: Building a Pipeline of Request Handlers

Using MongoDB with Brighter V10

Rafael Andrade — Mon, 24 Nov 2025 09:00:00 +0000

One of the new providers that Brighter V10 supports is MongoDB. In this article, I'll explore how to use it with the Inbox and Outbox patterns.

MongoDB

MongoDB is a popular NoSQL document database that stores data in flexible, JSON-like documents. Its scalability and performance make it an excellent choice for high-throughput applications, and its document model maps naturally to the message structures often used in distributed systems.

Project Overview

We will build a .NET 8+ service that consumes/produces messages from a Kafka topic, processes them using Brighter, and uses a MongoDB database as the persistent inbox and outbox to guarantee that all messages are sent. We also need to configure a distributed lock to avoid publishing duplicate messages when running this application in a multi-node environment.

Required Packages

You can choose between two setup styles:

Pure Brighter Packages:

Paramore.Brighter.Inbox.MongoDb
Paramore.Brighter.Outbox.MongoDb
Paramore.Brighter.Locking.MongoDb
Paramore.Brighter.MessagingGateway.Kafka
Paramore.Brighter.ServiceActivator.Extensions.Hosting
Paramore.Brighter.Extensions.DependencyInjection

Fluent Brighter wrappers:

Fluent.Brighter.Kafka
Fluent.Brighter.MongoDb
Paramore.Brighter.ServiceActivator.Extensions.Hosting

A Quick Brighter Recap

Before we dive into the configuration, let's quickly recap the core Brighter concepts.

Request: Command and Event

The first key concept is the IRequest interface, which is typically implemented as a Command or an Event. These objects are used by Brighter's IAmACommandProcessor to:

Send (expects exactly one handler)
Publish (expects zero or more handlers)
Post (sends to an external message broker)

Commands represent an action you want to execute, like CreateOrder. Events represent a fact or something that has already happened, like OrderCreated.

// A command to be executed
public class CreateOrder() : Command(Id.Random());

// An event that has occurred
public class OrderCreated() : Event(Id.Random());

// You can also implement IRequest directly
public class CustomRequest : IRequest { ... }

Message Mapper

When you Post or consume messages from an external broker, you need a message mapper. This component is responsible for mapping your IRequest object to and from a Brighter Message object (which includes the header and body for the broker).

Brighter v10+ defaults to a JSON message mapper, but you can provide your own custom implementation by implementing IAmAMessageMapper or IAmAMessageMapperAsync.

Request Handler

Last but not least is the RequestHandler (or RequestHandlerAsync). This is the class that contains your actual business logic for processing a specific IRequest. Brighter's pipeline model allows you to chain handlers and add attributes for cross-cutting concerns like retries, logging, and, of course, the inbox pattern.

Here's a simple synchronous handler:

public class GreetingHandler : RequestHandler<Greeting>
{
    public override Greeting Handle(Greeting @event)
    {
        Console.WriteLine("===== Hello, {0}", @event.Name);
        return base.Handle(@event);
    }
}

public class GreetingHandlerAsync : RequestHandlerAsync<GreetingAsync>
{
    public override Task<Greeting> HandleAsync(GreetingAsync @event, CancellationToken ct = default)
    {
        Console.WriteLine("===== Hello, {0}", @event.Name);
        return base.Handle(@event);
    }
}

Configuring Brighter with Kafka

Now for the main setup. We need to tell Brighter how to connect to Kafka (our broker).

You can configure Brighter using its core "pure" components or with the Fluent Brighter API. Both are shown here.

Using Fluent Brighter:

services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddFluentBrighter(opt => opt
        .UsingKafka(kf => kf
            .SetConnection(c => c
                .SetName("sample")
                .SetBootstrapServers("localhost:9092")
                .SetSecurityProtocol(SecurityProtocol.Plaintext)
                .SetSaslMechanisms(SaslMechanism.Plain))
            .UseSubscriptions(s => s
                .AddSubscription<OrderPlaced>(sb => sb
                    .SetTopic("order-placed-topic")
                    .SetConsumerGroupId("order-placed-topic-1")
                    .SetRequeueCount(3)
                    .CreateInfrastructureIfMissing()
                    .UseReactorMode())
                .AddSubscription<OrderPaid>(sb => sb
                    .SetTopic("order-paid-topic")
                    .SetConsumerGroupId("order-paid-topic-1")
                    .CreateInfrastructureIfMissing()
                    .UseReactorMode()))
              .UsePublications(p => p
                  .AddPublication<OrderPaid>(kp => kp
                      .SetTopic("order-paid-topic")
                      .CreateTopicIfMissing())
                  .AddPublication<OrderPlaced>(kp => kp
                      .SetTopic("order-placed-topic")
                      .CreateTopicIfMissing()))));

Using Pure Brighter:

var connection = new KafkaMessagingGatewayConfiguration
{
    Name = "sample",
    BootStrapServers = ["localhost:9092"],
    SecurityProtocol = SecurityProtocol.Plaintext,
    SaslMechanisms = SaslMechanism.Plain,
};

services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddConsumers(opt =>
    {
        opt.Subscriptions =
        [
            new KafkaSubscription<OrderPlaced>(
                new SubscriptionName("subscription-orderplaced"),
                new ChannelName("order-placed-queue"),
                new RoutingKey("order-placed-topic"),
                makeChannels: OnMissingChannel.Create,
                messagePumpType: MessagePumpType.Reactor,
                groupId: "test-1"),
            new KafkaSubscription<OrderPaid>(
                new SubscriptionName("subscription-orderpaid"),
                new ChannelName("order-paid-queue"),
                new RoutingKey("order-paid-topic"),
                makeChannels: OnMissingChannel.Create,
                messagePumpType: MessagePumpType.Reactor,
                groupId: "test-2"),
        ];

        opt.DefaultChannelFactory = new ChannelFactory(new KafkaMessageConsumerFactory(connection));
    })
    .AutoFromAssemblies()
    .AddProducers(opt =>
    {
        opt.ProducerRegistry = new KafkaProducerRegistryFactory(
            connection,
            [
                new KafkaPublication<OrderPaid>
                {
                    MakeChannels = OnMissingChannel.Create,
                    Topic = new RoutingKey("order-paid-topic"),
                },
                new KafkaPublication<OrderPlaced>
                {
                    MakeChannels = OnMissingChannel.Create,
                    Topic = new RoutingKey("order-placed-topic"),
                }
                ]).Create();
    });

MongoDB

For this project, I'll configure the inbox, outbox and distributed lock for MongoDB.

Using Fluent Brighter:

services
    .AddFluentBrighter(opt => opt
        .UsingMongoDb(pg => pg
            .SetConnection(db => db
                .SetConnectionString(connectionString)
                .SetDatabaseName("brightertests")
                .SetInbox("inbox")
                .SetOutbox("outbox")
                .SetLocking("locking")))
            .UseDistributedLock()
            .UseInbox()
            .UseOutbox())
 // ... other configuration (like Kafka) goes here
);

Using Pure Brighter:

var configuration = new MongoDbConfiguration(connectionString, "brighter")
{
    Inbox = new MongoDbCollectionConfiguration { Name = "inbox" },
    Outbox = new MongoDbCollectionConfiguration { Name = "outbox" },
    Locking = new MongoDbCollectionConfiguration { Name = "locking" },
};

services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddConsumers(opt =>
    {
        opt.InboxConfiguration = new InboxConfiguration(new MongoDbInbox(configuration));
    })
   .AddProducers(opt => 
   {
       opt.Outbox = new MongoDbOutbox(configuration);
       opt.DistributedLock = new MongoDbLockingProvider(configuration);
       opt.ConnectionProvider = typeof(MongoDbConnectionProvider);
       opt.TransactionProvider = typeof(MongoDbUnitOfWork);
   });

Example: Message and Request Handlers

Let's define the messages our service will use.

Messages

public class CreateNewOrder() : Command(Id.Random())
{
    public decimal Value { get; set; }
}

public class OrderPlaced() : Event(Id.Random())
{
    public string OrderId { get; set; } = string.Empty;

    public decimal Value { get; set; }
}

public class OrderPaid() : Event(Id.Random())
{
    public string OrderId { get; set; } = string.Empty;
}

Request Handlers

For these examples, I'll use synchronous handlers (RequestHandler).

CreateNewOrderHandler: This handler receives a CreateNewOrder command (via Send) and posts two new events, OrderPlaced and OrderPaid, to Kafka. The DepositPost method stores the messages in the outbox, allowing the background sweeper process to publish them to the broker asynchronously.

public class CreateNewOrderHandler(IAmACommandProcessor commandProcessor,
    ILogger<CreateNewOrderHandler> logger) : RequestHandler<CreateNewOrder>
{
    public override CreateNewOrder Handle(CreateNewOrder command)
    {
        try
        {
            var id = Uuid.NewAsString();
            logger.LogInformation("Creating a new order: {OrderId}", id);

            commandProcessor.DepositPost(new OrderPlaced { OrderId = id, Value = command.Value });
            commandProcessor.DepositPost(new OrderPaid { OrderId = id });
            return base.Handle(command);
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "Invalid data");
            throw;
        }
    }
}

OrderPaidHandler: This handler consumes the OrderPaid event from Kafka and simply logs it.

public class OrderPaidHandler(ILogger<OrderPaidHandler> logger) : RequestHandler<OrderPaid>
{
    public override OrderPaid Handle(OrderPaid command)
    {
        logger.LogInformation("{OrderId} paid", command.OrderId);
        return base.Handle(command);
    }
}

OrderPlaceHandler: This handler simulates a potential failure. If the order value is divisible by 3, it throws an exception. The [UseResiliencePipeline] attribute allows us to apply a retry policy (e.g., "kafka-policy" configured to retry 3 times).

public class OrderPlaceHandler(ILogger<OrderPlaceHandler> logger) : RequestHandler<OrderPlaced>
{
    [UseResiliencePipeline("kafka-policy", 1)]
    public override OrderPlaced Handle(OrderPlaced command)
    {
        logger.LogInformation("{OrderId} placed with value {OrderValue}", command.OrderId, command.Value);
        if (command.Value % 3 == 0)
        {
            logger.LogError("Simulate an error for {OrderId} with value {OrderValue}", command.OrderId, command.Value);
            throw new InvalidOperationException("invalid error");
        }

        return base.Handle(command);
    }
}

This handler demonstrates two important concepts working together:

Resilience: The [UseResiliencePipeline("kafka-policy", 1)] attribute adds in-handler retry logic (assuming you've defined a policy named "kafka-policy").
Inbox Pattern: When this handler fails, the message won't be stored in the inbox. This ensures that when Kafka offsets are reset, only genuinely unprocessed messages will be reprocessed, preventing duplicates.
Outbox Pattern: By using DepositPost, messages are persisted to the local outbox store before being published. This guarantees that messages are not lost even if the broker is temporarily unavailable.

Conclusion

The Inbox Pattern is essential for building resilient distributed systems. Paramore.Brighter, combined with its MongoDB inbox and Kafka integration, provides a clean, robust implementation for .NET applications.

This setup guarantees that your services can handle failures gracefully while maintaining data consistency across your distributed system.

The full code for this sample can be found on GitHub: https://github.com/lillo42/brighter-sample/tree/v10-mongodb

Usando MongoDB com Brighter V10

Rafael Andrade — Mon, 24 Nov 2025 09:00:00 +0000

Uma das novas implementação do Brighter V10 é o MongoDB. Neste artigo, explorarei como utilizá-lo com os padrões Inbox e Outbox.

MongoDB

O MongoDB é um popular banco de dados NoSQL de documentos que armazena dados em documentos flexíveis no formato JSON-like. Sua escalabilidade e desempenho o tornam uma excelente escolha para aplicações de alto throughput, e seu modelo de documentos mapeia naturalmente para estruturas de mensagens comumente usadas em sistemas distribuídos.

Visão Geral do Projeto

Construiremos um serviço .NET 8+ que consome/produz mensagens de um tópico Kafka, processa-as usando Brighter, e utiliza um banco de dados MongoDB como inbox e outbox persistentes para garantir que todas as mensagens sejam enviadas. Também precisamos configurar um lock distribuído para evitar a publicação de mensagens duplicadas quando esta aplicação estiver executando em um ambiente com múltiplos nós.

Pacotes Necessários

Você pode escolher entre dois estilos de configuração:

Pacotes Brighter Puros:
- Paramore.Brighter.Inbox.MongoDb
- Paramore.Brighter.Outbox.MongoDb
- Paramore.Brighter.Locking.MongoDb
- Paramore.Brighter.MessagingGateway.Kafka
- Paramore.Brighter.ServiceActivator.Extensions.Hosting
- Paramore.Brighter.Extensions.DependencyInjection
Wrappers Fluent Brighter:
- Fluent.Brighter.Kafka
- Fluent.Brighter.MongoDb
- Paramore.Brighter.ServiceActivator.Extensions.Hosting

Revisão Rápida do Brighter

Antes de mergulharmos na configuração, vamos recapitular rapidamente os conceitos fundamentais do Brighter.

Request: Command e Event

O primeiro conceito-chave é a interface IRequest, normalmente implementada como um Command ou um Event. Esses objetos são utilizados pelo IAmACommandProcessor do Brighter para:

Send (espera exatamente um handler)
Publish (espera zero ou mais handlers)
Post (envia para um broker de mensagens externo)

Commands representam uma ação que você deseja executar, como CreateOrder. Events representam um fato ou algo que já aconteceu, como OrderCreated.

// Um comando a ser executado
public class CreateOrder() : Command(Id.Random());

// Um evento que ocorreu
public class OrderCreated() : Event(Id.Random());

// Você também pode implementar IRequest diretamente
public class CustomRequest : IRequest { ... }

Message Mapper

Quando você realiza um Post ou consome mensagens de um broker externo, precisa de um message mapper. Este componente é responsável por mapear seu objeto IRequest para e de um objeto Message do Brighter (que inclui cabeçalho e corpo para o broker).

O Brighter v10+ usa por padrão um message mapper JSON, mas você pode fornecer sua própria implementação personalizada implementando IAmAMessageMapper ou IAmAMessageMapperAsync.

Request Handler

Por último, mas não menos importante, temos o RequestHandler (ou RequestHandlerAsync). Esta é a classe que contém sua lógica de negócios real para processar um IRequest específico. O modelo de pipeline do Brighter permite encadear handlers e adicionar atributos para preocupações transversais como retries, logging e, é claro, o padrão inbox.

Exemplo de handler síncrono simples:

public class GreetingHandler : RequestHandler<Greeting>
{
    public override Greeting Handle(Greeting @event)
    {
        Console.WriteLine("===== Hello, {0}", @event.Name);
        return base.Handle(@event);
    }
}

public class GreetingHandlerAsync : RequestHandlerAsync<GreetingAsync>
{
    public override Task<Greeting> HandleAsync(GreetingAsync @event, CancellationToken ct = default)
    {
        Console.WriteLine("===== Hello, {0}", @event.Name);
        return base.Handle(@event);
    }
}

Configurando Brighter com Kafka

Agora para a configuração principal. Precisamos dizer ao Brighter como se conectar ao Kafka (nosso broker).

Você pode configurar o Brighter usando seus componentes "puros" ou com a API Fluent Brighter. Ambos são mostrados aqui.

Usando Fluent Brighter:

services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddFluentBrighter(opt => opt
        .UsingKafka(kf => kf
            .SetConnection(c => c
                .SetName("sample")
                .SetBootstrapServers("localhost:9092")
                .SetSecurityProtocol(SecurityProtocol.Plaintext)
                .SetSaslMechanisms(SaslMechanism.Plain))
            .UseSubscriptions(s => s
                .AddSubscription<OrderPlaced>(sb => sb
                    .SetTopic("order-placed-topic")
                    .SetConsumerGroupId("order-placed-topic-1")
                    .SetRequeueCount(3)
                    .CreateInfrastructureIfMissing()
                    .UseReactorMode())
                .AddSubscription<OrderPaid>(sb => sb
                    .SetTopic("order-paid-topic")
                    .SetConsumerGroupId("order-paid-topic-1")
                    .CreateInfrastructureIfMissing()
                    .UseReactorMode()))
              .UsePublications(p => p
                  .AddPublication<OrderPaid>(kp => kp
                      .SetTopic("order-paid-topic")
                      .CreateTopicIfMissing())
                  .AddPublication<OrderPlaced>(kp => kp
                      .SetTopic("order-placed-topic")
                      .CreateTopicIfMissing()))));

Usando Brighter Puro:

var connection = new KafkaMessagingGatewayConfiguration
{
    Name = "sample",
    BootStrapServers = ["localhost:9092"],
    SecurityProtocol = SecurityProtocol.Plaintext,
    SaslMechanisms = SaslMechanism.Plain,
};

services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddConsumers(opt =>
    {
        opt.Subscriptions =
        [
            new KafkaSubscription<OrderPlaced>(
                new SubscriptionName("subscription-orderplaced"),
                new ChannelName("order-placed-queue"),
                new RoutingKey("order-placed-topic"),
                makeChannels: OnMissingChannel.Create,
                messagePumpType: MessagePumpType.Reactor,
                groupId: "test-1"),
            new KafkaSubscription<OrderPaid>(
                new SubscriptionName("subscription-orderpaid"),
                new ChannelName("order-paid-queue"),
                new RoutingKey("order-paid-topic"),
                makeChannels: OnMissingChannel.Create,
                messagePumpType: MessagePumpType.Reactor,
                groupId: "test-2"),
        ];

        opt.DefaultChannelFactory = new ChannelFactory(new KafkaMessageConsumerFactory(connection));
    })
    .AutoFromAssemblies()
    .AddProducers(opt =>
    {
        opt.ProducerRegistry = new KafkaProducerRegistryFactory(
            connection,
            [
                new KafkaPublication<OrderPaid>
                {
                    MakeChannels = OnMissingChannel.Create,
                    Topic = new RoutingKey("order-paid-topic"),
                },
                new KafkaPublication<OrderPlaced>
                {
                    MakeChannels = OnMissingChannel.Create,
                    Topic = new RoutingKey("order-placed-topic"),
                }
                ]).Create();
    });

MongoDB

Para este projeto, configurarei o inbox, outbox e lock distribuído para MongoDB.

Usando Fluent Brighter:

services
    .AddFluentBrighter(opt => opt
        .UsingMongoDb(pg => pg
            .SetConnection(db => db
                .SetConnectionString(connectionString)
                .SetDatabaseName("brightertests")
                .SetInbox("inbox")
                .SetOutbox("outbox")
                .SetLocking("locking")))
            .UseDistributedLock()
            .UseInbox()
            .UseOutbox())
 // ... outras configurações (como Kafka) ficam aqui
);

Usando Brighter Puro:

var configuration = new MongoDbConfiguration(connectionString, "brighter")
{
    Inbox = new MongoDbCollectionConfiguration { Name = "inbox" },
    Outbox = new MongoDbCollectionConfiguration { Name = "outbox" },
    Locking = new MongoDbCollectionConfiguration { Name = "locking" },
};

services
    .AddHostedService<ServiceActivatorHostedService>()
    .AddConsumers(opt =>
    {
        opt.InboxConfiguration = new InboxConfiguration(new MongoDbInbox(configuration));
    })
   .AddProducers(opt => 
   {
       opt.Outbox = new MongoDbOutbox(configuration);
       opt.DistributedLock = new MongoDbLockingProvider(configuration);
       opt.ConnectionProvider = typeof(MongoDbConnectionProvider);
       opt.TransactionProvider = typeof(MongoDbUnitOfWork);
   });

Exemplo: Mensagens e Request Handlers

Vamos definir as mensagens que nosso serviço utilizará.

Mensagens

public class CreateNewOrder() : Command(Id.Random())
{
    public decimal Value { get; set; }
}

public class OrderPlaced() : Event(Id.Random())
{
    public string OrderId { get; set; } = string.Empty;
    public decimal Value { get; set; }
}

public class OrderPaid() : Event(Id.Random())
{
    public string OrderId { get; set; } = string.Empty;
}

Request Handlers

Para estes exemplos, usarei handlers síncronos (RequestHandler).

CreateNewOrderHandler: Este handler recebe um comando CreateNewOrder (via Send) e publica dois novos eventos, OrderPlaced e OrderPaid, no Kafka. O método DepositPost armazena as mensagens no outbox, permitindo que o processo sweeper em background as publique no broker assincronamente.

public class CreateNewOrderHandler(IAmACommandProcessor commandProcessor,
    ILogger<CreateNewOrderHandler> logger) : RequestHandler<CreateNewOrder>
{
    public override CreateNewOrder Handle(CreateNewOrder command)
    {
        try
        {
            var id = Uuid.NewAsString();
            logger.LogInformation("Criando novo pedido: {OrderId}", id);

            commandProcessor.DepositPost(new OrderPlaced { OrderId = id, Value = command.Value });
            commandProcessor.DepositPost(new OrderPaid { OrderId = id });
            return base.Handle(command);
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "Dados inválidos");
            throw;
        }
    }
}

OrderPaidHandler: Este handler consome o evento OrderPaid do Kafka e simplesmente registra em log.

public class OrderPaidHandler(ILogger<OrderPaidHandler> logger) : RequestHandler<OrderPaid>
{
    public override OrderPaid Handle(OrderPaid command)
    {
        logger.LogInformation("Pedido {OrderId} pago", command.OrderId);
        return base.Handle(command);
    }
}

OrderPlaceHandler: Este handler simula uma falha potencial. Se o valor do pedido for divisível por 3, ele lança uma exceção. O atributo [UseResiliencePipeline] nos permite aplicar uma política de retry (ex: "kafka-policy" configurada para 3 tentativas).

public class OrderPlaceHandler(ILogger<OrderPlaceHandler> logger) : RequestHandler<OrderPlaced>
{
    [UseResiliencePipeline("kafka-policy", 1)]
    public override OrderPlaced Handle(OrderPlaced command)
    {
        logger.LogInformation("Pedido {OrderId} criado com valor {OrderValue}", command.OrderId, command.Value);
        if (command.Value % 3 == 0)
        {
            logger.LogError("Simulando erro para pedido {OrderId} com valor {OrderValue}", command.OrderId, command.Value);
            throw new InvalidOperationException("erro inválido");
        }

        return base.Handle(command);
    }
}

Este handler demonstra dois conceitos importantes trabalhando juntos:

Resiliência: O atributo [UseResiliencePipeline("kafka-policy", 1)] adiciona lógica de retry dentro do handler (assumindo que você definiu uma política chamada "kafka-policy").
Padrão Inbox: Quando este handler falha, a mensagem não é armazenada no inbox. Isso garante que, ao redefinir offsets do Kafka, apenas mensagens genuinamente não processadas sejam reprocessadas, evitando duplicatas.
Padrão Outbox: Ao usar DepositPost, as mensagens são persistidas no outbox local antes da publicação. Isso garante que as mensagens não sejam perdidas mesmo se o broker estiver temporariamente indisponível.

Conclusão

O Padrão Inbox é essencial para construir sistemas distribuídos resilientes. O Paramore.Brighter, combinado com seu inbox MongoDB e integração Kafka, oferece uma implementação limpa e robusta para aplicações .NET.

Esta configuração garante que seus serviços possam lidar com falhas com elegância enquanto mantêm a consistência de dados em todo o sistema distribuído.

O código completo desta amostra está disponível no GitHub: https://github.com/lillo42/brighter-sample/tree/v10-mongodb

A First Look at the Phoenix Framework

Rafael Andrade — Fri, 21 Nov 2025 09:00:00 +0000

As a .NET developer embarking on a journey to learn Elixir, one of the first questions is: "How do I build a web API?" In the Elixir ecosystem, the answer is overwhelmingly the Phoenix Framework. If you're familiar with ASP.NET Core, you'll find Phoenix to be a powerful and elegant counterpart.

In this article, I'll break down the key components of Phoenix and draw comparisons to the .NET world to help frame my understanding.

Overview

Phoenix is the web development framework for the Elixir language. It allows you to build modern web applications, including Web APIs, Web Sockets (for real-time functionality), and traditional Model View Controller (MVC) apps.

In the Microsoft world, it is the direct equivalent of ASP.NET Core. Phoenix is known for high developer productivity and exceptional application performance.

Plug: The HTTP Abstraction

Just as ASP.NET Core is built on top of the middleware concept, Phoenix is built on Plug.

Plug is both a specification for composable modules between web applications and an abstraction layer for web servers (like Cowboy or Bandit. The core concept is a unified connection (the %Plug.Conn{} struct, similar to HttpContext in .NET) that is transformed as it moves through a series of functions.

This is more granular than ASP.NET Core's middleware. A single Plug is a function that takes a connection and returns a connection, making them highly composable.

Pipelines: Organized Middleware

In Phoenix, you group Plugs into "pipelines" within your router. This is a cleaner and more declarative way to organize middleware compared to the app.UseMiddleware<T>() calls in the Program.cs file.

In Phoenix, we explicitly create a pipeline. You add your middleware (called "plugs")—such as authentication or validation—explicitly before the request hits your controller.

Phoenix Example:

defmodule HelloWeb.Router do
  use HelloWeb, :router

  pipeline :browser do
    plug :accepts, ["html"]
    plug :fetch_session
    plug :fetch_live_flash
    plug :put_root_layout, html: {HelloWeb.LayoutView, :root}
    plug :protect_from_forgery
    plug :put_secure_browser_headers
    plug HelloWeb.Plugs.Locale, "en"
  end

  scope "/", HelloWeb do
    pipe_through :browser

    get "/", PageController, :index
  end

In the sample above, we created a pipeline named :browser. Before a request can hit the PageController, it must pass through this pipeline. This feels much clearer than .NET because you can see, in a single place, exactly what steps a request goes through before reaching the controller.

Simplifying Controllers with Plugs

Plugs can also be used to clean up your controller logic. Look at this "messy" controller:

defmodule HelloWeb.MessageController do
  use HelloWeb, :controller

  def show(conn, params) do
    case Authenticator.find_user(conn) do
      {:ok, user} ->
        case find_message(params["id"]) do
          nil ->
            conn |> put_flash(:info, "That message wasn't found") |> redirect(to: ~p"/")
          message ->
            if Authorizer.can_access?(user, message) do
              render(conn, :show, page: message)
            else
              conn |> put_flash(:info, "You can't access that page") |> redirect(to: ~p"/")
            end
        end
      :error ->
        conn |> put_flash(:info, "You must be logged in") |> redirect(to: ~p"/")
    end
  end
end

The nesting makes the application flow difficult to follow. We can use Plug to refactor this into a functional pipeline:

defmodule HelloWeb.MessageController do
  use HelloWeb, :controller

  plug :authenticate
  plug :fetch_message
  plug :authorize_message

  def show(conn, params) do
    render(conn, :show, page: conn.assigns[:message])
  end

  defp authenticate(conn, _) do
    case Authenticator.find_user(conn) do
      {:ok, user} ->
        assign(conn, :user, user)
      :error ->
        conn |> put_flash(:info, "You must be logged in") |> redirect(to: ~p"/") |> halt()
    end
  end

  defp fetch_message(conn, _) do
    case find_message(conn.params["id"]) do
      nil ->
        conn |> put_flash(:info, "That message wasn't found") |> redirect(to: ~p"/") |> halt()
      message ->
        assign(conn, :message, message)
    end
  end

  defp authorize_message(conn, _) do
    if Authorizer.can_access?(conn.assigns[:user], conn.assigns[:message]) do
      conn
    else
      conn |> put_flash(:info, "You can't access that page") |> redirect(to: ~p"/") |> halt()
    end
  end
end

This makes our show method drastically clearer.

Routing

Phoenix routing feels very similar to "Minimal APIs" in .NET. You have a single file defining all routes, pointing them to the specific controller and pipeline that should be used.

defmodule HelloWeb.Router do
  use HelloWeb, :router

  pipeline :browser do
    plug :accepts, ["html"]
    plug :fetch_session
    plug :fetch_live_flash
    plug :put_root_layout, html: {HelloWeb.Layouts, :root}
    plug :protect_from_forgery
    plug :put_secure_browser_headers
  end

  pipeline :api do
    plug :accepts, ["json"]
  end

  scope "/", HelloWeb do
    pipe_through :browser

    get "/", PageController, :home
  end

  # Other scopes may use custom stacks.
  # scope "/api", HelloWeb do
  #   pipe_through :api
  # end

  # ...
end

Another excellent feature of Phoenix is the ability to list every route in the application via the CLI:

mix phx.routes

Output:

GET  /  HelloWeb.PageController :home

Controllers

Creating a controller in Phoenix is straightforward. You use the base module to bring in the necessary functionality and define actions as functions.

defmodule HelloWeb.PageController do
  use HelloWeb, :controller

  def home(conn, _params) do
    render(conn, :home)
  end
end

Each action takes a connection (conn) and parameters, and returns a (potentially modified) connection. This functional approach—transforming the connection through a series of steps—is a core pattern.

My Opinion: .NET vs. Elixir Ecosystems

Coming from the .NET world, I find myself constantly comparing the two.

It is refreshing that Elixir has one strong, dominant web framework. In other ecosystems, you might have to learn 2 or 3 different frameworks before deciding which to use. In Elixir, the whole community rallies around Phoenix.

Phoenix seems to shine brightest when we talk about real-time features (like LiveView). In .NET, we have SignalR and Blazor Server, but I don't see as many projects adopting them for the full stack compared to Phoenix adoption in Elixir.

The Ecosystem Factor .NET has a massive company behind it. This is both good and bad.

The Good: They have immense resources to invest in the ecosystem.
The Bad: It can stifle open-source innovation. For example, ASP.NET Core is maintained by Microsoft, so it is very difficult for a community-driven web framework to compete. Similarly, Entity Framework Core effectively replaced NHibernate, and Microsoft's native Dependency Injection reduced the need for libraries like Autofac or Ninject.

In the Elixir world, there is no single tech giant controlling the direction. The Elixir and Phoenix teams (which are relatively small) are doing an incredible job, and their work supports the open-source initiative rather than dominating it.

Reference

https://www.phoenixframework.org/
https://hexdocs.pm/phoenix/overview.html
https://en.wikipedia.org/wiki/Phoenix_(web_framework)
https://elixirschool.com/en/lessons/misc/plug

Um Primeiro Olhar sobre o Framework Phoenix

Rafael Andrade — Fri, 21 Nov 2025 09:00:00 +0000

Como desenvolvedor .NET embarcando em uma jornada para aprender Elixir, uma das primeiras perguntas é: "Como construo uma API web?" No ecossistema Elixir, a resposta é predominantemente o Framework Phoenix. Se você está familiarizado com ASP.NET Core, verá que o Phoenix é uma contraparte poderosa e elegante.

Neste artigo, vou detalhar os componentes-chave do Phoenix e fazer comparações com o mundo .NET para ajudar a estruturar meu entendimento.

Visão Geral

Phoenix é o framework de desenvolvimento web para a linguagem Elixir. Ele permite que você construa aplicações web modernas, incluindo APIs Web, Web Sockets (para funcionalidade em tempo real) e aplicativos tradicionais Model View Controller (MVC).

No mundo Microsoft, ele é o equivalente direto do ASP.NET Core. Phoenix é conhecido por alta produtividade do desenvolvedor e excepcional performance de aplicação.

Plug: A Abstração HTTP

Assim como o ASP.NET Core é construído em cima do conceito de middleware, o Phoenix é construído sobre o Plug.

Plug é tanto uma especificação para módulos compostos entre aplicações web quanto uma camada de abstração para servidores web (como Cowboy ou Bandit). O conceito central é uma conexão unificada (a struct %Plug.Conn{}, similar ao HttpContext no .NET) que é transformada conforme passa por uma série de funções.

Isso é mais granular que o middleware do ASP.NET Core. Um único Plug é uma função que recebe uma conexão e retorna uma conexão, tornando-os altamente compostáveis.

Pipelines: Middleware Organizado

No Phoenix, você agrupa Plugs em "pipelines" dentro do seu roteador. Esta é uma forma mais limpa e declarativa de organizar middleware comparado às chamadas app.UseMiddleware<T>() no arquivo Program.cs.

No Phoenix, criamos explicitamente um pipeline. Você adiciona seu middleware (chamado "plugs")—como autenticação ou validação—explicitamente antes que a requisição chegue ao seu controller.

Exemplo Phoenix:

defmodule HelloWeb.Router do
  use HelloWeb, :router

  pipeline :browser do
    plug :accepts, ["html"]
    plug :fetch_session
    plug :fetch_live_flash
    plug :put_root_layout, html: {HelloWeb.LayoutView, :root}
    plug :protect_from_forgery
    plug :put_secure_browser_headers
    plug HelloWeb.Plugs.Locale, "en"
  end

  scope "/", HelloWeb do
    pipe_through :browser

    get "/", PageController, :index
  end

Na amostra acima, criamos um pipeline nomeado :browser. Antes que uma requisição possa chegar ao PageController, ela precisa passar por este pipeline. Isso parece muito mais claro que no .NET porque você pode ver, em um único lugar, exatamente quais passos uma requisição percorre antes de chegar ao controller.

Simplificando Controllers com Plugs

Plugs também podem ser usados para limpar a lógica do seu controller. Veja este controller "bagunçado":

defmodule HelloWeb.MessageController do
  use HelloWeb, :controller

  def show(conn, params) do
    case Authenticator.find_user(conn) do
      {:ok, user} ->
        case find_message(params["id"]) do
          nil ->
            conn |> put_flash(:info, "Aquela mensagem não foi encontrada") |> redirect(to: ~p"/")
          message ->
            if Authorizer.can_access?(user, message) do
              render(conn, :show, page: message)
            else
              conn |> put_flash(:info, "Você não pode acessar essa página") |> redirect(to: ~p"/")
            end
        end
      :error ->
        conn |> put_flash(:info, "Você precisa estar logado") |> redirect(to: ~p"/")
    end
  end
end

O aninhamento dificulta o acompanhamento do fluxo da aplicação. Podemos usar Plug para refatorar isso em um pipeline funcional:

defmodule HelloWeb.MessageController do
  use HelloWeb, :controller

  plug :authenticate
  plug :fetch_message
  plug :authorize_message

  def show(conn, params) do
    render(conn, :show, page: conn.assigns[:message])
  end

  defp authenticate(conn, _) do
    case Authenticator.find_user(conn) do
      {:ok, user} ->
        assign(conn, :user, user)
      :error ->
        conn |> put_flash(:info, "Você precisa estar logado") |> redirect(to: ~p"/") |> halt()
    end
  end

  defp fetch_message(conn, _) do
    case find_message(conn.params["id"]) do
      nil ->
        conn |> put_flash(:info, "Aquela mensagem não foi encontrada") |> redirect(to: ~p"/") |> halt()
      message ->
        assign(conn, :message, message)
    end
  end

  defp authorize_message(conn, _) do
    if Authorizer.can_access?(conn.assigns[:user], conn.assigns[:message]) do
      conn
    else
      conn |> put_flash(:info, "Você não pode acessar essa página") |> redirect(to: ~p"/") |> halt()
    end
  end
end

Isso torna nosso método show drasticamente mais claro.

Rotas

O roteamento no Phoenix parece muito similar às "Minimal APIs" no .NET. Você tem um único arquivo definindo todas as rotas, apontando-as para o controller e pipeline específicos que devem ser usados.

defmodule HelloWeb.Router do
  use HelloWeb, :router

  pipeline :browser do
    plug :accepts, ["html"]
    plug :fetch_session
    plug :fetch_live_flash
    plug :put_root_layout, html: {HelloWeb.Layouts, :root}
    plug :protect_from_forgery
    plug :put_secure_browser_headers
  end

  pipeline :api do
    plug :accepts, ["json"]
  end

  scope "/", HelloWeb do
    pipe_through :browser

    get "/", PageController, :home
  end

  # Outros escopos podem usar pilhas personalizadas.
  # scope "/api", HelloWeb do
  #   pipe_through :api
  # end

  # ...
end

Outro recurso excelente do Phoenix é a capacidade de listar todas as rotas da aplicação via CLI:

mix phx.routes

Saída:

GET  /  HelloWeb.PageController :home

Controllers

Criar um controller no Phoenix é simples. Você usa o módulo base para trazer a funcionalidade necessária e define actions como funções.

defmodule HelloWeb.PageController do
  use HelloWeb, :controller

  def home(conn, _params) do
    render(conn, :home)
  end
end

Cada action recebe uma conexão (conn) e parâmetros, e retorna uma conexão (potencialmente modificada). Essa abordagem funcional—transformando a conexão através de uma série de passos—é um padrão fundamental.

Minha Opinião: Ecossistemas .NET vs. Elixir

Vindo do mundo .NET, encontro-me constantemente comparando os dois.

É refrescante que Elixir tenha um framework web forte e dominante. Em outros ecossistemas, você pode precisar aprender 2 ou 3 frameworks diferentes antes de decidir qual usar. Em Elixir, toda a comunidade se une em torno do Phoenix.

Phoenix parece brilhar mais quando falamos de recursos em tempo real (como LiveView). No .NET, temos SignalR e Blazor Server, mas não vejo tantos projetos adotando-os para o stack completo comparado à adoção do Phoenix no Elixir.

O Fator do Ecossistema: .NET tem uma empresa gigantesca por trás. Isso é tanto bom quanto ruim.

O Bom: Eles têm imensos recursos para investir no ecossistema.
O Ruim: Pode sufocar a inovação de código aberto. Por exemplo, ASP.NET Core é mantido pela Microsoft, então é muito difícil para um framework web impulsionado pela comunidade competir. Similarmente, Entity Framework Core efetivamente substituiu o NHibernate, e a Injeção de Dependência nativa da Microsoft reduziu a necessidade de bibliotecas como Autofac ou Ninject.

No mundo Elixir, não há um único gigante da tecnologia controlando a direção. As equipes do Elixir e Phoenix (que são relativamente pequenas) estão fazendo um trabalho incrível, e seu trabalho apoia a iniciativa de código aberto em vez de dominá-la.

Referências

https://www.phoenixframework.org/

https://hexdocs.pm/phoenix/overview.html

https://en.wikipedia.org/wiki/Phoenix_(web_framework)
https://elixirschool.com/en/lessons/misc/plug