WorkQueueProcessor.java

package com.redhat.demo.msg.beans;

import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.ejb.TransactionAttribute;
import javax.ejb.TransactionAttributeType;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.TextMessage;
import javax.jms.XAConnection;
import javax.jms.XAConnectionFactory;
import javax.jms.XASession;
import javax.naming.InitialContext;
import javax.transaction.TransactionManager;

@MessageDriven(activationConfig = {
		@ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
		@ActivationConfigProperty(propertyName = "destination", propertyValue = "/queue/WorkQueue") }, mappedName = "/queue/WorkQueue")
public class WorkQueueProcessor implements MessageListener {

	@TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
	public void onMessage(Message msg) {
		try {
			String text = ((TextMessage)msg).getText();
			System.out.println("Processing message : " + text);
			InitialContext ic = new InitialContext();
			XAConnectionFactory cf = (XAConnectionFactory)ic.lookup("XAConnectionFactory");
			TransactionManager tm = (TransactionManager)ic.lookup("java:/TransactionManager");
			Queue queue = (Queue)ic.lookup("/queue/SecondQueue");
			XAConnection conn = cf.createXAConnection();
			conn.start();
			XASession sess = conn.createXASession();
			tm.getTransaction().enlistResource(sess.getXAResource());
			MessageProducer prod = sess.createProducer(queue);
			prod.send(sess.createTextMessage("Second message : " + System.currentTimeMillis()));
			System.out.println("Finished sleeping.");
//			sess.close();
//			conn.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}