JMS producer consumer example

[s ravi chandran]

Hi,
I am trying to create a simple JMS producer consumer using solace messaging. I guess the API is not very famous, but I am trying a generic implementation. I am trying to create two endpoints, which will listen on one topic and send message on another. The issue comes with keeping the consumer alive to listen. I know I am missing something very basic here. Here is the code:

public class MessageService implements MessageListener{

    private SolConnectionFactoryImpl connFactory;
    private Connection conn;
    protected MessageProducer producer;
    protected MessageConsumer consumer;
    protected Session session;
    private String senderTopic;
    private String receiverTopic;
    private static int delay = 1000;

    public MessageService(Map<String, String> properties) throws Exception {

        connFactory = new SolConnectionFactoryImpl(new JMSProperties(new Hashtable()));
        connFactory.setHost(properties.get("host"));
        connFactory.setPort(Integer.valueOf(properties.get("port")));
        this.senderTopic = properties.get("senderqueue");
        this.receiverTopic = properties.get("receiverqueue");
        init();
    }

    private void init() throws JMSException {
        conn = connFactory.createConnection();
        session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Topic sendertopic = session.createTopic(senderTopic);
        Topic receivertopic = session.createTopic(receiverTopic);
        consumer = session.createConsumer(receivertopic);
        consumer.setMessageListener(this);
        producer = session.createProducer(sendertopic);

        //start the connection
        conn.start();

    }

    public void produce() {
        try {
            String something = "Something " + System.currentTimeMillis();
            System.out.println("producing  "+something);
            TextMessage txtMsg = session.createTextMessage();
            txtMsg.setText(something);
            producer.send(txtMsg);

        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onMessage(Message message) {

        if (!(message instanceof TextMessage)) {
            return;
        }
        TextMessage textMessage = (TextMessage) message;
        try {
            System.out.println("Consumer Received: " + textMessage.getText());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

Would someone please point out what is it that I am doing wrong here?

[Tim Moores]

What is "the issue"? How are you using this code?

You might also say a bit about "SolConnectionFactoryImpl", which is not a standard Java or JMS class.

[s ravi chandran]

Tim Moores wrote:What is "the issue"? How are you using this code?

You might also say a bit about "SolConnectionFactoryImpl", which is not a standard Java or JMS class.

This is sample code. If this works, I will develop the actual class using this as base. SolConnectionFactoryImpl is JMS implementation from Solace Message Service. The issue, is once I start the connection, the program exits. For an async listener to get message, the consumer should remain active. I am not getting how to do so.

[Paul Clapham]

You say "the program exits" but I don't see any program in your post. I assume that somewhere you have some code which creates a MessageService object and uses it as a MessageListener, so if you're asking why that exits immediately then it would be helpful for us to see its code.

[s ravi chandran]

Sorry for the missing part. It was a main method with these statements :

public static void main(String args[] )
{
Map<String, String> jmsProperties = new HashMap<String, String>();
jmsproperties.put("host", "www.host.com");
jmsproperties.put("port", "6000");
jmsproperties.put("senderqueue", "Channel1");
jmsproperties.put("receiverqueue", "Channel2");


new MessageService(jmsProperties);

}

This is the code that was at the end of the class.

[zlatan rahim]

In Jdev, now create a simple producer and consumer.

a) Create a new Application and Project.

b) Create a simple producer class that will send 5 messages to the queue.

package project1;

import javax.jms.*;
import javax.naming.*;
import java.util.Hashtable;

public class QueueProducer {

public static void main(String[] args) {
String queueName = "jms/MyFirstQueue";
String queueConnectionFactoryName = "jms/MyFirstQCF";
Context jndiContext = null;
QueueConnectionFactory queueConnectionFactory = null;
QueueConnection queueConnection = null;
QueueSession queueSession = null;
Queue queue = null;
QueueSender queueSender = null;
TextMessage message = null;
int noMessages = 5;

/*
* Set the environment for a connection to the OC4J instance
*/
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, "oracle.j2ee.rmi.RMIInitialContextFactory");
env.put(Context.SECURITY_PRINCIPAL, "oc4jadmin");
env.put(Context.SECURITY_CREDENTIALS, "welcome1");
env.put(Context.PROVIDER_URL, "opmn:ormi://localhost:6007:JMS/default");

/*
* Set the Context Object.
* Lookup the Queue Connection Factory.
* Lookup the JMS Destination.
*/
try {
jndiContext = new InitialContext(env);
queueConnectionFactory = (QueueConnectionFactory)
jndiContext.lookup(queueConnectionFactoryName);
queue = (Queue) jndiContext.lookup(queueName);
} catch (NamingException e) {
System.out.println("JNDI lookup failed: " +
e.toString());
System.exit(1);
}

/*
* Create connection.
* Create session from connection.
* Create sender.
* Create text message.
* Send messages.
* Send non text message to end text messages.
* Close connection.
*/
try {
queueConnection =
queueConnectionFactory.createQueueConnection();
queueSession =
queueConnection.createQueueSession(false,
Session.AUTO_ACKNOWLEDGE);
queueSender = queueSession.createSender(queue);
message = queueSession.createTextMessage();
for (int i = 0; i < noMessages; i++) {
message.setText("Message " + (i + 1));
System.out.println("Producing message: " +
message.getText());
queueSender.send(message);
}
queueSender.send(queueSession.createBytesMessage());;
} catch (JMSException e) {
System.out.println("Exception occurred: " +
e.toString());
} finally {
if (queueConnection != null) {
try {
queueConnection.close();
} catch (JMSException e) {
System.out.println("Closing error: " + e.toString());
}
}
}
}
}

c) Create a simple consumer class that will read all text messages from the queue and end once it reads a non text message:

package project1;

import java.util.Hashtable;

import javax.jms.*;

import javax.naming.*;

public class QueueConsumer {

public static void main(String[] args) {
String queueName = "jms/MyFirstQueue";
String queueConnectionFactoryName = "jms/MyFirstQCF";
Context jndiContext = null;
QueueConnectionFactory queueConnectionFactory = null;
QueueConnection queueConnection = null;
QueueSession queueSession = null;
Queue queue = null;
QueueReceiver queueReceiver = null;
TextMessage message = null;

/*
* Set the environment for a connection to the OC4J instance
*/
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY,
"oracle.j2ee.rmi.RMIInitialContextFactory");
env.put(Context.SECURITY_PRINCIPAL, "oc4jadmin");
env.put(Context.SECURITY_CREDENTIALS, "welcome1");
env.put(Context.PROVIDER_URL,
"opmn:ormi://localhost:6007:JMS/default");

/*
* Set the Context Object.
* Lookup the Queue Connection Factory.
* Lookup the JMS Destination.
*/
try {
jndiContext = new InitialContext(env);
queueConnectionFactory =
(QueueConnectionFactory)jndiContext.lookup(queueConnectionFactoryName);
queue = (Queue)jndiContext.lookup(queueName);
} catch (NamingException e) {
System.out.println("JNDI lookup failed: " + e.toString());
System.exit(1);
}

/*
* Create connection.
* Create session from connection.
* Create receiver.
* Receive all text messages from queue until non text message is receivied
* Close connection.
*/
try {
queueConnection = queueConnectionFactory.createQueueConnection();
queueSession =
queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
queueReceiver = queueSession.createReceiver(queue);
queueConnection.start();
while (true) {
Message m = queueReceiver.receive(1);
if (m != null) {
if (m instanceof TextMessage) {
message = (TextMessage)m;
System.out.println("Consuming message: " +
message.getText());
} else {
break;
}
}
}
} catch (JMSException e) {
System.out.println("Exception occurred: " + e.toString());
} finally {
if (queueConnection != null) {
try {
queueConnection.close();
} catch (JMSException e) {
System.out.println("Closing error: " + e.toString());
}
}
}
}
}

d) Add the following libraries to the project:
* J2EE
* Apache Ant

5) Run the QueueProducer class to send the message to the queue.

The is result:
Producing message: Message 1
Producing message: Message 2
Producing message: Message 3
Producing message: Message 4
Producing message: Message 5

6) Run the QueueConsumer class to receive the message from the queue.

This is the result:

Consuming message: Message 1
Consuming message: Message 2
Consuming message: Message 3
Consuming message: Message 4
Consuming message: Message 5

[s ravi chandran]

Thanks for the sample code. Here in the code, you are using a sync consumer. My requirement is actually an async consumer. Hence the use of a message listener. So, in place of using

Message m = queueReceiver.receive(1);

, i have used

 consumer.setMessageListener(this);

.