Exception while trying to consume a JMS Topic message using Spring Boot

Question

I am trying to consume message from ActiveMQ Topic. Below is the Code:

@Configuration
@EnableJms
public class Config {

@Value("${activemq.broker-url}")
private String brokerURL;

@Bean
public ActiveMQConnectionFactory activeMQConnectionFactory() {
    ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
    activeMQConnectionFactory.setBrokerURL(brokerURL);
    activeMQConnectionFactory.setTrustAllPackages(true);
    return activeMQConnectionFactory;
}

public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(){
    DefaultJmsListenerContainerFactory defaultJmsListenerContainerFactory = new DefaultJmsListenerContainerFactory();
    defaultJmsListenerContainerFactory.setConnectionFactory(activeMQConnectionFactory());
    defaultJmsListenerContainerFactory.setPubSubDomain(true);
    return defaultJmsListenerContainerFactory;
}

My Consumer Code:

@Component
@EnableJms
public class Consumer {

   @JmsListener(destination = "xml.inbound.topic", containerFactory = "jmsListenerContainerFactory")
   public void Processor(final Message xmlMessage) {

   }
}

The Exception which I am getting is:

Unsatisfied dependency expressed through field 'jmsTemplate'; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'jmsTemplate' defined in class path resource [com/investmentbank/equityfeedsprocessingrevised/config/Config.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.jms.core.JmsTemplate]: Factory method 'jmsTemplate' threw exception; nested exception is java.lang.ClassCastException: class org.springframework.jms.config.DefaultJmsListenerContainerFactory cannot be cast to class javax.jms.ConnectionFactory (org.springframework.jms.config.DefaultJmsListenerContainerFactory and javax.jms.ConnectionFactory are in unnamed module of loader 'app')

What am I doing wrong here ? Why the exception?

I have also tried :

@Bean
public DefaultMessageListenerContainer jmsListenerContainerFactory() {
        DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
        dmlc.setConnectionFactory(activeMQConnectionFactory());
        dmlc.setPubSubDomain(true);
        return dmlc;
}

The Exception which I get here is:

Unsatisfied dependency expressed through field 'jmsTemplate'; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'jmsTemplate' defined in class path resource [com/investmentbank/equityfeedsprocessingrevised/config/Config.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.jms.core.JmsTemplate]: Factory method 'jmsTemplate' threw exception; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'jmsListenerContainerFactory' defined in class path resource [com/investmentbank/equityfeedsprocessingrevised/config/Config.class]: Invocation of init method failed; nested exception is java.lang.IllegalArgumentException: Property 'destination' or 'destinationName' is required

Just to add my JmsTemplate code looks like this:

@Bean
public JmsTemplate jmsTemplate() {
JmsTemplate jmsTemplate = new JmsTemplate();
         jmsTemplate.setConnectionFactory(activeMQConnectionFactory());
         jmsTemplate.setPubSubDomain(true);
         return jmsTemplate;
}

I have implemented the JMS Topic Publisher using Apache Camel (Code):

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        CamelContext _ctx = new DefaultCamelContext(); 
        _ctx.addComponent("jms", JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));
        _ctx.addRoutes(new RouteBuilder() {

            public void configure() throws Exception {
                from("file:src/main/resources?fileName=data-sample_2.csv")
                .process(new MyTransformRevised1())
                .to("file:src/main/resources/?fileName=emp.xml")                
                .split(body().tokenizeXML("equityFeeds", null)).streaming().to("jms:topic:xml.inbound.topic");
            }

        });

I am able to see the number of messages in "Message Enqueued" column of my Topic name.

What is going wrong ? I have read various posts on net but not able to solve the issue. Kindly help in resolving the issue. I am not able to read the message on ActiveMQ Topic. Let me know if some information is missing from my end.

Answer 1

I solved this with the below Code:

    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(){
    DefaultJmsListenerContainerFactory defaultJmsListenerContainerFactory = new DefaultJmsListenerContainerFactory();
                defaultJmsListenerContainerFactory.setConnectionFactory(activeMQConnectionFactory());
defaultJmsListenerContainerFactory.setPubSubDomain(true);
return defaultJmsListenerContainerFactory;
        }

     @Bean
     public JmsTemplate jmsTemplate() {
        JmsTemplate jmsTemplate = new JmsTemplate();
        jmsTemplate.setConnectionFactory(activeMQConnectionFactory());
        jmsTemplate.setPubSubDomain(true);
        return jmsTemplate;
    }

    @JmsListener(destination = "${my.inboundTopicName}", containerFactory = "jmsListenerContainerFactory")
    public void myProcessor(final Message xmlMessage) {
    ///

    }

I want to add that I am experiencing some unusual behavior.

1. At the time when the messages are generated from the producer and if the consumer is up then only the messages are consumed by the consumer. I mean that if the messages are produced by the producer and enqueued by ActiveMQ and after some time I turn on the consumer then the messages are not consumed by the consumer.

2. . Also it takes a while before the messages are consumed by the consumer.

Answer 2

you could try to wrap the activeMQConnectionFactory in a CachingConnectionFactory and utilize the DefaultJmsListenerContainerFactoryConfigurer to configure the JmsListenerContainerFactory:

@Bean
ConnectionFactory connectionFactory() {
    return new CachingConnectionFactory(activeMQConnectionFactory());
}

@Bean
public JmsListenerContainerFactory<?> jmsListenerContainerFactory(ConnectionFactory connectionFactory,
                                    DefaultJmsListenerContainerFactoryConfigurer configurer) {

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

EDIT start:

could you try changing the JmsTemplate to:

@Bean
public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
JmsTemplate jmsTemplate = new JmsTemplate();
         jmsTemplate.setConnectionFactory(connectionFactory);
         jmsTemplate.setPubSubDomain(true);
         return jmsTemplate;
}