From 54362f7b211ea4d5c4b4854d70c95434c5ca2590 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 24 Apr 2013 14:12:14 -0400 Subject: [PATCH] AMQP-309 Fix Listener Channel Issue When an external transaction manager is assigned to the listener container, the fix for AMQP-260 corrected a problem where the listener's channel was not used by RabbitTemplates running on the same thread. This is required for operations performed by the template to participate in the same transaction. However, this was too encompassing in that this did not take into account that the template might be configured with a different connection factory. Add a test to show the template with a different connection factory incorrectly used the listener's channel. Add code to only use the listener channel if the client is requesting a channel from the same connection factory. Also, do not fetch a connection in the case where the consumer channel is used (actually a no-op because the single connection is cached). Improve debug log during channel registration: Registering consumer channelCached Rabbit Channel: AMQChannel(amqp://guest@127.0.1.1:5672/,1) from factory CachingConnectionFactory [channelCacheSize=2, host=localhost, port=5672, active=true] --- .../connection/ConnectionFactoryUtils.java | 9 +- .../SimpleMessageListenerContainer.java | 2 +- .../support/ConsumerChannelRegistry.java | 50 +++++++- .../listener/ExternalTxManagerTests.java | 117 +++++++++++++++++- 4 files changed, 168 insertions(+), 10 deletions(-) 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).