Spring Boot ActiveMQ Starter

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-activemq</artifactId>
  <version>3.5.3</version>
</dependency>

implementation 'org.springframework.boot:spring-boot-starter-activemq:3.5.3'

import org.springframework.boot.autoconfigure.jms.activemq.ActiveMQProperties;
import org.springframework.boot.autoconfigure.jms.activemq.ActiveMQConnectionFactoryCustomizer;
import org.springframework.boot.autoconfigure.jms.activemq.ActiveMQConnectionDetails;
import org.springframework.jms.core.JmsTemplate;
import jakarta.jms.ConnectionFactory;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

@SpringBootApplication
public class ActiveMQApplication {
    public static void main(String[] args) {
        SpringApplication.run(ActiveMQApplication.class, args);
    }
}

@Component
public class MessageService {
    private final JmsTemplate jmsTemplate;
    
    public MessageService(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }
    
    public void sendMessage(String destination, String message) {
        jmsTemplate.convertAndSend(destination, message);
    }
    
    public String receiveMessage(String destination) {
        return (String) jmsTemplate.receiveAndConvert(destination);
    }
}

# Basic connection settings
spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=secret

# Connection behavior
spring.activemq.close-timeout=15s
spring.activemq.send-timeout=0ms
spring.activemq.non-blocking-redelivery=false

# Embedded broker settings
spring.activemq.embedded.enabled=true

# Connection pooling
spring.activemq.pool.enabled=false
spring.activemq.pool.max-connections=1
spring.activemq.pool.max-sessions-per-connection=500
spring.activemq.pool.idle-timeout=30s

# Package trust settings
spring.activemq.packages.trust-all=false
spring.activemq.packages.trusted=com.example.messages

@ConfigurationProperties("spring.activemq")
public class ActiveMQProperties {
    
    /** URL of the ActiveMQ broker. Auto-generated by default. */
    public String getBrokerUrl();
    public void setBrokerUrl(String brokerUrl);
    
    /** Login user of the broker. */
    public String getUser();
    public void setUser(String user);
    
    /** Login password of the broker. */
    public String getPassword();
    public void setPassword(String password);
    
    /** Time to wait before considering a close complete. Default: 15s */
    public Duration getCloseTimeout();
    public void setCloseTimeout(Duration closeTimeout);
    
    /** Whether to stop message delivery before re-delivering messages from rolled back transaction. Default: false */
    public boolean isNonBlockingRedelivery();
    public void setNonBlockingRedelivery(boolean nonBlockingRedelivery);
    
    /** Time to wait on message sends for a response. Set to 0 to wait forever. Default: 0ms */
    public Duration getSendTimeout();
    public void setSendTimeout(Duration sendTimeout);
    
    /** Embedded broker configuration */
    public Embedded getEmbedded();
    
    /** Connection pool configuration */
    public JmsPoolConnectionFactoryProperties getPool();
    
    /** Package trust configuration */
    public Packages getPackages();
    
    /** Determines effective broker URL based on configuration and embedded settings */
    String determineBrokerUrl();
    
    /** Configuration for embedded ActiveMQ broker */
    public static class Embedded {
        /** Whether to enable embedded mode if ActiveMQ Broker is available. Default: true */
        public boolean isEnabled();
        public void setEnabled(boolean enabled);
    }
    
    /** Configuration for package trust settings */
    public static class Packages {
        /** Whether to trust all packages */
        public Boolean getTrustAll();
        public void setTrustAll(Boolean trustAll);
        
        /** List of specific packages to trust when not trusting all packages */
        public List<String> getTrusted();
        public void setTrusted(List<String> trusted);
    }
}

public interface ActiveMQConnectionDetails extends ConnectionDetails {
    
    /** Broker URL to use */
    String getBrokerUrl();
    
    /** Login user to authenticate to the broker, or null */
    String getUser();
    
    /** Login password to authenticate against the broker, or null */
    String getPassword();
}

@FunctionalInterface
public interface ActiveMQConnectionFactoryCustomizer {
    
    /**
     * Customize the ActiveMQConnectionFactory
     * @param factory the factory to customize
     */
    void customize(ActiveMQConnectionFactory factory);
}

@AutoConfiguration(before = JmsAutoConfiguration.class, after = JndiConnectionFactoryAutoConfiguration.class)
@ConditionalOnClass({ConnectionFactory.class, ActiveMQConnectionFactory.class})
@ConditionalOnMissingBean(ConnectionFactory.class)
@EnableConfigurationProperties({ActiveMQProperties.class, JmsProperties.class})
@Import({ActiveMQXAConnectionFactoryConfiguration.class, ActiveMQConnectionFactoryConfiguration.class})
public class ActiveMQAutoConfiguration {
    
    /** Creates ActiveMQConnectionDetails bean from properties */
    @Bean
    @ConditionalOnMissingBean
    ActiveMQConnectionDetails activemqConnectionDetails(ActiveMQProperties properties);
}

@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingBean(ConnectionFactory.class)
class ActiveMQConnectionFactoryConfiguration {
    
    /** Simple connection factory configuration (default) */
    static class SimpleConnectionFactoryConfiguration {
        
        /** Creates non-cached ActiveMQConnectionFactory */
        @Bean
        @ConditionalOnBooleanProperty(name = "spring.jms.cache.enabled", havingValue = false)
        ActiveMQConnectionFactory jmsConnectionFactory(
            ActiveMQProperties properties,
            ObjectProvider<ActiveMQConnectionFactoryCustomizer> factoryCustomizers,
            ActiveMQConnectionDetails connectionDetails
        );
        
        /** Cached connection factory configuration */
        static class CachingConnectionFactoryConfiguration {
            
            /** Creates cached ActiveMQConnectionFactory */
            @Bean
            CachingConnectionFactory jmsConnectionFactory(
                JmsProperties jmsProperties,
                ActiveMQProperties properties,
                ObjectProvider<ActiveMQConnectionFactoryCustomizer> factoryCustomizers,
                ActiveMQConnectionDetails connectionDetails
            );
        }
    }
    
    /** Pooled connection factory configuration */
    static class PooledConnectionFactoryConfiguration {
        
        /** Creates pooled JmsConnectionFactory */
        @Bean(destroyMethod = "stop")
        @ConditionalOnBooleanProperty("spring.activemq.pool.enabled")
        JmsPoolConnectionFactory jmsConnectionFactory(
            ActiveMQProperties properties,
            ObjectProvider<ActiveMQConnectionFactoryCustomizer> factoryCustomizers,
            ActiveMQConnectionDetails connectionDetails
        );
    }
}

@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(TransactionManager.class)
@ConditionalOnBean(XAConnectionFactoryWrapper.class)
@ConditionalOnMissingBean(ConnectionFactory.class)
class ActiveMQXAConnectionFactoryConfiguration {
    
    /** Creates XA-capable JMS connection factory */
    @Primary
    @Bean(name = {"jmsConnectionFactory", "xaJmsConnectionFactory"})
    ConnectionFactory jmsConnectionFactory(
        ActiveMQProperties properties,
        ObjectProvider<ActiveMQConnectionFactoryCustomizer> factoryCustomizers,
        XAConnectionFactoryWrapper wrapper,
        ActiveMQConnectionDetails connectionDetails
    ) throws Exception;
    
    /** Creates non-XA connection factory for non-transactional operations */
    @Bean
    @ConditionalOnBooleanProperty(name = "spring.activemq.pool.enabled", havingValue = false, matchIfMissing = true)
    ActiveMQConnectionFactory nonXaJmsConnectionFactory(
        ActiveMQProperties properties,
        ObjectProvider<ActiveMQConnectionFactoryCustomizer> factoryCustomizers,
        ActiveMQConnectionDetails connectionDetails
    );
}

class ActiveMQConnectionFactoryConfigurer {
    
    /**
     * Create a configurer instance
     * @param properties the ActiveMQ properties
     * @param factoryCustomizers list of connection factory customizers
     */
    ActiveMQConnectionFactoryConfigurer(
        ActiveMQProperties properties,
        List<ActiveMQConnectionFactoryCustomizer> factoryCustomizers
    );
    
    /**
     * Configure the given ActiveMQ connection factory
     * @param factory the factory to configure
     */
    void configure(ActiveMQConnectionFactory factory);
}

public class JmsPoolConnectionFactoryFactory {
    
    /**
     * Create a factory instance
     * @param properties the pool connection factory properties
     */
    public JmsPoolConnectionFactoryFactory(JmsPoolConnectionFactoryProperties properties);
    
    /**
     * Create a pooled connection factory based on the specified connection factory
     * @param connectionFactory the connection factory to wrap
     * @return a pooled connection factory
     */
    public JmsPoolConnectionFactory createPooledConnectionFactory(ConnectionFactory connectionFactory);
}

@Configuration
public class ActiveMQConfig {
    
    @Bean
    public ActiveMQConnectionFactoryCustomizer connectionFactoryCustomizer() {
        return factory -> {
            factory.setTrustedPackages(Arrays.asList("com.example.messages"));
            factory.setCloseTimeout(5000);
            factory.setOptimizeAcknowledge(true);
        };
    }
}

@Component
public class OrderService {
    
    private final JmsTemplate jmsTemplate;
    
    public OrderService(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }
    
    public void processOrder(Order order) {
        // Send order to processing queue
        jmsTemplate.convertAndSend("orders.processing", order);
    }
    
    @JmsListener(destination = "orders.completed")
    public void handleCompletedOrder(Order order) {
        // Handle completed order
        System.out.println("Order completed: " + order.getId());
    }
}

spring.activemq.pool.enabled=true
spring.activemq.pool.max-connections=10
spring.activemq.pool.max-sessions-per-connection=100
spring.activemq.pool.idle-timeout=30s

spring.activemq.embedded.enabled=true
# Embedded broker uses vm://localhost by default

spring.activemq.broker-url=tcp://activemq.production.com:61616
spring.activemq.user=myapp
spring.activemq.password=${ACTIVEMQ_PASSWORD}
spring.activemq.embedded.enabled=false

# Connection pooling for production
spring.activemq.pool.enabled=true
spring.activemq.pool.max-connections=20

/** Standard Java types */
import java.time.Duration;
import java.util.List;
import jakarta.jms.Connection;
import jakarta.jms.JMSException;
import jakarta.jms.XAConnection;
import jakarta.jms.XAConnectionFactory;
import jakarta.transaction.TransactionManager;
import org.springframework.boot.autoconfigure.service.connection.ConnectionDetails;
import org.springframework.boot.jms.XAConnectionFactoryWrapper;

/** Standard Jakarta JMS types */
interface ConnectionFactory {
    Connection createConnection() throws JMSException;
    Connection createConnection(String userName, String password) throws JMSException;
}

/** Apache ActiveMQ connection factory */
class ActiveMQConnectionFactory implements ConnectionFactory {
    public ActiveMQConnectionFactory();
    public ActiveMQConnectionFactory(String brokerURL);
    public ActiveMQConnectionFactory(String userName, String password, String brokerURL);
    
    public void setBrokerURL(String brokerURL);
    public void setUserName(String userName);
    public void setPassword(String password);
    public void setCloseTimeout(int closeTimeout);
    public void setNonBlockingRedelivery(boolean nonBlockingRedelivery);
    public void setSendTimeout(int sendTimeout);
    public void setTrustedPackages(List<String> trustedPackages);
    public void setTrustAllPackages(boolean trustAllPackages);
}

/** Apache ActiveMQ XA connection factory */
class ActiveMQXAConnectionFactory extends ActiveMQConnectionFactory implements XAConnectionFactory {
    public XAConnection createXAConnection() throws JMSException;
    public XAConnection createXAConnection(String userName, String password) throws JMSException;
}

/** Spring JMS cached connection factory */
class CachingConnectionFactory implements ConnectionFactory {
    public void setCacheConsumers(boolean cacheConsumers);
    public void setCacheProducers(boolean cacheProducers);
    public void setSessionCacheSize(int sessionCacheSize);
}

/** Pooled JMS connection factory */
class JmsPoolConnectionFactory implements ConnectionFactory {
    public void setMaxConnections(int maxConnections);
    public void setIdleTimeout(Duration idleTimeout);
    public void setMaxSessionsPerConnection(int maxSessionsPerConnection);
    public void stop();
}

/** Configuration properties for connection factory pooling */
class JmsPoolConnectionFactoryProperties {
    /** Whether a JmsPoolConnectionFactory should be created instead of regular ConnectionFactory */
    public boolean isEnabled();
    public void setEnabled(boolean enabled);
    
    /** Whether to block when connection requested and pool is full */
    public boolean isBlockIfFull();
    public void setBlockIfFull(boolean blockIfFull);
    
    /** Blocking period before throwing exception if pool still full */
    public Duration getBlockIfFullTimeout();
    public void setBlockIfFullTimeout(Duration blockIfFullTimeout);
    
    /** Connection idle timeout */
    public Duration getIdleTimeout();
    public void setIdleTimeout(Duration idleTimeout);
    
    /** Maximum number of pooled connections */
    public int getMaxConnections();
    public void setMaxConnections(int maxConnections);
    
    /** Maximum number of pooled sessions per connection */
    public int getMaxSessionsPerConnection();
    public void setMaxSessionsPerConnection(int maxSessionsPerConnection);
    
    /** Time between idle connection eviction thread runs */
    public Duration getTimeBetweenExpirationCheck();
    public void setTimeBetweenExpirationCheck(Duration timeBetweenExpirationCheck);
    
    /** Whether to use only one anonymous MessageProducer instance */
    public boolean isUseAnonymousProducers();
    public void setUseAnonymousProducers(boolean useAnonymousProducers);
}