Thanks. I took into account the gotchas and went ahead and started implementing it. I was mainly hoping to verify that I wasn't going to go off and implement something when there is already something available. I'm not a big fan of having to create and manage a thread but it doesn't seem like there is another prebuilt option.
As for the implementation I tried to use the sample code and xml snippets for spring to implement the solution and am now able to send and receive messages but for some reason I can't cause Spring to rollback the transaction. Here's a snippet of my application xml.
<bean id="eventListener" parent="jmsAbstractServiceProxy">
<property name="transactionAttributes">
<props>
<prop key="process*">PROPAGATION_REQUIRED</prop>
</props>
</property>
<property name="target">
<bean class="EventListener" init-method="init" destroy-method="destroy">
<property name="jmsTemplate">
<ref bean="jmsReceiveTemplate"/>
</property>
</bean>
</property>
</bean>
<bean id="jmsTransactionManager" class="org.springframework.jms.connection.JmsTransactionM anager">
<property name="connectionFactory" ref="jmsFactory" />
</bean>
<bean id="jmsAbstractServiceProxy" class="org.springframework.transaction.interceptor.Transa ctionProxyFactoryBean"
abstract="true">
<property name="transactionManager">
<ref bean="jmsTransactionManager"/>
</property>
</bean>
<bean id="jmsFactory" class="org.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory">
<bean class="org.activemq.ActiveMQConnectionFactory">
<property name="brokerURL">
<value>tcp://localhost:61616</value>
</property>
</bean>
</property>
</bean>
<bean id="jmsReceiveTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory">
<ref local="jmsFactory"/>
</property>
<property name="sessionTransacted">
<value>true</value>
</property>
</bean>
And the java code. Notice where I threw the IllegalArgumentException just to test the rollback. I assume that I should see the log messages over and over again as the message is resent.
import javax.jms.JMSException;
import javax.jms.TextMessage;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.jms.core.JmsTemplate;
public class EventListener implements Runnable
{
/**
* The number of milliseconds to wait before logging an infor message on shutdown
*/
private static final long STOP_WAIT_TIMEOUT = 1000;
/**
* The default number of milliseconds to wait before timing out on a receive message
*/
private static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
/**
* Template used to send messages
*/
private JmsTemplate jmsTemplate;
/**
* True if the listener should be running running
*/
private boolean running;
/**
* True if the listener thread has not stopped
*/
private boolean stopped;
/**
* The maximum time to wait receiving all the messages
*/
private long maximumBatchWaitTime = 5000L;
/**
* The batch size
*/
private int batchSize = 5;
/**
* The amount of time to wait before reattempting to send messages
*/
private long postJmsErrorWaitTimeout = 5000L;
/**
* Log used for logging
*/
private Log log = LogFactory.getLog(getClass());
/**
* Sets the JmsTemplate used to send messages
*
* @param jmsTemplate the JmsTemplate used to send messages
*/
public void setJmsTemplate(JmsTemplate jmsTemplate)
{
this.jmsTemplate = jmsTemplate;
jmsTemplate.setReceiveTimeout(DEFAULT_RECEIVE_TIME OUT);
}
/**
* Starts the listener thread
*/
public void init()
{
if (log.isDebugEnabled())
{
log.debug("Preparing to startup EventListener thread");
}
running = true;
new Thread(this).start();
}
/**
* Stops the listener thread
*/
public void destroy()
{
if (log.isDebugEnabled())
{
log.debug("Preparing to shutdown EventListener thread");
}
synchronized (this)
{
running = false;
while (!stopped)
{
if (log.isDebugEnabled())
{
log.debug("Waiting for EventListener to shutdown");
}
try
{
wait(STOP_WAIT_TIMEOUT);
}
catch (InterruptedException e)
{
log.warn("Thread interrupted on shutdown", e);
}
}
}
if (log.isDebugEnabled())
{
log.debug("EventListener thread stopped");
}
}
/**
* Processes listener messages asynchronously
*/
public void run()
{
if (log.isDebugEnabled())
{
log.debug("EventListener thread started");
}
while (running)
{
try
{
processMessages();
}
catch (JMSException e)
{
log.warn("JMS error processing messages; waiting [" + postJmsErrorWaitTimeout
+ "] before reattempting", e);
try
{
Thread.sleep(postJmsErrorWaitTimeout);
}
catch (InterruptedException e2)
{
log.warn("EventListener interrupted", e2);
}
}
catch (Throwable t)
{
log.error("Error in EventListener thread", t);
}
}
synchronized (this)
{
stopped = true;
notify();
}
}
/**
* Processes a batch of messages
* @throws JMSException if there is a JMS related error
*/
public void processMessages() throws JMSException
{
long startTime = System.currentTimeMillis();
int messageCount = 0;
while (running && (startTime > (System.currentTimeMillis() - maximumBatchWaitTime))
&& (messageCount < batchSize))
{
TextMessage message = (TextMessage) jmsTemplate.receive("test");
if (message != null)
{
messageCount++;
if (log.isDebugEnabled())
{
log.debug("Received Message [" + message.getText() + "]");
}
throw new IllegalArgumentException("Just a test");
}
}
}
}
Thanks for the help.
Eric