Merge pull request #89 from ghillert/AMQP-304

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java

@@ -20,7 +20,7 @@ import com.rabbitmq.client.Channel;
 
 /**
  * Simply a Connection.
- * @Dave Syer
+ * @author Dave Syer
  * @author Gary Russell
  * @since 1.0
  *

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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
@@ -37,6 +37,7 @@ import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
  * @author Mark Pollack
  * @author Dave Syer
  * @author Helena Edelson
+ * @author Gunnar Hillert
  */
 public class RabbitBrokerAdminIntegrationTests {
 
@@ -129,6 +130,7 @@ public class RabbitBrokerAdminIntegrationTests {
 		assertEquals("/", connectionFactory.getVirtualHost());
 		List<QueueInfo> queues = brokerAdmin.getQueues();
 		assertEquals(queue.getName(), queues.get(0).getName());
+		connectionFactory.destroy();
 	}
 
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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
@@ -25,11 +25,13 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 public final class ExchangeParserIntegrationTests {
 
 	@Rule

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/FederatedExchangeParserIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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
@@ -26,11 +26,13 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.test.BrokerFederated;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 public final class FederatedExchangeParserIntegrationTests {
 
 	@Rule

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MismatchedQueueDeclarationTests.java

@@ -29,6 +29,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.core.env.StandardEnvironment;
 
@@ -36,6 +37,7 @@ import com.rabbitmq.client.Channel;
 
 /**
  * @author Gary Russell
+ * @author Gunnar Hillert
  * @since 1.2
  *
  */
@@ -49,16 +51,18 @@ public class MismatchedQueueDeclarationTests {
 	private RabbitAdmin admin;
 
 	@Before
-	public void setup() {
+	public void setup() throws Exception {
 		connectionFactory = new SingleConnectionFactory();
 		this.admin = new RabbitAdmin(this.connectionFactory);
 		deleteQueues();
 	}
 
 	@After
-	public void deleteQueues() {
+	public void deleteQueues() throws Exception {
 		this.admin.deleteQueue("mismatch.foo");
 		this.admin.deleteQueue("mismatch.bar");
+
+		((DisposableBean) connectionFactory).destroy();
 	}
 
 	@Test @Ignore

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueArgumentsParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -24,6 +24,7 @@ import org.junit.runner.RunWith;
 import org.springframework.amqp.core.Queue;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
@@ -35,6 +36,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 public class QueueArgumentsParserTests {
 
 	@Autowired

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserIntegrationTests.java

@@ -32,6 +32,7 @@ import org.springframework.core.io.ClassPathResource;
 /**
  * @author Dave Syer
  * @author Gary Russell
+ * @author Gunnar Hillert
  * @since 1.0
  *
  */
@@ -54,8 +55,8 @@ public final class QueueParserIntegrationTests {
 
 		Queue queue = beanFactory.getBean("arguments", Queue.class);
 		assertNotNull(queue);
-
-		RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory(BrokerTestUtils.getPort()));
+		CachingConnectionFactory connectionFactory = new CachingConnectionFactory(BrokerTestUtils.getPort());
+		RabbitTemplate template = new RabbitTemplate(connectionFactory);
 		RabbitAdmin rabbitAdmin = new RabbitAdmin(template.getConnectionFactory());
 		rabbitAdmin.deleteQueue(queue.getName());
 		rabbitAdmin.declareQueue(queue);
@@ -67,6 +68,8 @@ public final class QueueParserIntegrationTests {
 		String result = (String) template.receiveAndConvert(queue.getName());
 		assertEquals(null, result);
 
+		connectionFactory.destroy();
+
 	}
 
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2002-2013 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.connection;
 
 import static org.junit.Assert.assertEquals;
@@ -27,11 +39,17 @@ import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.ShutdownListener;
 import com.rabbitmq.client.ShutdownSignalException;
 
+/**
+ * @author Dave Syer
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 public class CachingConnectionFactoryIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(CachingConnectionFactoryIntegrationTests.class);
 
-	private CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
+	private CachingConnectionFactory connectionFactory;
 
 	@Rule
 	public BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
@@ -41,13 +59,13 @@ public class CachingConnectionFactoryIntegrationTests {
 
 	@Before
 	public void open() {
+		connectionFactory = new CachingConnectionFactory();
 		connectionFactory.setPort(BrokerTestUtils.getPort());
 	}
 
 	@After
 	public void close() {
-		// Release resources
-		connectionFactory.reset();
+		connectionFactory.destroy();
 	}
 
 	@Test

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2002-2013 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.core;
 
 import static org.junit.Assert.assertFalse;
@@ -85,6 +97,7 @@ public class RabbitAdminIntegrationTests {
 		} finally {
 			// Need to release the connection so the exclusive queue is deleted
 			connectionFactory1.destroy();
+			connectionFactory2.destroy();
 		}
 	}
 
@@ -150,6 +163,8 @@ public class RabbitAdminIntegrationTests {
 		// Queue manually deleted
 		assertTrue(rabbitAdmin.deleteQueue(queue.getName()));
 		assertFalse(queueExists(queue));
+
+		connectionFactory.destroy();
 	}
 
 	@Test

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2002-2013 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.core;
 
 import static org.junit.Assert.assertEquals;
@@ -47,6 +59,7 @@ public class RabbitAdminTests {
 		rabbitAdmin.setApplicationContext(applicationContext);
 		rabbitAdmin.setAutoStartup(true);
 		rabbitAdmin.afterPropertiesSet();
+		connectionFactory.destroy();
 	}
 
 	@Test
@@ -61,6 +74,7 @@ public class RabbitAdminTests {
 		rabbitAdmin.afterPropertiesSet();
 		exception.expect(IllegalArgumentException.class);
 		rabbitAdmin.declareQueue();
+		connectionFactory.destroy();
 	}
 
 	@Test

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java

@@ -1,8 +1,22 @@
+/*
+ * Copyright 2002-2013 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.core;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 
+import org.junit.After;
+import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.springframework.amqp.core.AcknowledgeMode;
@@ -12,7 +26,6 @@ import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.TopicExchange;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
-import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.RabbitAccessor;
 import org.springframework.amqp.rabbit.listener.ActiveObjectCounter;
 import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer;
@@ -23,17 +36,34 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter;
 
 import com.rabbitmq.client.Channel;
 
+/**
+ * @author Dave Syer
+ * @author Gunnar Hillert
+ */
 public class RabbitBindingIntegrationTests {
 
 	private static Queue queue = new Queue("test.queue");
 
-	private ConnectionFactory connectionFactory = new CachingConnectionFactory(BrokerTestUtils.getPort());
+	private CachingConnectionFactory connectionFactory;
 
-	private RabbitTemplate template = new RabbitTemplate(connectionFactory );
+	private RabbitTemplate template;
 
 	@Rule
 	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue);
 
+	@Before
+	public void setup() {
+		connectionFactory = new CachingConnectionFactory(BrokerTestUtils.getPort());
+		template = new RabbitTemplate(connectionFactory);
+	}
+
+	@After
+	public void cleanUp() {
+		if (connectionFactory != null) {
+			connectionFactory.destroy();
+		}
+	}
+
 	@Test
 	public void testSendAndReceiveWithTopicSingleCallback() throws Exception {
 
@@ -123,7 +153,8 @@ public class RabbitBindingIntegrationTests {
 
 		admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("*.end"));
 
-		final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
+		final CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
+		final RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
 		template.setExchange(exchange.getName());
 
 		BlockingQueueConsumer consumer = template.execute(new ChannelCallback<BlockingQueueConsumer>() {
@@ -147,6 +178,7 @@ public class RabbitBindingIntegrationTests {
 		assertEquals("message", result);
 
 		consumer.stop();
+		cachingConnectionFactory.destroy();
 
 	}
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2012 the original author or authors.
+ * Copyright 2010-2013 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
@@ -29,6 +29,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.logging.Log;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -46,6 +47,7 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.TransactionException;
@@ -69,11 +71,16 @@ public class RabbitTemplateIntegrationTests {
 
 	@Before
 	public void create() {
-		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
+		final CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
 		connectionFactory.setPort(BrokerTestUtils.getPort());
 		template = new RabbitTemplate(connectionFactory);
 	}
 
+	@After
+	public void cleanup() throws Exception {
+		((DisposableBean) template.getConnectionFactory()).destroy();
+	}
+
 	@Rule
 	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(ROUTE);
 
@@ -131,13 +138,15 @@ public class RabbitTemplateIntegrationTests {
 
 	@Test
 	public void testSendAndReceiveTransactedWithUncachedConnection() throws Exception {
-		RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory());
+		final SingleConnectionFactory singleConnectionFactory = new SingleConnectionFactory();
+		RabbitTemplate template = new RabbitTemplate(singleConnectionFactory);
 		template.setChannelTransacted(true);
 		template.convertAndSend(ROUTE, "message");
 		String result = (String) template.receiveAndConvert(ROUTE);
 		assertEquals("message", result);
 		result = (String) template.receiveAndConvert(ROUTE);
 		assertEquals(null, result);
+		singleConnectionFactory.destroy();
 	}
 
 	@Test
@@ -298,7 +307,8 @@ public class RabbitTemplateIntegrationTests {
 
 	@Test
 	public void testAtomicSendAndReceive() throws Exception {
-		final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
+		final CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
+		final RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
 		template.setRoutingKey(ROUTE);
 		template.setQueue(ROUTE);
 		ExecutorService executor = Executors.newFixedThreadPool(1);
@@ -328,11 +338,12 @@ public class RabbitTemplateIntegrationTests {
 		// Message was consumed so nothing left on queue
 		reply = template.receive();
 		assertEquals(null, reply);
+		cachingConnectionFactory.destroy();
 	}
 
 	@Test
 	public void testAtomicSendAndReceiveExternalExecutor() throws Exception {
-		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
+		final CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
 		ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
 		final String execName = "make-sure-exec-passed-in";
 		exec.setBeanName(execName);
@@ -398,11 +409,13 @@ public class RabbitTemplateIntegrationTests {
 		assertEquals(null, reply);
 
 		assertTrue(execConfiguredOk.get());
+		connectionFactory.destroy();
 	}
 
 	@Test
 	public void testAtomicSendAndReceiveWithRoutingKey() throws Exception {
-		final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
+		final CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
+		final RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
 		ExecutorService executor = Executors.newFixedThreadPool(1);
 		// Set up a consumer to respond to our producer
 		Future<Message> received = executor.submit(new Callable<Message>() {
@@ -430,11 +443,13 @@ public class RabbitTemplateIntegrationTests {
 		// Message was consumed so nothing left on queue
 		reply = template.receive(ROUTE);
 		assertEquals(null, reply);
+		cachingConnectionFactory.destroy();
 	}
 
 	@Test
 	public void testAtomicSendAndReceiveWithExchangeAndRoutingKey() throws Exception {
-		final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
+		final CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
+		final RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
 		ExecutorService executor = Executors.newFixedThreadPool(1);
 		// Set up a consumer to respond to our producer
 		Future<Message> received = executor.submit(new Callable<Message>() {
@@ -462,11 +477,13 @@ public class RabbitTemplateIntegrationTests {
 		// Message was consumed so nothing left on queue
 		reply = template.receive(ROUTE);
 		assertEquals(null, reply);
+		cachingConnectionFactory.destroy();
 	}
 
 	@Test
 	public void testAtomicSendAndReceiveWithConversion() throws Exception {
-		final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
+		final CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
+		final RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
 		template.setRoutingKey(ROUTE);
 		template.setQueue(ROUTE);
 		ExecutorService executor = Executors.newFixedThreadPool(1);
@@ -494,6 +511,7 @@ public class RabbitTemplateIntegrationTests {
 		// Message was consumed so nothing left on queue
 		result = (String) template.receiveAndConvert();
 		assertEquals(null, result);
+		cachingConnectionFactory.destroy();
 	}
 
 	@Test
@@ -556,7 +574,8 @@ public class RabbitTemplateIntegrationTests {
 
 	@Test
 	public void testAtomicSendAndReceiveWithConversionAndMessagePostProcessor() throws Exception {
-		final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
+		final CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
+		final RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
 		template.setRoutingKey(ROUTE);
 		template.setQueue(ROUTE);
 		ExecutorService executor = Executors.newFixedThreadPool(1);
@@ -594,6 +613,7 @@ public class RabbitTemplateIntegrationTests {
 		// Message was consumed so nothing left on queue
 		result = (String) template.receiveAndConvert();
 		assertEquals(null, result);
+		cachingConnectionFactory.destroy();
 	}
 
 	@Test

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2010-2013 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.core;
 
 import static org.junit.Assert.assertEquals;
@@ -21,11 +33,17 @@ import org.springframework.transaction.support.DefaultTransactionStatus;
 import org.springframework.transaction.support.TransactionCallback;
 import org.springframework.transaction.support.TransactionTemplate;
 
+/**
+ * @author Dave Syer
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 public class RabbitTemplatePerformanceIntegrationTests {
 
 	private static final String ROUTE = "test.queue";
 
-	private RabbitTemplate template = new RabbitTemplate();
+	private final RabbitTemplate template = new RabbitTemplate();
 
 	@Rule
 	public RepeatProcessor repeat = new RepeatProcessor(4);
@@ -54,7 +72,7 @@ public class RabbitTemplatePerformanceIntegrationTests {
 
 	@After
 	public void cleanUp() {
-		if (repeat.isInitialized()) {
+		if (!repeat.isFinalizing()) {
 			return;
 		}
 		if (connectionFactory != null) {

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2012 the original author or authors.
+ * Copyright 2010-2013 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
@@ -36,6 +36,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -58,6 +59,12 @@ import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 
+/**
+ * @author Gary Russell
+ * @author Gunar Hillert
+ * @since 1.1
+ *
+ */
 public class RabbitTemplatePublisherCallbacksIntegrationTests {
 
 	private static final String ROUTE = "test.queue";
@@ -91,6 +98,21 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
 		templateWithReturnsEnabled = new RabbitTemplate(connectionFactoryWithReturnsEnabled);
 	}
 
+	@After
+	public void cleanUp() {
+		if (connectionFactory != null) {
+			connectionFactory.destroy();
+		}
+
+		if (connectionFactoryWithConfirmsEnabled != null) {
+			connectionFactoryWithConfirmsEnabled.destroy();
+		}
+
+		if (connectionFactoryWithReturnsEnabled != null) {
+			connectionFactoryWithReturnsEnabled.destroy();
+		}
+	}
+
 	@Rule
 	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(ROUTE);
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2010-2013 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.listener;
 
 import static org.junit.Assert.assertNull;
@@ -14,6 +26,12 @@ import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
 
+/**
+ * @author Dave Syer
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 public class BlockingQueueConsumerIntegrationTests {
 
 	private static Queue queue = new Queue("test.queue");
@@ -38,12 +56,12 @@ public class BlockingQueueConsumerIntegrationTests {
 				new DefaultMessagePropertiesConverter(), new ActiveObjectCounter<BlockingQueueConsumer>(),
 				AcknowledgeMode.AUTO, true, 1, queue.getName());
 		blockingQueueConsumer.start();
-		connectionFactory.destroy();
 
 		// TODO: make this into a proper assertion. An exception can be thrown here by the Rabbit client and printed to
 		// stderr without being rethrown (so hard to make a test fail).
 		blockingQueueConsumer.stop();
 		assertNull(template.receiveAndConvert(queue.getName()));
+		connectionFactory.destroy();
 
 	}
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerBrokerInterruptionIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2010-2013 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.listener;
 
 import static org.junit.Assert.assertEquals;
@@ -34,22 +46,28 @@ import org.springframework.amqp.rabbit.test.EnvironmentAvailable;
 
 import com.rabbitmq.client.Channel;
 
+/**
+ * @author Dave Syer
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 public class MessageListenerBrokerInterruptionIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(MessageListenerBrokerInterruptionIntegrationTests.class);
 
 	// Ensure queue is durable, or it won't survive the broker restart
-	private Queue queue = new Queue("test.queue", true);
+	private final Queue queue = new Queue("test.queue", true);
 
-	private int concurrentConsumers = 2;
+	private final int concurrentConsumers = 2;
 
-	private int messageCount = 60;
+	private final int messageCount = 60;
 
-	private int txSize = 1;
+	private final int txSize = 1;
 
-	private boolean transactional = false;
+	private final boolean transactional = false;
 
-	private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
+	private final AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
 
 	private SimpleMessageListenerContainer container;
 
@@ -65,9 +83,9 @@ public class MessageListenerBrokerInterruptionIntegrationTests {
 	@Rule
 	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue);
 
-	private ConnectionFactory connectionFactory;
+	private CachingConnectionFactory connectionFactory;
 
-	private RabbitBrokerAdmin brokerAdmin;
+	private final RabbitBrokerAdmin brokerAdmin;
 
 	public MessageListenerBrokerInterruptionIntegrationTests() throws Exception {
 		FileUtils.deleteDirectory(new File("target/rabbitmq"));
@@ -102,6 +120,10 @@ public class MessageListenerBrokerInterruptionIntegrationTests {
 			brokerAdmin.stopNode();
 			// Remove all trace of the durable queue...
 			FileUtils.deleteDirectory(new File("target/rabbitmq"));
+
+			if (this.connectionFactory != null) {
+				this.connectionFactory.destroy();
+			}
 		}
 	}
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2010-2013 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.listener;
 
 import static org.junit.Assert.assertNull;
@@ -28,10 +40,17 @@ import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.util.ErrorHandler;
 
 import com.rabbitmq.client.Channel;
 
+/**
+ * @author Dave Syer
+ * @author Gunar Hillert
+ * @since 1.0
+ *
+ */
 public class MessageListenerContainerErrorHandlerIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(MessageListenerContainerErrorHandlerIntegrationTests.class);
@@ -39,7 +58,7 @@ public class MessageListenerContainerErrorHandlerIntegrationTests {
 	private static Queue queue = new Queue("test.queue");
 
 	// Mock error handler
-	private ErrorHandler errorHandler = mock(ErrorHandler.class);
+	private final ErrorHandler errorHandler = mock(ErrorHandler.class);
 
 	@Rule
 	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue);
@@ -153,6 +172,8 @@ public class MessageListenerContainerErrorHandlerIntegrationTests {
 		} finally {
 			container.shutdown();
 		}
+
+		((DisposableBean) template.getConnectionFactory()).destroy();
 	}
 
 	private RabbitTemplate createTemplate(int concurrentConsumers) {
@@ -169,8 +190,8 @@ public class MessageListenerContainerErrorHandlerIntegrationTests {
 	// Helper classes
 	// ///////////////
 	public static class PojoThrowingExceptionListener {
-		private CountDownLatch latch;
-		private Throwable exception;
+		private final CountDownLatch latch;
+		private final Throwable exception;
 
 		public PojoThrowingExceptionListener(CountDownLatch latch, Throwable exception) {
 			this.latch = latch;
@@ -189,8 +210,8 @@ public class MessageListenerContainerErrorHandlerIntegrationTests {
 	}
 
 	public static class ThrowingExceptionListener implements MessageListener {
-		private CountDownLatch latch;
-		private RuntimeException exception;
+		private final CountDownLatch latch;
+		private final RuntimeException exception;
 
 		public ThrowingExceptionListener(CountDownLatch latch, RuntimeException exception) {
 			this.latch = latch;
@@ -214,8 +235,8 @@ public class MessageListenerContainerErrorHandlerIntegrationTests {
 	}
 
 	public static class ThrowingExceptionChannelAwareListener implements ChannelAwareMessageListener {
-		private CountDownLatch latch;
-		private Exception exception;
+		private final CountDownLatch latch;
+		private final Exception exception;
 
 		public ThrowingExceptionChannelAwareListener(CountDownLatch latch, Exception exception) {
 			this.latch = latch;

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2010-2013 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.listener;
 
 import static org.junit.Assert.assertEquals;
@@ -22,7 +34,15 @@ import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
+import org.springframework.beans.factory.DisposableBean;
 
+/**
+ * @author Dave Syer
+ * @author Gary Russell
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 public class MessageListenerContainerLifecycleIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(MessageListenerContainerLifecycleIntegrationTests.class);
@@ -217,10 +237,11 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 		assertEquals(0, container.getActiveConsumerCount());
 		assertNull(template.receiveAndConvert(queue.getName()));
 
+		((DisposableBean) template.getConnectionFactory()).destroy();
 	}
 
 	public static class PojoListener {
-		private AtomicInteger count = new AtomicInteger();
+		private final AtomicInteger count = new AtomicInteger();
 
 		private CountDownLatch latch;
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerMultipleQueueIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2013 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
@@ -41,6 +41,7 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter;
 
 /**
  * @author Mark Fisher
+ * @author Gunnar Hillert
  */
 public class MessageListenerContainerMultipleQueueIntegrationTests {
 
@@ -142,6 +143,8 @@ public class MessageListenerContainerMultipleQueueIntegrationTests {
 		}
 		assertNull(template.receiveAndConvert(queue1.getName()));
 		assertNull(template.receiveAndConvert(queue2.getName()));
+
+		connectionFactory.destroy();
 	}
 
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2002-2013 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.listener;
 
 import static org.junit.Assert.assertEquals;
@@ -32,10 +44,18 @@ import org.springframework.amqp.rabbit.test.RepeatProcessor;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.amqp.utils.SerializationUtils;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.retry.policy.MapRetryContextCache;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.test.annotation.Repeat;
 
+/**
+ * @author Dave Syer
+ * @author Gary Russell
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 public class MessageListenerContainerRetryIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(MessageListenerContainerRetryIntegrationTests.class);
@@ -236,13 +256,15 @@ public class MessageListenerContainerRetryIntegrationTests {
 
 		} finally {
 			container.shutdown();
+			((DisposableBean) template.getConnectionFactory()).destroy();
+
 			assertEquals(0, container.getActiveConsumerCount());
 		}
 
 	}
 
 	public static class PojoListener {
-		private AtomicInteger count = new AtomicInteger();
+		private final AtomicInteger count = new AtomicInteger();
 		private final int failFrequency;
 
 		public PojoListener(int failFrequency) {

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2002-2013 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.listener;
 
 import static org.junit.Assert.assertNull;
@@ -23,22 +35,29 @@ import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
+import org.springframework.beans.factory.DisposableBean;
 
 import com.rabbitmq.client.Channel;
 
+/**
+ * @author Dave Syer
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 public class MessageListenerManualAckIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(MessageListenerManualAckIntegrationTests.class);
 
-	private Queue queue = new Queue("test.queue");
+	private final Queue queue = new Queue("test.queue");
 
-	private RabbitTemplate template = new RabbitTemplate();
+	private final RabbitTemplate template = new RabbitTemplate();
 
-	private int concurrentConsumers = 1;
+	private final int concurrentConsumers = 1;
 
-	private int messageCount = 50;
+	private final int messageCount = 50;
 
-	private int txSize = 1;
+	private final int txSize = 1;
 
 	private boolean transactional = false;
 
@@ -67,6 +86,8 @@ public class MessageListenerManualAckIntegrationTests {
 		if (container != null) {
 			container.shutdown();
 		}
+
+		((DisposableBean) template.getConnectionFactory()).destroy();
 	}
 
 	@Test

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java

@@ -31,20 +31,27 @@ import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
+import org.springframework.beans.factory.DisposableBean;
 
 import com.rabbitmq.client.Channel;
 
+/**
+ * @author Dave Syer
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(MessageListenerRecoveryCachingConnectionIntegrationTests.class);
 
-	private Queue queue = new Queue("test.queue");
+	private final Queue queue = new Queue("test.queue");
 
-	private Queue sendQueue = new Queue("test.send");
+	private final Queue sendQueue = new Queue("test.send");
 
 	private int concurrentConsumers = 1;
 
-	private int messageCount = 10;
+	private final int messageCount = 10;
 
 	private boolean transactional = false;
 
@@ -104,6 +111,8 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 		assertEquals("bar", new String(bytes));
 		assertEquals(null, template.receiveAndConvert(queue.getName()));
 
+		((DisposableBean) connectionFactory).destroy();
+
 	}
 
 	@Test
@@ -134,17 +143,21 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 		// Sending of bar message is also rolled back
 		assertNull(template.receiveAndConvert(sendQueue.getName()));
 
+		((DisposableBean) connectionFactory).destroy();
+
 	}
 
 	@Test
 	public void testListenerRecoversFromBogusDoubleAck() throws Exception {
 
-		RabbitTemplate template = new RabbitTemplate(createConnectionFactory());
+		ConnectionFactory connectionFactory1 = createConnectionFactory();
+		RabbitTemplate template = new RabbitTemplate(connectionFactory1);
 
 		acknowledgeMode = AcknowledgeMode.MANUAL;
 
 		CountDownLatch latch = new CountDownLatch(messageCount);
-		container = createContainer(queue.getName(), new ManualAckListener(latch), createConnectionFactory());
+		ConnectionFactory connectionFactory2 = createConnectionFactory();
+		container = createContainer(queue.getName(), new ManualAckListener(latch), connectionFactory2);
 		for (int i = 0; i < messageCount; i++) {
 			template.convertAndSend(queue.getName(), i + "foo");
 		}
@@ -156,15 +169,19 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 
 		assertNull(template.receiveAndConvert(queue.getName()));
 
+		((DisposableBean) connectionFactory1).destroy();
+		((DisposableBean) connectionFactory2).destroy();
 	}
 
 	@Test
 	public void testListenerRecoversFromClosedChannel() throws Exception {
 
-		RabbitTemplate template = new RabbitTemplate(createConnectionFactory());
+		ConnectionFactory connectionFactory1 = createConnectionFactory();
+		RabbitTemplate template = new RabbitTemplate(connectionFactory1);
 
 		CountDownLatch latch = new CountDownLatch(messageCount);
-		container = createContainer(queue.getName(), new AbortChannelListener(latch), createConnectionFactory());
+		ConnectionFactory connectionFactory2 = createConnectionFactory();
+		container = createContainer(queue.getName(), new AbortChannelListener(latch), connectionFactory2);
 		for (int i = 0; i < messageCount; i++) {
 			template.convertAndSend(queue.getName(), i + "foo");
 		}
@@ -176,15 +193,19 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 
 		assertNull(template.receiveAndConvert(queue.getName()));
 
+		((DisposableBean) connectionFactory1).destroy();
+		((DisposableBean) connectionFactory2).destroy();
 	}
 
 	@Test
 	public void testListenerRecoversFromClosedChannelAndStop() throws Exception {
 
-		RabbitTemplate template = new RabbitTemplate(createConnectionFactory());
+		ConnectionFactory connectionFactory1 = createConnectionFactory();
+		RabbitTemplate template = new RabbitTemplate(connectionFactory1);
 
 		CountDownLatch latch = new CountDownLatch(messageCount);
-		container = createContainer(queue.getName(), new AbortChannelListener(latch), createConnectionFactory());
+		ConnectionFactory connectionFactory2 = createConnectionFactory();
+		container = createContainer(queue.getName(), new AbortChannelListener(latch), connectionFactory2);
 		int n = 0;
 		while (n++ < 100 && container.getActiveConsumerCount() != concurrentConsumers) {
 			Thread.sleep(50L);
@@ -206,18 +227,22 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 		container.stop();
 		assertEquals(0, container.getActiveConsumerCount());
 
+		((DisposableBean) connectionFactory1).destroy();
+		((DisposableBean) connectionFactory2).destroy();
+
 	}
 
 	@Test
 	public void testListenerRecoversFromClosedConnection() throws Exception {
 
-		RabbitTemplate template = new RabbitTemplate(createConnectionFactory());
+		ConnectionFactory connectionFactory1 = createConnectionFactory();
+		RabbitTemplate template = new RabbitTemplate(connectionFactory1);
 
 		CountDownLatch latch = new CountDownLatch(messageCount);
-		ConnectionFactory connectionFactory = createConnectionFactory();
+		ConnectionFactory connectionFactory2 = createConnectionFactory();
 		container = createContainer(queue.getName(),
-				new CloseConnectionListener((ConnectionProxy) connectionFactory.createConnection(), latch),
-				connectionFactory);
+				new CloseConnectionListener((ConnectionProxy) connectionFactory2.createConnection(), latch),
+				connectionFactory2);
 		for (int i = 0; i < messageCount; i++) {
 			template.convertAndSend(queue.getName(), i + "foo");
 		}
@@ -229,6 +254,9 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 
 		assertNull(template.receiveAndConvert(queue.getName()));
 
+		((DisposableBean) connectionFactory1).destroy();
+		((DisposableBean) connectionFactory2).destroy();
+
 	}
 
 	@Test
@@ -254,13 +282,24 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 
 		assertNull(template.receiveAndConvert(queue.getName()));
 
+		((DisposableBean) connectionFactory).destroy();
+
 	}
 
 	@Test(expected = AmqpIllegalStateException.class)
 	public void testListenerDoesNotRecoverFromMissingQueue() throws Exception {
+
 		concurrentConsumers = 3;
 		CountDownLatch latch = new CountDownLatch(messageCount);
-		container = createContainer("nonexistent", new VanillaListener(latch), createConnectionFactory());
+
+		ConnectionFactory connectionFactory = createConnectionFactory();
+
+		try {
+			container = createContainer("nonexistent", new VanillaListener(latch), connectionFactory);
+		}
+		finally {
+			((DisposableBean) connectionFactory).destroy();
+		}
 	}
 
 	@Test(expected = AmqpIllegalStateException.class)
@@ -271,7 +310,14 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 		 */
 		concurrentConsumers = 1;
 		CountDownLatch latch = new CountDownLatch(messageCount);
-		container = createContainer("nonexistent", new VanillaListener(latch), createConnectionFactory());
+
+		ConnectionFactory connectionFactory = createConnectionFactory();
+		try {
+			container = createContainer("nonexistent", new VanillaListener(latch), connectionFactory);
+		}
+		finally {
+			((DisposableBean) connectionFactory).destroy();
+		}
 	}
 
 	private int getTimeout() {
@@ -293,7 +339,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 
 	public static class ManualAckListener implements ChannelAwareMessageListener {
 
-		private AtomicBoolean failed = new AtomicBoolean(false);
+		private final AtomicBoolean failed = new AtomicBoolean(false);
 
 		private final CountDownLatch latch;
 
@@ -348,7 +394,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 
 	public static class AbortChannelListener implements ChannelAwareMessageListener {
 
-		private AtomicBoolean failed = new AtomicBoolean(false);
+		private final AtomicBoolean failed = new AtomicBoolean(false);
 
 		private final CountDownLatch latch;
 
@@ -370,7 +416,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 
 	public static class CloseConnectionListener implements ChannelAwareMessageListener {
 
-		private AtomicBoolean failed = new AtomicBoolean(false);
+		private final AtomicBoolean failed = new AtomicBoolean(false);
 
 		private final CountDownLatch latch;
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryRepeatIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2002-2013 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.listener;
 
 import static org.junit.Assert.assertNull;
@@ -13,7 +25,6 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.log4j.Level;
 import org.junit.After;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.springframework.amqp.core.AcknowledgeMode;
@@ -28,6 +39,7 @@ import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
 import org.springframework.amqp.rabbit.test.RepeatProcessor;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.test.annotation.Repeat;
 
 import com.rabbitmq.client.Channel;
@@ -36,26 +48,26 @@ import com.rabbitmq.client.Channel;
  * Long-running test created to facilitate profiling of SimpleMessageListenerContainer.
  *
  * @author Dave Syer
+ * @author Gunnar Hillert
  *
  */
-@Ignore
 public class MessageListenerRecoveryRepeatIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(MessageListenerRecoveryRepeatIntegrationTests.class);
 
-	private Queue queue = new Queue("test.queue");
+	private final Queue queue = new Queue("test.queue");
 
-	private Queue sendQueue = new Queue("test.send");
+	private final Queue sendQueue = new Queue("test.send");
 
-	private int concurrentConsumers = 1;
+	private final int concurrentConsumers = 1;
 
-	private int messageCount = 2;
+	private final int messageCount = 2;
 
-	private int txSize = 1;
+	private final int txSize = 1;
 
-	private boolean transactional = false;
+	private final boolean transactional = false;
 
-	private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
+	private final AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
 
 	private SimpleMessageListenerContainer container;
 
@@ -92,6 +104,9 @@ public class MessageListenerRecoveryRepeatIntegrationTests {
 			if (container != null) {
 				container.shutdown();
 			}
+			if (connectionFactory != null) {
+				((DisposableBean) connectionFactory).destroy();
+			}
 		}
 	}
 
@@ -144,7 +159,7 @@ public class MessageListenerRecoveryRepeatIntegrationTests {
 
 	private static class CloseConnectionListener implements ChannelAwareMessageListener {
 
-		private AtomicBoolean failed = new AtomicBoolean(false);
+		private final AtomicBoolean failed = new AtomicBoolean(false);
 
 		private CountDownLatch latch;
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2002-2013 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.listener;
 
 import static org.junit.Assert.assertNull;
@@ -23,22 +35,29 @@ import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
+import org.springframework.beans.factory.DisposableBean;
 
 import com.rabbitmq.client.Channel;
 
+/**
+ * @author Dave Syer
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 public class MessageListenerTxSizeIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(MessageListenerTxSizeIntegrationTests.class);
 
-	private Queue queue = new Queue("test.queue");
+	private final Queue queue = new Queue("test.queue");
 
-	private RabbitTemplate template = new RabbitTemplate();
+	private final RabbitTemplate template = new RabbitTemplate();
 
-	private int concurrentConsumers = 1;
+	private final int concurrentConsumers = 1;
 
-	private int messageCount = 12;
+	private final int messageCount = 12;
 
-	private int txSize = 4;
+	private final int txSize = 4;
 
 	private boolean transactional = true;
 
@@ -67,6 +86,9 @@ public class MessageListenerTxSizeIntegrationTests {
 		if (container != null) {
 			container.shutdown();
 		}
+
+		((DisposableBean) template.getConnectionFactory()).destroy();
+
 	}
 
 	@Test
@@ -115,7 +137,7 @@ public class MessageListenerTxSizeIntegrationTests {
 
 	public class TestListener implements ChannelAwareMessageListener {
 
-		private ThreadLocal<Integer> count = new ThreadLocal<Integer>();
+		private final ThreadLocal<Integer> count = new ThreadLocal<Integer>();
 
 		private final CountDownLatch latch;
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java

@@ -1,3 +1,15 @@
+/*
+ * Copyright 2002-2013 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.listener;
 
 import static org.junit.Assert.assertNotNull;
@@ -31,6 +43,7 @@ import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.TransactionException;
 import org.springframework.transaction.support.AbstractPlatformTransactionManager;
@@ -38,14 +51,20 @@ import org.springframework.transaction.support.DefaultTransactionStatus;
 
 import com.rabbitmq.client.Channel;
 
+/**
+ * @author Dave Syer
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 @RunWith(Parameterized.class)
 public class SimpleMessageListenerContainerIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(SimpleMessageListenerContainerIntegrationTests.class);
 
-	private Queue queue = new Queue("test.queue");
+	private final Queue queue = new Queue("test.queue");
 
-	private RabbitTemplate template = new RabbitTemplate();
+	private final RabbitTemplate template = new RabbitTemplate();
 
 	private final int concurrentConsumers;
 
@@ -140,6 +159,7 @@ public class SimpleMessageListenerContainerIntegrationTests {
 		if (container != null) {
 			container.shutdown();
 		}
+		((DisposableBean) template.getConnectionFactory()).destroy();
 	}
 
 	@Test
@@ -235,7 +255,7 @@ public class SimpleMessageListenerContainerIntegrationTests {
 	}
 
 	public static class PojoListener {
-		private AtomicInteger count = new AtomicInteger();
+		private final AtomicInteger count = new AtomicInteger();
 
 		private final CountDownLatch latch;
 
@@ -266,7 +286,7 @@ public class SimpleMessageListenerContainerIntegrationTests {
 	}
 
 	public static class Listener implements MessageListener {
-		private AtomicInteger count = new AtomicInteger();
+		private final AtomicInteger count = new AtomicInteger();
 
 		private final CountDownLatch latch;
 
@@ -298,7 +318,7 @@ public class SimpleMessageListenerContainerIntegrationTests {
 	}
 
 	public static class ChannelAwareListener implements ChannelAwareMessageListener {
-		private AtomicInteger count = new AtomicInteger();
+		private final AtomicInteger count = new AtomicInteger();
 
 		private final CountDownLatch latch;
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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
@@ -28,6 +28,7 @@ import org.springframework.transaction.support.DefaultTransactionStatus;
 
 /**
  * @author David Syer
+ * @author Gunnar Hillert
  *
  */
 public class SimpleMessageListenerContainerTests {
@@ -37,7 +38,8 @@ public class SimpleMessageListenerContainerTests {
 
 	@Test
 	public void testInconsistentTransactionConfiguration() throws Exception {
-		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(new SingleConnectionFactory());
+		final SingleConnectionFactory singleConnectionFactory = new SingleConnectionFactory();
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(singleConnectionFactory);
 		container.setMessageListener(new MessageListenerAdapter(this));
 		container.setQueueNames("foo");
 		container.setChannelTransacted(false);
@@ -45,32 +47,38 @@ public class SimpleMessageListenerContainerTests {
 		container.setTransactionManager(new TestTransactionManager());
 		expectedException.expect(IllegalStateException.class);
 		container.afterPropertiesSet();
+		singleConnectionFactory.destroy();
 	}
 
 	@Test
 	public void testInconsistentAcknowledgeConfiguration() throws Exception {
-		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(new SingleConnectionFactory());
+		final SingleConnectionFactory singleConnectionFactory = new SingleConnectionFactory();
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(singleConnectionFactory);
 		container.setMessageListener(new MessageListenerAdapter(this));
 		container.setQueueNames("foo");
 		container.setChannelTransacted(true);
 		container.setAcknowledgeMode(AcknowledgeMode.NONE);
 		expectedException.expect(IllegalStateException.class);
 		container.afterPropertiesSet();
+		singleConnectionFactory.destroy();
 	}
 
 	@Test
 	public void testDefaultConsumerCount() throws Exception {
-		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(new SingleConnectionFactory());
+		final SingleConnectionFactory singleConnectionFactory = new SingleConnectionFactory();
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(singleConnectionFactory);
 		container.setMessageListener(new MessageListenerAdapter(this));
 		container.setQueueNames("foo");
 		container.setAutoStartup(false);
 		container.afterPropertiesSet();
 		assertEquals(1, ReflectionTestUtils.getField(container, "concurrentConsumers"));
+		singleConnectionFactory.destroy();
 	}
 
 	@Test
 	public void testLazyConsumerCount() throws Exception {
-		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(new SingleConnectionFactory()) {
+		final SingleConnectionFactory singleConnectionFactory = new SingleConnectionFactory();
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(singleConnectionFactory) {
 			@Override
 			protected void doStart() throws Exception {
 				// do nothing
@@ -78,6 +86,7 @@ public class SimpleMessageListenerContainerTests {
 		};
 		container.start();
 		assertEquals(1, ReflectionTestUtils.getField(container, "concurrentConsumers"));
+		singleConnectionFactory.destroy();
 	}
 
 	@SuppressWarnings("serial")

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/StopStartIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -28,16 +28,19 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Gary Russell
+ * @author Gunnar Hillert
  * @since 1.1.3
  *
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 public class StopStartIntegrationTests {
 
 	@Rule

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011-2012 by the original author(s).
+ * Copyright (c) 2011-2013 by the original author(s).
  *
  * 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
@@ -33,6 +33,7 @@ import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.Log4jConfigurer;
@@ -40,9 +41,11 @@ import org.springframework.util.Log4jConfigurer;
 /**
  * @author Jon Brisbin <jbrisbin@vmware.com>
  * @author Gary Russell
+ * @author Gunnar Hillert
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration(locations = { "org.springframework.amqp.rabbit.log4j" }, loader = AnnotationConfigContextLoader.class)
+@DirtiesContext
 public class AmqpAppenderIntegrationTests {
 
 	@Rule

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManagerIntegrationTests.java

@@ -1,18 +1,38 @@
+/*
+ * Copyright (c) 2011-2013 by the original author(s).
+ *
+ * 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 static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
+import org.junit.After;
 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.beans.factory.DisposableBean;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.support.TransactionCallback;
 import org.springframework.transaction.support.TransactionTemplate;
 
+/**
+ * @author David Syer
+ * @author Gunnar Hillert
+ * @since 1.0
+ *
+ */
 public class RabbitTransactionManagerIntegrationTests {
 
 	private static final String ROUTE = "test.queue";
@@ -33,6 +53,11 @@ public class RabbitTransactionManagerIntegrationTests {
 		transactionTemplate = new TransactionTemplate(transactionManager);
 	}
 
+	@After
+	public void cleanup() throws Exception {
+		((DisposableBean) this.template.getConnectionFactory()).destroy();
+	}
+
 	@Test
 	public void testSendAndReceiveInTransaction() throws Exception {
 		String result = transactionTemplate.execute(new TransactionCallback<String>() {