diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java index e8d9f993..4c15086f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java @@ -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); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 5375460e..d93b4c19 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/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 diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ConsumerChannelRegistry.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ConsumerChannelRegistry.java index 1357a973..9e7ae240 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ConsumerChannelRegistry.java +++ b/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 consumerChannel = new ThreadLocal(); + private static final ThreadLocal consumerChannel = new ThreadLocal(); /** * 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; + } + } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ExternalTxManagerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ExternalTxManagerTests.java index 7705fef5..031973a3 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ExternalTxManagerTests.java +++ b/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 tooManyChannels = new AtomicReference(); + + doAnswer(new Answer(){ + 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 = new AtomicReference(); + + doAnswer(new Answer() { + + @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() { + + @Override + public String answer(InvocationOnMock invocation) throws Throwable { + commitLatch.countDown(); + return null; + } + }).when(listenerChannel).txCommit(); + doAnswer(new Answer() { + + @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).