INT-136: Weaken checks for concurrent initialization

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java

@@ -207,24 +207,27 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
 				return;
 			}
 
-			// Prevent stack overflow...
-			final AtomicBoolean initializing = new AtomicBoolean(false);
-
 			connectionFactory.addConnectionListener(new ConnectionListener() {
 
-				private volatile boolean initialized = false;
+				// Prevent stack overflow...
+				private AtomicBoolean initializing = new AtomicBoolean(false);
 
 				public void onCreate(Connection connection) {
-					if (!initializing.compareAndSet(false, true) || initialized) {
+					if (!initializing.compareAndSet(false, true)) {
+						// If we are already initializing, we don't need to do it again...
 						return;
 					}
-					initialize();
-					initializing.compareAndSet(true, false);
-					initialized = true;
+					try {
+						// ...but it is possible for this to happen twice in the same ConnectionFactory (if more than
+						// one concurrent Connection is allowed). It's idempotent, so no big deal (a bit of network
+						// chatter).  If anyone has a problem with it: use auto-startup="false".
+						initialize();
+					} finally {
+						initializing.compareAndSet(true, false);
+					}
 				}
 
 				public void onClose(Connection connection) {
-					initialized = false;
 				}
 
 			});

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java

@@ -101,6 +101,50 @@ public class RabbitAdminIntegrationTests {
 
 	}
 
+	@Test
+	public void testStartupWithNonDurable() throws Exception {
+
+		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>();
+
+		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());
+				connectionHolder.set(channel.getConnection());
+				return result != null;
+			}
+		});
+		assertTrue("Expected Queue to exist", exists);
+
+		assertTrue(queueExists(connectionHolder.get(), queue));
+		
+		// simulate broker going down and coming back up...
+		rabbitAdmin.deleteQueue(queue.getName());
+		connectionFactory.destroy();
+		assertFalse(queueExists(null, queue));
+
+		// Broker auto-deleted queue, but it is re-created by the connection listener
+		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);
+
+		assertTrue(queueExists(connectionHolder.get(), queue));
+		assertTrue(rabbitAdmin.deleteQueue(queue.getName()));
+		assertFalse(queueExists(null, queue));
+
+	}
+
 	/**
 	 * Use native Rabbit API to test queue, bypassing all the connection and channel caching and callbacks in Spring
 	 * AMQP.