diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java index 259d3c14..98a24205 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java @@ -31,8 +31,11 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di protected final Log logger = LogFactory.getLog(getClass()); private final com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory; + private final CompositeConnectionListener connectionListener = new CompositeConnectionListener(); + private final CompositeChannelListener channelListener = new CompositeChannelListener(); + /** * Create a new SingleConnectionFactory for the given target ConnectionFactory. * @param rabbitConnectionFactory the target ConnectionFactory @@ -79,9 +82,18 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di * * @return the connection listener */ - protected CompositeConnectionListener getConnectionListener() { + protected ConnectionListener getConnectionListener() { return connectionListener; } + + /** + * A composite channel listener to be used by subclasses when creating and closing channels. + * + * @return the channel listener + */ + protected ChannelListener getChannelListener() { + return channelListener; + } public void setConnectionListeners(List listeners) { this.connectionListener.setDelegates(listeners); @@ -91,6 +103,14 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di this.connectionListener.addDelegate(listener); } + public void setChannelListeners(List listeners) { + this.channelListener.setDelegates(listeners); + } + + public void addChannelListener(ChannelListener listener) { + this.channelListener.addDelegate(listener); + } + final protected Connection createBareConnection() { try { return new SimpleConnection(this.rabbitConnectionFactory.newConnection()); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 0e230d48..59014896 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -21,7 +21,6 @@ import java.util.LinkedList; import java.util.List; import org.springframework.amqp.AmqpException; -import org.springframework.beans.factory.DisposableBean; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -46,7 +45,7 @@ import com.rabbitmq.client.Channel; * @author Mark Fisher * @author Dave Syer */ -public class CachingConnectionFactory extends AbstractConnectionFactory implements DisposableBean { +public class CachingConnectionFactory extends AbstractConnectionFactory { private int channelCacheSize = 1; @@ -61,8 +60,6 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); - private final CompositeChannelListener channelListener = new CompositeChannelListener(); - /** * Create a new CachingConnectionFactory initializing the hostname to be the value returned from * InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception. @@ -112,14 +109,6 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen super(rabbitConnectionFactory); } - public void setChannelListeners(List listeners) { - this.channelListener.setDelegates(listeners); - } - - public void addChannelListener(ChannelListener listener) { - this.channelListener.addDelegate(listener); - } - public void setChannelCacheSize(int sessionCacheSize) { Assert.isTrue(sessionCacheSize >= 1, "Channel cache size must be 1 or higher"); this.channelCacheSize = sessionCacheSize; @@ -129,6 +118,22 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen return this.channelCacheSize; } + public void setConnectionListeners(List listeners) { + super.setConnectionListeners(listeners); + // If the connection is already alive we assume that the new listeners want to be notified + if (this.connection != null) { + this.getConnectionListener().onCreate(this.connection); + } + } + + public void addConnectionListener(ConnectionListener listener) { + super.addConnectionListener(listener); + // If the connection is already alive we assume that the new listener wants to be notified + if (this.connection != null) { + listener.onCreate(this.connection); + } + } + private Channel getChannel(boolean transactional) { LinkedList channelList = transactional ? this.cachedChannelsTransactional : this.cachedChannelsNonTransactional; @@ -153,14 +158,14 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen if (logger.isDebugEnabled()) { logger.debug("Creating cached Rabbit Channel from " + targetChannel); } - channelListener.onCreate(targetChannel, transactional); + getChannelListener().onCreate(targetChannel, transactional); return (ChannelProxy) Proxy.newProxyInstance(ChannelProxy.class.getClassLoader(), new Class[] { ChannelProxy.class }, new CachedChannelInvocationHandler(targetChannel, channelList, transactional)); } private Channel createBareChannel(boolean transactional) { - if (this.connection==null || !this.connection.isOpen()) { + if (this.connection == null || !this.connection.isOpen()) { this.connection = null; // Use createConnection here not doCreateConnection so that the old one is properly disposed createConnection(); @@ -280,9 +285,12 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen return this.target; } try { - synchronized (targetMonitor) { - if (this.target == null) { - this.target = createBareChannel(transactional); + if (this.target == null || !this.target.isOpen()) { + this.target = null; + synchronized (targetMonitor) { + if (this.target == null) { + this.target = createBareChannel(transactional); + } } } return method.invoke(this.target, args); @@ -306,9 +314,9 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen * @param proxy the channel to close */ private void logicalClose(ChannelProxy proxy) throws Exception { - if (this.target!=null && !this.target.isOpen()) { + if (this.target != null && !this.target.isOpen()) { synchronized (targetMonitor) { - if (this.target!=null && !this.target.isOpen()) { + if (this.target != null && !this.target.isOpen()) { this.target = null; return; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeChannelListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeChannelListener.java index 28f28d22..4718fca8 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeChannelListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeChannelListener.java @@ -28,9 +28,9 @@ public class CompositeChannelListener implements ChannelListener { private List delegates = new ArrayList(); - public void onCreate(Channel connection, boolean transactional) { + public void onCreate(Channel channel, boolean transactional) { for (ChannelListener delegate : delegates) { - delegate.onCreate(connection, transactional); + delegate.onCreate(channel, transactional); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java index 8d63f5d5..667d2d5f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java @@ -165,6 +165,7 @@ public class SingleConnectionFactory extends AbstractConnectionFactory { } } Channel channel = target.createChannel(transactional); + getChannelListener().onCreate(channel, transactional); return channel; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index a15273eb..7ba62acc 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -1,17 +1,14 @@ /* * Copyright 2002-2011 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. + * + * 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; @@ -87,7 +84,8 @@ public class BlockingQueueConsumer { * Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker * until it is started. */ - public BlockingQueueConsumer(ConnectionFactory connectionFactory, MessagePropertiesConverter messagePropertiesConverter, + public BlockingQueueConsumer(ConnectionFactory connectionFactory, + MessagePropertiesConverter messagePropertiesConverter, ActiveObjectCounter activeObjectCounter, AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, String... queues) { this.connectionFactory = connectionFactory; @@ -172,15 +170,20 @@ public class BlockingQueueConsumer { } public void start() throws AmqpException { + if (logger.isDebugEnabled()) { + logger.debug("Starting consumer " + this); + } this.channel = ConnectionFactoryUtils.getTransactionalResourceHolder(connectionFactory, transactional) .getChannel(); this.consumer = new InternalConsumer(channel); - this.activeObjectCounter.add(this); this.deliveryTags.clear(); + this.activeObjectCounter.add(this); try { - // Set basicQos before calling basicConsume (it is ignored if we are not transactional and the broker will - // send blocks of 100 messages) - channel.basicQos(prefetchCount); + if (!acknowledgeMode.isAutoAck()) { + // Set basicQos before calling basicConsume (otherwise if we are not acking the broker + // will send blocks of 100 messages) + channel.basicQos(prefetchCount); + } for (int i = 0; i < queues.length; i++) { channel.queueDeclarePassive(queues[i]); } @@ -193,7 +196,7 @@ public class BlockingQueueConsumer { for (int i = 0; i < queues.length; i++) { channel.basicConsume(queues[i], acknowledgeMode.isAutoAck(), consumer); if (logger.isDebugEnabled()) { - logger.debug("Started " + this); + logger.debug("Started on queue '" + queues[i] + "': " + this); } } } catch (IOException e) { @@ -341,7 +344,7 @@ public class BlockingQueueConsumer { try { boolean ackRequired = !acknowledgeMode.isAutoAck() && !acknowledgeMode.isManual(); - + if (ackRequired) { if (transactional && !locallyTransacted) { @@ -350,13 +353,12 @@ public class BlockingQueueConsumer { // could be synchronized with an external transaction for (Long deliveryTag : deliveryTags) { ConnectionFactoryUtils.registerDeliveryTag(connectionFactory, channel, deliveryTag); - } + } - } else { if (!deliveryTags.isEmpty()) { - long deliveryTag = new ArrayList(deliveryTags).get(deliveryTags.size()-1); + long deliveryTag = new ArrayList(deliveryTags).get(deliveryTags.size() - 1); channel.basicAck(deliveryTag, true); } 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 9425bff6..7ad6ea16 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 @@ -106,11 +106,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private ContainerDelegate proxy = delegate; - /** * Default constructor for convenient dependency injection via setters. */ - public SimpleMessageListenerContainer() { } + public SimpleMessageListenerContainer() { + } /** * Create a listener container from the connection factory (mandatory). @@ -121,7 +121,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta this.setConnectionFactory(connectionFactory); } - /** *

* Public setter for the {@link Advice} to apply to listener executions. If {@link #setTxSize(int) txSize>1} then @@ -378,7 +377,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta try { // Need to recycle the channel in this consumer consumer.stop(); - // Ensure consumer counts are correct (another is not going + // Ensure consumer counts are correct (another is going // to start because of the exception, but // we haven't counted down yet) this.cancellationLock.release(consumer); @@ -444,7 +443,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } - + return consumer.commitIfNecessary(isChannelLocallyTransacted(channel)); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactoryTests.java new file mode 100644 index 00000000..747b640d --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactoryTests.java @@ -0,0 +1,135 @@ +package org.springframework.amqp.rabbit.connection; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.Arrays; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import com.rabbitmq.client.ConnectionFactory; + +/** + * @author Dave Syer + */ +public abstract class AbstractConnectionFactoryTests { + + protected abstract AbstractConnectionFactory createConnectionFactory(ConnectionFactory mockConnectionFactory); + @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 AtomicInteger called = new AtomicInteger(0); + AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory); + connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() { + public void onCreate(Connection connection) { + called.incrementAndGet(); + } + public void onClose(Connection connection) { + called.decrementAndGet(); + } + })); + + Connection con = connectionFactory.createConnection(); + assertEquals(1, called.get()); + + con.close(); + assertEquals(1, called.get()); + verify(mockConnection, never()).close(); + + connectionFactory.createConnection(); + assertEquals(1, called.get()); + + connectionFactory.destroy(); + assertEquals(0, called.get()); + verify(mockConnection, atLeastOnce()).close(); + + verify(mockConnectionFactory, times(1)).newConnection(); + + } + + @Test + public void testWithListenerRegisteredAfterOpen() 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 AtomicInteger called = new AtomicInteger(0); + AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory); + Connection con = connectionFactory.createConnection(); + assertEquals(0, called.get()); + + connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() { + public void onCreate(Connection connection) { + called.incrementAndGet(); + } + public void onClose(Connection connection) { + called.decrementAndGet(); + } + })); + assertEquals(1, called.get()); + + con.close(); + assertEquals(1, called.get()); + verify(mockConnection, never()).close(); + + connectionFactory.createConnection(); + assertEquals(1, called.get()); + + connectionFactory.destroy(); + assertEquals(0, called.get()); + verify(mockConnection, atLeastOnce()).close(); + + verify(mockConnectionFactory, times(1)).newConnection(); + + } + + @Test + public void testCloseInvalidConnection() throws Exception { + + com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); + com.rabbitmq.client.Connection mockConnection1 = mock(com.rabbitmq.client.Connection.class); + com.rabbitmq.client.Connection mockConnection2 = mock(com.rabbitmq.client.Connection.class); + + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection1).thenReturn(mockConnection2); + // simulate a dead connection + when(mockConnection1.isOpen()).thenReturn(false); + + AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory); + + Connection connection = connectionFactory.createConnection(); + // the dead connection should be discarded + connection.createChannel(false); + verify(mockConnectionFactory, times(2)).newConnection(); + verify(mockConnection2, times(1)).createChannel(); + + connectionFactory.destroy(); + verify(mockConnection2, times(1)).close(); + + } + + @Test + public void testDestroyBeforeUsed() throws Exception { + + com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); + + AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory); + connectionFactory.destroy(); + + verify(mockConnectionFactory, never()).newConnection(); + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java index 42ba829c..7d3a52fd 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java @@ -19,15 +19,21 @@ import org.junit.Test; import org.springframework.test.util.ReflectionTestUtils; import com.rabbitmq.client.Channel; +import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.GetResponse; /** * @author Mark Pollack + * @author Dave Syer */ -public class CachingConnectionFactoryTests { +public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTests { + + @Override + protected AbstractConnectionFactory createConnectionFactory(ConnectionFactory connectionFactory) { + return new CachingConnectionFactory(connectionFactory); + } @Test - public void testWithConnectionFactoryDefaults() throws IOException { com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); @@ -341,60 +347,26 @@ public class CachingConnectionFactoryTests { Assert.assertNotSame(channel3, channel2); } - @Test - public void testWithConnectionListener() 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 AtomicInteger called = new AtomicInteger(0); - CachingConnectionFactory connectionFactory = new CachingConnectionFactory(mockConnectionFactory); - connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() { - public void onCreate(Connection connection) { - called.incrementAndGet(); - } - public void onClose(Connection connection) { - called.decrementAndGet(); - } - })); - - Connection con = connectionFactory.createConnection(); - assertEquals(1, called.get()); - - con.close(); - assertEquals(1, called.get()); - verify(mockConnection, never()).close(); - - connectionFactory.createConnection(); - assertEquals(1, called.get()); - - connectionFactory.destroy(); - assertEquals(0, called.get()); - verify(mockConnection, atLeastOnce()).close(); - - verify(mockConnectionFactory).newConnection(); - - } - @Test public void testWithChannelListener() throws IOException { com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); + Channel mockChannel = mock(Channel.class); when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); + when(mockChannel.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(mockChannel); final AtomicInteger called = new AtomicInteger(0); - CachingConnectionFactory connectionFactory = new CachingConnectionFactory(mockConnectionFactory); + AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory); connectionFactory.setChannelListeners(Arrays.asList(new ChannelListener() { public void onCreate(Channel channel, boolean transactional) { called.incrementAndGet(); } })); - connectionFactory.setChannelCacheSize(1); + ((CachingConnectionFactory)connectionFactory).setChannelCacheSize(1); Connection con = connectionFactory.createConnection(); Channel channel = con.createChannel(false); @@ -414,5 +386,4 @@ public class CachingConnectionFactoryTests { verify(mockConnectionFactory).newConnection(); } - } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java index 00dcfbd9..a9c82dfd 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java @@ -4,7 +4,6 @@ import static org.junit.Assert.assertEquals; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -14,119 +13,55 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.ConnectionFactory; + /** * @author Dave Syer */ -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 AtomicInteger called = new AtomicInteger(0); - SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory); - connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() { - public void onCreate(Connection connection) { - called.incrementAndGet(); - } - public void onClose(Connection connection) { - called.decrementAndGet(); - } - })); - - Connection con = connectionFactory.createConnection(); - assertEquals(1, called.get()); - - con.close(); - assertEquals(1, called.get()); - verify(mockConnection, never()).close(); - - connectionFactory.createConnection(); - assertEquals(1, called.get()); - - connectionFactory.destroy(); - assertEquals(0, called.get()); - verify(mockConnection, atLeastOnce()).close(); - - verify(mockConnectionFactory, times(1)).newConnection(); - - } +public class SingleConnectionFactoryTests extends AbstractConnectionFactoryTests { + @Override + protected AbstractConnectionFactory createConnectionFactory(ConnectionFactory connectionFactory) { + return new SingleConnectionFactory(connectionFactory); + } + @Test - public void testWithListenerRegisteredAfterOpen() throws IOException { + public void testWithChannelListener() throws IOException { com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); + Channel mockChannel = mock(Channel.class); when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(mockChannel); final AtomicInteger called = new AtomicInteger(0); - SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory); - Connection con = connectionFactory.createConnection(); - assertEquals(0, called.get()); - - connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() { - public void onCreate(Connection connection) { + AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory); + connectionFactory.setChannelListeners(Arrays.asList(new ChannelListener() { + public void onCreate(Channel channel, boolean transactional) { called.incrementAndGet(); } - public void onClose(Connection connection) { - called.decrementAndGet(); - } })); - assertEquals(1, called.get()); - con.close(); + Connection con = connectionFactory.createConnection(); + Channel channel = con.createChannel(false); assertEquals(1, called.get()); + channel.close(); + + con.close(); verify(mockConnection, never()).close(); connectionFactory.createConnection(); - assertEquals(1, called.get()); + con.createChannel(false); + assertEquals(2, called.get()); connectionFactory.destroy(); - assertEquals(0, called.get()); verify(mockConnection, atLeastOnce()).close(); - verify(mockConnectionFactory, times(1)).newConnection(); + verify(mockConnectionFactory).newConnection(); } - @Test - public void testCloseInvalidConnection() throws Exception { - - com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); - com.rabbitmq.client.Connection mockConnection1 = mock(com.rabbitmq.client.Connection.class); - com.rabbitmq.client.Connection mockConnection2 = mock(com.rabbitmq.client.Connection.class); - - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection1).thenReturn(mockConnection2); - // simulate a dead connection - when(mockConnection1.isOpen()).thenReturn(false); - - SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory); - - Connection connection = connectionFactory.createConnection(); - // the dead connection should be discarded - connection.createChannel(false); - verify(mockConnectionFactory, times(2)).newConnection(); - verify(mockConnection2, times(1)).createChannel(); - - connectionFactory.destroy(); - verify(mockConnection2, times(1)).close(); - - } - - @Test - public void testDestroyBeforeUsed() throws Exception { - - com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); - - SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory); - connectionFactory.destroy(); - - verify(mockConnectionFactory, never()).newConnection(); - } - } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java index 99cca6c3..5ba69641 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java @@ -160,7 +160,7 @@ public class MessageListenerContainerLifecycleIntegrationTests { assertEquals(concurrentConsumers, container.getActiveConsumerCount()); container.stop(); - Thread.sleep(500L); + Thread.sleep(1000L); assertEquals(0, container.getActiveConsumerCount()); if (!transactional) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java index bc2c9ec3..7916fdac 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java @@ -53,8 +53,8 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { private SimpleMessageListenerContainer container; @Rule - public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.DEBUG, RabbitTemplate.class, - SimpleMessageListenerContainer.class, BlockingQueueConsumer.class); + public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.INFO, RabbitTemplate.class, + SimpleMessageListenerContainer.class, BlockingQueueConsumer.class, CachingConnectionFactory.class); @Rule public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue, sendQueue);