Publish-Subscribe Channel no Mule: um evento para vários consumidores com o Anypoint MQ

Quando um evento interessa a mais de um sistema, o Publish-Subscribe Channel entrega uma cópia a cada interessado sem que quem publica precise conhecê-los. Esta página mostra o padrão no livro Enterprise Integration Patterns e no Mule 4, com um message exchange do Anypoint MQ (o recurso do broker, não o Anypoint Exchange dos assets), no cenário de pedidos da série. Como o message exchange só existe no broker, parte do padrão é configuração do Anypoint MQ, fora do XML da aplicação.

Um pedido criado que três sistemas precisam receber

A API de pedidos recebe um pedido, grava e precisa avisar três sistemas: o pagamento cobra, o estoque reserva os itens e a nota fiscal é emitida. O livro descreve o mesmo problema de forma geral: uma aplicação usa mensagens para anunciar eventos e precisa que cada evento chegue a todos os receptores interessados (Publish-Subscribe Channel).

O cenário acrescenta uma restrição: a API de pedidos não deveria saber quem são os interessados, nem mudar quando aparecer um quarto.

No livro, cada assinante recebe a sua cópia uma vez

O desenho abaixo refaz o diagrama do livro e responde a uma pergunta: o que acontece com um evento publicado num canal publish-subscribe?

A solução do livro é publicar o evento num Publish-Subscribe Channel, que entrega uma cópia dele a cada receptor. O canal tem uma entrada que se divide em várias saídas, uma por assinante, e põe uma cópia do evento em cada saída. Cada saída tem um único assinante, que só pode consumir aquela mensagem uma vez; assim, cada assinante recebe o evento uma vez, e a cópia consumida sai do seu canal (Publish-Subscribe Channel).

Para o assinante que está fora do ar, o livro tem outro padrão, o Durable Subscriber: o sistema de mensagens guarda o que foi publicado e entrega quando ele volta (Durable Subscriber).

No Anypoint MQ, o canal é um message exchange com filas ligadas

No Anypoint MQ, quem faz o papel do canal é o message exchange, um destino do broker, como uma fila. Você liga filas a um message exchange e, por padrão, ele envia cada mensagem a todas as filas ligadas (message exchanges). Cada ligação entre o message exchange e uma fila se chama binding (glossário do Anypoint MQ).

O message exchange aceita só a operação Publish, nunca Consume: o Anypoint MQ repassa a mensagem às filas ligadas, e a aplicação consome da fila. Quem tenta consumir direto do message exchange recebe da API REST do Anypoint MQ o erro The destination should be a queue. A mesma página traz os limites: a mensagem enviada a um message exchange tem no máximo 10 MB, e um message exchange aceita até 450 filas ligadas, ou até 25 quando há regras de roteamento (message exchanges).

Comparando com o livro, cada fila ligada faz o papel da saída de um assinante, e ela também cobre o caso do Durable Subscriber. A cópia fica na fila até ser consumida, dentro do tempo de vida (TTL) configurado na fila, que vai até duas semanas (filas).

O pedido criado no message exchange do Anypoint MQ

O diagrama responde onde fica cada peça: o que é da aplicação Mule e o que é configuração do broker.

A API de pedidos publica o evento no message exchange order-created. Três bindings ligam três filas a ele, uma por consumidor. Cada consumidor é um flow com um Subscriber na sua fila, numa aplicação Mule que pode ser separada das outras. A API conhece só o nome do message exchange; cada consumidor conhece só a sua fila.

Os bindings ficam no Anypoint MQ, não na aplicação

Nenhum XML da aplicação diz quais filas recebem o evento. As filas são criadas antes, na página Destinations do Anypoint MQ. Depois, ao criar o message exchange ou nas configurações dele, você marca as filas que quer ligar (message exchanges), e o Anypoint MQ cria e mantém os bindings a partir dessa escolha (glossário do Anypoint MQ). Por isso o quarto interessado recebe uma fila e um binding, e a API de pedidos continua igual. Como o Anypoint MQ repassa às filas ligadas as mensagens publicadas no message exchange, uma fila ligada depois recebe só o que for publicado a partir do binding, não os eventos anteriores (message exchanges).

A conexão e a publicação na API de pedidos

Os três blocos desta seção e da próxima são configuração de referência, com elementos e atributos tirados da documentação do Anypoint MQ Connector. Eles não rodam no repositório mulesoft-patterns-examples: precisam de um broker do Anypoint MQ, e o conector é gratuito, mas a conexão com o serviço exige licença (Anypoint MQ Connector).

A configuração global aponta para a URL da região e usa o ID e o segredo de uma client app do Anypoint MQ (referência do conector). O segredo vem de uma propriedade segura: o prefixo ${secure::…} só funciona com o módulo Secure Properties e um secure-properties:config na aplicação.

<!-- Configuração de referência: não roda no repositório de exemplos (precisa de um broker do Anypoint MQ) -->
<anypoint-mq:config name="Anypoint_MQ_Config">
    <anypoint-mq:connection url="${anypoint.mq.url}"
        clientId="${anypoint.mq.clientId}"
        clientSecret="${secure::anypoint.mq.clientSecret}"/>
</anypoint-mq:config>

Depois de gravar o pedido, o flow da API publica o evento. Para o conector, o message exchange é um destino como uma fila: a operação Publish recebe o nome dele em destination, e o corpo da mensagem pode ser montado com DataWeave em anypoint-mq:body (Publish).

<!-- Configuração de referência: não roda no repositório de exemplos (precisa de um broker do Anypoint MQ) -->
<anypoint-mq:publish config-ref="Anypoint_MQ_Config" destination="order-created">
    <anypoint-mq:body><![CDATA[#[output application/json ---
    {
        orderId: payload.orderId,
        customerId: payload.customerId,
        items: payload.items
    }]]]></anypoint-mq:body>
</anypoint-mq:publish>

Um consumidor por fila, com confirmação automática

O consumidor de pagamento escuta a fila dele com o Subscriber. Os de estoque e de nota fiscal têm a mesma forma, cada um na sua fila.

<!-- Configuração de referência: não roda no repositório de exemplos (precisa de um broker do Anypoint MQ) -->
<flow name="payment-consumer">
    <anypoint-mq:subscriber config-ref="Anypoint_MQ_Config"
        destination="order-created-payment"
        acknowledgementMode="AUTO"/>
    <flow-ref name="charge-order"/>
</flow>

AUTO já é o padrão do Subscriber e está escrito aqui só para ficar visível (referência do conector). Nesse modo, o conector confirma a mensagem (ACK) só se o flow terminar sem erro. Se o flow terminar com um erro propagado, a mensagem não é confirmada (NACK) e volta para a fila para ser entregue de novo (Subscriber). O modo IMMEDIATE confirma antes do processamento: a mensagem sai da fila e não volta se o processamento falhar. No MANUAL, o próprio flow chama as operações ACK ou NACK com o ackToken da mensagem (ACK e NACK).

O que muda em relação a uma fila e a três chamadas HTTP

Uma fila sozinha não resolve este cenário. Num Point-to-Point Channel, só um receptor consome cada mensagem; se três consumidores lerem a mesma fila, eles disputam as mensagens, e cada pedido chega a apenas um deles (Point-to-Point Channel). Isso serve para dividir carga entre instâncias do mesmo consumidor (Competing Consumers), não para avisar três sistemas diferentes.

Três chamadas HTTP resolvem a entrega, mas a API de pedidos passaria a conhecer o endereço dos três sistemas, e cada interessado novo seria uma mudança nela. A falha também fica com quem chama. Se a nota fiscal estiver fora do ar, a chamada falha na hora, e a API decide se tenta de novo, desfaz o resto ou devolve erro ao cliente.

Com o message exchange, cada consumidor tem a sua fila, e a confirmação acontece fila a fila. Se o consumidor de nota fiscal falhar, a cópia dele volta para a fila order-created-invoice para nova entrega; as cópias do pagamento e do estoque estão em outras filas e seguem normalmente (message exchanges, ACK e NACK). A API de pedidos já recebeu a confirmação do Publish e não fica sabendo da falha.

Cuidados com o message exchange e as filas

• Idempotência em cada consumidor. O livro fala em uma entrega por assinante, mas as filas standard do Anypoint MQ garantem entrega “pelo menos uma vez” e, de vez em quando, entregam a mesma mensagem mais de uma vez (filas). Cada consumidor trata o duplicado do seu lado: cobrar duas vezes e reservar o estoque duas vezes são erros diferentes, em sistemas diferentes. O idempotent receiver mostra como.
• A DLQ é da fila. A dead letter queue é configurada na fila, não no message exchange: cada fila ligada pode ter a sua, como a order-created-invoice-dlq da fila de nota fiscal, com o seu número de tentativas de entrega antes do desvio, 10 por padrão. Sem DLQ atribuída, a mensagem volta à fila até o TTL (filas). O dead letter channel mostra essa configuração.
• O Publish não confirma a entrega nas filas. Um Publish que deu certo confirma só que o Anypoint MQ aceitou a mensagem para o message exchange. Se nenhuma fila recebe a mensagem, as configurações de reentrega e de DLQ das filas não valem, porque a mensagem nunca entrou numa fila (message exchanges).
• Regras de roteamento mudam o padrão. Com uma regra num binding, a fila recebe só as mensagens cujas propriedades ou headers casam com a regra; os bindings sem regra continuam recebendo tudo. Se todos os bindings tiverem regra e a mensagem não casar com nenhuma, ela não vai para fila nenhuma (regras de roteamento). O canal deixa de entregar tudo a todos e passa a escolher os destinos pelas propriedades da mensagem: vira roteamento seletivo.

Quando uma fila ou uma chamada basta

• Um único consumidor. Com um interessado só, uma fila basta. Um message exchange com uma fila ligada não distribui nada e só cria outro lugar onde procurar a mensagem que sumiu. Se um segundo consumidor aparecer, dá para criar o message exchange depois, ligar a fila que já existe a ele (message exchanges) e trocar o destination do Publish.
• Quem publica precisa de uma resposta. O message exchange só confirma que aceitou a mensagem. Se a API de pedidos precisa responder ao cliente com o resultado da cobrança, publicar um evento não traz essa resposta; uma chamada síncrona ao pagamento resolve.

Os outros padrões do cenário de pedidos, e o componente do Mule que implementa cada um, estão no catálogo de padrões de integração no Mule.