2

I am trying to create some Spring Boot code that uses DefaultMessageListenerContainer to listen to messages from IBM MQ.

I can create the MQQueueConnectionFactory and send and receive messages using JmsTemplate but this is intended to by high throughput and want use a listener instead of polling.

I have consolidate much of the code down to a Component so I hope I have all that is relevant.

If I schedule the receiveMessage method, it picks up queued messages so I know the sendMessage method is queuing messages.

@Component
class AllInOneTest {

    private MessagingConfiguration.QueueConfig config;
    private MQQueueConnectionFactory           connectionFactory;
    private JmsTemplate                        jmsTemplate;
    private DefaultMessageListenerContainer    listenerContainer;
    private final Logger                       logger = LoggerFactory.getLogger(getClass());

    public AllInOneTest(MessagingManager manager) throws JMSException {
        String detailsName = "default";
        config = manager.getMessagingDetails(detailsName).getConfig();

        logger.debug("AllInOneTest Initializing Connection Factory: {}", detailsName);
        connectionFactory = new MQQueueConnectionFactory();
        connectionFactory.setHostName(config.getHost());
        connectionFactory.setPort(config.getPort());
        connectionFactory.setTransportType(config.getTransportType());
        connectionFactory.setQueueManager(config.getQueueManager());
        connectionFactory.setChannel(config.getChannel());

        logger.debug("AllInOneTest Initializing Message Listener: {}", detailsName);
        DefaultMessageListenerContainer defaultListener = new DefaultMessageListenerContainer();
        defaultListener.setConnectionFactory(connectionFactory);

        defaultListener.setExceptionListener((ee) -> {
            logger.warn(String.format("AllInOneTest Message Listener Error: %s", detailsName), ee);
        });

        defaultListener.setDestinationResolver((session, name, pubSub) -> {
            Destination ret = session.createQueue(name);

            logger.debug("AllInOneTest Created Listener Destination: {}", ret);

            return ret;
        });

        defaultListener.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                logger.info("AllInOneTest Listening For Message: {}", message);
            }
        });

        // TODO Configure subscription.
        // defaultListener.setSubscriptionDurable(true);
        // defaultListener.setSubscriptionName("masher-service");

        // TODO Configure concurrency.
        // defaultListener.setConcurrency(config.getConcurrency());

        // TODO Configure transaction.
        // defaultListener.setSessionTransacted(config.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE);

        listenerContainer = defaultListener;

        logger.debug("AllInOneTest Initializing JMS Template: {}", detailsName);
        jmsTemplate = new JmsTemplate(connectionFactory);
        jmsTemplate.setMessageConverter(new SpringToJMSMessageConverter());
        jmsTemplate.setReceiveTimeout(1000L);
        jmsTemplate.setDefaultDestinationName(config.getOutputQueue());
        jmsTemplate.setDestinationResolver((session, name, pubSub) -> {
            Destination ret = session.createQueue(name);

            logger.debug("AllInOneTest Created JMS Template Destination: {}", ret);

            return ret;
        });

        listenerContainer.setDestinationName(config.getOutputQueue());

        logger.debug("AllInOneTest Starting Message Listener: {} on {}", detailsName, config.getOutputQueue());
        listenerContainer.start();
    }

    // @Scheduled(fixedRate = 500L)
    public void receiveMessage() {
        Object message = jmsTemplate.receiveAndConvert();
        if (message != null) {
            logger.info("AllInOneTest Received: {}", message);
        }
    }

    @Scheduled(fixedRate = 1500L)
    public void sendMessage() {
        int count = counter.incrementAndGet();
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(String.format("JMS Masher Message %d %s %s", count,
                new SimpleDateFormat("HH:mm:ss.SSS").format(new Date()), UUID.randomUUID().toString())).build();

        logger.info("AllInOneTest Sending: {} [{}]", message.getPayload(), message.getHeaders());

        jmsTemplate.convertAndSend(config.getInputQueue(), message);
    }

}

I am calling DefaultMessageListenerContainer.start() but I get the sense it is not "starting" and I must be missing something.

The DestinationResolver is called for the JmsTemplate but not the DefaultMessageListenerContainer.

I don't see any exceptions in the console.

Thanks for any help, Wes.

1 Answer 1

1

DefaultMessageListenerContainer defaultListener = new DefaultMessageListenerContainer();

When you create the container programmatically, rather than having Spring manage it as a @Bean, you have to call afterPropertiesSet() (after you have set all the properties, and before you start() it).

This is true for many Spring components. It's generally better to let Spring manage them.

2
  • Ahhh. I had see that in some unrelated samples. I have a @Bean and @JmsListener implementation working but I may have a requirement that I need to configure one or more listeners based on configuration. I am using Spring so I don't need to write procedural code but requirements don't always behave. Thanks.
    – Wes
    Jun 15, 2017 at 15:22
  • 1
    That's fine, understood; but whenever instantiating Spring-provided components yourself, you should always check if they implement InitializingBean and, if they do, call afterPropertiesSet(). Jun 15, 2017 at 15:27

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.