Aggregator

Combina uma série de mensagens relacionadas (mesmo correlationId) em uma única mensagem consolidada, usando uma condição de conclusão — contagem, timeout, ou ambos. É o padrão inverso do Splitter: aqui muitas mensagens viram uma; no Splitter, uma vira muitas.

passos: 6
arquivos: 8
testes: 2
tecnologias: 5

Infraestrutura realTypeScriptBunElysiaRabbitMQDocker

Cenário Relações Comparar

Baixar cartão

Cenário

Ao final do dia, cada transação liquidada (PIX, TED, boleto) gera uma confirmação publicada na exchange settlement, identificada por businessDate — o correlationId do lote. O worker agregador junta as confirmações de cada data até chegar o total esperado ou até a janela vencer, e publica um relatório de fechamento (EOD) em settlement-report.

Como funciona

6 passos

Junta partes relacionadas numa só.

Esconde cada passo: lembre antes de tocar para revelar.

Vocabulário compartilhado

ack+12

mandatory+5

amountCents+2

transactionId+1

Content Filter

total+1

Splitter

recovery+1

RabbitMQ — liquidação de pagamentos com retry e DLQ

01
Cada confirmação traz businessDate, transactionId, total e amountCents, validada por schema — a inválida vai para a DLQ
02
A primeira confirmação de uma data abre a janela e o timer; a confirmação é confirmada (ack) assim que entra na janela
03
Reentrega com o mesmo transactionId substitui a si mesma em vez de inflar a contagem
04
Ao chegar total, a janela fecha por count e cancela o timer; se o timer vence antes, fecha por timeout com transactionsPending
05
O relatório sai uma única vez, por confirm channel com mandatory, e o estado da data é descartado
06
O worker roda em processo próprio, com recovery do amqplib e probes /health/live e /health/ready

apps/messaging-streaming/aggregator

8 arquivos

src/

aggregation_settlement.tsMáquina de estado da janela: contagem, timeout e deduplicação por transactionId
worker_aggregator.tsWorker: consome confirmações, publica relatórios, probes e desligamento limpo
topology_settlement.tsExchange, filas, DLX, schemas e publish confirmado com mandatory
config_settlement.tsLê e valida a configuração no boot
demo.tsdemoSobe o worker e fecha uma data por contagem, outra por timeout e manda uma mensagem inválida à DLQ
aggregation_settlement.test.tstesteUnitário: fronteira da contagem e da janela, relatório único, reentrega
worker_aggregator.test.tstesteIntegração com RabbitMQ real: relatório, timeout, DLQ e mandatory