JMS v5.2.1-SNAPSHOT

On this page

JMS

1 Introduction

This project includes integration between Micronaut and Java Messaging Service (JMS).

Supported implementations include Active MQ "Classic", Active MQ Artemis, and AWS SQS, and configuring other JMS providers is straightforward.

2 Release History

For this project, you can find a list of releases (with release notes) here:

https://github.com/micronaut-projects/micronaut-jms/releases

3 Breaking Changes

5.0.0

The way of configuring unsupported brokers has changed. See the Customizing Brokers section.

4.0.0

Removal of `javax.jms` namespace

Version 4 switches from the deprecated javax.jms namespace to the new Jakarta EE jakarta.jms namespace.

ActiveMQ Classic

The ActiveMQ Classic client has been updated to version 6 to support the new namespace.

ActiveMQ Artemis

The client dependency for ActiveMq Artemis has changed from org.apache.activemq:artemis-jms-client to org.apache.activemq:artemis-jakarta-client which uses the new namespace.

Deprecations

The annotation io.micronaut.jms.annotations.Queue removes the following attributes that were deprecated previously. This is to align the implementation with the JMS model and the messaging libraries' presumptions.
- String concurrency() default "1-1"
- String executor() default ""
The annotation io.micronaut.jms.annotations.Topic removes the following attributes that were deprecated previously. This is to align the implementation with the JMS model and the messaging libraries' presumptions.
- String executor() default ""
The following classes in the io.micronaut.jms.listener package were deprecated previously are now removed.
- ConcurrentMessageHandler
- JMSListenerContainer
- JMSListenerContainerFactory

4 JMS Quick Start

To add support for JMS to a project, add a dependency to your build, depending on the JMS implementation(s) you require:

implementation("io.micronaut.jms:micronaut-jms-activemq-classic")

<dependency>
    <groupId>io.micronaut.jms</groupId>
    <artifactId>micronaut-jms-activemq-classic</artifactId>
</dependency>

implementation("io.micronaut.jms:micronaut-jms-activemq-artemis")

<dependency>
    <groupId>io.micronaut.jms</groupId>
    <artifactId>micronaut-jms-activemq-artemis</artifactId>
</dependency>

implementation("io.micronaut.jms:micronaut-jms-sqs")

<dependency>
    <groupId>io.micronaut.jms</groupId>
    <artifactId>micronaut-jms-sqs</artifactId>
</dependency>

Note that you can use multiple providers in a project.

Creating a JMS Producer with @JMSProducer

To create a JMS client that sends messages, define an interface annotated with JMSProducer, for example:

Note	In Python the producer is an abstract class (`ABC`) whose `@abstractmethod` methods have `…` bodies; the parameter binding annotations are applied with `typing.Annotated`.

imports

import io.micronaut.jms.annotations.JMSProducer;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.messaging.annotation.MessageBody;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME) // 
public interface TextProducer {

    @Queue("queue_text") // 
    void send(@MessageBody String body); // 
}

imports

from abc import ABC, abstractmethod
from typing import Annotated

from micronaut.jms.annotations import JMSProducer, Queue
from micronaut.messaging.annotation import MessageBody

@JMSProducer("activeMqConnectionFactory")  # 
class TextProducer(ABC):

    @Queue("queue_text")  # 
    @abstractmethod
    def send(self, body: Annotated[str, MessageBody]) -> None:  # 
        ...

imports

import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME) // 
interface TextProducer {

    @Queue("queue_text") // 
    fun send(@MessageBody body: String) // 
}

imports

import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME) // 
interface TextProducer {

    @Queue('queue_text') // 
    void send(@MessageBody String body) // 
}

The JMSProducer annotation designates this interface as a client.
The @Queue annotation indicates which queue the message should be published to.
The send method accepts a single parameter which is the payload of the message.

At compile time Micronaut will produce an implementation of the interface. You can retrieve an instance of TextProducer either by looking up the bean from the ApplicationContext or with dependency injection:

TextProducer textProducer = applicationContext.getBean(TextProducer.class);
textProducer.send("quickstart");

text_producer = self.application_context.getBean(TextProducer)
text_producer.send("quickstart")

val textProducer = applicationContext.getBean(TextProducer::class.java)
textProducer.send("quickstart")

def textProducer = applicationContext.getBean(TextProducer)
textProducer.send 'quickstart'

Creating a JMS Consumer with @JMSListener

To listen to JMS messages you can use the @JMSListener annotation to define a message listener.

The following example will listen for messages published by the TextProducer in the previous section:

imports

import io.micronaut.jms.annotations.JMSListener;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.messaging.annotation.MessageBody;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSListener(CONNECTION_FACTORY_BEAN_NAME) // 
public class TextConsumer {

    List<String> messages = Collections.synchronizedList(new ArrayList<>());

    @Queue(value = "queue_text") // 
    public void receive(@MessageBody String body) { // 
        messages.add(body);
    }
}

imports

from typing import Annotated

from micronaut.jms.annotations import JMSListener, Queue
from micronaut.messaging.annotation import MessageBody

@JMSListener("activeMqConnectionFactory")  # 
class TextConsumer:

    def __init__(self) -> None:
        self.messages: list[str] = []

    @Queue("queue_text")  # 
    def receive(self, body: Annotated[str, MessageBody]) -> None:  # 
        self.messages.append(body)

imports

import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import java.util.*

@JMSListener(CONNECTION_FACTORY_BEAN_NAME) // 
class TextConsumer {

    val messages: MutableList<String> = Collections.synchronizedList(ArrayList())

    @Queue(value = "queue_text") // 
    fun receive(@MessageBody body: String) { // 
        messages.add(body)
    }
}

imports

import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSListener(CONNECTION_FACTORY_BEAN_NAME) // 
class TextConsumer {

    List<String> messages = [].asSynchronized()

    @Queue(value = 'queue_text') // 
    void receive(@MessageBody String body) { // 
        messages << body
    }
}

The @JMSListener is used to designate the bean as a message listener.
The @Queue annotation is used to indicate which queue to subscribe to.
The receive method accepts a single parameter which is the payload of the message.

5 Configuration

5.1 Configuring ActiveMQ "Classic"

To use ActiveMQ "Classic", add the following properties to your application.yml

micronaut.jms.activemq.classic.enabled=true
micronaut.jms.activemq.classic.connection-string=tcp://host:port

micronaut:
  jms:
    activemq:
      classic:
        enabled: true
        connection-string: 'tcp://host:port'

[micronaut.jms.activemq.classic]
enabled = true
connection-string = "tcp://host:port"

micronaut {
  jms {
    activemq {
      classic {
        enabled = true
        connectionString = 'tcp://host:port'
      }
    }
  }
}

{
  micronaut = {
    jms = {
      activemq = {
        classic = {
          enabled = true
          connection-string = "tcp://host:port"
        }
      }
    }
  }
}

{
  "micronaut": {
    "jms": {
      "activemq": {
        "classic": {
          "enabled": true,
          "connection-string": "tcp://host:port"
        }
      }
    }
  }
}

and the following to your build.gradle or pom.xml

implementation("io.micronaut.jms:micronaut-jms-activemq-classic")

<dependency>
    <groupId>io.micronaut.jms</groupId>
    <artifactId>micronaut-jms-activemq-classic</artifactId>
</dependency>

The JMS ConnectionFactory will be an instance of ActiveMQConnectionFactory. Only the broker URL is configurable in application.yml but further customization is possible with a BeanCreatedEventListener.

5.2 Configuring ActiveMQ Artemis

To use ActiveMQ Artemis, add the following properties to your application.yml

micronaut.jms.activemq.artemis.enabled=true
micronaut.jms.activemq.artemis.connection-string=tcp://host:port

micronaut:
  jms:
    activemq:
      artemis:
        enabled: true
        connection-string: 'tcp://host:port'

[micronaut.jms.activemq.artemis]
enabled = true
connection-string = "tcp://host:port"

micronaut {
  jms {
    activemq {
      artemis {
        enabled = true
        connectionString = 'tcp://host:port'
      }
    }
  }
}

{
  micronaut = {
    jms = {
      activemq = {
        artemis = {
          enabled = true
          connection-string = "tcp://host:port"
        }
      }
    }
  }
}

{
  "micronaut": {
    "jms": {
      "activemq": {
        "artemis": {
          "enabled": true,
          "connection-string": "tcp://host:port"
        }
      }
    }
  }
}

and the following to your build.gradle or pom.xml

implementation("io.micronaut.jms:micronaut-jms-activemq-artemis")

<dependency>
    <groupId>io.micronaut.jms</groupId>
    <artifactId>micronaut-jms-activemq-artemis</artifactId>
</dependency>

The JMS ConnectionFactory will be an instance of ActiveMQJMSConnectionFactory. Only the broker URL is configurable in application.yml but further customization is possible with a BeanCreatedEventListener.

5.3 Configuring SQS

To use Amazon SQS, add the following property to your application.yml

micronaut.jms.sqs.enabled=true

micronaut:
  jms:
    sqs:
      enabled: true

[micronaut.jms.sqs]
enabled = true

micronaut {
  jms {
    sqs {
      enabled = true
    }
  }
}

{
  micronaut = {
    jms = {
      sqs = {
        enabled = true
      }
    }
  }
}

{
  "micronaut": {
    "jms": {
      "sqs": {
        "enabled": true
      }
    }
  }
}

and the following to your build.gradle or pom.xml

implementation("io.micronaut.jms:micronaut-jms-sqs")

<dependency>
    <groupId>io.micronaut.jms</groupId>
    <artifactId>micronaut-jms-sqs</artifactId>
</dependency>

See the Micronaut AWS docs for more information.

Note that it’s not necessary to add a dependency on io.micronaut.aws:micronaut-aws-sdk-v2 as it’s already a transitive dependency of io.micronaut.jms:micronaut-jms-sqs.

The JMS ConnectionFactory will be an instance of com.amazon.sqs.javamessaging.SQSConnectionFactory; you can customize that with a BeanCreatedEventListener.

Tip	See the guide for Connect a Micronaut JMS Application to an AWS SQS Queue to learn more.

5.4 Configuring Unsupported Brokers

Using an unsupported JMS broker is simple; you simply need to instantiate a ConnectionFactory and JMSConnectionPool in a Bean factory. See the Replacing the ConnectionFactory section below for an example.

With this in place you can configure consumers and producers that use the connection factory by referencing the bean name ("myConnectionFactory" in the example) in @JMSProducer and @JMSListener annotations on your producer interfaces and consumer classes.

5.5 Customizing Brokers

Although only the broker URL is configurable in application.yml, there are potentially many configurable properties in the instantiated ConnectionFactory. Also, you may wish to use a different ConnectionFactory implementation class. Any customization you need can be done by intercepting the creation of the ConnectionFactory with a BeanCreatedEventListener.

Customizing the ConnectionFactory

Here we intercept the creation of the ActiveMQ "Classic" ConnectionFactory by overriding the default to enable asynchronous message sending:

imports

import io.micronaut.context.event.BeanCreatedEvent;
import io.micronaut.context.event.BeanCreatedEventListener;
import jakarta.inject.Singleton;
import org.apache.activemq.ActiveMQConnectionFactory;

import jakarta.jms.ConnectionFactory;

@Singleton
public class CustomizeBrokerJMSConnectionPoolListener implements BeanCreatedEventListener<ConnectionFactory> {

    @Override
    public ConnectionFactory onCreated(BeanCreatedEvent<ConnectionFactory> event) {

        ConnectionFactory connectionFactory = event.getBean();
        if (connectionFactory instanceof ActiveMQConnectionFactory) {
            ActiveMQConnectionFactory amqcf = (ActiveMQConnectionFactory) connectionFactory;
            amqcf.setUseAsyncSend(true);
        }

        return connectionFactory;
    }
}

imports

from jakarta.inject import Singleton
from jakarta.jms import ConnectionFactory
from micronaut.context.event import BeanCreatedEvent, BeanCreatedEventListener
from org.apache.activemq import ActiveMQConnectionFactory

@Singleton
class CustomizeBrokerJMSConnectionPoolListener(BeanCreatedEventListener[ConnectionFactory]):

    def onCreated(self, event: BeanCreatedEvent[ConnectionFactory]) -> ConnectionFactory:
        connection_factory = event.getBean()
        if isinstance(connection_factory, ActiveMQConnectionFactory):
            connection_factory.setUseAsyncSend(True)
        return connection_factory

imports

import io.micronaut.context.event.BeanCreatedEvent
import io.micronaut.context.event.BeanCreatedEventListener
import jakarta.inject.Singleton
import org.apache.activemq.ActiveMQConnectionFactory
import jakarta.jms.ConnectionFactory

@Singleton
class CustomizeBrokerJMSConnectionPoolListener : BeanCreatedEventListener<ConnectionFactory> {

    override fun onCreated(event: BeanCreatedEvent<ConnectionFactory>): ConnectionFactory {
        val connectionFactory = event.bean
        if (connectionFactory is ActiveMQConnectionFactory) {
            connectionFactory.isUseAsyncSend = true
        }
        return connectionFactory
    }
}

imports

import groovy.transform.CompileStatic
import io.micronaut.context.event.BeanCreatedEvent
import io.micronaut.context.event.BeanCreatedEventListener
import jakarta.inject.Singleton
import org.apache.activemq.ActiveMQConnectionFactory

import jakarta.jms.ConnectionFactory

@CompileStatic
@Singleton
class CustomizeBrokerJMSConnectionPoolListener implements BeanCreatedEventListener<ConnectionFactory> {

    @Override
    ConnectionFactory onCreated(BeanCreatedEvent<ConnectionFactory> event) {

        ConnectionFactory connectionFactory = event.bean
        if (connectionFactory instanceof ActiveMQConnectionFactory) {
            connectionFactory.useAsyncSend = true
        }

        return connectionFactory
    }
}

Replacing the ConnectionFactory

Here we discard the default instance and replace it with a custom implementation, in this case an instance of XAConnectionFactory for distributed transaction support.

imports

import io.micronaut.context.annotation.Factory;
import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.activemq.classic.configuration.properties.ActiveMqClassicConfigurationProperties;
import io.micronaut.jms.annotations.JMSConnectionFactory;
import jakarta.inject.Singleton;
import org.apache.activemq.ActiveMQXAConnectionFactory;

@Factory
class ActiveMQXAConnectionBeanFactory {
    @JMSConnectionFactory("activeMQXAConnectionFactory")
    @Singleton
    ActiveMQXAConnectionFactory createActiveMQXAConnectionFactory(ActiveMqClassicConfigurationProperties amqConfig) {
        return new ActiveMQXAConnectionFactory(amqConfig.getConnectionString());
    }
}

imports

from jakarta.inject import Singleton
from micronaut.context.annotation import Factory, Requires
from micronaut.jms.activemq.classic.configuration.properties import ActiveMqClassicConfigurationProperties
from micronaut.jms.annotations import JMSConnectionFactory
from org.apache.activemq import ActiveMQXAConnectionFactory

@Factory
class ActiveMQXAConnectionBeanFactory:

    @JMSConnectionFactory("activeMQXAConnectionFactory")
    @Singleton
    def create_active_mq_xa_connection_factory(self, amq_config: ActiveMqClassicConfigurationProperties) -> ActiveMQXAConnectionFactory:
        return ActiveMQXAConnectionFactory(amq_config.getConnectionString())

imports

import io.micronaut.context.annotation.Factory
import io.micronaut.jms.activemq.classic.configuration.properties.ActiveMqClassicConfigurationProperties
import io.micronaut.jms.annotations.JMSConnectionFactory
import jakarta.inject.Singleton
import org.apache.activemq.ActiveMQXAConnectionFactory

@Factory
internal class ActiveMQXAConnectionBeanFactory {
    @JMSConnectionFactory("activeMQXAConnectionFactory")
    @Singleton
    fun createActiveMQXAConnectionFactory(amqConfig: ActiveMqClassicConfigurationProperties): ActiveMQXAConnectionFactory {
        return ActiveMQXAConnectionFactory(amqConfig.getConnectionString())
    }
}

imports

import groovy.transform.CompileStatic
import io.micronaut.jms.annotations.JMSConnectionFactory
import jakarta.inject.Singleton
import io.micronaut.context.annotation.Factory
import io.micronaut.context.annotation.Requires
import io.micronaut.jms.activemq.classic.configuration.properties.ActiveMqClassicConfigurationProperties
import org.apache.activemq.ActiveMQXAConnectionFactory

@Factory
@CompileStatic
class ActiveMQXAConnectionBeanFactory {
    @JMSConnectionFactory("activeMQXAConnectionFactory")
    @Singleton
    ActiveMQXAConnectionFactory createActiveMQXAConnectionFactory(ActiveMqClassicConfigurationProperties amqConfig) {
        new ActiveMQXAConnectionFactory(amqConfig.getConnectionString())
    }
}

and, then, create a JMSConnectionPool in another factory:

imports

import io.micronaut.context.annotation.EachBean;
import io.micronaut.context.annotation.Factory;
import io.micronaut.context.annotation.Primary;
import io.micronaut.jms.configuration.properties.JMSConfigurationProperties;
import io.micronaut.jms.pool.JMSConnectionPool;
import jakarta.inject.Singleton;
import org.apache.activemq.ActiveMQXAConnectionFactory;

@Factory
class JmsConnectionPoolActiveMQXAConnectionFactory {
    private final JMSConfigurationProperties properties;
    JmsConnectionPoolActiveMQXAConnectionFactory(JMSConfigurationProperties properties) {
        this.properties = properties;
    }

    @EachBean(ActiveMQXAConnectionFactory.class)
    @Singleton
    @Primary
    JMSConnectionPool createJmsConnectionPool(ActiveMQXAConnectionFactory connectionFactory) {
        return new JMSConnectionPool(connectionFactory, properties.getInitialPoolSize(), properties.getMaxPoolSize());
    }
}

imports

from jakarta.inject import Singleton
from micronaut.context.annotation import EachBean, Factory, Primary
from micronaut.jms.configuration.properties import JMSConfigurationProperties
from micronaut.jms.pool import JMSConnectionPool
from org.apache.activemq import ActiveMQXAConnectionFactory

@Factory
class JmsConnectionPoolActiveMQXAConnectionFactory:

    def __init__(self, properties: JMSConfigurationProperties):
        self.properties = properties

    @EachBean(ActiveMQXAConnectionFactory)
    @Singleton
    @Primary
    def create_jms_connection_pool(self, connection_factory: ActiveMQXAConnectionFactory) -> JMSConnectionPool:
        return JMSConnectionPool(connection_factory, self.properties.getInitialPoolSize(), self.properties.getMaxPoolSize())

imports

import io.micronaut.context.annotation.EachBean
import io.micronaut.context.annotation.Factory
import io.micronaut.context.annotation.Primary
import io.micronaut.jms.configuration.properties.JMSConfigurationProperties
import io.micronaut.jms.pool.JMSConnectionPool
import jakarta.inject.Singleton
import org.apache.activemq.ActiveMQXAConnectionFactory

@Factory
internal class JmsConnectionPoolActiveMQXAConnectionFactory(val properties: JMSConfigurationProperties) {
    @EachBean(ActiveMQXAConnectionFactory::class)
    @Singleton
    @Primary
    fun createJmsConnectionPool(connectionFactory: ActiveMQXAConnectionFactory): JMSConnectionPool {
        return JMSConnectionPool(connectionFactory, properties.getInitialPoolSize(), properties.getMaxPoolSize())
    }
}

imports

import io.micronaut.context.annotation.EachBean
import io.micronaut.context.annotation.Factory
import io.micronaut.context.annotation.Primary
import io.micronaut.core.annotation.Internal
import io.micronaut.jms.configuration.properties.JMSConfigurationProperties
import io.micronaut.jms.pool.JMSConnectionPool
import jakarta.inject.Singleton
import org.apache.activemq.ActiveMQXAConnectionFactory

@Factory
@Internal
class JmsConnectionPoolActiveMQXAConnectionFactory {
    private final JMSConfigurationProperties properties

    /**
     * @param properties JMS Configuration
     */
    JmsConnectionPoolActiveMQXAConnectionFactory(JMSConfigurationProperties properties) {
        this.properties = properties
    }

    /**
     * Creates a {@link JMSConnectionPool} from each registered {@link ActiveMQXAConnectionFactory} in the context.
     * @param connectionFactory Connection Factory
     * @return JMS Connection Pool
     */
    @EachBean(ActiveMQXAConnectionFactory.class)
    @Singleton
    @Primary
    JMSConnectionPool createJmsConnectionPool(ActiveMQXAConnectionFactory connectionFactory) {
        return new JMSConnectionPool(connectionFactory, properties.getInitialPoolSize(), properties.getMaxPoolSize())
    }
}

6 Parameter Binding

Methods in classes/interfaces annotated with @JMSProducer and @JMSListener use parameter binding annotations to access the JMS Message, message body, and message headers.

@MessageBody

The MessageBody annotation is required for the method argument that will be serialized/deserialized as the JMS Message body.

@MessageHeader

The MessageHeader annotation is required for any method arguments that will be serialized/deserialized as JMS headers or non-JMS headers (message properties).

In this example, the message body is a Map and the method supports sending the JMSCorrelationID along with several custom header values:

imports

import org.jspecify.annotations.Nullable;
import io.micronaut.jms.annotations.JMSProducer;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.messaging.annotation.MessageBody;
import io.micronaut.messaging.annotation.MessageHeader;
import java.io.Serializable;
import java.util.Map;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;
import static io.micronaut.jms.model.JMSHeaders.JMS_CORRELATION_ID;

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
public interface MapProducer {

    @Queue("queue_map")
    void send(@MessageBody Map<String, Serializable> body,
              @MessageHeader(JMS_CORRELATION_ID) @Nullable String correlationId,
              @MessageHeader("CustomStringHeader") @Nullable String stringHeader,
              @MessageHeader("CustomBooleanHeader") boolean booleanHeader,
              @MessageHeader("CustomByteHeader") byte byteHeader,
              @MessageHeader("CustomShortHeader") short shortHeader,
              @MessageHeader("CustomIntegerHeader") int intHeader,
              @MessageHeader("CustomLongHeader") long longHeader,
              @MessageHeader("CustomFloatHeader") float floatHeader,
              @MessageHeader("CustomDoubleHeader") double doubleHeader);
}

imports

from abc import ABC, abstractmethod
from typing import Annotated

from java.lang import Byte, Float, Long, Short
from micronaut.jms.annotations import JMSProducer, Queue
from micronaut.messaging.annotation import MessageBody, MessageHeader

@JMSProducer("activeMqConnectionFactory")
class MapProducer(ABC):

    @Queue("queue_map")
    @abstractmethod
    def send(self, body: Annotated[dict[str, object], MessageBody],
             correlation_id: Annotated[str | None, MessageHeader("JMSCorrelationID")],
             string_header: Annotated[str | None, MessageHeader("CustomStringHeader")],
             boolean_header: Annotated[bool, MessageHeader("CustomBooleanHeader")],
             byte_header: Annotated[Byte, MessageHeader("CustomByteHeader")],
             short_header: Annotated[Short, MessageHeader("CustomShortHeader")],
             int_header: Annotated[int, MessageHeader("CustomIntegerHeader")],
             long_header: Annotated[Long, MessageHeader("CustomLongHeader")],
             float_header: Annotated[Float, MessageHeader("CustomFloatHeader")],
             double_header: Annotated[float, MessageHeader("CustomDoubleHeader")]) -> None:
        ...

imports

import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import io.micronaut.messaging.annotation.MessageHeader
import io.micronaut.jms.model.JMSHeaders.JMS_CORRELATION_ID
import java.io.Serializable

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
interface MapProducer {
    @Queue("queue_map")
    fun send(
        @MessageBody body: Map<String, Serializable?>,
        @MessageHeader(JMS_CORRELATION_ID) correlationId: String?,
        @MessageHeader("CustomStringHeader") stringHeader: String?,
        @MessageHeader("CustomBooleanHeader") booleanHeader: Boolean,
        @MessageHeader("CustomByteHeader") byteHeader: Byte,
        @MessageHeader("CustomShortHeader") shortHeader: Short,
        @MessageHeader("CustomIntegerHeader") intHeader: Int,
        @MessageHeader("CustomLongHeader") longHeader: Long,
        @MessageHeader("CustomFloatHeader") floatHeader: Float,
        @MessageHeader("CustomDoubleHeader") doubleHeader: Double
    )
}

imports

import org.jspecify.annotations.Nullable
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import io.micronaut.messaging.annotation.MessageHeader

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import static io.micronaut.jms.model.JMSHeaders.JMS_CORRELATION_ID

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
interface MapProducer {

    @Queue('queue_map')
    void send(@MessageBody Map<String, Serializable> body,
              @MessageHeader(JMS_CORRELATION_ID) @Nullable String correlationId,
              @MessageHeader('CustomStringHeader') @Nullable String stringHeader,
              @MessageHeader('CustomBooleanHeader') boolean booleanHeader,
              @MessageHeader('CustomByteHeader') byte byteHeader,
              @MessageHeader('CustomShortHeader') short shortHeader,
              @MessageHeader('CustomIntegerHeader') int intHeader,
              @MessageHeader('CustomLongHeader') long longHeader,
              @MessageHeader('CustomFloatHeader') float floatHeader,
              @MessageHeader('CustomDoubleHeader') double doubleHeader)
}

This consumer can access the user-supplied header values as well as several headers that can only be set by the JMS provider, e.g. JMSMessageID, JMSPriority, etc.:

imports

import org.jspecify.annotations.Nullable;
import io.micronaut.jms.annotations.JMSListener;
import io.micronaut.jms.annotations.Message;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.messaging.annotation.MessageBody;
import io.micronaut.messaging.annotation.MessageHeader;

import jakarta.jms.Destination;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;
import static io.micronaut.jms.model.JMSHeaders.JMS_CORRELATION_ID;
import static io.micronaut.jms.model.JMSHeaders.JMS_DELIVERY_MODE;
import static io.micronaut.jms.model.JMSHeaders.JMS_DESTINATION;
import static io.micronaut.jms.model.JMSHeaders.JMS_EXPIRATION;
import static io.micronaut.jms.model.JMSHeaders.JMS_MESSAGE_ID;
import static io.micronaut.jms.model.JMSHeaders.JMS_PRIORITY;
import static io.micronaut.jms.model.JMSHeaders.JMS_REDELIVERED;
import static io.micronaut.jms.model.JMSHeaders.JMS_REPLY_TO;
import static io.micronaut.jms.model.JMSHeaders.JMS_TIMESTAMP;
import static io.micronaut.jms.model.JMSHeaders.JMS_TYPE;

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
public class MapConsumer {

    List<Map<String, Serializable>> messageBodies = Collections.synchronizedList(new ArrayList<>());
    List<Map<String, Object>> messageHeaders = Collections.synchronizedList(new ArrayList<>());
    List<jakarta.jms.Message> messages = Collections.synchronizedList(new ArrayList<>());

    @Queue(value = "queue_map")
    public void receive(@MessageBody Map<String, Serializable> body,
                        @Message jakarta.jms.Message message,
                        @MessageHeader(JMS_CORRELATION_ID) @Nullable String correlationId,
                        @MessageHeader(JMS_DELIVERY_MODE) int deliveryMode,
                        @MessageHeader(JMS_DESTINATION) Destination destination,
                        @MessageHeader(JMS_EXPIRATION) long expiration,
                        @MessageHeader(JMS_MESSAGE_ID) String messageId,
                        @MessageHeader(JMS_PRIORITY) int priority,
                        @MessageHeader(JMS_REDELIVERED) boolean redelivered,
                        @MessageHeader(JMS_REPLY_TO) @Nullable Destination replyTo,
                        @MessageHeader(JMS_TIMESTAMP) long timestamp,
                        @MessageHeader(JMS_TYPE) @Nullable String type,
                        @MessageHeader("CustomStringHeader") @Nullable String stringHeader,
                        @MessageHeader("CustomBooleanHeader") boolean booleanHeader,
                        @MessageHeader("CustomByteHeader") byte byteHeader,
                        @MessageHeader("CustomShortHeader") short shortHeader,
                        @MessageHeader("CustomIntegerHeader") int intHeader,
                        @MessageHeader("CustomLongHeader") long longHeader,
                        @MessageHeader("CustomFloatHeader") float floatHeader,
                        @MessageHeader("CustomDoubleHeader") double doubleHeader) {

        Map<String, Object> headerValues = new HashMap<>();
        headerValues.put(JMS_CORRELATION_ID, correlationId);
        headerValues.put(JMS_DELIVERY_MODE, deliveryMode);
        headerValues.put(JMS_DESTINATION, destination);
        headerValues.put(JMS_EXPIRATION, expiration);
        headerValues.put(JMS_MESSAGE_ID, messageId);
        headerValues.put(JMS_PRIORITY, priority);
        headerValues.put(JMS_REDELIVERED, redelivered);
        headerValues.put(JMS_REPLY_TO, replyTo);
        headerValues.put(JMS_TIMESTAMP, timestamp);
        headerValues.put(JMS_TYPE, type);
        headerValues.put("CustomStringHeader", stringHeader);
        headerValues.put("CustomBooleanHeader", booleanHeader);
        headerValues.put("CustomByteHeader", byteHeader);
        headerValues.put("CustomShortHeader", shortHeader);
        headerValues.put("CustomIntegerHeader", intHeader);
        headerValues.put("CustomLongHeader", longHeader);
        headerValues.put("CustomFloatHeader", floatHeader);
        headerValues.put("CustomDoubleHeader", doubleHeader);

        messageHeaders.add(headerValues);
        messageBodies.add(body);
        messages.add(message);
    }
}

imports

from typing import Annotated

from jakarta.jms import Destination
from jakarta.jms import Message as JmsMessage
from java.lang import Byte, Float, Long, Short
from micronaut.jms.annotations import JMSListener, Message, Queue
from micronaut.messaging.annotation import MessageBody, MessageHeader

@JMSListener("activeMqConnectionFactory")
class MapConsumer:

    def __init__(self) -> None:
        self.message_bodies: list[dict[str, object]] = []
        self.message_headers: list[dict[str, object]] = []
        self.messages: list[JmsMessage] = []

    @Queue("queue_map")
    def receive(self, body: Annotated[dict[str, object], MessageBody],
                message: Annotated[JmsMessage, Message],
                correlation_id: Annotated[str | None, MessageHeader("JMSCorrelationID")],
                delivery_mode: Annotated[int, MessageHeader("JMSDeliveryMode")],
                destination: Annotated[Destination, MessageHeader("JMSDestination")],
                expiration: Annotated[Long, MessageHeader("JMSExpiration")],
                message_id: Annotated[str, MessageHeader("JMSMessageID")],
                priority: Annotated[int, MessageHeader("JMSPriority")],
                redelivered: Annotated[bool, MessageHeader("JMSRedelivered")],
                reply_to: Annotated[Destination | None, MessageHeader("JMSReplyTo")],
                timestamp: Annotated[Long, MessageHeader("JMSTimestamp")],
                message_type: Annotated[str | None, MessageHeader("JMSType")],
                string_header: Annotated[str | None, MessageHeader("CustomStringHeader")],
                boolean_header: Annotated[bool, MessageHeader("CustomBooleanHeader")],
                byte_header: Annotated[Byte, MessageHeader("CustomByteHeader")],
                short_header: Annotated[Short, MessageHeader("CustomShortHeader")],
                int_header: Annotated[int, MessageHeader("CustomIntegerHeader")],
                long_header: Annotated[Long, MessageHeader("CustomLongHeader")],
                float_header: Annotated[Float, MessageHeader("CustomFloatHeader")],
                double_header: Annotated[float, MessageHeader("CustomDoubleHeader")]) -> None:

        header_values = {
            "JMSCorrelationID": correlation_id,
            "JMSDeliveryMode": delivery_mode,
            "JMSDestination": destination,
            "JMSExpiration": expiration,
            "JMSMessageID": message_id,
            "JMSPriority": priority,
            "JMSRedelivered": redelivered,
            "JMSReplyTo": reply_to,
            "JMSTimestamp": timestamp,
            "JMSType": message_type,
            "CustomStringHeader": string_header,
            "CustomBooleanHeader": boolean_header,
            "CustomByteHeader": byte_header,
            "CustomShortHeader": short_header,
            "CustomIntegerHeader": int_header,
            "CustomLongHeader": long_header,
            "CustomFloatHeader": float_header,
            "CustomDoubleHeader": double_header,
        }

        self.message_headers.append(header_values)
        self.message_bodies.append(body)
        self.messages.append(message)

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Message
import io.micronaut.jms.annotations.Queue
import io.micronaut.jms.model.JMSHeaders.*
import io.micronaut.messaging.annotation.MessageBody
import io.micronaut.messaging.annotation.MessageHeader
import jakarta.jms.Destination
import java.io.Serializable

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class MapConsumer {

    val messageBodies = mutableListOf<Map<String, Serializable>>()
    val messageHeaders = mutableListOf<Map<String, Any?>>()
    val messages = mutableListOf<jakarta.jms.Message>()

    @Queue(value = "queue_map")
    fun receive(
        @MessageBody body: Map<String, Serializable>,
        @Message message: jakarta.jms.Message,
        @MessageHeader(JMS_CORRELATION_ID) correlationId: String?,
        @MessageHeader(JMS_DELIVERY_MODE) deliveryMode: Int,
        @MessageHeader(JMS_DESTINATION) destination: Destination?,
        @MessageHeader(JMS_EXPIRATION) expiration: Long,
        @MessageHeader(JMS_MESSAGE_ID) messageId: String?,
        @MessageHeader(JMS_PRIORITY) priority: Int,
        @MessageHeader(JMS_REDELIVERED) redelivered: Boolean,
        @MessageHeader(JMS_REPLY_TO) replyTo: Destination?,
        @MessageHeader(JMS_TIMESTAMP) timestamp: Long,
        @MessageHeader(JMS_TYPE) type: String?,
        @MessageHeader("CustomStringHeader") stringHeader: String?,
        @MessageHeader("CustomBooleanHeader") booleanHeader: Boolean,
        @MessageHeader("CustomByteHeader") byteHeader: Byte,
        @MessageHeader("CustomShortHeader") shortHeader: Short,
        @MessageHeader("CustomIntegerHeader") intHeader: Int,
        @MessageHeader("CustomLongHeader") longHeader: Long,
        @MessageHeader("CustomFloatHeader") floatHeader: Float,
        @MessageHeader("CustomDoubleHeader") doubleHeader: Double
    ) {
        val headerValues = mapOf(
            JMS_CORRELATION_ID to correlationId,
            JMS_DELIVERY_MODE to deliveryMode,
            JMS_DESTINATION to destination,
            JMS_EXPIRATION to expiration,
            JMS_MESSAGE_ID to messageId,
            JMS_PRIORITY to priority,
            JMS_REDELIVERED to redelivered,
            JMS_REPLY_TO to replyTo,
            JMS_TIMESTAMP to timestamp,
            JMS_TYPE to type,
            "CustomStringHeader" to stringHeader,
            "CustomBooleanHeader" to booleanHeader,
            "CustomByteHeader" to byteHeader,
            "CustomShortHeader" to shortHeader,
            "CustomIntegerHeader" to intHeader,
            "CustomLongHeader" to longHeader,
            "CustomFloatHeader" to floatHeader,
            "CustomDoubleHeader" to doubleHeader)
        messageHeaders.add(headerValues)
        messageBodies.add(body)
        messages.add(message)
    }
}

imports

import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Message
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import io.micronaut.messaging.annotation.MessageHeader
import jakarta.jms.Destination

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import static io.micronaut.jms.model.JMSHeaders.JMS_CORRELATION_ID
import static io.micronaut.jms.model.JMSHeaders.JMS_DELIVERY_MODE
import static io.micronaut.jms.model.JMSHeaders.JMS_DESTINATION
import static io.micronaut.jms.model.JMSHeaders.JMS_EXPIRATION
import static io.micronaut.jms.model.JMSHeaders.JMS_MESSAGE_ID
import static io.micronaut.jms.model.JMSHeaders.JMS_PRIORITY
import static io.micronaut.jms.model.JMSHeaders.JMS_REDELIVERED
import static io.micronaut.jms.model.JMSHeaders.JMS_REPLY_TO
import static io.micronaut.jms.model.JMSHeaders.JMS_TIMESTAMP
import static io.micronaut.jms.model.JMSHeaders.JMS_TYPE

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class MapConsumer {

    List<Map<String, Serializable>> messageBodies = [].asSynchronized()
    List<Map<String, Object>> messageHeaders =  [].asSynchronized()
    List<jakarta.jms.Message> messages =  [].asSynchronized()

    @Queue(value = 'queue_map')
    void receive(@MessageBody Map<String, Serializable> body,
                 @Message jakarta.jms.Message message,
                 @MessageHeader(JMS_CORRELATION_ID) @Nullable String correlationId,
                 @MessageHeader(JMS_DELIVERY_MODE) int deliveryMode,
                 @MessageHeader(JMS_DESTINATION) Destination destination,
                 @MessageHeader(JMS_EXPIRATION) long expiration,
                 @MessageHeader(JMS_MESSAGE_ID) String messageId,
                 @MessageHeader(JMS_PRIORITY) int priority,
                 @MessageHeader(JMS_REDELIVERED) boolean redelivered,
                 @MessageHeader(JMS_REPLY_TO) @Nullable Destination replyTo,
                 @MessageHeader(JMS_TIMESTAMP) long timestamp,
                 @MessageHeader(JMS_TYPE) @Nullable String type,
                 @MessageHeader('CustomStringHeader') @Nullable String stringHeader,
                 @MessageHeader('CustomBooleanHeader') boolean booleanHeader,
                 @MessageHeader('CustomByteHeader') byte byteHeader,
                 @MessageHeader('CustomShortHeader') short shortHeader,
                 @MessageHeader('CustomIntegerHeader') int intHeader,
                 @MessageHeader('CustomLongHeader') long longHeader,
                 @MessageHeader('CustomFloatHeader') float floatHeader,
                 @MessageHeader('CustomDoubleHeader') double doubleHeader) {

        Map<String, Object> headerValues = [
                (JMS_CORRELATION_ID): correlationId,
                (JMS_DELIVERY_MODE): deliveryMode,
                (JMS_DESTINATION): destination,
                (JMS_EXPIRATION): expiration,
                (JMS_MESSAGE_ID): messageId,
                (JMS_PRIORITY): priority,
                (JMS_REDELIVERED): redelivered,
                (JMS_REPLY_TO): replyTo,
                (JMS_TIMESTAMP): timestamp,
                (JMS_TYPE): type,
                CustomStringHeader: stringHeader,
                CustomBooleanHeader: booleanHeader,
                CustomByteHeader: byteHeader,
                CustomShortHeader: shortHeader,
                CustomIntegerHeader: intHeader,
                CustomLongHeader: longHeader,
                CustomFloatHeader: floatHeader,
                CustomDoubleHeader: doubleHeader]

        messageHeaders << headerValues
        messageBodies << body
        messages << message
    }
}

Session

A listener method can declare a parameter of type jakarta.jms.Session, without a binding annotation, to receive the session that consumed the message, for example to reply on that session. A parameter typed as a provider subtype of Session, such as Oracle’s oracle.jakarta.jms.AQjmsSession from aqapi-jakarta, receives the session when it is of that type. When it is not, the application fails to start. Oracle’s oracle.jms.AQjmsSession, from aqapi, implements javax.jms.Session instead, and is not recognised as a session: like any other listener parameter without a binding annotation, it fails the application at startup.

With transacted = true, the listener commits the session when the method returns and rolls it back when the method throws, so messages sent on the session commit together with the receipt of the message. With Oracle Transactional Event Queues, SQL run on the session’s database connection commits with them too. The following listener confirms each order in the same transaction as its receipt:

imports

import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.annotations.JMSListener;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.messaging.annotation.MessageBody;
import jakarta.jms.JMSException;
import jakarta.jms.MessageProducer;
import jakarta.jms.Session;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
public class OrderListener {

    final List<String> confirmations = Collections.synchronizedList(new ArrayList<>());

    @Queue(value = "orders", transacted = true)
    public void receive(@MessageBody String order, Session session) throws JMSException { // 
        try (MessageProducer producer = session.createProducer(session.createQueue("order_confirmations"))) { // 
            producer.send(session.createTextMessage("confirmed " + order));
        }
    }

    @Queue("order_confirmations")
    public void confirmed(@MessageBody String confirmation) {
        confirmations.add(confirmation);
    }
}

imports

from jakarta.jms import Session
from micronaut.context.annotation import Requires
from micronaut.jms.annotations import JMSListener, Queue
from micronaut.messaging.annotation import MessageBody

@JMSListener("activeMqConnectionFactory")
class OrderListener:

    def __init__(self) -> None:
        self.confirmations: list[str] = []

    @Queue("orders", transacted=True)
    def receive(self, order: Annotated[str, MessageBody], session: Session) -> None:  # 
        producer = session.createProducer(session.createQueue("order_confirmations"))  # 
        try:
            producer.send(session.createTextMessage("confirmed " + order))
        finally:
            producer.close()

    @Queue("order_confirmations")
    def confirmed(self, confirmation: Annotated[str, MessageBody]) -> None:
        self.confirmations.append(confirmation)

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import jakarta.jms.Session
import java.util.Collections

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class OrderListener {

    val confirmations: MutableList<String> = Collections.synchronizedList(ArrayList())

    @Queue(value = "orders", transacted = true)
    fun receive(@MessageBody order: String, session: Session) { // 
        session.createProducer(session.createQueue("order_confirmations")).use { producer -> // 
            producer.send(session.createTextMessage("confirmed $order"))
        }
    }

    @Queue("order_confirmations")
    fun confirmed(@MessageBody confirmation: String) {
        confirmations.add(confirmation)
    }
}

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import jakarta.jms.Session

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class OrderListener {

    final List<String> confirmations = Collections.synchronizedList([])

    @Queue(value = 'orders', transacted = true)
    void receive(@MessageBody String order, Session session) { // 
        session.createProducer(session.createQueue('order_confirmations')).withCloseable { producer -> // 
            producer.send(session.createTextMessage("confirmed ${order}".toString()))
        }
    }

    @Queue('order_confirmations')
    void confirmed(@MessageBody String confirmation) {
        confirmations << confirmation
    }
}

The Session parameter receives the session of the listener.
The reply is sent with a producer of that session, and commits with the receipt of the order.

7 Serializers and Deserializers

By default, message bodies are converted by DefaultSerializerDeserializer: strings become text messages, maps map messages, byte arrays bytes messages, other serializable objects (lists and object arrays included) object messages, and other objects JSON text messages. A destination can use another conversion with a named Serializer or Deserializer bean:

imports

import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.serdes.Deserializer;
import io.micronaut.jms.serdes.Serializer;
import io.micronaut.messaging.exceptions.MessagingException;
import jakarta.inject.Named;
import jakarta.inject.Singleton;
import jakarta.jms.JMSException;
import jakarta.jms.Message;
import jakarta.jms.Session;
import jakarta.jms.TextMessage;

import java.util.Arrays;
import java.util.List;

@Singleton
@Named("csv") // 
public class CsvSerdes implements Serializer, Deserializer {

    @Override
    @SuppressWarnings("unchecked")
    public Message serialize(Session session, Object body) {
        try {
            return session.createTextMessage(String.join(",", (List<String>) body));
        } catch (JMSException e) {
            throw new MessagingException("Unable to create the message", e);
        }
    }

    @Override
    public <T> T deserialize(Message message, Class<T> type) {
        try {
            return type.cast(Arrays.asList(((TextMessage) message).getText().split(",")));
        } catch (JMSException e) {
            throw new MessagingException("Unable to read the message", e);
        }
    }
}

imports

from jakarta.inject import Named, Singleton
from jakarta.jms import Message, Session
from java.util import Arrays
from micronaut.context.annotation import Requires
from micronaut.jms.serdes import Deserializer, Serializer

@Singleton
@Named("csv")  # 
class CsvSerdes(Serializer, Deserializer):

    def serialize(self, session: Session, body: object) -> Message:
        return session.createTextMessage(",".join(body))

    def deserialize(self, message: Message, type: type) -> object:
        return Arrays.asList(*message.getText().split(","))

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.serdes.Deserializer
import io.micronaut.jms.serdes.Serializer
import jakarta.inject.Named
import jakarta.inject.Singleton
import jakarta.jms.Message
import jakarta.jms.Session
import jakarta.jms.TextMessage

@Singleton
@Named("csv") // 
class CsvSerdes : Serializer, Deserializer {

    override fun serialize(session: Session, body: Any): Message =
        session.createTextMessage((body as List<*>).joinToString(","))

    override fun <T> deserialize(message: Message, type: Class<T>): T =
        type.cast((message as TextMessage).text.split(","))
}

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.serdes.Deserializer
import io.micronaut.jms.serdes.Serializer
import jakarta.inject.Named
import jakarta.inject.Singleton
import jakarta.jms.Message
import jakarta.jms.Session
import jakarta.jms.TextMessage

@Singleton
@Named('csv') // 
class CsvSerdes implements Serializer, Deserializer {

    @Override
    Message serialize(Session session, Object body) {
        session.createTextMessage(((List<String>) body).join(','))
    }

    @Override
    <T> T deserialize(Message message, Class<T> type) {
        type.cast(((TextMessage) message).text.split(',').toList())
    }
}

The bean name is the name given to the serializer and deserializer members of @Queue and @Topic. A bean can implement either interface, or both.

The serializer member of a producer method names the bean that creates the messages it sends:

imports

import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.annotations.JMSProducer;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.messaging.annotation.MessageBody;

import java.util.List;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
public interface TagProducer {

    @Queue(value = "tags", serializer = "csv") // 
    void send(@MessageBody List<String> tags);
}

imports

from micronaut.context.annotation import Requires
from micronaut.jms.annotations import JMSProducer, Queue
from micronaut.messaging.annotation import MessageBody

@JMSProducer("activeMqConnectionFactory")
class TagProducer(ABC):

    @Queue("tags", serializer="csv")  # 
    @abstractmethod
    def send(self, tags: Annotated[list[str], MessageBody]) -> None:
        ...

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
interface TagProducer {

    @Queue(value = "tags", serializer = "csv") // 
    fun send(@MessageBody tags: List<String>)
}

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
interface TagProducer {

    @Queue(value = 'tags', serializer = 'csv') // 
    void send(@MessageBody List<String> tags)
}

The messages of this method are created by the csv serializer.

The deserializer member of a listener method names the bean that converts the messages it receives into its @MessageBody argument. The other arguments are bound as usual:

imports

import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.annotations.JMSListener;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.messaging.annotation.MessageBody;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
public class TagListener {

    final List<List<String>> received = new CopyOnWriteArrayList<>();

    @Queue(value = "tags", deserializer = "csv") // 
    public void receive(@MessageBody List<String> tags) {
        received.add(tags);
    }
}

imports

from micronaut.context.annotation import Requires
from micronaut.jms.annotations import JMSListener, Queue
from micronaut.messaging.annotation import MessageBody

@JMSListener("activeMqConnectionFactory")
class TagListener:

    def __init__(self) -> None:
        self.received: list[list[str]] = []

    @Queue("tags", deserializer="csv")  # 
    def receive(self, tags: Annotated[list[str], MessageBody]) -> None:
        self.received.append(list(tags))

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import java.util.concurrent.CopyOnWriteArrayList

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class TagListener {

    val received: MutableList<List<String>> = CopyOnWriteArrayList()

    @Queue(value = "tags", deserializer = "csv") // 
    fun receive(@MessageBody tags: List<String>) {
        received.add(tags)
    }
}

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody

import java.util.concurrent.CopyOnWriteArrayList

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class TagListener {

    final List<List<String>> received = new CopyOnWriteArrayList<>()

    @Queue(value = 'tags', deserializer = 'csv') // 
    void receive(@MessageBody List<String> tags) {
        received << tags
    }
}

The bodies of the messages of this method are read by the csv deserializer.

8 Error Handlers

There are multiple ways to inject custom error handling into your JMS Listeners as shown here:

imports

import io.micronaut.jms.annotations.JMSListener;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.jms.listener.JMSListenerErrorHandler;
import io.micronaut.messaging.annotation.MessageBody;
import jakarta.inject.Singleton;

import jakarta.jms.Message;
import jakarta.jms.Session;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.atomic.AtomicInteger;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSListener(value = CONNECTION_FACTORY_BEAN_NAME, errorHandlers = {AccumulatingErrorHandler.class}) // 
class ErrorThrowingConsumer {

    Collection<String> messages = Collections.synchronizedSet(new HashSet<>());

    @Queue(value = "error-queue", errorHandlers = {CountingErrorHandler.class}) // 
    void receive(@MessageBody String message) {
        if ("throw an error".equalsIgnoreCase(message)) {
            throw new RuntimeException("this is an error"); // 
        }
        messages.add(message);
    }
}

imports

from typing import Annotated

from jakarta.inject import Singleton
from jakarta.jms import Message, Session
from java.lang import RuntimeException, Throwable
from micronaut.context.annotation import Requires
from micronaut.jms.annotations import JMSListener, Queue
from micronaut.jms.listener import JMSListenerErrorHandler
from micronaut.messaging.annotation import MessageBody

@JMSListener("activeMqConnectionFactory", errorHandlers=[AccumulatingErrorHandler])  # 
class ErrorThrowingConsumer:

    def __init__(self) -> None:
        self.messages: set[str] = set()

    @Queue("error-queue", errorHandlers=[CountingErrorHandler])  # 
    def receive(self, message: Annotated[str, MessageBody]) -> None:
        if message.lower() == "throw an error":
            raise RuntimeException("this is an error")  # 
        self.messages.add(message)

imports

import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.jms.listener.JMSListenerErrorHandler
import io.micronaut.messaging.annotation.MessageBody
import jakarta.inject.Singleton
import jakarta.jms.Message
import jakarta.jms.Session
import java.util.*
import java.util.concurrent.atomic.AtomicInteger

@JMSListener(
        CONNECTION_FACTORY_BEAN_NAME,
        errorHandlers = [AccumulatingErrorHandler::class]) // 
class ErrorThrowingConsumer {

    val processed: MutableList<String> = mutableListOf()

    @Queue(
            value = "error-queue",
            errorHandlers = [CountingErrorHandler::class]) // 
    fun receive(@MessageBody message: String) {
        if (message == "throw an error") {
            throw RuntimeException("This is an unexpected error.") // 
        }
        processed.add(message)
    }
}

imports

import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.jms.listener.JMSListenerErrorHandler
import io.micronaut.messaging.annotation.MessageBody
import jakarta.inject.Singleton
import jakarta.jms.Message
import jakarta.jms.Session

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSListener(value = CONNECTION_FACTORY_BEAN_NAME, errorHandlers = [AccumulatingErrorHandler.class]) // 
class ErrorThrowingConsumer {

    Collection<String> messages = Collections.synchronizedSet(new HashSet<String>())

    @Queue(value = "error-queue", errorHandlers = [CountingErrorHandler.class]) // 
    void receive(@MessageBody String message) {
        if (message == "throw an error") {
            throw new RuntimeException("this is an error") // 
        }
        messages.add(message)
    }
}

You can add the errorHandlers arguments to your @JMSListener to add custom error handling logic to all @Queue or @Topic annotated methods within that class.
You can add the errorHandlers argument to your @Queue or @Topic annotated method to inject custom error handling logic only for messages received on that specific queue or topic.
By default you will always have an io.micronaut.jms.listener.LoggingJMSListenerErrorHandler which will log the thrown exception at ERROR level.

9 Success Handlers

There are multiple ways to inject custom success handling into your JMS Listeners as shown here:

imports

import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.annotations.JMSListener;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.jms.listener.JMSListenerSuccessHandler;
import io.micronaut.messaging.annotation.MessageBody;
import jakarta.inject.Singleton;

import jakarta.jms.JMSException;
import jakarta.jms.Message;
import jakarta.jms.Session;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.atomic.AtomicInteger;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSListener(value = CONNECTION_FACTORY_BEAN_NAME, successHandlers = {AccumulatingSuccessHandler.class}) // 
class SuccessHandlingConsumer {

    Collection<String> messages = Collections.synchronizedSet(new HashSet<>());

    @Queue(value = "success-queue", successHandlers = {CountingSuccessHandler.class})  // 
    void receive(@MessageBody String message) throws JMSException {
        messages.add(message);
    }
}

imports

from typing import Annotated

from jakarta.inject import Singleton
from jakarta.jms import Message, Session
from micronaut.context.annotation import Requires
from micronaut.jms.annotations import JMSListener, Queue
from micronaut.jms.listener import JMSListenerSuccessHandler
from micronaut.messaging.annotation import MessageBody

@JMSListener("activeMqConnectionFactory", successHandlers=[AccumulatingSuccessHandler])  # 
class SuccessHandlingConsumer:

    def __init__(self) -> None:
        self.messages: set[str] = set()

    @Queue("success-queue", successHandlers=[CountingSuccessHandler])  # 
    def receive(self, message: Annotated[str, MessageBody]) -> None:
        self.messages.add(message)

imports

import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.jms.listener.JMSListenerSuccessHandler
import io.micronaut.messaging.annotation.MessageBody
import jakarta.inject.Singleton
import jakarta.jms.Message
import jakarta.jms.Session
import java.util.*
import java.util.concurrent.atomic.AtomicInteger

@JMSListener(
        CONNECTION_FACTORY_BEAN_NAME,
        successHandlers = [AccumulatingSuccessHandler::class]) // 
class SuccessHandlingConsumer {

    val processed: MutableList<String> = mutableListOf()

    @Queue(
            value = "success-queue",
            successHandlers = [CountingSuccessHandler::class]) // 
    fun receive(@MessageBody message: String) {
        processed.add(message)
    }
}

imports

import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.jms.listener.JMSListenerSuccessHandler
import io.micronaut.messaging.annotation.MessageBody
import jakarta.inject.Singleton
import jakarta.jms.Message
import jakarta.jms.Session

import java.util.concurrent.atomic.AtomicInteger

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSListener(value = CONNECTION_FACTORY_BEAN_NAME, successHandlers = [AccumulatingSuccessHandler.class]) // 
class SuccessHandlingConsumer {

    Collection<String> messages = Collections.synchronizedSet(new HashSet<String>())

    @Queue(value = "success-queue", successHandlers = [CountingSuccessHandler.class]) // 
    void receive(@MessageBody String message) {
        messages.add(message)
    }
}

You can add the successHandlers arguments to your @JMSListener to add custom success handling logic to all @Queue or @Topic annotated methods within that class.
You can add the successHandlers argument to your @Queue or @Topic annotated method to inject custom success handling logic only for messages received on that specific queue or topic

10 Message Selectors

Message Selectors allow you to filter the messages a @JMSListener method will receive. The selector can use all message header and properties for filtering except the message content.

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
public class SelectorConsumer {

    List<String> messageBodiesTrue = Collections.synchronizedList(new ArrayList<>());

    List<String> messageBodiesFalse = Collections.synchronizedList(new ArrayList<>());

    List<String> messageBodiesTopic = Collections.synchronizedList(new ArrayList<>());

    @Queue(value = "selector_queue", messageSelector = "CustomBooleanHeader=true")
    public void receive(@MessageBody String body) {
        messageBodiesTrue.add(body);
    }

    @Queue(value = "selector_queue", messageSelector = "CustomBooleanHeader=false")
    public void receive2(@MessageBody String body) {
        messageBodiesFalse.add(body);
    }

    @Topic(value = "selector_topic", messageSelector = "CustomBooleanHeader=true")
    public void receiveTopic(@MessageBody String body) {
        messageBodiesTopic.add(body);
    }
}

@JMSListener("activeMqConnectionFactory")
class SelectorConsumer:

    def __init__(self) -> None:
        self.message_bodies_true: list[str] = []
        self.message_bodies_false: list[str] = []
        self.message_bodies_topic: list[str] = []

    @Queue("selector_queue", messageSelector="CustomBooleanHeader=true")
    def receive(self, body: Annotated[str, MessageBody]) -> None:
        self.message_bodies_true.append(body)

    @Queue("selector_queue", messageSelector="CustomBooleanHeader=false")
    def receive2(self, body: Annotated[str, MessageBody]) -> None:
        self.message_bodies_false.append(body)

    @Topic("selector_topic", messageSelector="CustomBooleanHeader=true")
    def receive_topic(self, body: Annotated[str, MessageBody]) -> None:
        self.message_bodies_topic.append(body)

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class SelectorConsumer {

    val messageBodiesTrue: MutableList<String> = Collections.synchronizedList(ArrayList())

    val messageBodiesFalse: MutableList<String> = Collections.synchronizedList(ArrayList())

    val messageBodiesTopic: MutableList<String> = Collections.synchronizedList(ArrayList())

    @Queue(value = "selector_queue", messageSelector = "CustomBooleanHeader=true")
    fun receive(@MessageBody body: String) {
        messageBodiesTrue.add(body)
    }

    @Queue(value = "selector_queue", messageSelector = "CustomBooleanHeader=false")
    fun receive2(@MessageBody body: String) {
        messageBodiesFalse.add(body)
    }

    @Topic(value = "selector_topic", messageSelector = "CustomBooleanHeader=true")
    fun receiveTopic(@MessageBody body: String) {
        messageBodiesTopic.add(body)
    }
}

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class SelectorConsumer {

    List<String> messageBodiesTrue = [].asSynchronized()

    List<String> messageBodiesFalse = [].asSynchronized()

    List<String> messageBodiesTopic = [].asSynchronized()

    @Queue(value = 'selector_queue', messageSelector = 'CustomBooleanHeader=true')
    void receive(@MessageBody String body) {
        messageBodiesTrue << body
    }

    @Queue(value = 'selector_queue', messageSelector = 'CustomBooleanHeader=false')
    void receive2(@MessageBody String body) {
        messageBodiesFalse << body
    }

    @Topic(value = 'selector_topic', messageSelector = 'CustomBooleanHeader=true')
    void receiveTopic(@MessageBody String body) {
        messageBodiesTopic << body
    }
}

More examples on how to use Message Selector can be found here.

11 Delivery Mode and Priority

Producer methods set the delivery mode and priority of the messages they send with the deliveryMode and priority members of @Queue and @Topic. Messages are PERSISTENT with the JMS default priority, 4, unless a producer method sets otherwise. Listeners ignore both members.

imports

import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.annotations.JMSProducer;
import io.micronaut.jms.annotations.MessagePriority;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.jms.model.JMSDeliveryMode;
import io.micronaut.messaging.annotation.MessageBody;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
public interface AlertProducer {

    @Queue(value = "alerts", deliveryMode = JMSDeliveryMode.NON_PERSISTENT, priority = 7) // 
    void send(@MessageBody String alert);

    @Queue(value = "alerts", deliveryMode = JMSDeliveryMode.NON_PERSISTENT)
    void sendWithPriority(@MessageBody String alert, @MessagePriority int priority); // 
}

imports

from micronaut.context.annotation import Requires
from micronaut.jms.annotations import JMSProducer, MessagePriority, Queue
from micronaut.messaging.annotation import MessageBody

@JMSProducer("activeMqConnectionFactory")
class AlertProducer(ABC):

    @Queue("alerts", deliveryMode="NON_PERSISTENT", priority=7)  # 
    @abstractmethod
    def send(self, alert: Annotated[str, MessageBody]) -> None:
        ...

    @Queue("alerts", deliveryMode="NON_PERSISTENT")
    @abstractmethod
    def send_with_priority(self, alert: Annotated[str, MessageBody],
                           priority: Annotated[int, MessagePriority]) -> None:  # 
        ...

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.MessagePriority
import io.micronaut.jms.annotations.Queue
import io.micronaut.jms.model.JMSDeliveryMode
import io.micronaut.messaging.annotation.MessageBody

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
interface AlertProducer {

    @Queue(value = "alerts", deliveryMode = JMSDeliveryMode.NON_PERSISTENT, priority = 7) // 
    fun send(@MessageBody alert: String)

    @Queue(value = "alerts", deliveryMode = JMSDeliveryMode.NON_PERSISTENT)
    fun sendWithPriority(@MessageBody alert: String, @MessagePriority priority: Int) // 
}

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.MessagePriority
import io.micronaut.jms.annotations.Queue
import io.micronaut.jms.model.JMSDeliveryMode
import io.micronaut.messaging.annotation.MessageBody

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
interface AlertProducer {

    @Queue(value = 'alerts', deliveryMode = JMSDeliveryMode.NON_PERSISTENT, priority = 7) // 
    void send(@MessageBody String alert)

    @Queue(value = 'alerts', deliveryMode = JMSDeliveryMode.NON_PERSISTENT)
    void sendWithPriority(@MessageBody String alert, @MessagePriority int priority) // 
}

Messages sent by this method are not persisted by the broker and have priority 7.
A @MessagePriority argument sets the priority of each call, from 0 (lowest) to 9. A @MessageDeliveryMode argument, a JMSDeliveryMode or a jakarta.jms.DeliveryMode value, sets the delivery mode of each call in the same way, and @MessageTTL the time to live in milliseconds.

A @MessageDeliveryMode argument can also be the name of a delivery mode, such as "NON_PERSISTENT", and a @MessagePriority argument a String of digits, such as "7". Arguments of other types fail the call. An invalid delivery mode or a priority outside 0 to 9 fails the call with a MessagingClientException.

Listeners read both values with the JMSPriority and JMSDeliveryMode headers, as described under Parameter Binding.

Note

Before 5.2.0, producer methods sent the priority of the message that was just created, which some brokers, such as ActiveMQ Classic, report as 0. They now send 4 unless they set another priority. On a broker that orders messages by priority, these messages are now delivered ahead of messages still sent with priority 0.

Note	Amazon SQS has no delivery modes or priorities, and the SQS JMS client ignores them: `jms-sqs` producers send their messages without the delivery mode, the priority or the `@MessageTTL` time to live.

12 Asynchronous Producers and Producer Sessions

Asynchronous producer methods

Producer methods can return a CompletableFuture or another CompletionStage, or a Reactive Streams Publisher such as a Reactor Mono, instead of void:

imports

import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.annotations.JMSProducer;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.messaging.annotation.MessageBody;
import org.reactivestreams.Publisher;

import java.util.concurrent.CompletableFuture;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
public interface ReceiptProducer {

    @Queue("receipts")
    CompletableFuture<Void> sendAsync(@MessageBody String receipt); // 

    @Queue("receipts")
    Publisher<Void> sendReactive(@MessageBody String receipt); // 
}

imports

from java.util.concurrent import CompletableFuture
from micronaut.context.annotation import Requires
from micronaut.jms.annotations import JMSProducer, Queue
from micronaut.messaging.annotation import MessageBody
from org.reactivestreams import Publisher

@JMSProducer("activeMqConnectionFactory")
class ReceiptProducer(ABC):

    @Queue("receipts")
    @abstractmethod
    def send_async(self, receipt: Annotated[str, MessageBody]) -> CompletableFuture[None]:  # 
        ...

    @Queue("receipts")
    @abstractmethod
    def send_reactive(self, receipt: Annotated[str, MessageBody]) -> Publisher[None]:  # 
        ...

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import org.reactivestreams.Publisher
import java.util.concurrent.CompletableFuture

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
interface ReceiptProducer {

    @Queue("receipts")
    fun sendAsync(@MessageBody receipt: String): CompletableFuture<Void> // 

    @Queue("receipts")
    fun sendReactive(@MessageBody receipt: String): Publisher<Void> // 
}

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import org.reactivestreams.Publisher

import java.util.concurrent.CompletableFuture

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
interface ReceiptProducer {

    @Queue('receipts')
    CompletableFuture<Void> sendAsync(@MessageBody String receipt) // 

    @Queue('receipts')
    Publisher<Void> sendReactive(@MessageBody String receipt) // 
}

The message is sent on the blocking executor, since JMS sends block. The future completes once the message is sent, or completes exceptionally if the send fails.
The message is sent when the publisher is subscribed to. The publisher then completes without emitting an item, or signals the error.

The propagated context of the caller, for example the current trace with Micronaut Tracing, follows the send to the executor. It is the context of the code that called the method, also when a returned publisher is subscribed to on another thread. Invalid arguments, such as an unknown delivery mode, complete the future exceptionally or make the publisher signal the error, rather than being thrown by the call.

Producer sessions

Producer methods reuse their sessions. For each connection factory, they open one connection and keep up to max-idle idle sessions for each session mode, plain or transacted. Each send borrows a session, so concurrent sends use different sessions, and a session on which a send failed is closed rather than reused. The connection and its sessions are closed with the application context.

At most max-active sessions of a connection factory are borrowed at once, of all session modes. When all of them are borrowed, for example during a burst of asynchronous sends, a send waits up to max-wait for another send to release its session. A session released while sends wait is kept for them, so a burst reuses at most max-active sessions of each session mode rather than closing and opening sessions. Once no send waits, the idle sessions above max-idle are closed: when they are released, or within 30 seconds otherwise. If none is released in time, the send fails with a MessagingClientException: a synchronous method throws it, a returned future completes exceptionally and a returned publisher signals the error. Nothing is sent in that case.

If a session cannot be created, or the JMS provider reports to the connection’s ExceptionListener that the connection failed, the connection is replaced. Its idle sessions are closed, and sessions that sends are still using are closed when they are released, rather than reused. The old connection is closed once the last of its sessions is, so sends in progress on it are not interrupted.

micronaut.jms.producer-sessions.enabled=true
micronaut.jms.producer-sessions.max-idle=8
micronaut.jms.producer-sessions.max-active=32
micronaut.jms.producer-sessions.max-wait=10s

micronaut:
  jms:
    producer-sessions:
      enabled: true
      max-idle: 8
      max-active: 32
      max-wait: 10s

[micronaut.jms.producer-sessions]
enabled = true
max-idle = 8
max-active = 32
max-wait = "10s"

micronaut {
  jms {
    producerSessions {
      enabled = true
      maxIdle = 8
      maxActive = 32
      maxWait = '10s'
    }
  }
}

{
  micronaut = {
    jms = {
      producer-sessions = {
        enabled = true
        max-idle = 8
        max-active = 32
        max-wait = "10s"
      }
    }
  }
}

{
  "micronaut": {
    "jms": {
      "producer-sessions": {
        "enabled": true,
        "max-idle": 8,
        "max-active": 32,
        "max-wait": "10s"
      }
    }
  }
}

With enabled: false, each send takes a connection from the connection pool and opens a session for that message, as in earlier releases. A JmsProducer created by the application reuses sessions when it is given a ProducerSessionPool.

13 Listener Concurrency and Transacted Sends

A listener method consumes messages on one session, one message at a time. To handle messages concurrently, set consumers on @Queue to the number of consumers to start, each with its own session. The consumers of a listener method share one connection, so the method takes a single connection from the pool however many consumers it has:

imports

import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.annotations.JMSListener;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.messaging.annotation.MessageBody;
import jakarta.jms.Session;

import java.util.Collections;
import java.util.IdentityHashMap;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
public class JobListener {

    final Set<Session> sessions = Collections.synchronizedSet(Collections.newSetFromMap(new IdentityHashMap<>()));
    final AtomicInteger processed = new AtomicInteger();

    @Queue(value = "jobs", consumers = 3) // 
    public void process(@MessageBody String job, Session session) throws InterruptedException {
        sessions.add(session); // 
        Thread.sleep(50);
        processed.incrementAndGet();
    }
}

imports

from jakarta.jms import Session
from java.lang import System
from micronaut.context.annotation import Requires
from micronaut.jms.annotations import JMSListener, Queue
from micronaut.messaging.annotation import MessageBody

@JMSListener("activeMqConnectionFactory")
class JobListener:

    def __init__(self) -> None:
        self.lock = threading.Lock()
        self.session_ids: set[int] = set()
        self.processed = 0

    @Queue("jobs", consumers=3)  # 
    def process(self, job: Annotated[str, MessageBody], session: Session) -> None:
        with self.lock:
            self.session_ids.add(System.identityHashCode(session))  # 
        time.sleep(0.05)
        with self.lock:
            self.processed += 1

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import jakarta.jms.Session
import java.util.Collections
import java.util.IdentityHashMap
import java.util.concurrent.atomic.AtomicInteger

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class JobListener {

    val sessions: MutableSet<Session> = Collections.synchronizedSet(Collections.newSetFromMap(IdentityHashMap()))
    val processed = AtomicInteger()

    @Queue(value = "jobs", consumers = 3) // 
    fun process(@MessageBody job: String, session: Session) {
        sessions.add(session) // 
        Thread.sleep(50)
        processed.incrementAndGet()
    }
}

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody
import jakarta.jms.Session

import java.util.concurrent.atomic.AtomicInteger

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class JobListener {

    final Set<Session> sessions = Collections.synchronizedSet(Collections.newSetFromMap(new IdentityHashMap<Session, Boolean>()))
    final AtomicInteger processed = new AtomicInteger()

    @Queue(value = 'jobs', consumers = 3) // 
    void process(@MessageBody String job, Session session) {
        sessions << session // 
        Thread.sleep(50)
        processed.incrementAndGet()
    }
}

Three consumers receive the messages of the jobs queue, so up to three jobs run at once.
Each consumer has its own session, which a Session parameter receives.

Before setting consumers above 1, note that:

The listener bean is still a single instance, and its method now runs on several threads at once, one per consumer. The listener must be thread-safe: the example guards its state with a synchronized set and an atomic counter (a lock in Python).
Messages are no longer processed in the order they were sent, since each consumer handles the next message it receives while the others are still busy.
If a consumer fails to start, the method keeps none of its consumers: the consumers already started are stopped and their connection is given back (returned to the pool, or closed if it is dedicated). A JMSException is logged and the application starts without the method’s consumers; any other failure stops startup.

On a topic, every consumer of a plain subscription receives every message, so consumers above 1 requires a shared subscription (see Durable and Shared Topic Subscriptions), whose consumers share the messages. An unshared durable subscription allows only one consumer. A shared subscription with a client ID can have several consumers: they share the one connection that holds the client ID.

Producer methods send each message on a session of their own. With transacted = true, the session is transacted and the send is committed before the method returns:

imports

import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.annotations.JMSProducer;
import io.micronaut.jms.annotations.Queue;
import io.micronaut.messaging.annotation.MessageBody;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
public interface JobProducer {

    @Queue(value = "jobs", transacted = true) // 
    void submit(@MessageBody String job);
}

imports

from micronaut.context.annotation import Requires
from micronaut.jms.annotations import JMSProducer, Queue
from micronaut.messaging.annotation import MessageBody

@JMSProducer("activeMqConnectionFactory")
class JobProducer(ABC):

    @Queue("jobs", transacted=True)  # 
    @abstractmethod
    def submit(self, job: Annotated[str, MessageBody]) -> None:
        ...

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
interface JobProducer {

    @Queue(value = "jobs", transacted = true) // 
    fun submit(@MessageBody job: String)
}

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.annotations.JMSProducer
import io.micronaut.jms.annotations.Queue
import io.micronaut.messaging.annotation.MessageBody

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSProducer(CONNECTION_FACTORY_BEAN_NAME)
interface JobProducer {

    @Queue(value = 'jobs', transacted = true) // 
    void submit(@MessageBody String job)
}

The message is sent on a transacted session and committed.

To send messages in the transaction of a transacted listener, so that they are committed or rolled back with the message it received, send them with a producer of the listener’s Session, as shown under Parameter Binding.

14 Health, Metrics and Tracing

Health

With the micronaut-management module, the health endpoint has a jms indicator. For each JMS connection factory, it opens a new connection, not one from the pool, on the blocking executor. The indicator is UP when every connection factory connects, with the JMS provider name and version of each, and DOWN with the error otherwise. It only checks that a connection can be opened, which doesn’t reach the broker with every provider: an ActiveMQ Classic failover: URL connects in the background, and the SQS connection factory makes no AWS call, so the indicator reports UP for them while the broker is unreachable.

Add the dependency:

implementation("io.micronaut:micronaut-management")

<dependency>
    <groupId>io.micronaut</groupId>
    <artifactId>micronaut-management</artifactId>
</dependency>

Disable the indicator with:

endpoints.health.jms.enabled=false

endpoints:
  health:
    jms:
      enabled: false

[endpoints.health.jms]
enabled = false

endpoints {
  health {
    jms {
      enabled = false
    }
  }
}

{
  endpoints = {
    health = {
      jms = {
        enabled = false
      }
    }
  }
}

{
  "endpoints": {
    "health": {
      "jms": {
        "enabled": false
      }
    }
  }
}

Metrics

With a Micrometer MeterRegistry, for example from the Micronaut Micrometer module, the following meters are recorded:

Meter Description

Meter	Description
`jms.listener.messages`	Counter of the messages that listeners process
`jms.listener.duration`	Timer of the processing of a message, from the start of the listener method to the end of the success or error handlers
`jms.producer.messages`	Counter of the messages that producer methods and `JmsProducer` send
`jms.producer.duration`	Timer of the sending of a message, from taking the connection to the commit of a transacted send, so a send that fails to connect is counted as a failure too

jms.listener.messages

Counter of the messages that listeners process

jms.listener.duration

Timer of the processing of a message, from the start of the listener method to the end of the success or error handlers

jms.producer.messages

Counter of the messages that producer methods and JmsProducer send

jms.producer.duration

Timer of the sending of a message, from taking the connection to the commit of a transacted send, so a send that fails to connect is counted as a failure too

The meters are tagged with the destination and the result: success, or failure with the exception class name. Disable the metrics with:

micronaut.metrics.binders.jms.enabled=false

micronaut:
  metrics:
    binders:
      jms:
        enabled: false

[micronaut.metrics.binders.jms]
enabled = false

micronaut {
  metrics {
    binders {
      jms {
        enabled = false
      }
    }
  }
}

{
  micronaut = {
    metrics = {
      binders = {
        jms = {
          enabled = false
        }
      }
    }
  }
}

{
  "micronaut": {
    "metrics": {
      "binders": {
        "jms": {
          "enabled": false
        }
      }
    }
  }
}

Tracing

To trace messages with OpenTelemetry, add the micronaut-jms-tracing-opentelemetry module, which uses the OpenTelemetry bean of Micronaut Tracing:

implementation("io.micronaut.jms:micronaut-jms-tracing-opentelemetry")

<dependency>
    <groupId>io.micronaut.jms</groupId>
    <artifactId>micronaut-jms-tracing-opentelemetry</artifactId>
</dependency>

Messages are then traced:

Sending a message creates a PRODUCER span named send <destination>, a child of the current span. It starts before the connection is taken, so a send that fails to connect ends it with an error too. The trace context is propagated to the receiver in message properties by the configured propagators.
Processing a message creates a CONSUMER span named process <destination> that continues the trace of the sender. It is the current span while the listener method and the success and error handlers run, so spans created by the listener method are its children. It is also current in Micronaut’s propagated context, so it follows the listener method to the threads that the context propagates to. A failure sets the span’s status to ERROR and records the exception.

The spans have the messaging.system (jms), messaging.destination.name, messaging.operation.name, messaging.operation.type and messaging.message.id attributes. JMS property names must be Java identifiers, so the dashes of propagation headers such as those of B3 are replaced with __dash__ in property names, as the OpenTelemetry Java agent does. Disable tracing with:

micronaut.jms.tracing.enabled=false

micronaut:
  jms:
    tracing:
      enabled: false

[micronaut.jms.tracing]
enabled = false

micronaut {
  jms {
    tracing {
      enabled = false
    }
  }
}

{
  micronaut = {
    jms = {
      tracing = {
        enabled = false
      }
    }
  }
}

{
  "micronaut": {
    "jms": {
      "tracing": {
        "enabled": false
      }
    }
  }
}

Note

The OpenTelemetry Java agent has its own JMS instrumentation, which also creates spans for the messages that are sent and processed. Micronaut Tracing uses the agent’s OpenTelemetry when the agent is attached, so an application that runs with the agent should not add micronaut-jms-tracing-opentelemetry, or each message would be traced twice.

Custom observers

The metrics and the tracing are beans of type JMSMessageObserver. Other beans of that type are also called before a listener processes a message and before a producer sends one, and the observation each returns is completed afterwards with the error, if any. A producer’s observation starts before the connection is taken, and its beforeSend method is called with the message once the message is created, so properties set on it are sent. A listener’s observation can give, from its propagatedContext method, the propagated context that the listener method and the handlers run in. Observers are called in order, and their observations are completed in the reverse order. A JmsProducer created by the application only calls the observers added with addObservers.

15 Durable and Shared Topic Subscriptions

By default a @Topic listener method creates a non-durable subscription: it receives only the messages published while it is connected, and every listener receives every message.

Set subscription to name the subscription, then choose its kind with durable and shared:

durable shared Subscription

`durable`	`shared`	Subscription
`false`	`false`	Non-durable (the default). Do not set `subscription`.
`true`	`false`	Durable, unshared (`Session.createDurableSubscriber`). The broker keeps messages for the subscription while the listener is disconnected.
`false`	`true`	Shared, non-durable (`Session.createSharedConsumer`, JMS 2.0). Every consumer of the subscription, in this or another application instance, receives a share of the messages.
`true`	`true`	Shared, durable (`Session.createSharedDurableConsumer`, JMS 2.0). Messages are split between the consumers and kept while all of them are disconnected.

false

Non-durable (the default). Do not set subscription.

true

false

Durable, unshared (Session.createDurableSubscriber). The broker keeps messages for the subscription while the listener is disconnected.

false

true

Shared, non-durable (Session.createSharedConsumer, JMS 2.0). Every consumer of the subscription, in this or another application instance, receives a share of the messages.

true

Shared, durable (Session.createSharedDurableConsumer, JMS 2.0). Messages are split between the consumers and kept while all of them are disconnected.

The following listener keeps an audit of every order, including the ones published while the application was down:

imports

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import io.micronaut.context.annotation.Requires;
import io.micronaut.jms.annotations.JMSListener;
import io.micronaut.jms.annotations.Topic;
import io.micronaut.messaging.annotation.MessageBody;

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME;

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
public class AuditListener {

    final List<String> audited = new CopyOnWriteArrayList<>();

    @Topic(value = "orders", subscription = "order-audit", durable = true) // 
    public void audit(@MessageBody String order) {
        audited.add(order);
    }
}

imports

from micronaut.context.annotation import Requires
from micronaut.jms.annotations import JMSListener, Topic
from micronaut.messaging.annotation import MessageBody

@JMSListener("activeMqConnectionFactory")
class AuditListener:

    def __init__(self) -> None:
        self.audited: list[str] = []

    @Topic("orders", subscription="order-audit", durable=True)  # 
    def audit(self, order: Annotated[str, MessageBody]) -> None:
        self.audited.append(order)

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Topic
import io.micronaut.messaging.annotation.MessageBody
import java.util.concurrent.CopyOnWriteArrayList

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class AuditListener {

    val audited: MutableList<String> = CopyOnWriteArrayList()

    @Topic(value = "orders", subscription = "order-audit", durable = true) // 
    fun audit(@MessageBody order: String) {
        audited.add(order)
    }
}

imports

import io.micronaut.context.annotation.Requires
import io.micronaut.jms.annotations.JMSListener
import io.micronaut.jms.annotations.Topic
import io.micronaut.messaging.annotation.MessageBody

import java.util.concurrent.CopyOnWriteArrayList

import static io.micronaut.jms.activemq.classic.configuration.ActiveMqClassicConfiguration.CONNECTION_FACTORY_BEAN_NAME

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
class AuditListener {

    final List<String> audited = new CopyOnWriteArrayList<>()

    @Topic(value = 'orders', subscription = 'order-audit', durable = true) // 
    void audit(@MessageBody String order) {
        audited << order
    }
}

subscription names the durable subscription, and durable = true asks the broker to keep its messages.

Client IDs

JMS identifies an unshared durable subscription by the connection’s client ID together with the subscription name, and a client ID can be set only once per connection. Micronaut JMS therefore gives each unshared durable listener its own connection, outside the connection pool, with the client ID set from clientId. When clientId is not set, the subscription name is the client ID.

Set clientId when a broker administrator has assigned one, or to give a shared subscription a client ID. A listener with a clientId always gets its own connection. Shared subscriptions without one use a pooled connection.

Note	A client ID can be in use by only one connection at a time. Do not run two instances of an application with the same unshared durable listener against one broker; use a shared durable subscription to spread the messages between instances.

durable, shared and clientId require subscription, and subscription requires durable or shared. Micronaut JMS fails at startup if either is missing. Shared subscriptions need a JMS 2.0 broker, such as ActiveMQ Artemis. ActiveMQ "Classic" supports only unshared durable subscriptions: a shared listener on it fails to register, which is logged like any other registration failure.

16 Listener Session Recovery

A listener recovers when the JMS provider reports that the session of one of its consumers died, so that the consumer doesn’t stop receiving messages until the application restarts.

What Triggers Recovery

When the JMS provider reports an error through the ExceptionListener of a listener’s connection, each consumer of that connection checks its session by looking up its destination. Only a consumer whose session fails that check recovers. Errors that leave the session usable don’t recreate anything, such as the rejected asynchronous send that ActiveMQ "Classic" reports while its connection still works.

For example, when the database session of an Oracle AQ listener is disconnected, AQ reports JMS-120: Dequeue failed, and later JMS-112: Connection is invalid on each of its pings. A new session on the same connection still works, so the listener recovers.

What Is Recreated

The consumer closes its session and MessageConsumer and creates them again on the same connection, with the same message selector and subscription. A durable or shared subscription resumes on the same subscription, and the provider redelivers the messages that weren’t acknowledged.

If the session or consumer can’t be created, the consumer tries again after a delay that doubles after each attempt, up to a maximum. It keeps trying until it succeeds or the application shuts down, since the provider may report nothing more once the failed session is closed. io.micronaut.jms.listener.JMSListenerRegistry logs the first attempts as warnings and later ones as errors.

micronaut.jms.listener-recovery.enabled=true
micronaut.jms.listener-recovery.warn-attempts=10
micronaut.jms.listener-recovery.initial-delay=1s
micronaut.jms.listener-recovery.max-delay=1m

micronaut:
  jms:
    listener-recovery:
      enabled: true
      warn-attempts: 10
      initial-delay: 1s
      max-delay: 1m

[micronaut.jms.listener-recovery]
enabled = true
warn-attempts = 10
initial-delay = "1s"
max-delay = "1m"

micronaut {
  jms {
    listenerRecovery {
      enabled = true
      warnAttempts = 10
      initialDelay = '1s'
      maxDelay = '1m'
    }
  }
}

{
  micronaut = {
    jms = {
      listener-recovery = {
        enabled = true
        warn-attempts = 10
        initial-delay = "1s"
        max-delay = "1m"
      }
    }
  }
}

{
  "micronaut": {
    "jms": {
      "listener-recovery": {
        "enabled": true,
        "warn-attempts": 10,
        "initial-delay": "1s",
        "max-delay": "1m"
      }
    }
  }
}

enabled turns recovery on or off.
warn-attempts is the number of failed attempts logged as warnings. Later attempts are logged as errors.
initial-delay is the delay after the first failed attempt. It must be positive.
max-delay is the longest delay between two attempts. It must not be shorter than initial-delay.

An invalid value fails the registration of the listeners with a ConfigurationException. Only the first attempt logged as an error includes the stack trace, unless debug logging is enabled for io.micronaut.jms.listener.JMSListenerRegistry.

Limits

A non-durable subscription loses the messages published while its consumer is recreated.
A connection that fails as a whole isn’t replaced. For example, when an ActiveMQ broker restarts without a failover transport, only the session is recreated, on the same connection.
A consumer that is still recovering doesn’t change the jms health indicator.

17 GraalVM support

Micronaut JMS is compatible with GraalVM. Everything is handled automatically by the library so users don’t need any special configuration.

Note	See the section on GraalVM in the user guide for more information.

18 Repository

You can find the source code of this project in this repository:

https://github.com/micronaut-projects/micronaut-jms

JMS

1 Introduction

2 Release History

3 Breaking Changes

5.0.0

4.0.0

Removal of javax.jms namespace

ActiveMQ Classic

ActiveMQ Artemis

Deprecations

4 JMS Quick Start

Creating a JMS Producer with @JMSProducer

Creating a JMS Consumer with @JMSListener

5 Configuration

5.1 Configuring ActiveMQ "Classic"

5.2 Configuring ActiveMQ Artemis

5.3 Configuring SQS

5.4 Configuring Unsupported Brokers

5.5 Customizing Brokers

Customizing the ConnectionFactory

Replacing the ConnectionFactory

6 Parameter Binding

@MessageBody

@MessageHeader

Session

7 Serializers and Deserializers

8 Error Handlers

9 Success Handlers

10 Message Selectors

11 Delivery Mode and Priority

12 Asynchronous Producers and Producer Sessions

Asynchronous producer methods

Producer sessions

13 Listener Concurrency and Transacted Sends

14 Health, Metrics and Tracing

Health

Metrics

Tracing

Custom observers

15 Durable and Shared Topic Subscriptions

Client IDs

16 Listener Session Recovery

What Triggers Recovery

What Is Recreated

Limits

17 GraalVM support

18 Repository

Removal of `javax.jms` namespace