AMQP-150: fix some internals of connection factories so the RabbitAdmin behaves correctly

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

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);

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();
+
+	}
+	
 }

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();
+	}
+
 }

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<Connection> connectionHolder = new AtomicReference<Connection>();
 
 		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
-		// Force RabbitAdmin to initialize the queue
 		boolean exists = rabbitTemplate.execute(new ChannelCallback<Boolean>() {
 			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<Boolean>() {
+			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<Connection> connectionHolder = new AtomicReference<Connection>();
@@ -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();
+			}
 		}
 	}
 

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);