diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java index f889b6ce..3847e29c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java @@ -17,7 +17,6 @@ import java.io.IOException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.amqp.AmqpException; import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.rabbit.support.RabbitUtils; import org.springframework.transaction.support.ResourceHolderSynchronization; @@ -29,15 +28,12 @@ import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; /** - * Helper class for managing a Spring based Rabbit - * {@link org.springframework.amqp.rabbit.connection.ConnectionFactory}, in - * particular for obtaining transactional Rabbit resources for a given - * ConnectionFactory. + * Helper class for managing a Spring based Rabbit {@link org.springframework.amqp.rabbit.connection.ConnectionFactory}, + * in particular for obtaining transactional Rabbit resources for a given ConnectionFactory. * *

- * Mainly for internal use within the framework. Used by - * {@link org.springframework.amqp.rabbit.core.RabbitTemplate} as well as - * {@link org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer}. + * Mainly for internal use within the framework. Used by {@link org.springframework.amqp.rabbit.core.RabbitTemplate} as + * well as {@link org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer}. * * @author Mark Fisher */ @@ -54,41 +50,38 @@ public class ConnectionFactoryUtils { } try { con.close(); - } - catch (Throwable ex) { + } catch (Throwable ex) { logger.debug("Could not close RabbitMQ Connection", ex); } } /** - * Determine whether the given RabbitMQ Channel is transactional, that is, - * bound to the current thread by Spring's transaction facilities. + * Determine whether the given RabbitMQ Channel is transactional, that is, bound to the current thread by Spring's + * transaction facilities. * @param channel the RabbitMQ Channel to check - * @param cf the RabbitMQ ConnectionFactory that the Channel originated from + * @param connectionFactory the RabbitMQ ConnectionFactory that the Channel originated from * @return whether the Channel is transactional */ - public static boolean isChannelTransactional(Channel channel, ConnectionFactory cf) { - if (channel == null || cf == null) { + public static boolean isChannelTransactional(Channel channel, ConnectionFactory connectionFactory) { + if (channel == null || connectionFactory == null) { return false; } - RabbitResourceHolder resourceHolder = (RabbitResourceHolder) TransactionSynchronizationManager.getResource(cf); + RabbitResourceHolder resourceHolder = (RabbitResourceHolder) TransactionSynchronizationManager + .getResource(connectionFactory); return (resourceHolder != null && resourceHolder.containsChannel(channel)); } /** - * Obtain a RabbitMQ Channel that is synchronized with the current - * transaction, if any. + * Obtain a RabbitMQ Channel that is synchronized with the current transaction, if any. * @param cf the ConnectionFactory to obtain a Channel for - * @param synchedLocalTransactionAllowed whether to allow for a local - * RabbitMQ transaction that is synchronized with a Spring-managed - * transaction (where the main transaction might be a JDBC-based one for a - * specific DataSource, for example), with the RabbitMQ transaction - * committing right after the main transaction. If not allowed, the given - * ConnectionFactory needs to handle transaction enlistment underneath the - * covers. + * @param synchedLocalTransactionAllowed whether to allow for a local RabbitMQ transaction that is synchronized with + * a Spring-managed transaction (where the main transaction might be a JDBC-based one for a specific DataSource, for + * example), with the RabbitMQ transaction committing right after the main transaction. If not allowed, the given + * ConnectionFactory needs to handle transaction enlistment underneath the covers. * @return the transactional Channel, or null if none found */ - public static RabbitResourceHolder getTransactionalResourceHolder(final ConnectionFactory cf, final boolean synchedLocalTransactionAllowed) { + public static RabbitResourceHolder getTransactionalResourceHolder(final ConnectionFactory cf, + final boolean synchedLocalTransactionAllowed) { return doGetTransactionalResourceHolder(cf, new ResourceFactory() { public Channel getChannel(RabbitResourceHolder holder) { @@ -114,15 +107,14 @@ public class ConnectionFactoryUtils { } /** - * Obtain a RabbitMQ Channel that is synchronized with the current - * transaction, if any. - * @param connectionFactory the RabbitMQ ConnectionFactory to bind for (used - * as TransactionSynchronizationManager key) - * @param resourceFactory the ResourceFactory to use for extracting or - * creating RabbitMQ resources + * Obtain a RabbitMQ Channel that is synchronized with the current transaction, if any. + * @param connectionFactory the RabbitMQ ConnectionFactory to bind for (used as TransactionSynchronizationManager + * key) + * @param resourceFactory the ResourceFactory to use for extracting or creating RabbitMQ resources * @return the transactional Channel, or null if none found */ - private static RabbitResourceHolder doGetTransactionalResourceHolder(ConnectionFactory connectionFactory, ResourceFactory resourceFactory) { + private static RabbitResourceHolder doGetTransactionalResourceHolder(ConnectionFactory connectionFactory, + ResourceFactory resourceFactory) { Assert.notNull(connectionFactory, "ConnectionFactory must not be null"); Assert.notNull(resourceFactory, "ResourceFactory must not be null"); @@ -151,7 +143,8 @@ public class ConnectionFactoryUtils { resourceHolderToUse.addChannel(channel, con); if (resourceHolderToUse != resourceHolder) { - bindResourceToTransaction(resourceHolderToUse, connectionFactory, resourceFactory.isSynchedLocalTransactionAllowed()); + bindResourceToTransaction(resourceHolderToUse, connectionFactory, + resourceFactory.isSynchedLocalTransactionAllowed()); } if (isChannelTransactional(channel, connectionFactory)) { // It is externally transacted and was just created so we want to start the transaction @@ -160,21 +153,18 @@ public class ConnectionFactoryUtils { return resourceHolderToUse; - } - catch (IOException ex) { + } catch (IOException ex) { if (channel != null) { try { channel.close(); - } - catch (Throwable ex2) { + } catch (Throwable ex2) { // ignore } } if (con != null) { try { con.close(); - } - catch (Throwable ex2) { + } catch (Throwable ex2) { // ignore } } @@ -190,12 +180,14 @@ public class ConnectionFactoryUtils { releaseConnection(resourceHolder.getConnection()); } - public static void bindResourceToTransaction(RabbitResourceHolder resourceHolder, ConnectionFactory connectionFactory, boolean synched) { - if (TransactionSynchronizationManager.hasResource(connectionFactory) || !TransactionSynchronizationManager.isActualTransactionActive()) { + public static void bindResourceToTransaction(RabbitResourceHolder resourceHolder, + ConnectionFactory connectionFactory, boolean synched) { + if (TransactionSynchronizationManager.hasResource(connectionFactory) + || !TransactionSynchronizationManager.isActualTransactionActive()) { return; } TransactionSynchronizationManager.bindResource(connectionFactory, resourceHolder); - resourceHolder.setSynchronizedWithTransaction(true); + resourceHolder.setSynchronizedWithTransaction(true); if (TransactionSynchronizationManager.isSynchronizationActive()) { TransactionSynchronizationManager.registerSynchronization(new RabbitResourceSynchronization(resourceHolder, connectionFactory, synched)); @@ -218,38 +210,34 @@ public class ConnectionFactoryUtils { } /** - * Callback interface for resource creation. Serving as argument for the - * doGetTransactionalChannel method. + * Callback interface for resource creation. Serving as argument for the doGetTransactionalChannel + * method. */ public interface ResourceFactory { /** * Fetch an appropriate Channel from the given RabbitResourceHolder. * @param holder the RabbitResourceHolder - * @return an appropriate Channel fetched from the holder, or - * null if none found + * @return an appropriate Channel fetched from the holder, or null if none found */ Channel getChannel(RabbitResourceHolder holder); /** * Fetch an appropriate Connection from the given RabbitResourceHolder. * @param holder the RabbitResourceHolder - * @return an appropriate Connection fetched from the holder, or - * null if none found + * @return an appropriate Connection fetched from the holder, or null if none found */ Connection getConnection(RabbitResourceHolder holder); /** - * Create a new RabbitMQ Connection for registration with a - * RabbitResourceHolder. + * Create a new RabbitMQ Connection for registration with a RabbitResourceHolder. * @return the new RabbitMQ Connection * @throws IOException if thrown by RabbitMQ API methods */ Connection createConnection() throws IOException; /** - * Create a new RabbitMQ Session for registration with a - * RabbitResourceHolder. + * Create a new RabbitMQ Session for registration with a RabbitResourceHolder. * @param con the RabbitMQ Connection to create a Channel for * @return the new RabbitMQ Channel * @throws IOException if thrown by RabbitMQ API methods @@ -257,26 +245,22 @@ public class ConnectionFactoryUtils { Channel createChannel(Connection con) throws IOException; /** - * Return whether to allow for a local RabbitMQ transaction that is - * synchronized with a Spring-managed transaction (where the main - * transaction might be a JDBC-based one for a specific DataSource, for - * example), with the RabbitMQ transaction committing right after the - * main transaction. - * @return whether to allow for synchronizing a local RabbitMQ - * transaction + * Return whether to allow for a local RabbitMQ transaction that is synchronized with a Spring-managed + * transaction (where the main transaction might be a JDBC-based one for a specific DataSource, for example), + * with the RabbitMQ transaction committing right after the main transaction. + * @return whether to allow for synchronizing a local RabbitMQ transaction */ boolean isSynchedLocalTransactionAllowed(); } /** - * Callback for resource cleanup at the end of a non-native RabbitMQ - * transaction (e.g. when participating in a JtaTransactionManager - * transaction). + * Callback for resource cleanup at the end of a non-native RabbitMQ transaction (e.g. when participating in a + * JtaTransactionManager transaction). * @see org.springframework.transaction.jta.JtaTransactionManager */ private static class RabbitResourceSynchronization extends ResourceHolderSynchronization { - + private final boolean transacted; private final RabbitResourceHolder resourceHolder; @@ -292,12 +276,7 @@ public class ConnectionFactoryUtils { } protected void processResourceAfterCommit(RabbitResourceHolder resourceHolder) { - try { - resourceHolder.commitAll(); - } - catch (IOException e) { - throw new AmqpException("failed to commit RabbitMQ transaction", e); - } + resourceHolder.commitAll(); } @Override diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java index 6c676d21..17893804 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java @@ -21,9 +21,11 @@ import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.amqp.AmqpException; import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.support.RabbitUtils; +import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager; import org.springframework.transaction.support.ResourceHolderSupport; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -122,14 +124,18 @@ public class RabbitResourceHolder extends ResourceHolderSupport { return (!this.channels.isEmpty() ? this.channels.get(0) : null); } - public void commitAll() throws IOException { - for (Channel channel : this.channels) { - if (deliveryTags.containsKey(channel)) { - for (Long deliveryTag : deliveryTags.get(channel)) { - channel.basicAck(deliveryTag, false); + public void commitAll() throws AmqpException { + try { + for (Channel channel : this.channels) { + if (deliveryTags.containsKey(channel)) { + for (Long deliveryTag : deliveryTags.get(channel)) { + channel.basicAck(deliveryTag, false); + } } + channel.txCommit(); } - channel.txCommit(); + } catch (IOException e) { + throw new AmqpException("failed to commit RabbitMQ transaction", e); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java new file mode 100644 index 00000000..d7a8ff6b --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java @@ -0,0 +1,220 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package org.springframework.amqp.rabbit.transaction; + +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils; +import org.springframework.amqp.rabbit.connection.RabbitResourceHolder; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.transaction.CannotCreateTransactionException; +import org.springframework.transaction.InvalidIsolationLevelException; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.support.AbstractPlatformTransactionManager; +import org.springframework.transaction.support.DefaultTransactionStatus; +import org.springframework.transaction.support.ResourceTransactionManager; +import org.springframework.transaction.support.SmartTransactionObject; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import com.rabbitmq.client.Connection; + +/** + * {@link org.springframework.transaction.PlatformTransactionManager} implementation for a single Rabbit + * {@link ConnectionFactory}. Binds a Rabbit Channel from the specified ConnectionFactory to the thread, potentially + * allowing for one thread-bound channel per ConnectionFactory. + * + *

+ * This local strategy is an alternative to executing Rabbit operations within, and synchronized with, external + * transactions. This strategy is not able to provide XA transactions, for example in order to share transactions + * between messaging and database access. + * + *

+ * Application code is required to retrieve the transactional Rabbit resources via + * {@link ConnectionFactoryUtils#getTransactionalResourceHolder(ConnectionFactory, boolean)} instead of a standard + * {@link Connection#createChannel()} call with subsequent Channel creation. Spring's {@link RabbitTemplate} will + * autodetect a thread-bound Channel and automatically participate in it. + * + *

+ * The use of {@link CachingConnectionFactory} as a target for this transaction manager is strongly recommended. + * CachingConnectionFactory uses a single Rabbit Connection for all Rabbit access in order to avoid the overhead of + * repeated Connection creation, as well as maintaining a cache of Channels. Each transaction will then share the same + * Rabbit Connection, while still using its own individual Rabbit Channel. + * + *

+ * Transaction synchronization is turned off by default, as this manager might be used alongside a datastore-based + * Spring transaction manager such as the JDBC {@link org.springframework.jdbc.datasource.DataSourceTransactionManager}, + * which has stronger needs for synchronization. + * + * @author Dave Syer + * + */ +@SuppressWarnings("serial") +public class RabbitTransactionManager extends AbstractPlatformTransactionManager implements ResourceTransactionManager, + InitializingBean { + + private ConnectionFactory connectionFactory; + + /** + * Create a new RabbitTransactionManager for bean-style usage. + *

+ * Note: The ConnectionFactory has to be set before using the instance. This constructor can be used to prepare a + * RabbitTemplate via a BeanFactory, typically setting the ConnectionFactory via setConnectionFactory. + *

+ * Turns off transaction synchronization by default, as this manager might be used alongside a datastore-based + * Spring transaction manager like DataSourceTransactionManager, which has stronger needs for synchronization. Only + * one manager is allowed to drive synchronization at any point of time. + * @see #setConnectionFactory + * @see #setTransactionSynchronization + */ + public RabbitTransactionManager() { + setTransactionSynchronization(SYNCHRONIZATION_NEVER); + } + + /** + * Create a new RabbitTransactionManager, given a ConnectionFactory. + * @param connectionFactory the ConnectionFactory to use + */ + public RabbitTransactionManager(ConnectionFactory connectionFactory) { + this(); + this.connectionFactory = connectionFactory; + afterPropertiesSet(); + } + + /** + * @param connectionFactory the connectionFactory to set + */ + public void setConnectionFactory(ConnectionFactory connectionFactory) { + this.connectionFactory = connectionFactory; + } + + /** + * @return the connectionFactory + */ + public ConnectionFactory getConnectionFactory() { + return connectionFactory; + } + + /** + * Make sure the ConnectionFactory has been set. + */ + public void afterPropertiesSet() { + if (getConnectionFactory() == null) { + throw new IllegalArgumentException("Property 'connectionFactory' is required"); + } + } + + public Object getResourceFactory() { + return getConnectionFactory(); + } + + protected Object doGetTransaction() { + RabbitTransactionObject txObject = new RabbitTransactionObject(); + txObject.setResourceHolder((RabbitResourceHolder) TransactionSynchronizationManager + .getResource(getConnectionFactory())); + return txObject; + } + + protected boolean isExistingTransaction(Object transaction) { + RabbitTransactionObject txObject = (RabbitTransactionObject) transaction; + return (txObject.getResourceHolder() != null); + } + + protected void doBegin(Object transaction, TransactionDefinition definition) { + if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) { + throw new InvalidIsolationLevelException("AMQP does not support an isolation level concept"); + } + RabbitTransactionObject txObject = (RabbitTransactionObject) transaction; + RabbitResourceHolder resourceHolder = null; + try { + resourceHolder = ConnectionFactoryUtils.getTransactionalResourceHolder(getConnectionFactory(), true); + if (logger.isDebugEnabled()) { + logger.debug("Created AMQP transaction on channel [" + resourceHolder.getChannel() + "]"); + } + resourceHolder.declareTransactional(); + txObject.setResourceHolder(resourceHolder); + txObject.getResourceHolder().setSynchronizedWithTransaction(true); + int timeout = determineTimeout(definition); + if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) { + txObject.getResourceHolder().setTimeoutInSeconds(timeout); + } + TransactionSynchronizationManager.bindResource(getConnectionFactory(), txObject.getResourceHolder()); + } catch (AmqpException ex) { + if (resourceHolder != null) { + ConnectionFactoryUtils.releaseResources(resourceHolder); + } + throw new CannotCreateTransactionException("Could not create AMQP transaction", ex); + } + } + + protected Object doSuspend(Object transaction) { + RabbitTransactionObject txObject = (RabbitTransactionObject) transaction; + txObject.setResourceHolder(null); + return TransactionSynchronizationManager.unbindResource(getConnectionFactory()); + } + + protected void doResume(Object transaction, Object suspendedResources) { + RabbitResourceHolder conHolder = (RabbitResourceHolder) suspendedResources; + TransactionSynchronizationManager.bindResource(getConnectionFactory(), conHolder); + } + + protected void doCommit(DefaultTransactionStatus status) { + RabbitTransactionObject txObject = (RabbitTransactionObject) status.getTransaction(); + RabbitResourceHolder resourceHolder = txObject.getResourceHolder(); + resourceHolder.commitAll(); + } + + protected void doRollback(DefaultTransactionStatus status) { + RabbitTransactionObject txObject = (RabbitTransactionObject) status.getTransaction(); + RabbitResourceHolder resourceHolder = txObject.getResourceHolder(); + resourceHolder.rollbackAll(); + } + + protected void doSetRollbackOnly(DefaultTransactionStatus status) { + RabbitTransactionObject txObject = (RabbitTransactionObject) status.getTransaction(); + txObject.getResourceHolder().setRollbackOnly(); + } + + protected void doCleanupAfterCompletion(Object transaction) { + RabbitTransactionObject txObject = (RabbitTransactionObject) transaction; + TransactionSynchronizationManager.unbindResource(getConnectionFactory()); + txObject.getResourceHolder().closeAll(); + txObject.getResourceHolder().clear(); + } + + /** + * Rabbit transaction object, representing a RabbitResourceHolder. Used as transaction object by + * RabbitTransactionManager. + * @see RabbitResourceHolder + */ + private static class RabbitTransactionObject implements SmartTransactionObject { + + private RabbitResourceHolder resourceHolder; + + public void setResourceHolder(RabbitResourceHolder resourceHolder) { + this.resourceHolder = resourceHolder; + } + + public RabbitResourceHolder getResourceHolder() { + return this.resourceHolder; + } + + public boolean isRollbackOnly() { + return this.resourceHolder.isRollbackOnly(); + } + + public void flush() { + // no-op + } + } +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManagerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManagerIntegrationTests.java new file mode 100644 index 00000000..3304d8d8 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManagerIntegrationTests.java @@ -0,0 +1,124 @@ +package org.springframework.amqp.rabbit.transaction; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; + +public class RabbitTransactionManagerIntegrationTests { + + private static final String ROUTE = "test.queue"; + + private RabbitTemplate template; + + private TransactionTemplate transactionTemplate; + + @Rule + public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(ROUTE); + + @Before + public void init() { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + template = new RabbitTemplate(connectionFactory); + template.setChannelTransacted(true); + RabbitTransactionManager transactionManager = new RabbitTransactionManager(connectionFactory); + transactionTemplate = new TransactionTemplate(transactionManager); + } + + @Test + public void testSendAndReceiveInTransaction() throws Exception { + String result = transactionTemplate.execute(new TransactionCallback() { + public String doInTransaction(TransactionStatus status) { + template.convertAndSend(ROUTE, "message"); + return (String) template.receiveAndConvert(ROUTE); + } + }); + assertEquals(null, result); + result = (String) template.receiveAndConvert(ROUTE); + assertEquals("message", result); + } + + @Test + public void testReceiveInTransaction() throws Exception { + template.convertAndSend(ROUTE, "message"); + String result = transactionTemplate.execute(new TransactionCallback() { + public String doInTransaction(TransactionStatus status) { + return (String) template.receiveAndConvert(ROUTE); + } + }); + assertEquals("message", result); + result = (String) template.receiveAndConvert(ROUTE); + assertEquals(null, result); + } + + @Test + public void testReceiveInTransactionWithRollback() throws Exception { + template.setChannelTransacted(true); // Makes receive (and send in + // principle) transactional + template.convertAndSend(ROUTE, "message"); + try { + transactionTemplate.execute(new TransactionCallback() { + public String doInTransaction(TransactionStatus status) { + template.receiveAndConvert(ROUTE); + throw new PlannedException(); + } + }); + fail("Expected PlannedException"); + } catch (PlannedException e) { + // Expected + } + String result = (String) template.receiveAndConvert(ROUTE); + assertEquals("message", result); + result = (String) template.receiveAndConvert(ROUTE); + assertEquals(null, result); + } + + @Test + public void testSendInTransaction() throws Exception { + template.setChannelTransacted(true); + transactionTemplate.execute(new TransactionCallback() { + public Void doInTransaction(TransactionStatus status) { + template.convertAndSend(ROUTE, "message"); + return null; + } + }); + String result = (String) template.receiveAndConvert(ROUTE); + assertEquals("message", result); + result = (String) template.receiveAndConvert(ROUTE); + assertEquals(null, result); + } + + @Test + public void testSendInTransactionWithRollback() throws Exception { + template.setChannelTransacted(true); + try { + transactionTemplate.execute(new TransactionCallback() { + public Void doInTransaction(TransactionStatus status) { + template.convertAndSend(ROUTE, "message"); + throw new PlannedException(); + } + }); + fail("Expected PlannedException"); + } catch (PlannedException e) { + // Expected + } + String result = (String) template.receiveAndConvert(ROUTE); + assertEquals(null, result); + } + + @SuppressWarnings("serial") + private class PlannedException extends RuntimeException { + public PlannedException() { + super("Planned"); + } + } + +}