Spring Boot – Messaging with JMS

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-artemis</artifactId>
</dependency>

docker run -p 61616:61616 -e AMQ_USER=admin -e AMQ_PASSWORD=admin \
  apache/activemq-artemis:latest-alpine

/**
 * <p>{@code @Profile("messaging")} is doing real work here rather than decorating. A
 * {@code @JmsListener} container starts polling as soon as the context is up, so merely having the
 * Artemis starter on the classpath with a listener declared would make the application try to reach
 * a broker at every startup — and fill the log with connection retries on any machine that has not
 * got one. Gating the listener means the default experience is unchanged: MySQL, and nothing else.
 */
@Configuration
@EnableJms
@Profile("messaging")
public class MessagingConfig { }

/** The queue orders are announced on. A constant, so producer and consumer cannot disagree. */
public static final String ORDER_QUEUE = "pizza.orders";

/**
 * The dead-letter destination.
 *
 * <p>Without somewhere for poison messages to go, a message that always fails is redelivered
 * forever: it blocks the queue, burns CPU and fills the log, and the underlying bug is hidden
 * behind the noise. Artemis will move a message here after its redelivery attempts are
 * exhausted.
 */
public static final String ORDER_DLQ = "DLQ.pizza.orders";

/**
 * <p>Out of the box, JMS moves a {@code SimpleMessageConverter} payload: String, byte[],
 * Serializable, Map. Sending a record such as {@link OrderMessage} through that either fails or
 * drags Java serialisation into your wire format, which then couples both ends of the queue to your
 * class files. JSON keeps the contract readable and lets the consumer be written in something other
 * than Java.
 */
@Bean
public MessageConverter jacksonJmsMessageConverter() {
    JacksonJsonMessageConverter converter = new JacksonJsonMessageConverter();
    converter.setTargetType(MessageType.TEXT);
    // The consumer needs to know which class to deserialise into. This names the JMS string
    // property carrying that type id — without it, the receiving side gets a LinkedHashMap.
    converter.setTypeIdPropertyName("_type");
    return converter;
}

/**
 * <p>Taking Boot's {@code Configurer} first and then overriding is the pattern to copy: it
 * keeps every sensible default Boot computed (including the converter above) instead of
 * silently discarding them, which is what building a bare factory by hand does.
 */
@Bean
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(
        ConnectionFactory connectionFactory,
        DefaultJmsListenerContainerFactoryConfigurer configurer) {

    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    configurer.configure(factory, connectionFactory);

    // Transacted sessions: the message is only acknowledged once the listener returns
    // normally. Throw, and it goes back on the queue for redelivery — which is what makes the
    // dead-letter queue above meaningful.
    factory.setSessionTransacted(true);
    factory.setConcurrency("1-3");
    return factory;
}

/**
 * <p>A separate type from {@link OrderPlacedEvent} on purpose. The in-process event can change
 * whenever both sides are recompiled together; this one is a <b>wire contract</b> that some other
 * service — possibly not even a Java one — deserialises. Letting an internal refactor rename a
 * field on a published message is how you break a consumer you have never met.
 */
public record OrderMessage(
        UUID orderId, String customerEmail, BigDecimal total, String orderType) {

    public static OrderMessage from(OrderPlacedEvent event) {
        return new OrderMessage(
                event.orderPublicId(), event.contactEmail(), event.total(),
                event.orderType().name());
    }
}

public void publish(OrderMessage message) {
    JmsTemplate jms = jmsTemplateProvider.getIfAvailable();
    if (jms == null) {
        log.debug("Messaging is off — not publishing order {}", message.orderId());
        return;
    }

    try {
        // convertAndSend runs the payload through the MessageConverter from MessagingConfig,
        // so what actually goes on the wire is JSON plus a _type property.
        jms.convertAndSend(MessagingConfig.ORDER_QUEUE, message);
    } catch (Exception ex) {
        // The order is already committed and paid for. A broker outage must not turn that into
        // a customer-visible failure, so this is logged and swallowed — the same call the
        // confirmation email makes.
        log.error("Could not publish order {} — the order is unaffected", message.orderId(), ex);
    }
}

/**
 * <p>AFTER_COMMIT for the same reason the email is: publishing inside the transaction would let
 * a consumer read the order from the database before the insert is visible, and act on an order
 * that a rollback is about to erase. Message brokers have no idea your transaction exists.
 */
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void publishToQueue(OrderPlacedEvent event) {
    messagePublisher.publish(OrderMessage.from(event));
}

/**
 * ⚠️ <b>A listener must be idempotent.</b> JMS guarantees at-least-once delivery, not
 * exactly-once: a broker restart or a slow acknowledgement can hand you the same message twice.
 * Anything with an external effect — charging a card, sending an email — has to be safe to repeat,
 * usually by recording the message id and ignoring one that has already been seen.
 */
@JmsListener(destination = MessagingConfig.ORDER_QUEUE,
             containerFactory = "jmsListenerContainerFactory")
public void onOrderPlaced(@Payload OrderMessage message) {
    log.info("Kitchen received order {} for {} ({} {})",
            message.orderId(), message.customerEmail(), message.total(), message.orderType());

    // Throwing here would roll the transacted session back and the broker would redeliver.
    // After the configured redelivery attempts, Artemis moves the message to
    // MessagingConfig.ORDER_DLQ rather than looping on it forever.
}

spring.artemis.broker-url=tcp://localhost:61616
spring.jms.listener.session.acknowledge-mode=client
spring.jms.listener.min-concurrency=1
spring.jms.listener.max-concurrency=3