Merge pull request #95 from garyrussell/AMQP-309

* garyrussell-AMQP-309:
  AMQP-309 Fix Listener Channel Issue

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

@@ -117,12 +117,15 @@ public class ConnectionFactoryUtils {
 		Connection connection = resourceFactory.getConnection(resourceHolderToUse);
 		Channel channel = null;
 		try {
-			boolean isExistingCon = (connection != null);
-			if (!isExistingCon) {
+			/*
+			 * If we are in a listener container, first see if there's a channel registered
+			 * for this consumer and the consumer is using the same connection factory.
+			 */
+			channel = ConsumerChannelRegistry.getConsumerChannel(connectionFactory);
+			if (channel == null && connection == null) {
 				connection = resourceFactory.createConnection();
 				resourceHolderToUse.addConnection(connection);
 			}
-			channel = ConsumerChannelRegistry.getConsumerChannel();
 			if (channel == null) {
 				channel = resourceFactory.createChannel(connection);
 			}

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

@@ -537,7 +537,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 					 * Register the consumer's channel so it will be used by the transaction manager
 					 * if it's an instance of RabbitTransactionManager.
 					 */
-					ConsumerChannelRegistry.registerConsumerChannel(consumer.getChannel());
+					ConsumerChannelRegistry.registerConsumerChannel(consumer.getChannel(), getConnectionFactory());
 				}
 
 				// Always better to stop receiving as soon as possible if

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ConsumerChannelRegistry.java

@@ -17,6 +17,7 @@ package org.springframework.amqp.rabbit.support;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 
 import com.rabbitmq.client.Channel;
 
@@ -35,7 +36,7 @@ public class ConsumerChannelRegistry {
 
 	private static final Log logger = LogFactory.getLog(ConsumerChannelRegistry.class);
 
-	private static final ThreadLocal<Channel> consumerChannel = new ThreadLocal<Channel>();
+	private static final ThreadLocal<ChannelHolder> consumerChannel = new ThreadLocal<ChannelHolder>();
 
 	/**
 	 * If a listener container is configured to use a RabbitTransactionManager, the
@@ -46,11 +47,12 @@ public class ConsumerChannelRegistry {
 	 * to wire in a RabbitTransactionManager.
 	 * @param channel
 	 */
-	public static void registerConsumerChannel(Channel channel) {
+	public static void registerConsumerChannel(Channel channel, ConnectionFactory connectionFactory) {
 		if (logger.isDebugEnabled()) {
-			logger.debug("Registering consumer channel" + channel);
+			logger.debug("Registering consumer channel" + channel + " from factory " +
+							connectionFactory);
 		}
-		consumerChannel.set(channel);
+		consumerChannel.set(new ChannelHolder(channel, connectionFactory));
 	}
 
 	/**
@@ -69,7 +71,45 @@ public class ConsumerChannelRegistry {
 	 * channel for this consumer.
 	 */
 	public static Channel getConsumerChannel() {
-		return consumerChannel.get();
+		ChannelHolder channelHolder = consumerChannel.get();
+		Channel channel = null;
+		if (channelHolder != null) {
+			channel = channelHolder.getChannel();
+		}
+		return channel;
 	}
 
+	/**
+	 * See registerConsumerChannel. This method is called to retrieve the
+	 * channel for this consumer if the connection factory matches.
+	 * @param connectionFactory The connection factory.
+	 */
+	public static Channel getConsumerChannel(ConnectionFactory connectionFactory) {
+		ChannelHolder channelHolder = consumerChannel.get();
+		Channel channel = null;
+		if (channelHolder != null && channelHolder.getConnectionFactory() == connectionFactory) {
+			channel = channelHolder.getChannel();
+		}
+		return channel;
+	}
+
+	private static class ChannelHolder {
+
+		private final Channel channel;
+
+		private final ConnectionFactory connectionFactory;
+
+		private ChannelHolder(Channel channel, ConnectionFactory connectionFactory) {
+			this.channel = channel;
+			this.connectionFactory = connectionFactory;
+		}
+
+		private Channel getChannel() {
+			return channel;
+		}
+
+		private ConnectionFactory getConnectionFactory() {
+			return connectionFactory;
+		}
+	}
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ExternalTxManagerTests.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.
@@ -157,6 +157,121 @@ public class ExternalTxManagerTests {
 
 	}
 
+	/**
+	 * Verifies that an up-stack RabbitTemplate does not use the listener's
+	 * channel when it has its own connection factory.
+	 */
+	@Test
+	public void testMessageListenerTemplateUsesDifferentConnectionFactory() throws Exception {
+		ConnectionFactory listenerConnectionFactory = mock(ConnectionFactory.class);
+		ConnectionFactory templateConnectionFactory = mock(ConnectionFactory.class);
+		Connection listenerConnection = mock(Connection.class);
+		Connection templateConnection = mock(Connection.class);
+		final Channel listenerChannel = mock(Channel.class);
+		Channel templateChannel = mock(Channel.class);
+		when(listenerChannel.isOpen()).thenReturn(true);
+		when(templateChannel.isOpen()).thenReturn(true);
+
+		final CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(listenerConnectionFactory);
+		final CachingConnectionFactory cachingTemplateConnectionFactory = new CachingConnectionFactory(templateConnectionFactory);
+
+		when(listenerConnectionFactory.newConnection((ExecutorService) null)).thenReturn(listenerConnection);
+		when(listenerConnection.isOpen()).thenReturn(true);
+		when(templateConnectionFactory.newConnection((ExecutorService) null)).thenReturn(templateConnection);
+		when(templateConnection.isOpen()).thenReturn(true);
+		when(templateConnection.createChannel()).thenReturn(templateChannel);
+
+		final AtomicReference<Exception> tooManyChannels = new AtomicReference<Exception>();
+
+		doAnswer(new Answer<Channel>(){
+			boolean done;
+			@Override
+			public Channel answer(InvocationOnMock invocation) throws Throwable {
+				if (!done) {
+					done = true;
+					return listenerChannel;
+				}
+				tooManyChannels.set(new Exception("More than one channel requested"));
+				Channel channel = mock(Channel.class);
+				when(channel.isOpen()).thenReturn(true);
+				return channel;
+			}
+		}).when(listenerConnection).createChannel();
+
+		final AtomicReference<Consumer> consumer = new AtomicReference<Consumer>();
+
+		doAnswer(new Answer<String>() {
+
+			@Override
+			public String answer(InvocationOnMock invocation) throws Throwable {
+				consumer.set((Consumer) invocation.getArguments()[2]);
+				return null;
+			}
+		}).when(listenerChannel)
+			.basicConsume(Mockito.anyString(), Mockito.anyBoolean(), Mockito.any(Consumer.class));
+
+		final CountDownLatch commitLatch = new CountDownLatch(2);
+		doAnswer(new Answer<String>() {
+
+			@Override
+			public String answer(InvocationOnMock invocation) throws Throwable {
+				commitLatch.countDown();
+				return null;
+			}
+		}).when(listenerChannel).txCommit();
+		doAnswer(new Answer<String>() {
+
+			@Override
+			public String answer(InvocationOnMock invocation) throws Throwable {
+				commitLatch.countDown();
+				return null;
+			}
+		}).when(templateChannel).txCommit();
+
+		final CountDownLatch latch = new CountDownLatch(1);
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cachingConnectionFactory);
+		container.setMessageListener(new MessageListener() {
+			public void onMessage(Message message) {
+				RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingTemplateConnectionFactory);
+				rabbitTemplate.setChannelTransacted(true);
+				// should use same channel as container
+				rabbitTemplate.convertAndSend("foo", "bar", "baz");
+				latch.countDown();
+			}
+		});
+		container.setQueueNames("queue");
+		container.setChannelTransacted(true);
+		container.setShutdownTimeout(100);
+		container.setTransactionManager(new DummyTxManager());
+		container.afterPropertiesSet();
+		container.start();
+
+		consumer.get().handleDelivery("qux", new Envelope(1, false, "foo", "bar"), new BasicProperties(), new byte[] {0});
+
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+		Exception e = tooManyChannels.get();
+		if (e != null) {
+			throw e;
+		}
+
+		verify(listenerConnection, Mockito.times(1)).createChannel();
+		verify(templateConnection, Mockito.times(1)).createChannel();
+		assertTrue(commitLatch.await(10, TimeUnit.SECONDS));
+		verify(listenerChannel).txCommit();
+		verify(templateChannel).basicPublish(Mockito.anyString(), Mockito.anyString(), Mockito.anyBoolean(),
+				Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
+		verify(templateChannel).txCommit();
+
+		// verify close() was never called on the channel
+		DirectFieldAccessor dfa = new DirectFieldAccessor(cachingConnectionFactory);
+		List<?> channels = (List<?>) dfa.getPropertyValue("cachedChannelsTransactional");
+		assertEquals(0, channels.size());
+
+		container.stop();
+
+	}
+
 	/**
 	 * Verifies that an up-stack RabbitTemplate uses the listener's
 	 * channel (ChannelAwareMessageListener).