Publish-Subscribe Channel in Mule: one event for several consumers with Anypoint MQ

When more than one system cares about an event, the Publish-Subscribe Channel delivers a copy to each interested party without the publisher having to know them. This page shows the pattern in the book Enterprise Integration Patterns and in Mule 4, with an Anypoint MQ message exchange (the broker resource, not the Anypoint Exchange where assets live), in the series’ order scenario. Because the message exchange exists only on the broker, part of the pattern is Anypoint MQ configuration, outside the application’s XML.

A created order that three systems need to receive

The orders API receives an order, saves it and has to notify three systems: payment charges it, inventory reserves the items and the invoice gets issued. The book describes the same problem in general terms: an application uses messages to announce events and needs each event to reach every interested receiver (Publish-Subscribe Channel).

The scenario adds a constraint: the orders API should not know who the interested parties are, nor change when a fourth one shows up.

In the book, each subscriber gets its copy once

The drawing below redraws the book’s diagram and answers one question: what happens to an event published on a publish-subscribe channel?

The book’s solution is to publish the event on a Publish-Subscribe Channel, which delivers a copy of it to each receiver. The channel has one input that splits into several outputs, one per subscriber, and puts a copy of the event on each output. Each output has a single subscriber, which can consume that message only once; so each subscriber gets the event once, and the consumed copy disappears from its channel (Publish-Subscribe Channel).

For a subscriber that is down, the book has another pattern, the Durable Subscriber: the messaging system keeps what was published and delivers it when the subscriber comes back (Durable Subscriber).

In Anypoint MQ, the channel is a message exchange with bound queues

In Anypoint MQ, the message exchange plays the channel’s role: it is a broker destination, like a queue. You bind queues to a message exchange and, by default, it sends each message to every bound queue (message exchanges). Each link between the message exchange and a queue is called a binding (Anypoint MQ glossary).

A message exchange accepts only the Publish operation, never Consume: Anypoint MQ forwards the message to the bound queues, and the application consumes from the queue. An attempt to consume directly from the message exchange gets the error The destination should be a queue. from the Anypoint MQ REST API. The same page lists the limits: a message sent to a message exchange can be at most 10 MB, and a message exchange accepts up to 450 bound queues, or up to 25 when there are routing rules (message exchanges).

Compared with the book, each bound queue plays the role of a subscriber’s output, and it also covers the Durable Subscriber case. The copy stays on the queue until it is consumed, within the time to live (TTL) configured on the queue, which can be up to two weeks (queues).

The order-created event on the Anypoint MQ message exchange

The diagram answers where each piece lives: what belongs to the Mule application and what is broker configuration.

The orders API publishes the event to the order-created message exchange. Three bindings connect three queues to it, one per consumer. Each consumer is a flow with a Subscriber on its own queue, in a Mule application that can be separate from the others. The API knows only the message exchange’s name; each consumer knows only its own queue.

Bindings live in Anypoint MQ, not in the application

No XML in the application says which queues receive the event. The queues are created first, on the Anypoint MQ Destinations page. Then, when you create the message exchange or in its settings, you select the queues you want to bind (message exchanges), and Anypoint MQ creates and maintains the bindings from that choice (Anypoint MQ glossary). That is why a fourth interested party gets a queue and a binding, and the orders API stays the same. Because Anypoint MQ forwards to the bound queues the messages published to the message exchange, a queue bound later receives only what is published after the binding, not earlier events (message exchanges).

The connection and the publish in the orders API

The three blocks in this section and the next are reference configuration, with elements and attributes taken from the Anypoint MQ Connector documentation. They do not run in the mulesoft-patterns-examples repository: they need an Anypoint MQ broker, and while the connector is free, connecting to the service requires a license (Anypoint MQ Connector).

The global configuration points to the region’s URL and uses the ID and secret of an Anypoint MQ client app (connector reference). The secret comes from a secure property: the ${secure::…} prefix works only with the Secure Properties module and a secure-properties:config in the application.

<!-- Reference configuration: does not run in the examples repository (needs an Anypoint MQ broker) -->
<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>

After saving the order, the API’s flow publishes the event. To the connector, the message exchange is a destination like a queue: the Publish operation takes its name in destination, and the message body can be built with DataWeave in anypoint-mq:body (Publish).

<!-- Reference configuration: does not run in the examples repository (needs an Anypoint MQ broker) -->
<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>

One consumer per queue, with automatic acknowledgment

The payment consumer listens on its queue with the Subscriber. The inventory and invoice consumers have the same shape, each on its own queue.

<!-- Reference configuration: does not run in the examples repository (needs an Anypoint MQ broker) -->
<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 is already the Subscriber’s default and is written here only to make it visible (connector reference). In this mode, the connector acknowledges the message (ACK) only if the flow finishes without an error. If the flow ends with a propagated error, the message is not acknowledged (NACK) and goes back to the queue to be delivered again (Subscriber). IMMEDIATE mode acknowledges before processing: the message leaves the queue and does not come back if processing fails. In MANUAL, the flow itself calls the ACK or NACK operations with the message’s ackToken (ACK and NACK).

What changes compared with one queue or three HTTP calls

A single queue does not solve this scenario. On a Point-to-Point Channel, only one receiver consumes each message; if three consumers read the same queue, they compete for the messages, and each order reaches only one of them (Point-to-Point Channel). That is useful for spreading load across instances of the same consumer (Competing Consumers), not for notifying three different systems.

Three HTTP calls handle delivery, but the orders API would then know the address of all three systems, and every new interested party would mean a change to it. Failure also stays with the caller. If invoicing is down, the call fails immediately, and the API decides whether to retry, undo the rest or return an error to the client.

With the message exchange, each consumer has its own queue, and acknowledgment happens queue by queue. If the invoice consumer fails, its copy goes back to the order-created-invoice queue for redelivery; the payment and inventory copies are on other queues and carry on normally (message exchanges, ACK and NACK). The orders API has already received the Publish confirmation and never learns about the failure.

Things to watch with the message exchange and the queues

• Idempotency in each consumer. The book talks about one delivery per subscriber, but Anypoint MQ standard queues guarantee “at least once” delivery and occasionally deliver the same message more than once (queues). Each consumer handles duplicates on its own side: charging twice and reserving stock twice are different errors, in different systems. The idempotent receiver page shows how.
• The DLQ belongs to the queue. The dead letter queue is configured on the queue, not on the message exchange: each bound queue can have its own, such as order-created-invoice-dlq for the invoice queue, with its own number of delivery attempts before rerouting, 10 by default. With no DLQ assigned, the message keeps going back to the queue until its TTL runs out (queues). The dead letter channel page shows this configuration.
• Publish does not confirm delivery to the queues. A successful Publish confirms only that Anypoint MQ accepted the message for the message exchange. If no queue receives the message, the queues’ redelivery and DLQ settings do not apply, because the message never entered a queue (message exchanges).
• Routing rules change the pattern. With a rule on a binding, the queue receives only the messages whose properties or headers match the rule; bindings without a rule still receive everything. If every binding has a rule and the message matches none, it goes to no queue at all (routing rules). The channel stops delivering everything to everyone and starts choosing destinations by the message’s properties: it becomes selective routing.

When a queue or a call is enough

• A single consumer. With only one interested party, a queue is enough. A message exchange with one bound queue distributes nothing and just adds another place to look for the message that went missing. If a second consumer shows up, you can create the message exchange later, bind the existing queue to it (message exchanges) and change the Publish destination.
• The publisher needs a reply. The message exchange only confirms that it accepted the message. If the orders API has to answer the client with the result of the charge, publishing an event does not bring that answer back; a synchronous call to payment does.

The other patterns in the order scenario, and the Mule component that implements each one, are in the catalog of integration patterns in Mule.