Jms Transaction and Threads

[Dominic Marinaro Soundranayagam]

Hi Team
I Have a main thread which will poll the messages from the Queue
and create a child thread to process it .The requirement is Both the parent and the child should be in one transaction unit.
Here is my application config.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:amq="http://activemq.apache.org/schema/core" xmlns:jms="http://www.springframework.org/schema/jms"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.2.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd
http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-2.5.xsd">


	<!--
		allows for ${} replacement in the spring xml configuration from the
		system.properties file on the classpath
	-->
	<context:property-placeholder location="classpath:config/jms.properties" />

	<!-- creates an activemq connection factory using the amq namespace -->
	<amq:connectionFactory id="amqConnectionFactory"
		brokerURL="${jms.activeMQ.url}" userName="${jms.activeMQ.username}"
		password="${jms.activeMQ.password}" />
	<!--
		CachingConnectionFactory Definition, sessionCacheSize property is the
		number of sessions to cache
	-->
	<bean id="connectionFactory"
		class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
		<constructor-arg ref="amqConnectionFactory" />
	</bean>

	<!-- JmsTemplate Definition -->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<constructor-arg ref="connectionFactory" />
	</bean>

	<bean id="messageSenderService" class="com.obh.ncs.core.utility.QueueUtility"
		scope="singleton">
		<property name="jmsTemplate" ref="jmsTemplate" />
	</bean>

	<bean id="jmsTransactionManager"
		class="org.springframework.jms.connection.JmsTransactionManager">
		<property name="connectionFactory" ref="connectionFactory" />
	</bean>
	
	<bean id="loader"
		class="com.obh.ncs.core.utility.M1OperatorLoader">
		<constructor-arg ref="genericConnector"></constructor-arg>
		
	</bean>
	
	<bean id="genericConnector"
		class="com.obh.ncs.core.utility.M1GenericConnector">
		<constructor-arg ref="messageSenderService"></constructor-arg>
		<constructor-arg ref="jmsTransactionManager"></constructor-arg>
	</bean>
	
	
</beans>

Code for M1GenericConnector

public class M1GenericConnector implements Runnable {

    private QueueUtility queueConnector;

    private JmsTransactionManager transactionManager;

    public M1GenericConnector(final QueueUtility queueConnector, final JmsTransactionManager manager) {
        this.transactionManager = manager;
        this.queueConnector = queueConnector;
    }

    /**
     * This is main processor method of this service utility. Which will have the following logics 1.Check Whether the
     * current window size is capable of getting new Tasks in order to assign to TaskExecutor. 2.Check Whether the cron
     * Job scheduler is triggering this Service. 3.Retrive the message from the QueueMessage Utility. 4.Construct the
     * New Task Thread
     * 
     */
    @Override
    public void run() {
        while (true) {

            DefaultTransactionDefinition def = new DefaultTransactionDefinition();

            def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);

            TransactionStatus status = transactionManager.getTransaction(def);

            Integer queueValue = null;
            try {

                Message value = (Message) queueConnector.retrive("TestQueue");
                queueValue = (Integer) value.getObjectProperty("DEST_ID");
                System.out.println("Message Retrived" + queueValue);

                // If queueValue is Null means there are no messages
                // in
                // the destination

                ConnectorTask connectorTask = new ConnectorTask(Integer.valueOf(queueValue), transactionManager);

                Thread thread = new Thread(connectorTask);
                thread.start();
                // System.out.println("Picking Up " + queueValue + " Time " + System.currentTimeMillis());

            } catch (JMSException e) {
                throw new RuntimeException();
            } catch (Exception e) {
                transactionManager.rollback(status);
                System.out.println("M1 Generic Connector Roll Back");
            }

        }

    }

}

Code For ConnectorTask

public class ConnectorTask implements Runnable {

    private Integer valueOf;
    // private TransactionStatus status;
    private JmsTransactionManager transactionManager;

    public ConnectorTask(Integer valueOf, JmsTransactionManager transactionManager) {
        this.valueOf = valueOf;
        // this.status = status;
        this.transactionManager = transactionManager;
    }

    @Override
    public void run() {
        // TODO Auto-generated method stub
        DefaultTransactionDefinition def = new DefaultTransactionDefinition();
        Integer item = System.identityHashCode(def);
        // explicitly setting the transaction name is something that can only be done programmatically
        // def.setName("SomeTxName78" + item);

        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);

        TransactionStatus status = transactionManager.getTransaction(def);

        try {
            if (valueOf % 2 == 1) {
                throw new RuntimeException();
            }
        } catch (Exception ex) {
            transactionManager.rollback(status);
            // System.out.println("Connector Task Roll Back" + status.toString());
        }
        transactionManager.commit(status);
        System.out.println("Connector Task Commited Sucessflly" + status.toString());

    }

}

Code for QueueUtility
[CODE]

public class QueueUtility {

private JmsTemplate jmsTemplate;

public JmsTemplate getJmsTemplate() {
return jmsTemplate;
}

public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
this.jmsTemplate.setReceiveTimeout(1500);
}

public Message retrive(final String destinationName){
return jmsTemplate.receive(destinationName);
}

public void sendMessage(final String destinationName ,final String message){
jmsTemplate.convertAndSend(destinationName, message);
}

public void sendMessage(final String destinationName ,final Map<String,Object> message){
jmsTemplate.send(destinationName, new MessageCreator() {

@Override
public Message createMessage(Session arg0) throws JMSException {
// TODO Auto-generated method stub
MapMessage mapmessage = arg0.createMapMessage();
for (Map.Entry<String, Object> entry : message.entrySet()) {
mapmessage.setObjectProperty(entry.getKey(), entry.getValue());
}
return mapmessage;
}
});
}

}
[/CODE]

Code for M1OperatorLoader

public class M1OperatorLoader {

    public M1OperatorLoader(final M1GenericConnector genericConnectorParam) {

        Thread connectorThread = new Thread(genericConnectorParam);
        connectorThread.start();
    }

}

But the Poblem is in the Queue there were 5 messages each messages DEST_ID value is 0 ,1,2,3 respectively.So according to the program two messages should be rollback and remain should poll off from queue(commited)
But when i run this program no message get off from the in bound queue....
Please help me out what is the actual problem in my code