AMQP-103, AMQP-126: add tests for re-declaration and fix bug exposed in tx handling

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

@@ -32,6 +32,12 @@ public class CompositeConnectionListener implements ConnectionListener {
 		}	
 	}
 
+	public void onClose(Connection connection) {
+		for (ConnectionListener delegate : delegates) {
+			delegate.onClose(connection);
+		}	
+	}
+
 	public void setDelegates(List<? extends ConnectionListener> delegates) {
 		this.delegates = new ArrayList<ConnectionListener>(delegates);
 	}

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

@@ -23,4 +23,6 @@ public interface ConnectionListener {
 	
 	void onCreate(Connection connection);
 
+	void onClose(Connection connection);
+
 }

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

@@ -147,6 +147,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
 	public final void destroy() {
 		synchronized (this.connectionMonitor) {
 			if (this.targetConnection != null) {
+				listener.onClose(targetConnection);
 				RabbitUtils.closeConnection(this.targetConnection);
 			}
 			this.targetConnection = null;

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java

@@ -223,6 +223,10 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
 					initialized = true;
 				}
 
+				public void onClose(Connection connection) {
+					initialized = false;
+				}
+
 			});
 
 			this.running = true;

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java

@@ -124,7 +124,7 @@ public class BlockingQueueConsumer {
 	 * @throws ShutdownSignalException if the connection is shut down while waiting
 	 */
 	public Message nextMessage() throws InterruptedException, ShutdownSignalException {
-		logger.debug("Retrieving delivery for " + this);
+		logger.trace("Retrieving delivery for " + this);
 		return handle(queue.take());
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java

@@ -35,10 +35,11 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.jmx.export.annotation.ManagedMetric;
 import org.springframework.jmx.support.MetricType;
 import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
-import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
 import org.springframework.transaction.interceptor.TransactionAttribute;
-import org.springframework.transaction.interceptor.TransactionInterceptor;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.Assert;
 
 import com.rabbitmq.client.Channel;
@@ -237,16 +238,10 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	}
 
 	private void initializeProxy() {
-		if (advices.length == 0 && transactionManager == null) {
+		if (advices.length == 0) {
 			return;
 		}
 		ProxyFactory factory = new ProxyFactory();
-		if (transactionManager != null) {
-			MatchAlwaysTransactionAttributeSource txAttributeSource = new MatchAlwaysTransactionAttributeSource();
-			txAttributeSource.setTransactionAttribute(transactionAttribute);
-			Advice txAdvice = new TransactionInterceptor(transactionManager, txAttributeSource);
-			factory.addAdvisor(new DefaultPointcutAdvisor(Pointcut.TRUE, txAdvice));
-		}
 		for (Advice advice : getAdvices()) {
 			factory.addAdvisor(new DefaultPointcutAdvisor(Pointcut.TRUE, advice));
 		}
@@ -391,26 +386,44 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		}
 	}
 
-	private boolean receiveAndExecute(BlockingQueueConsumer consumer) throws Throwable {
+	private boolean receiveAndExecute(final BlockingQueueConsumer consumer) throws Throwable {
+
+		if (transactionManager != null) {
+			try {
+				return new TransactionTemplate(transactionManager, transactionAttribute)
+				.execute(new TransactionCallback<Boolean>() {
+					public Boolean doInTransaction(TransactionStatus status) {
+						ConnectionFactoryUtils.bindResourceToTransaction(
+								new RabbitResourceHolder(consumer.getChannel()), getConnectionFactory(), true);
+						try {
+							return doReceiveAndExecute(consumer);
+						} catch (RuntimeException e) {
+							throw e;
+						} catch (Throwable e) {
+							throw new WrappedTransactionException(e);
+						}
+					}
+				});
+			} catch (WrappedTransactionException e) {
+				throw e.getCause();
+			}
+		}
+
+		return doReceiveAndExecute(consumer);
+
+	}
+
+	private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Throwable {
 
 		Channel channel = consumer.getChannel();
 
-		int totalMsgCount = 0;
-
-		ConnectionFactory connectionFactory = getConnectionFactory();
-		if (getAcknowledgeMode().isTransactionAllowed()) {
-			ConnectionFactoryUtils
-					.bindResourceToTransaction(new RabbitResourceHolder(channel), connectionFactory, true);
-		}
-
 		for (int i = 0; i < txSize; i++) {
 
-			logger.debug("Waiting for message from consumer.");
+			logger.trace("Waiting for message from consumer.");
 			Message message = consumer.nextMessage(receiveTimeout);
 			if (message == null) {
 				return false;
 			}
-			totalMsgCount++;
 			executeListener(channel, message);
 
 		}
@@ -525,7 +538,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		}
 
 	}
-	
+
 	@Override
 	protected void invokeListener(Channel channel, Message message) throws Exception {
 		proxy.invokeListener(channel, message);
@@ -549,5 +562,12 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 			throw new IllegalStateException("Unrecoverable interruption on consumer restart");
 		}
 	}
+	
+	@SuppressWarnings("serial")
+	private static class WrappedTransactionException extends RuntimeException {
+		public WrappedTransactionException(Throwable cause) {
+			super(cause);
+		}
+	}
 
 }

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

@@ -294,19 +294,18 @@ public class CachingConnectionFactoryTests {
 
 		Connection con = ccf.createConnection();
 
-		Channel channel1 = con.createChannel(false); // This will return a
-														// Spring AOP proxy that
-														// surpresses calls to
-														// close
-		Channel channel2 = con.createChannel(false); // "
+		// This will return a proxy that surpresses calls to close
+		Channel channel1 = con.createChannel(false); 
+		Channel channel2 = con.createChannel(false);
 
-		channel1.close(); // should be ignored, and add last into channel cache.
-		channel2.close(); // "
+		// Should be ignored, and add last into channel cache.
+		channel1.close(); 
+		channel2.close();
 
-		Channel ch1 = con.createChannel(false); // remove first entry in cache
-												// (channel1)
-		Channel ch2 = con.createChannel(false); // remove first entry in cache
-												// (channel2)
+		// remove first entry in cache (channel1)
+		Channel ch1 = con.createChannel(false);
+		// remove first entry in cache (channel2)
+		Channel ch2 = con.createChannel(false); 
 
 		Assert.assertSame(ch1, channel1);
 		Assert.assertSame(ch2, channel2);
@@ -314,9 +313,8 @@ public class CachingConnectionFactoryTests {
 		Channel target1 = ((ChannelProxy) ch1).getTargetChannel();
 		Channel target2 = ((ChannelProxy) ch2).getTargetChannel();
 
-		Assert.assertNotSame(target1, target2); // make sure mokito returned
-												// different mocks for the
-												// channel
+		// make sure mokito returned different mocks for the channel
+		Assert.assertNotSame(target1, target2); 
 
 		ch1.close();
 		ch2.close();
@@ -331,5 +329,14 @@ public class CachingConnectionFactoryTests {
 		// verify(mockChannel1).close();
 		verify(mockChannel2, times(1)).close();
 
+		// After destroy we can get a new connection
+		Connection con1 = ccf.createConnection();
+		Assert.assertNotSame(con, con1);
+
+		// This will return a proxy that surpresses calls to close
+		Channel channel3 = con.createChannel(false); 
+		Assert.assertNotSame(channel3, channel1);
+		Assert.assertNotSame(channel3, channel2);
 	}
+
 }

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

@@ -1,6 +1,8 @@
 package org.springframework.amqp.rabbit.connection;
 
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
@@ -18,29 +20,35 @@ import org.junit.Test;
 public class SingleConnectionFactoryTests {
 
 	@Test
-
 	public void testWithListener() throws IOException {
 
 		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
 		com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
-		
+
 		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
-		
+
 		final AtomicBoolean called = new AtomicBoolean(false);
-		SingleConnectionFactory ccf = new SingleConnectionFactory(mockConnectionFactory);
-		ccf.setConnectionListeners(Arrays.asList(new ConnectionListener(){
+		SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory);
+		connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() {
 			public void onCreate(Connection connection) {
 				called.set(true);
-			}	
+			}
+			public void onClose(Connection connection) {
+				called.set(false);
+			}
 		}));
-		Connection con = ccf.createConnection();
-		
+
+		Connection con = connectionFactory.createConnection();
 		assertTrue(called.get());
-		
+
 		con.close();
-		
+		assertTrue(called.get());
 		verify(mockConnection, never()).close();
-		
+
+		connectionFactory.destroy();
+		assertFalse(called.get());
+		verify(mockConnection, atLeastOnce()).close();
+
 	}
 
 }

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

@@ -1,7 +1,12 @@
 package org.springframework.amqp.rabbit.core;
 
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.After;
+import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.springframework.amqp.core.Queue;
@@ -10,29 +15,113 @@ import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.context.support.GenericApplicationContext;
 
+import com.rabbitmq.client.AMQP.Queue.DeclareOk;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
 public class RabbitAdminIntegrationTests {
 
-	private static Queue queue = new Queue("test.queue");
-
 	private CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
-	
+
 	@Rule
 	public BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
-	
+
+	private GenericApplicationContext context;
+
+	private RabbitAdmin rabbitAdmin;
+
 	public RabbitAdminIntegrationTests() {
 		connectionFactory.setPort(BrokerTestUtils.getPort());
 	}
-	
+
+	@Before
+	public void init() {
+		context = new GenericApplicationContext();
+		rabbitAdmin = new RabbitAdmin(connectionFactory);
+		rabbitAdmin.setApplicationContext(context);
+		rabbitAdmin.setAutoStartup(true);
+	}
+
+	@After
+	public void close() {
+		if (context != null) {
+			context.close();
+		}
+	}
+
 	@Test
 	public void testStartupWithBroker() throws Exception {
-		GenericApplicationContext applicationContext = new GenericApplicationContext();
-		applicationContext.getBeanFactory().registerSingleton("foo", queue);
-		RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
-		rabbitAdmin.setApplicationContext(applicationContext);
-		rabbitAdmin.setAutoStartup(true);
+		Queue queue = new Queue("test.queue");
+		context.getBeanFactory().registerSingleton("foo", queue);
 		rabbitAdmin.deleteQueue(queue.getName());
 		rabbitAdmin.afterPropertiesSet();
 		assertTrue(rabbitAdmin.deleteQueue(queue.getName()));
 	}
 
+	@Test
+	public void testStartupWithAutodelete() throws Exception {
+
+		final Queue queue = new Queue("test.queue", false, true, true);
+		context.getBeanFactory().registerSingleton("foo", queue);
+		rabbitAdmin.deleteQueue(queue.getName());
+		rabbitAdmin.afterPropertiesSet();
+		
+		final AtomicReference<Connection> connectionHolder = new AtomicReference<Connection>();
+
+		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
+		// Force RabbitAdmin to initialize the queue
+		boolean exists = rabbitTemplate.execute(new ChannelCallback<Boolean>() {
+			public Boolean doInRabbit(Channel channel) throws Exception {
+				DeclareOk result = channel.queueDeclarePassive(queue.getName());
+				connectionHolder.set(channel.getConnection());
+				return result != null;
+			}
+		});
+		assertTrue("Expected Queue to exist", exists);
+
+		assertTrue(queueExists(connectionHolder.get(), queue));
+		connectionFactory.destroy();
+		// Broker now deletes queue (only verifiable in native API)
+		assertFalse(queueExists(null, queue));
+
+		// Broker auto-deleted queue, but it is re-created by the connection listener
+		exists = rabbitTemplate.execute(new ChannelCallback<Boolean>() {
+			public Boolean doInRabbit(Channel channel) throws Exception {
+				DeclareOk result = channel.queueDeclarePassive(queue.getName());
+				connectionHolder.set(channel.getConnection());
+				return result != null;
+			}
+		});
+		assertTrue("Expected Queue to exist", exists);
+
+		assertTrue(queueExists(connectionHolder.get(), queue));
+		assertTrue(rabbitAdmin.deleteQueue(queue.getName()));
+		assertFalse(queueExists(null, queue));
+
+	}
+
+	/**
+	 * Use native Rabbit API to test queue, bypassing all the connection and channel caching and callbacks in Spring
+	 * AMQP.
+	 * 
+	 * @param connection the raw connection to use
+	 * @param queue the Queue to test
+	 * @return true if the queue exists
+	 */
+	private boolean queueExists(Connection connection, Queue queue) throws Exception {
+		if (connection==null) {
+			ConnectionFactory connectionFactory = new ConnectionFactory();
+			connectionFactory.setPort(BrokerTestUtils.getPort());
+			connection = connectionFactory.newConnection();
+		}
+		Channel channel = connection.createChannel();
+		try {
+			DeclareOk result = channel.queueDeclarePassive(queue.getName());
+			return result != null;
+		} catch (Exception e) {
+			return false;
+		}
+	}
+
 }

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

@@ -12,7 +12,6 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.log4j.Level;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -30,7 +29,6 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
 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.transaction.TransactionDefinition;
 import org.springframework.transaction.TransactionException;
 import org.springframework.transaction.support.AbstractPlatformTransactionManager;
@@ -51,9 +49,9 @@ public class SimpleMessageListenerContainerIntegrationTests {
 
 	private final AcknowledgeMode acknowledgeMode;
 
-	@Rule
-	public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.ERROR, RabbitTemplate.class,
-			SimpleMessageListenerContainer.class, BlockingQueueConsumer.class);
+	// @Rule
+	// public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.ERROR, RabbitTemplate.class,
+	// SimpleMessageListenerContainer.class, BlockingQueueConsumer.class);
 
 	@Rule
 	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue);
@@ -96,7 +94,8 @@ public class SimpleMessageListenerContainerIntegrationTests {
 				);
 	}
 
-	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, boolean transactional, int txSize) {
+	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode,
+			boolean transactional, int txSize) {
 		// "i" is just a counter to make it easier to identify the test in the log
 		return new Object[] { messageCount, concurrency, acknowledgeMode, transactional, txSize, false };
 	}
@@ -106,7 +105,8 @@ public class SimpleMessageListenerContainerIntegrationTests {
 		return params(i, messageCount, concurrency, acknowledgeMode, acknowledgeMode.isTransactionAllowed(), txSize);
 	}
 
-	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, boolean transactional) {
+	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode,
+			boolean transactional) {
 		return params(i, messageCount, concurrency, acknowledgeMode, transactional, 1);
 	}
 
@@ -141,37 +141,37 @@ public class SimpleMessageListenerContainerIntegrationTests {
 		CountDownLatch latch = new CountDownLatch(messageCount);
 		doSunnyDayTest(latch, new MessageListenerAdapter(new PojoListener(latch)));
 	}
-	
+
 	@Test
 	public void testListenerSunnyDay() throws Exception {
 		CountDownLatch latch = new CountDownLatch(messageCount);
 		doSunnyDayTest(latch, new Listener(latch));
 	}
-	
+
 	@Test
 	public void testChannelAwareListenerSunnyDay() throws Exception {
 		CountDownLatch latch = new CountDownLatch(messageCount);
 		doSunnyDayTest(latch, new ChannelAwareListener(latch));
 	}
-	
+
 	@Test
 	public void testPojoListenerWithException() throws Exception {
 		CountDownLatch latch = new CountDownLatch(messageCount);
 		doListenerWithExceptionTest(latch, new MessageListenerAdapter(new PojoListener(latch, true)));
 	}
-	
+
 	@Test
 	public void testListenerWithException() throws Exception {
 		CountDownLatch latch = new CountDownLatch(messageCount);
 		doListenerWithExceptionTest(latch, new Listener(latch, true));
 	}
-	
+
 	@Test
 	public void testChannelAwareListenerWithException() throws Exception {
 		CountDownLatch latch = new CountDownLatch(messageCount);
 		doListenerWithExceptionTest(latch, new ChannelAwareListener(latch, true));
 	}
-	
+
 	private void doSunnyDayTest(CountDownLatch latch, Object listener) throws Exception {
 		container = createContainer(listener);
 		for (int i = 0; i < messageCount; i++) {
@@ -181,7 +181,7 @@ public class SimpleMessageListenerContainerIntegrationTests {
 		assertTrue("Timed out waiting for message", waited);
 		assertNull(template.receiveAndConvert(queue.getName()));
 	}
-	
+
 	private void doListenerWithExceptionTest(CountDownLatch latch, Object listener) throws Exception {
 		container = createContainer(listener);
 		if (acknowledgeMode.isTransactionAllowed()) {
@@ -248,7 +248,7 @@ public class SimpleMessageListenerContainerIntegrationTests {
 			try {
 				int counter = count.getAndIncrement();
 				if (logger.isDebugEnabled() && counter % 500 == 0) {
-					logger.debug(value + counter);
+					logger.debug("Handling: " + value + ":" + counter + " - " + latch);
 				}
 				if (fail) {
 					throw new RuntimeException("Planned failure");
@@ -258,23 +258,23 @@ public class SimpleMessageListenerContainerIntegrationTests {
 			}
 		}
 	}
-	
+
 	public static class Listener implements MessageListener {
 		private AtomicInteger count = new AtomicInteger();
-		
+
 		private final CountDownLatch latch;
-		
+
 		private final boolean fail;
-		
+
 		public Listener(CountDownLatch latch) {
 			this(latch, false);
 		}
-		
+
 		public Listener(CountDownLatch latch, boolean fail) {
 			this.latch = latch;
 			this.fail = fail;
 		}
-		
+
 		public void onMessage(Message message) {
 			String value = new String(message.getBody());
 			try {
@@ -290,23 +290,23 @@ public class SimpleMessageListenerContainerIntegrationTests {
 			}
 		}
 	}
-	
+
 	public static class ChannelAwareListener implements ChannelAwareMessageListener {
 		private AtomicInteger count = new AtomicInteger();
-		
+
 		private final CountDownLatch latch;
-		
+
 		private final boolean fail;
-		
+
 		public ChannelAwareListener(CountDownLatch latch) {
 			this(latch, false);
 		}
-		
+
 		public ChannelAwareListener(CountDownLatch latch, boolean fail) {
 			this.latch = latch;
 			this.fail = fail;
 		}
-		
+
 		public void onMessage(Message message, Channel channel) throws Exception {
 			String value = new String(message.getBody());
 			try {

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java

@@ -39,6 +39,15 @@ public class BrokerTestUtils {
 		return DEFAULT_PORT;
 	}
 
+	/**
+	 * The port that the tracer is listening on (e.g. as input for a {@link ConnectionFactory}).
+	 * 
+	 * @return a port number
+	 */
+	public static int getTracerPort() {
+		return TRACER_PORT;
+	}
+
 	/**
 	 * An alternative port number than can safely be used to stop and start a broker, even when one is already running
 	 * on the standard port as a privileged user. Useful for tests involving {@link RabbitBrokerAdmin} on UN*X.