Skip to main content

Outbox | Padrões de Projeto

  • Outbox | Padrões de Projeto

alt text

Fluxo Outbox + SAGA​

          ┌───────────────────────────────┐
          │          SERVICE A             │
          │        (Order Service)         │
          └───────────────┬───────────────┘
                          │
                 (1) cria pedido
                          │
          ┌───────────────v───────────────┐
          │        BANCO DO SERVICE A      │
          │                               │
          │  orders         outbox         │
          │  - id           - id           │
          │  - status       - event_type   │
          │  ...            - payload      │
          │                - status=PENDING│
          └───────────────┬───────────────┘
                          │  (2) mesma transação:
                          │      grava pedido + outbox
                          │
          ┌───────────────v───────────────┐
          │     OUTBOX PUBLISHER/WORKER    │
          │  lê outbox e publica mensagem  │
          └───────────────┬───────────────┘
                          │ (3) publica evento
                          v
                ┌───────────────────┐
                │   MESSAGE BROKER   │
                │ (Kafka/RabbitMQ)   │
                └─────────┬─────────┘
                          │
                          │ (4) outros serviços consomem
          ┌───────────────v───────────────┐
          │          SERVICE B             │
          │     (Payment / Stock etc.)     │
          └───────────────┬───────────────┘
                          │
                    processa etapa
                          │
                    publica novo evento
                          v
                ┌───────────────────┐
                │   MESSAGE BROKER   │
                └───────────────────┘
                          │
                          v
              (fluxo continua = SAGA)

Intenção​

  • O Outbox pattern permite publicar eventos de forma confiável. A ideia é que, quando sua API publica mensagens, ela não as envie diretamente. Em vez disso, as mensagens são persistidas em uma tabela do banco de dados (na mesma transação em que o pedido for gravado) e depois disso, um worker consome esses registros e publica essas mensagens no broker, em intervalos de tempo predefinidos.

Exemplo​

  • Essa classe é configurada como interceptor do EFCore e quando é chamado o SaveChanges ele entra em ação

public sealed class ConvertDomainEventsToOutboxMessagesInterceptor
     : SaveChangesInterceptor
{
    public override ValueTask<InterceptionResult<int>> SavingChangesAsync(
        DbContextEventData eventData,
        InterceptionResult<int> result,
        CancellationToken cancellationToken = default)
    {
        DbContext? dbContext = eventData.Context;

        if (dbContext is null)
        {
            return base.SavingChangesAsync(eventData, result, cancellationToken);
        }

        var outboxMessages = dbContext.ChangeTracker
            .Entries<AggregateRoot>()
            .Select(x => x.Entity)
            .SelectMany(aggregateRoot =>
            {
                var domainEvents = aggregateRoot.GetDomainEvents();

                aggregateRoot.ClearDomainEvents();

                return domainEvents;
            })
            .Select(domainEvent => new OutboxMessage
            {
                Id = Guid.NewGuid(),
                OccurredOnUtc = DateTime.UtcNow,
                Type = domainEvent.GetType().Name,
                Content = JsonConvert.SerializeObject(
                    domainEvent,
                    new JsonSerializerSettings
                    {
                        TypeNameHandling = TypeNameHandling.All
                    })
            })
            .ToList();

        dbContext.Set<OutboxMessage>().AddRange(outboxMessages);

        return base.SavingChangesAsync(eventData, result, cancellationToken);
    }
}

  • Depois é criado um job para realizar o envio desse evento

[DisallowConcurrentExecution]
public class ProcessOutboxMessagesJob : IJob
{
    private readonly ApplicationDbContext _dbContext;
    private readonly IPublisher _publisher;

    public ProcessOutboxMessagesJob(ApplicationDbContext dbContext, IPublisher publisher)
    {
        _dbContext = dbContext;
        _publisher = publisher;
    }

    public async Task Execute(IJobExecutionContext context)
    {
        List<OutboxMessage> messages = await _dbContext
            .Set<OutboxMessage>()
            .Where(m => m.ProcessedOnUtc == null)
            .Take(20)
            .ToListAsync(context.CancellationToken);

        foreach (OutboxMessage outboxMessage in messages)
        {
            IDomainEvent? domainEvent = JsonConvert
                .DeserializeObject<IDomainEvent>(
                    outboxMessage.Content,
                    new JsonSerializerSettings
                    {
                        TypeNameHandling = TypeNameHandling.All
                    });

            if (domainEvent is null)
            {
                continue;
            }

            await _publisher.Publish(domainEvent, context.CancellationToken);

            outboxMessage.ProcessedOnUtc = DateTime.UtcNow;
        }

        await _dbContext.SaveChangesAsync();
    }
}

Anotações​