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 a4a9634f..15057825 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 @@ -154,7 +154,7 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements targetConnection = new ChannelCachingConnectionProxy(super.doCreateConnection()); return targetConnection; } - + /** * Reset the Channel cache and underlying shared Connection, to be reinitialized on next access. */ @@ -172,6 +172,7 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements } this.active = true; super.reset(); + this.targetConnection = null; } @Override 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 b3416536..7ca2b30f 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 @@ -46,7 +46,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); - + private final CompositeConnectionListener listener = new CompositeConnectionListener(); /** @@ -64,7 +64,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea public SingleConnectionFactory(int port) { this(null, port); } - + /** * Create a new SingleConnectionFactory given a host name. * @param hostname the host name to connect to @@ -72,7 +72,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea public SingleConnectionFactory(String hostname) { this(hostname, com.rabbitmq.client.ConnectionFactory.DEFAULT_AMQP_PORT); } - + /** * Create a new SingleConnectionFactory given a host name. * @param hostname the host name to connect to @@ -139,9 +139,11 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea public final Connection createConnection() throws AmqpException { synchronized (this.connectionMonitor) { if (this.connection == null) { - this.connection = new SharedConnectionProxy(doCreateConnection()); + Connection target = doCreateConnection(); + this.connection = new SharedConnectionProxy(target); + // invoke the listener *after* this.connection is assigned + listener.onCreate(target); } - this.listener.onCreate(connection); } return this.connection; } @@ -154,8 +156,10 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea */ public final void destroy() { synchronized (this.connectionMonitor) { - this.connection.destroy(); - this.connection = null; + if (connection != null) { + this.connection.destroy(); + this.connection = null; + } } reset(); } @@ -173,7 +177,8 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea * @return the new Connection */ protected Connection doCreateConnection() { - return createBareConnection(); + Connection connection = createBareConnection(); + return connection; } private Connection createBareConnection() { @@ -217,11 +222,12 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea } public Channel createChannel(boolean transactional) { - if (target==null || !target.isOpen()) { + if (target == null || !target.isOpen()) { synchronized (this) { - if (target==null || !target.isOpen()) { + if (target == null || !target.isOpen()) { logger.debug("Detected closed connection. Opening a new one before creating Channel."); target = createBareConnection(); + listener.onCreate(target); } } } @@ -231,7 +237,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea public void close() { } - + public void destroy() { if (this.target != null) { listener.onClose(target); 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 409f3529..c99d0625 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 @@ -9,7 +9,9 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.IOException; +import java.util.Arrays; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; import junit.framework.Assert; @@ -339,4 +341,41 @@ public class CachingConnectionFactoryTests { Assert.assertNotSame(channel3, channel2); } + @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); + 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, times(1)).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 9a4a7d70..f100a6b4 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 @@ -1,7 +1,6 @@ package org.springframework.amqp.rabbit.connection; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; +import static org.junit.Assert.assertEquals; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -11,7 +10,7 @@ import static org.mockito.Mockito.when; import java.io.IOException; import java.util.Arrays; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; @@ -28,28 +27,33 @@ public class SingleConnectionFactoryTests { when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); - final AtomicBoolean called = new AtomicBoolean(false); + final AtomicInteger called = new AtomicInteger(0); SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory); connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() { public void onCreate(Connection connection) { - called.set(true); + called.incrementAndGet(); } public void onClose(Connection connection) { - called.set(false); + called.decrementAndGet(); } })); Connection con = connectionFactory.createConnection(); - assertTrue(called.get()); + assertEquals(1, called.get()); con.close(); - assertTrue(called.get()); + assertEquals(1, called.get()); verify(mockConnection, never()).close(); + + connectionFactory.createConnection(); + assertEquals(1, called.get()); connectionFactory.destroy(); - assertFalse(called.get()); + assertEquals(0, called.get()); verify(mockConnection, atLeastOnce()).close(); + verify(mockConnectionFactory, times(1)).newConnection(); + } @Test @@ -76,4 +80,15 @@ public class SingleConnectionFactoryTests { } + @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/core/RabbitAdminIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java index e32c0a35..2ba1ce08 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java @@ -3,6 +3,7 @@ package org.springframework.amqp.rabbit.core; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import java.io.IOException; import java.util.concurrent.atomic.AtomicReference; import org.junit.After; @@ -40,6 +41,9 @@ public class RabbitAdminIntegrationTests { public void init() { context = new GenericApplicationContext(); rabbitAdmin = new RabbitAdmin(connectionFactory); + rabbitAdmin.deleteQueue("test.queue"); + // Force connection factory to forget that it has been used to delete the queue + connectionFactory.destroy(); rabbitAdmin.setApplicationContext(context); rabbitAdmin.setAutoStartup(true); } @@ -49,14 +53,17 @@ public class RabbitAdminIntegrationTests { if (context != null) { context.close(); } + if (connectionFactory!=null) { + connectionFactory.destroy(); + } } @Test - public void testStartupWithBroker() throws Exception { + public void testStartupWithLazyDeclaration() throws Exception { Queue queue = new Queue("test.queue"); context.getBeanFactory().registerSingleton("foo", queue); - rabbitAdmin.deleteQueue(queue.getName()); rabbitAdmin.afterPropertiesSet(); + // A new connection is initialized so the queue is declared assertTrue(rabbitAdmin.deleteQueue(queue.getName())); } @@ -87,7 +94,6 @@ public class RabbitAdminIntegrationTests { CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(); connectionFactory2.setPort(BrokerTestUtils.getPort()); Queue queue = new Queue("test.queue", false, false, true); - rabbitAdmin.deleteQueue(queue.getName()); new RabbitAdmin(connectionFactory1).declareQueue(queue); new RabbitAdmin(connectionFactory2).declareQueue(queue); connectionFactory1.destroy(); @@ -99,13 +105,11 @@ public class RabbitAdminIntegrationTests { final Queue queue = new Queue("test.queue", false, true, true); context.getBeanFactory().registerSingleton("foo", queue); - rabbitAdmin.deleteQueue(queue.getName()); rabbitAdmin.afterPropertiesSet(); final AtomicReference connectionHolder = new AtomicReference(); RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); - // Force RabbitAdmin to initialize the queue boolean exists = rabbitTemplate.execute(new ChannelCallback() { public Boolean doInRabbit(Channel channel) throws Exception { DeclareOk result = channel.queueDeclarePassive(queue.getName()); @@ -116,6 +120,16 @@ public class RabbitAdminIntegrationTests { assertTrue("Expected Queue to exist", exists); assertTrue(queueExists(connectionHolder.get(), queue)); + + exists = rabbitTemplate.execute(new ChannelCallback() { + 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); + connectionFactory.destroy(); // Broker now deletes queue (only verifiable in native API) assertFalse(queueExists(null, queue)); @@ -141,7 +155,6 @@ public class RabbitAdminIntegrationTests { final Queue queue = new Queue("test.queue", false, false, false); context.getBeanFactory().registerSingleton("foo", queue); - rabbitAdmin.deleteQueue(queue.getName()); rabbitAdmin.afterPropertiesSet(); final AtomicReference connectionHolder = new AtomicReference(); @@ -189,17 +202,25 @@ public class RabbitAdminIntegrationTests { * @return true if the queue exists */ private boolean queueExists(Connection connection, Queue queue) throws Exception { - if (connection == null) { + Connection target = connection; + if (target == null) { ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setPort(BrokerTestUtils.getPort()); - connection = connectionFactory.newConnection(); + target = connectionFactory.newConnection(); } - Channel channel = connection.createChannel(); + Channel channel = target.createChannel(); try { DeclareOk result = channel.queueDeclarePassive(queue.getName()); return result != null; - } catch (Exception e) { + } catch (IOException e) { + if (e.getCause().getMessage().contains("RESOURCE_LOCKED")) { + return true; + } return false; + } finally { + if (connection==null) { + target.close(); + } } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java index 61980675..dbcd9ec9 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java @@ -135,9 +135,10 @@ public class BrokerRunning extends TestWatchman { Assume.assumeTrue(brokerOffline); } + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + try { - CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); connectionFactory.setPort(port); if (StringUtils.hasText(hostName)) { connectionFactory.setHost(hostName); @@ -171,6 +172,8 @@ public class BrokerRunning extends TestWatchman { if (assumeOnline) { Assume.assumeNoException(e); } + } finally { + connectionFactory.destroy(); } return super.apply(base, method, target);