INT-4320: Fix AMQP channels to declare on demand

JIRA: https://jira.spring.io/browse/INT-4320

To avoid "fail fast" situation when Broker isn't available during
application init phase, move all the declarations in
the AMQP Message Channels to the `ConnectionListener.onCreate()`.
In addition call the declaration from the `Lifecycle.start()`

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java

@@ -16,10 +16,15 @@
 
 package org.springframework.integration.amqp.channel;
 
+import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.rabbit.connection.Connection;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionListener;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.integration.amqp.support.MappingUtils;
@@ -29,9 +34,12 @@ import org.springframework.util.Assert;
 
 /**
  * @author Mark Fisher
+ * @author Artem Bilan
+ *
  * @since 2.1
  */
-public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
+public abstract class AbstractAmqpChannel extends AbstractMessageChannel
+		implements DisposableBean, ConnectionListener {
 
 	private final AmqpTemplate amqpTemplate;
 
@@ -41,14 +49,20 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 
 	private final AmqpHeaderMapper inboundHeaderMapper;
 
-	private volatile boolean extractPayload;
+	private AmqpAdmin admin;
 
-	private volatile boolean loggingEnabled = true;
+	private ConnectionFactory connectionFactory;
+
+	private boolean extractPayload;
+
+	private boolean loggingEnabled = true;
 
 	private MessageDeliveryMode defaultDeliveryMode;
 
 	private boolean headersMappedLast;
 
+	private volatile boolean initialized;
+
 	/**
 	 * Construct an instance with the supplied template and default header mappers
 	 * used if the template is a {@link RabbitTemplate} and the message is mapped.
@@ -149,7 +163,6 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 	/**
 	 * Subclasses may override this method to return an Exchange name.
 	 * By default, Messages will be sent to the no-name Direct Exchange.
-	 *
 	 * @return The exchange name.
 	 */
 	protected String getExchangeName() {
@@ -159,7 +172,6 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 	/**
 	 * Subclasses may override this method to return a routing key.
 	 * By default, there will be no routing key (empty string).
-	 *
 	 * @return The routing key.
 	 */
 	protected String getRoutingKey() {
@@ -178,6 +190,41 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 		return this.rabbitTemplate;
 	}
 
+	protected final void setAdmin(AmqpAdmin admin) {
+		this.admin = admin;
+	}
+
+	protected final void setConnectionFactory(ConnectionFactory connectionFactory) {
+		this.connectionFactory = connectionFactory;
+	}
+
+	protected AmqpAdmin getAdmin() {
+		return this.admin;
+	}
+
+	protected ConnectionFactory getConnectionFactory() {
+		return this.connectionFactory;
+	}
+
+	@Override
+	protected void onInit() throws Exception {
+		super.onInit();
+		if (!this.initialized && this.rabbitTemplate != null) {
+			if (this.connectionFactory != null) {
+				this.connectionFactory.addConnectionListener(this);
+			}
+		}
+		this.initialized = true;
+	}
+
+	@Override
+	public void destroy() throws Exception {
+		if (this.connectionFactory != null) {
+			this.connectionFactory.removeConnectionListener(this);
+			this.initialized = false;
+		}
+	}
+
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
 		if (this.extractPayload) {
@@ -191,4 +238,15 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 		return true;
 	}
 
+	@Override
+	public void onCreate(Connection connection) {
+		doDeclares();
+	}
+
+	@Override
+	public void onClose(Connection connection) {
+	}
+
+	protected abstract void  doDeclares();
+
 }

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -21,19 +21,18 @@ import java.util.Map;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.amqp.AmqpConnectException;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.MessageListener;
-import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
-import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.integration.context.IntegrationProperties;
 import org.springframework.integration.dispatcher.AbstractDispatcher;
 import org.springframework.integration.dispatcher.MessageDispatcher;
@@ -49,10 +48,11 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.1
  */
 abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
-		implements SubscribableChannel, SmartLifecycle, DisposableBean {
+		implements SubscribableChannel, SmartLifecycle {
 
 	private final String channelName;
 
@@ -64,9 +64,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 
 	private volatile Integer maxSubscribers;
 
-	private final AmqpAdmin admin;
-
-	private final ConnectionFactory connectionFactory;
+	private volatile boolean declared;
 
 	/**
 	 * Construct an instance with the supplied name, container and template; default header
@@ -109,14 +107,8 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 	protected AbstractSubscribableAmqpChannel(String channelName,
 			AbstractMessageListenerContainer container,
 			AmqpTemplate amqpTemplate, boolean isPubSub) {
-		super(amqpTemplate);
-		Assert.notNull(container, "container must not be null");
-		Assert.hasText(channelName, "channel name must not be empty");
-		this.channelName = channelName;
-		this.container = container;
-		this.isPubSub = isPubSub;
-		this.connectionFactory = container.getConnectionFactory();
-		this.admin = new RabbitAdmin(this.connectionFactory);
+		this(channelName, container, amqpTemplate, isPubSub,
+				DefaultAmqpHeaderMapper.outboundMapper(), DefaultAmqpHeaderMapper.inboundMapper());
 	}
 
 	/**
@@ -141,8 +133,8 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 		this.channelName = channelName;
 		this.container = container;
 		this.isPubSub = isPubSub;
-		this.connectionFactory = container.getConnectionFactory();
-		this.admin = new RabbitAdmin(this.connectionFactory);
+		setConnectionFactory(container.getConnectionFactory());
+		setAdmin(new RabbitAdmin(getConnectionFactory()));
 	}
 
 	/**
@@ -157,14 +149,6 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 		}
 	}
 
-	protected AmqpAdmin getAdmin() {
-		return this.admin;
-	}
-
-	protected ConnectionFactory getConnectionFactory() {
-		return this.connectionFactory;
-	}
-
 	@Override
 	public boolean subscribe(MessageHandler handler) {
 		return this.dispatcher.addHandler(handler);
@@ -181,12 +165,12 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 		this.dispatcher = this.createDispatcher();
 		if (this.maxSubscribers == null) {
 			this.maxSubscribers = this.getIntegrationProperty(this.isPubSub ?
-					IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS :
-					IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS,
+							IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS :
+							IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS,
 					Integer.class);
 		}
 		setMaxSubscribers(this.maxSubscribers);
-		String queue = this.obtainQueueName(this.admin, this.channelName);
+		String queue = obtainQueueName(this.channelName);
 		this.container.setQueueNames(queue);
 		MessageConverter converter = (this.getAmqpTemplate() instanceof RabbitTemplate)
 				? ((RabbitTemplate) this.getAmqpTemplate()).getMessageConverter()
@@ -221,6 +205,16 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 
 	@Override
 	public void start() {
+		if (!this.declared) {
+			try {
+				doDeclares();
+				this.declared = true;
+			}
+			catch (AmqpConnectException e) {
+				logger.info("Broker not available; cannot check queue declarations. " +
+						"Postponed to the next connection create...");
+			}
+		}
 		if (this.container != null) {
 			this.container.start();
 		}
@@ -230,6 +224,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 	public void stop() {
 		if (this.container != null) {
 			this.container.stop();
+			this.declared = false;
 		}
 	}
 
@@ -237,19 +232,22 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 	public void stop(Runnable callback) {
 		if (this.container != null) {
 			this.container.stop(callback);
+			this.declared = false;
 		}
 	}
 
 	@Override
 	public void destroy() throws Exception {
+		super.destroy();
 		if (this.container != null) {
 			this.container.destroy();
+			this.declared = false;
 		}
 	}
 
 	protected abstract AbstractDispatcher createDispatcher();
 
-	protected abstract String obtainQueueName(AmqpAdmin admin, String channelName);
+	protected abstract String obtainQueueName(String channelName);
 
 
 	private static final class DispatchingMessageListener implements MessageListener {

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -28,12 +28,12 @@ import org.springframework.integration.dispatcher.UnicastingDispatcher;
 /**
  * @author Mark Fisher
  * @author Artem Bilan
+ *
  * @since 2.1
  */
 public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmqpChannel {
 
-	private volatile String queueName;
-
+	private volatile Queue queue;
 
 	/**
 	 * Construct an instance with the supplied name, container and template; default header
@@ -72,18 +72,16 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq
 	 * @param queueName The queue name.
 	 */
 	public void setQueueName(String queueName) {
-		this.queueName = queueName;
+		this.queue = new Queue(queueName);
 	}
 
 	@Override
-	protected String obtainQueueName(AmqpAdmin admin, String channelName) {
-		if (this.queueName == null) {
-			this.queueName = channelName;
+	protected String obtainQueueName(String channelName) {
+		if (this.queue == null) {
+			this.queue = new Queue(channelName);
 		}
-		if (admin.getQueueProperties(this.queueName) == null) {
-			admin.declareQueue(new Queue(this.queueName));
-		}
-		return this.queueName;
+
+		return this.queue.getName();
 	}
 
 	@Override
@@ -95,7 +93,15 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq
 
 	@Override
 	protected String getRoutingKey() {
-		return this.queueName;
+		return this.queue != null ? this.queue.getName() : super.getRoutingKey();
+	}
+
+	@Override
+	protected void doDeclares() {
+		AmqpAdmin admin = getAdmin();
+		if (admin != null && this.queue != null && admin.getQueueProperties(this.queue.getName()) == null) {
+			admin.declareQueue(this.queue);
+		}
 	}
 
 }

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java

@@ -24,6 +24,7 @@ import java.util.Map;
 import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
@@ -51,12 +52,12 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 
 	private final String channelName;
 
-	private volatile String queueName;
-
-	private volatile AmqpAdmin amqpAdmin;
+	private volatile Queue queue;
 
 	private volatile int executorInterceptorsSize;
 
+	private volatile boolean declared;
+
 	/**
 	 * Construct an instance with the supplied name, template and default header mappers
 	 * used if the template is a {@link RabbitTemplate} and the message is mapped.
@@ -91,22 +92,20 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 	 * Provide an explicitly configured queue name. If this is not provided, then a Queue will be created
 	 * implicitly with the channelName as its name. The implicit creation will require that either an AmqpAdmin
 	 * instance has been provided or that the configured AmqpTemplate is an instance of RabbitTemplate.
-	 *
 	 * @param queueName The queue name.
 	 */
 	public void setQueueName(String queueName) {
-		this.queueName = queueName;
+		this.queue = new Queue(queueName);
 	}
 
 	/**
 	 * Provide an instance of AmqpAdmin for implicitly declaring Queues if the queueName is not provided.
 	 * When providing a RabbitTemplate implementation, this is not strictly necessary since a RabbitAdmin
 	 * instance can be created from the template's ConnectionFactory reference.
-	 *
 	 * @param amqpAdmin The amqp admin.
 	 */
 	public void setAmqpAdmin(AmqpAdmin amqpAdmin) {
-		this.amqpAdmin = amqpAdmin;
+		setAdmin(amqpAdmin);
 	}
 
 	@Override
@@ -131,22 +130,33 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 
 	@Override
 	protected String getRoutingKey() {
-		return this.queueName;
+		return this.queue != null ? this.queue.getName() : super.getRoutingKey();
 	}
 
 	@Override
 	protected void onInit() throws Exception {
-		super.onInit();
-		AmqpTemplate amqpTemplate = this.getAmqpTemplate();
-		if (this.queueName == null) {
-			if (this.amqpAdmin == null && amqpTemplate instanceof RabbitTemplate) {
-				this.amqpAdmin = new RabbitAdmin(((RabbitTemplate) amqpTemplate).getConnectionFactory());
+		AmqpTemplate amqpTemplate = getAmqpTemplate();
+		if (this.queue == null) {
+			if (getAdmin() == null && amqpTemplate instanceof RabbitTemplate) {
+				ConnectionFactory connectionFactory = ((RabbitTemplate) amqpTemplate).getConnectionFactory();
+				setAdmin(new RabbitAdmin(connectionFactory));
+				setConnectionFactory(connectionFactory);
 			}
-			Assert.notNull(this.amqpAdmin,
+
+			Assert.notNull(getAdmin(),
 					"If no queueName is configured explicitly, an AmqpAdmin instance must be provided, " +
 							"or the AmqpTemplate must be a RabbitTemplate since the Queue needs to be declared.");
-			this.queueName = this.channelName;
-			this.amqpAdmin.declareQueue(new Queue(this.queueName));
+
+			this.queue = new Queue(this.channelName);
+		}
+		super.onInit();
+	}
+
+	@Override
+	protected void doDeclares() {
+		AmqpAdmin admin = getAdmin();
+		if (admin != null && admin.getQueueProperties(this.queue.getName()) == null) {
+			admin.declareQueue(this.queue);
 		}
 	}
 
@@ -218,22 +228,27 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 	}
 
 	protected Object performReceive(Long timeout) {
+		if (!this.declared) {
+			doDeclares();
+			this.declared = true;
+		}
+
 		if (!isExtractPayload()) {
 			if (timeout == null) {
-				return getAmqpTemplate().receiveAndConvert(this.queueName);
+				return getAmqpTemplate().receiveAndConvert(this.queue.getName());
 			}
 			else {
-				return getAmqpTemplate().receiveAndConvert(this.queueName, timeout);
+				return getAmqpTemplate().receiveAndConvert(this.queue.getName(), timeout);
 			}
 		}
 		else {
 			RabbitTemplate rabbitTemplate = getRabbitTemplate();
 			org.springframework.amqp.core.Message message;
 			if (timeout == null) {
-				message = rabbitTemplate.receive(this.queueName);
+				message = rabbitTemplate.receive(this.queue.getName());
 			}
 			else {
-				message = rabbitTemplate.receive(this.queueName, timeout);
+				message = rabbitTemplate.receive(this.queue.getName(), timeout);
 			}
 
 			if (message != null) {

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -23,10 +23,6 @@ import org.springframework.amqp.core.Binding;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.FanoutExchange;
 import org.springframework.amqp.core.Queue;
-import org.springframework.amqp.rabbit.connection.Connection;
-import org.springframework.amqp.rabbit.connection.ConnectionFactory;
-import org.springframework.amqp.rabbit.connection.ConnectionListener;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.dispatcher.AbstractDispatcher;
@@ -35,9 +31,11 @@ import org.springframework.integration.dispatcher.BroadcastingDispatcher;
 /**
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.1
  */
-public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel implements ConnectionListener {
+public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel {
 
 	private volatile FanoutExchange exchange;
 
@@ -45,8 +43,6 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
 
 	private volatile Binding binding;
 
-	private volatile boolean initialized;
-
 	/**
 	 * Construct an instance with the supplied name, container and template; default header
 	 * mappers will be used if the message is mapped.
@@ -89,37 +85,20 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
 	}
 
 	@Override
-	protected String obtainQueueName(AmqpAdmin admin, String channelName) {
+	protected String getExchangeName() {
+		return (this.exchange != null) ? this.exchange.getName() : "";
+	}
+
+	@Override
+	protected String obtainQueueName(String channelName) {
 		if (this.exchange == null) {
 			String exchangeName = "si.fanout." + channelName;
 			this.exchange = new FanoutExchange(exchangeName);
 		}
-		admin.declareExchange(this.exchange);
-		admin.declareQueue(this.queue);
-		this.binding = BindingBuilder.bind(this.queue).to(this.exchange);
-		admin.declareBinding(this.binding);
-		if (!this.initialized && this.getAmqpTemplate() instanceof RabbitTemplate) {
-			ConnectionFactory connectionFactory = this.getConnectionFactory();
-			if (connectionFactory != null) {
-				connectionFactory.addConnectionListener(this);
-			}
-		}
-		this.initialized = true;
-		return this.queue.getName();
-	}
 
-	private void doDeclares() {
-		if (this.isRunning()) {
-			AmqpAdmin admin = this.getAdmin();
-			if (admin != null) {
-				if (this.queue != null) {
-					admin.declareQueue(this.queue);
-				}
-				if (this.binding != null) {
-					admin.declareBinding(this.binding);
-				}
-			}
-		}
+		this.binding = BindingBuilder.bind(this.queue).to(this.exchange);
+
+		return this.queue.getName();
 	}
 
 	@Override
@@ -130,32 +109,19 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
 	}
 
 	@Override
-	protected String getExchangeName() {
-		return (this.exchange != null) ? this.exchange.getName() : "";
-	}
-
-	@Override
-	public void destroy() throws Exception {
-		super.destroy();
-		if (this.getConnectionFactory() != null) {
-			this.getConnectionFactory().removeConnectionListener(this);
-			this.initialized = false;
+	protected void doDeclares() {
+		AmqpAdmin admin = getAdmin();
+		if (admin != null) {
+			if (admin.getQueueProperties(this.queue.getName()) == null) {
+				admin.declareQueue(this.queue);
+			}
+			if (this.exchange != null) {
+				admin.declareExchange(this.exchange);
+			}
+			if (this.binding != null) {
+				admin.declareBinding(this.binding);
+			}
 		}
 	}
 
-	@Override
-	public void start() {
-		this.doDeclares(); // connection may have been lost while we were stopped
-		super.start();
-	}
-
-	@Override
-	public void onCreate(Connection connection) {
-		doDeclares();
-	}
-
-	@Override
-	public void onClose(Connection connection) {
-	}
-
 }

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java

@@ -100,7 +100,7 @@ public class ChannelTests {
 	private PollableChannel out;
 
 	@Autowired
-	private CachingConnectionFactory factory;
+	private CachingConnectionFactory connectionFactory;
 
 	@Autowired
 	private AmqpHeaderMapper mapperIn;
@@ -115,7 +115,6 @@ public class ChannelTests {
 	}
 
 	@Test
-	@DirtiesContext
 	public void pubSubLostConnectionTest() throws Exception {
 		final CyclicBarrier latch = new CyclicBarrier(2);
 		channel.subscribe(message -> {
@@ -130,13 +129,15 @@ public class ChannelTests {
 		latch.reset();
 		BlockingQueueConsumer consumer = (BlockingQueueConsumer) TestUtils.getPropertyValue(this.channel,
 				"container.consumers", Set.class).iterator().next();
-		factory.destroy();
+		connectionFactory.destroy();
 		waitForNewConsumer(this.channel, consumer);
 		this.channel.send(new GenericMessage<String>("bar"));
 		latch.await(10, TimeUnit.SECONDS);
 		this.channel.destroy();
 		this.pubSubWithEP.destroy();
-		assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size());
+		this.withEP.destroy();
+		this.pollableWithEP.destroy();
+		assertEquals(0, TestUtils.getPropertyValue(connectionFactory, "connectionListener.delegates", Collection.class).size());
 	}
 
 	@SuppressWarnings("unchecked")
@@ -166,25 +167,31 @@ public class ChannelTests {
 	 */
 	@Test
 	public void channelDeclarationTests() {
-		RabbitAdmin admin = new RabbitAdmin(this.factory);
+		RabbitAdmin admin = new RabbitAdmin(this.connectionFactory);
 		admin.deleteQueue("implicit");
-		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.factory);
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
+		container.setAutoStartup(false);
 		AmqpTemplate amqpTemplate = mock(AmqpTemplate.class);
 		PointToPointSubscribableAmqpChannel channel = new PointToPointSubscribableAmqpChannel("implicit", container,
 				amqpTemplate);
 		channel.setBeanFactory(mock(BeanFactory.class));
 		channel.afterPropertiesSet();
+		channel.onCreate(null);
+
 		assertNotNull(admin.getQueueProperties("implicit"));
-		admin.deleteQueue("implicit");
 
 		admin.deleteQueue("explicit");
 		channel.setQueueName("explicit");
 		channel.afterPropertiesSet();
+		channel.onCreate(null);
+
 		assertNotNull(admin.getQueueProperties("explicit"));
 
 		admin.deleteQueue("explicit");
 		admin.declareQueue(new Queue("explicit", false)); // verify no declaration if exists with non-standard props
 		channel.afterPropertiesSet();
+		channel.onCreate(null);
+
 		assertNotNull(admin.getQueueProperties("explicit"));
 		admin.deleteQueue("explicit");
 	}
@@ -193,7 +200,7 @@ public class ChannelTests {
 	public void testAmqpChannelFactoryBean() throws Exception {
 		AmqpChannelFactoryBean channelFactoryBean = new AmqpChannelFactoryBean();
 		channelFactoryBean.setBeanFactory(mock(BeanFactory.class));
-		channelFactoryBean.setConnectionFactory(this.factory);
+		channelFactoryBean.setConnectionFactory(this.connectionFactory);
 		channelFactoryBean.setBeanName("testChannel");
 		channelFactoryBean.afterPropertiesSet();
 		AbstractAmqpChannel channel = channelFactoryBean.getObject();
@@ -201,14 +208,14 @@ public class ChannelTests {
 
 		channelFactoryBean = new AmqpChannelFactoryBean();
 		channelFactoryBean.setBeanFactory(mock(BeanFactory.class));
-		channelFactoryBean.setConnectionFactory(this.factory);
+		channelFactoryBean.setConnectionFactory(this.connectionFactory);
 		channelFactoryBean.setBeanName("testChannel");
 		channelFactoryBean.setPubSub(true);
 		channelFactoryBean.afterPropertiesSet();
 		channel = channelFactoryBean.getObject();
 		assertThat(channel, instanceOf(PublishSubscribeAmqpChannel.class));
 
-		RabbitAdmin rabbitAdmin = new RabbitAdmin(this.factory);
+		RabbitAdmin rabbitAdmin = new RabbitAdmin(this.connectionFactory);
 		rabbitAdmin.deleteQueue("testChannel");
 		rabbitAdmin.deleteExchange("si.fanout.testChannel");
 	}
@@ -241,11 +248,11 @@ public class ChannelTests {
 
 	@Test
 	public void messageConversionTests() throws Exception {
-		RabbitTemplate amqpTemplate = new RabbitTemplate(this.factory);
+		RabbitTemplate amqpTemplate = new RabbitTemplate(this.connectionFactory);
 		MessageConverter messageConverter = mock(MessageConverter.class);
 		amqpTemplate.setMessageConverter(messageConverter);
 		PointToPointSubscribableAmqpChannel channel = new PointToPointSubscribableAmqpChannel("testConvertFail",
-				new SimpleMessageListenerContainer(this.factory), amqpTemplate);
+				new SimpleMessageListenerContainer(this.connectionFactory), amqpTemplate);
 		channel.afterPropertiesSet();
 		MessageListener listener = TestUtils.getPropertyValue(channel, "container.messageListener",
 				MessageListener.class);

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java

@@ -35,11 +35,9 @@ import java.util.List;
 import org.apache.commons.logging.Log;
 import org.junit.Test;
 
-import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
-import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.Connection;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
@@ -55,6 +53,7 @@ import com.rabbitmq.client.Channel;
 /**
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.1
  *
  */
@@ -103,15 +102,8 @@ public class DispatcherHasNoSubscribersTests {
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
 		container.setConnectionFactory(connectionFactory);
 		AmqpTemplate amqpTemplate = mock(AmqpTemplate.class);
-		final Queue queue = new Queue("noSubscribersQueue");
 		PublishSubscribeAmqpChannel amqpChannel = new PublishSubscribeAmqpChannel("noSubscribersChannel",
-				container, amqpTemplate) {
-			@Override
-			protected String obtainQueueName(AmqpAdmin admin,
-					String channelName) {
-				return queue.getName();
-			}
-		};
+				container, amqpTemplate);
 		amqpChannel.setBeanName("noSubscribersChannel");
 		amqpChannel.setBeanFactory(mock(BeanFactory.class));
 		amqpChannel.afterPropertiesSet();

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml

@@ -1,12 +1,8 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:amqp="http://www.springframework.org/schema/integration/amqp"
-	xmlns:int="http://www.springframework.org/schema/integration"
-	xmlns:rabbit="http://www.springframework.org/schema/rabbit"
-	xsi:schemaLocation="http://www.springframework.org/schema/integration/amqp http://www.springframework.org/schema/integration/amqp/spring-integration-amqp.xsd
-		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
-		http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:amqp="http://www.springframework.org/schema/integration/amqp"
+	   xsi:schemaLocation="http://www.springframework.org/schema/integration/amqp http://www.springframework.org/schema/integration/amqp/spring-integration-amqp.xsd
 		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
 
 	<amqp:channel id="channelWithInterceptor">
@@ -18,26 +14,36 @@
 	<bean id="rabbitConnectionFactory" class="org.springframework.integration.amqp.StubRabbitConnectionFactory"/>
 
 	<amqp:channel id="channelWithSubscriberLimit" max-subscribers="1" missing-queues-fatal="false"
-			template-channel-transacted="true" consumers-per-queue="2" />
+				  template-channel-transacted="true" consumers-per-queue="2"
+				  recovery-interval="0"
+				  shutdown-timeout="0"/>
 
-	<amqp:publish-subscribe-channel id="pubSub" />
+	<amqp:publish-subscribe-channel id="pubSub"
+									recovery-interval="0"
+									shutdown-timeout="0"/>
 
 	<amqp:channel id="withEP" extract-payload="true" default-delivery-mode="NON_PERSISTENT"
-		inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" />
+				  inbound-header-mapper="inMapper" outbound-header-mapper="outMapper"
+				  recovery-interval="0"
+				  shutdown-timeout="0"/>
 
 	<amqp:channel id="pollableWithEP" extract-payload="true" message-driven="false"
-		headers-last="true"
-		inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" />
+				  headers-last="true"
+				  inbound-header-mapper="inMapper" outbound-header-mapper="outMapper"
+				  recovery-interval="0"
+				  shutdown-timeout="0"/>
 
 	<amqp:publish-subscribe-channel id="pubSubWithEP" extract-payload="true"
-		inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" />
+									inbound-header-mapper="inMapper" outbound-header-mapper="outMapper"
+									recovery-interval="0"
+									shutdown-timeout="0"/>
 
 	<bean id="inMapper" class="org.mockito.Mockito" factory-method="mock">
-		<constructor-arg value="org.springframework.integration.amqp.support.AmqpHeaderMapper" />
+		<constructor-arg value="org.springframework.integration.amqp.support.AmqpHeaderMapper"/>
 	</bean>
 
 	<bean id="outMapper" class="org.mockito.Mockito" factory-method="mock">
-		<constructor-arg value="org.springframework.integration.amqp.support.AmqpHeaderMapper" />
+		<constructor-arg value="org.springframework.integration.amqp.support.AmqpHeaderMapper"/>
 	</bean>
 
 </beans>