JMS Consumer to Synchronously receive Message from Remote Producer using Glassfish and OpenMQ

Question

I am using Glassfish and OpenMQ for a remote Consumer instance to look up a message Producer's queue and process requests synchronously.

According to McIntosh's answer on Synchronous Consumer with JMS Queue, synchronous message receiving can be handled via scheduling. I plan on doing this, but I have only seen examples of connecting to a message queue by way of an asynchronous Message Driven Bean (MDB), as shown below:

import javax.jms.MessageListener;
import javax.annotation.Resource;
import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.ejb.MessageDrivenContext;

@MessageDriven(activationConfig = {
@ActivationConfigProperty(propertyName = "addressList", propertyValue = "mq://localhost:54020/"), //found in Producer server's domain.xml as JMS_PROVIDER_PORT
@ActivationConfigProperty(propertyName = "destinationLookup", propertyValue = "jms/ProducerRequestMessageQueue"),
@ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue")})

public class ConsumerNode extends Node implements MessageListener  {

    @Resource
    private MessageDrivenContext _mdc;

    public ConsumerNode() {
        super();
    }

    @Override
    public void onMessage(Message message) {
        //process message...
    }
}

How do I connect to the remote Producer queue without implementing MessageListener and setting up as a Message Driven Bean?

Answer 1

Finally figured it out. I was needing the imq.jar mq library located at:

%GLASSFISH_HOME%/mq/lib/

The following code answers my question with core pieces being the init() and onMessage() methods:

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.ejb.Singleton;
import javax.ejb.Schedule;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.ObjectMessage;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;
import javax.jms.Connection;
import javax.jms.Queue;

//from imq.jar
import com.sun.messaging.ConnectionConfiguration;
import com.sun.messaging.ConnectionFactory;

@Singleton
public class SyncNode {
    private ConnectionFactory _producerRequestFactory;
    private Connection _connection;
    private Session _session;
    private Queue _producerRequestMessageQueue;
    private MessageConsumer _consumer;

    @PostConstruct
    void init() {
        try {
            _producerRequestFactory = new ConnectionFactory();
            _producerRequestFactory.setProperty(ConnectionConfiguration.imqBrokerHostName, "localhost");
            _producerRequestFactory.setProperty(ConnectionConfiguration.imqBrokerHostPort, "56527"); //56527 is JMS_PROVIDER_PORT found in producer's domain.xml in domain config directory
            _connection = _producerRequestFactory.createConnection();
            _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            _producerRequestMessageQueue = _session.createQueue("ProducerRequestMessageQueue"); //name of the queue that the producer sends messages to.
            _consumer = _session.createConsumer(_producerRequestMessageQueue);
            _connection.start();
        } catch (JMSException ex) {
            //handle exception
        }
    }

    @PreDestroy
    void cleanup() {
        try {
            _consumer.close();
            _session.close();
            _connection.close();
        } catch (JMSException ex) {
            //handle exception
        }
    }

    @Schedule(hour = "*", minute = "*", second = "*/10", persistent = false)
    public void onMessage() {
        try {
            _connection.start(); 
            Message message = _consumer.receive();
            //handle message
        } catch (JMSException ex) {
           //handle exception
        }
    }
}

Two code examples helped me:

1. This answer taught me the proper annotations to escape using the main method.
2. This tutorial helped make things click on how to connect to another server, or port in my case.