AMQP-49: add first cut of RabbitTransactionManager

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.
  * 
  * <p>
- * 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 <code>null</code> 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 <code>null</code> 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
-	 * <code>doGetTransactionalChannel</code> method.
+	 * Callback interface for resource creation. Serving as argument for the <code>doGetTransactionalChannel</code>
+	 * 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
-		 * <code>null</code> if none found
+		 * @return an appropriate Channel fetched from the holder, or <code>null</code> 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
-		 * <code>null</code> if none found
+		 * @return an appropriate Connection fetched from the holder, or <code>null</code> 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<RabbitResourceHolder, Object> {
-		
+
 		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

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);
 		}
 	}
 

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.
+ * 
+ * <p>
+ * This local strategy is an alternative to executing Rabbit operations within, and synchronized with, external
+ * transactions. This strategy is <i>not</i> able to provide XA transactions, for example in order to share transactions
+ * between messaging and database access.
+ * 
+ * <p>
+ * 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.
+ * 
+ * <p>
+ * <b>The use of {@link CachingConnectionFactory} as a target for this transaction manager is strongly recommended.</b>
+ * 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.
+ * 
+ * <p>
+ * 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.
+	 * <p>
+	 * 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.
+	 * <p>
+	 * 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
+		}
+	}
+}

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<String>() {
+			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<String>() {
+			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<String>() {
+				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<Void>() {
+			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<Void>() {
+				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");
+		}
+	}
+
+}