Refactored AMQP channels

added namespace support

moved afterPropertiesSet call into FactoryBean

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

@@ -36,13 +36,29 @@ 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.
+	 */
+	protected String getExchangeName() {
+		return "";
+	}
+
+	/**
+	 * Subclasses may override this method to return a routing key.
+	 * By default, there will be no routing key (empty string).
+	 */
+	protected String getRoutingKey() {
+		return "";
+	}
+
 	AmqpTemplate getAmqpTemplate() {
 		return this.amqpTemplate;
 	}
 
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
-		this.amqpTemplate.convertAndSend(message);
+		this.amqpTemplate.convertAndSend(this.getExchangeName(), this.getRoutingKey(), message);
 		return true;
 	}
 

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

@@ -18,25 +18,23 @@ package org.springframework.integration.amqp.channel;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
+import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;
-import org.springframework.amqp.core.Binding;
-import org.springframework.amqp.core.BindingBuilder;
-import org.springframework.amqp.core.FanoutExchange;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+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.Message;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.core.SubscribableChannel;
-import org.springframework.integration.dispatcher.BroadcastingDispatcher;
 import org.springframework.integration.dispatcher.MessageDispatcher;
-import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
-import org.springframework.integration.dispatcher.UnicastingDispatcher;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.util.Assert;
 
@@ -44,24 +42,21 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @since 2.1
  */
-public class SubscribableAmqpChannel extends AbstractAmqpChannel implements SubscribableChannel, SmartLifecycle, DisposableBean {
+abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel implements SubscribableChannel, SmartLifecycle, DisposableBean {
 
 	private final String channelName;
 
 	private final SimpleMessageListenerContainer container;
 
-	private final boolean isPubSub;
-
 	private volatile MessageDispatcher dispatcher;
 
 
-	public SubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate, boolean isPubSub) {
+	public AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) {
 		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;
 	}
 
 
@@ -76,67 +71,45 @@ public class SubscribableAmqpChannel extends AbstractAmqpChannel implements Subs
 	@Override
 	public void onInit() throws Exception {
 		super.onInit();
-		this.configureDispatcher();
-		AmqpTemplate amqpTemplate = this.getAmqpTemplate();
-		if (!(amqpTemplate instanceof RabbitTemplate)) {
-			throw new IllegalArgumentException("AmqpTemplate must be a RabbitTemplate");
-		}
-		RabbitTemplate rabbitTemplate = (RabbitTemplate) amqpTemplate;
-		RabbitAdmin admin = new RabbitAdmin(rabbitTemplate.getConnectionFactory());
-		if (this.isPubSub) {
-			FanoutExchange exchange = new FanoutExchange("si.fanout." + this.channelName);
-			admin.declareExchange(exchange);
-			Queue queue = admin.declareQueue();
-			Binding binding = BindingBuilder.bind(queue).to(exchange);
-			admin.declareBinding(binding);
-			this.container.setQueues(queue);
-			rabbitTemplate.setExchange(exchange.getName());
-		}
-		else {
-			String queueName = "si." + this.channelName;
-			Queue queue = new Queue(queueName);
-			admin.declareQueue(queue);
-			this.container.setQueues(queue);
-			rabbitTemplate.setRoutingKey(queueName);
-		}
-		MessageListener listener = new DispatchingMessageListener(rabbitTemplate, this.dispatcher);
+		this.dispatcher = this.createDispatcher();
+		AmqpAdmin admin = new RabbitAdmin(this.container.getConnectionFactory());
+		Queue queue = this.initializeQueue(admin, this.channelName);
+		this.container.setQueues(queue);
+		MessageConverter converter = (this.getAmqpTemplate() instanceof RabbitTemplate)
+				? ((RabbitTemplate) this.getAmqpTemplate()).getMessageConverter()
+				: new SimpleMessageConverter();
+		MessageListener listener = new DispatchingMessageListener(converter, this.dispatcher);
 		this.container.setMessageListener(listener);
 		if (!this.container.isActive()) {
 			this.container.afterPropertiesSet();
 		}
-		rabbitTemplate.afterPropertiesSet();
 	}
 
-	private void configureDispatcher() {
-		if (this.isPubSub) {
-			this.dispatcher = new BroadcastingDispatcher();
-		}
-		else {
-			UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher();
-			unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy());
-			this.dispatcher = unicastingDispatcher;
-		}
-	}
+	protected abstract MessageDispatcher createDispatcher();
+
+	protected abstract Queue initializeQueue(AmqpAdmin admin, String channelName);
 
 
 	private static class DispatchingMessageListener implements MessageListener {
 
 		private final Log logger = LogFactory.getLog(this.getClass());
 
-		private final RabbitTemplate rabbitTemplate;
-
 		private final MessageDispatcher dispatcher;
 
+		private final MessageConverter converter;
 
-		private DispatchingMessageListener(RabbitTemplate rabbitTemplate, MessageDispatcher dispatcher) {
-			this.rabbitTemplate = rabbitTemplate;
+
+		private DispatchingMessageListener(MessageConverter converter, MessageDispatcher dispatcher) {
+			Assert.notNull(converter, "MessageConverter must not be null");
+			Assert.notNull(dispatcher, "MessageDispatcher must not be null");
+			this.converter = converter;
 			this.dispatcher = dispatcher;
 		}
 
 
 		public void onMessage(org.springframework.amqp.core.Message message) {
 			try {
-				Object converted = this.rabbitTemplate.getMessageConverter().fromMessage(message);
+				Object converted = this.converter.fromMessage(message);
 				if (converted != null) {
 					Message<?> messageToSend = (converted instanceof Message<?>) ? (Message<?>) converted
 							: MessageBuilder.withPayload(converted).build();

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

@@ -0,0 +1,71 @@
+/*
+ * 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.
+ */
+
+package org.springframework.integration.amqp.channel;
+
+import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.integration.dispatcher.MessageDispatcher;
+import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
+import org.springframework.integration.dispatcher.UnicastingDispatcher;
+
+/**
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmqpChannel {
+
+	private volatile String queueName;
+
+
+	public PointToPointSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) {
+		super(channelName, container, amqpTemplate);
+	}
+
+
+	/**
+	 * Provide a Queue name to be used. If this is not provided,
+	 * the Queue's name will be the same as the channel name.
+	 */
+	public void setQueueName(String queueName) {
+		this.queueName = queueName;
+	}
+
+	@Override
+	protected Queue initializeQueue(AmqpAdmin admin, String channelName) {
+		if (this.queueName == null) {
+			this.queueName = channelName;
+		}
+		Queue queue = new Queue(this.queueName);
+		admin.declareQueue(queue);
+		return queue;
+	}
+
+	@Override
+	protected MessageDispatcher createDispatcher() {
+		UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher();
+		unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy());
+		return unicastingDispatcher;
+	}
+
+	@Override
+	protected String getRoutingKey() {
+		return this.queueName;
+	}
+
+}

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.amqp.channel;
 
+import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
@@ -26,6 +27,10 @@ import org.springframework.integration.support.MessageBuilder;
 import org.springframework.util.Assert;
 
 /**
+ * A {@link PollableChannel} implementation that is backed by an AMQP Queue.
+ * Messages will be sent to the default (no-name) exchange with that Queue's
+ * name as the routing key.
+ * 
  * @author Mark Fisher
  * @since 2.1
  */
@@ -33,6 +38,10 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
 
 	private final String channelName;
 
+	private volatile String queueName;
+
+	private volatile AmqpAdmin amqpAdmin;
+
 
 	public PollableAmqpChannel(String channelName, AmqpTemplate amqpTemplate) {
 		super(amqpTemplate);
@@ -41,26 +50,49 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
 	}
 
 
+	/**
+	 * 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.
+	 */
+	public void setQueueName(String queueName) {
+		this.queueName = 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.
+	 */
+	public void setAmqpAdmin(AmqpAdmin amqpAdmin) {
+		this.amqpAdmin = amqpAdmin;
+	}
+
 	@Override
 	protected void onInit() throws Exception {
 		AmqpTemplate amqpTemplate = this.getAmqpTemplate();
-		if (!(amqpTemplate instanceof RabbitTemplate)) {
-			throw new IllegalArgumentException("AmqpTemplate must be a RabbitTemplate");
+		if (this.queueName == null) {
+			if (this.amqpAdmin == null && amqpTemplate instanceof RabbitTemplate) {
+				this.amqpAdmin = new RabbitAdmin(((RabbitTemplate) amqpTemplate).getConnectionFactory());				
+			}
+			Assert.notNull(this.amqpAdmin,
+					"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));
 		}
-		RabbitTemplate rabbitTemplate = (RabbitTemplate) amqpTemplate;
-		RabbitAdmin admin = new RabbitAdmin(rabbitTemplate.getConnectionFactory());
-		String queueName = "si." + this.channelName;
-		Queue queue = new Queue(queueName);
-		admin.declareQueue(queue);
-		rabbitTemplate.setRoutingKey(queueName);
-		rabbitTemplate.setQueue(queueName);
+	}
+
+	@Override
+	protected String getRoutingKey() {
+		return this.queueName;
 	}
 
 	public Message<?> receive() {
 		if (!this.getInterceptors().preReceive(this)) {
  			return null;
  		}
-		Object object = this.getAmqpTemplate().receiveAndConvert();
+		Object object = this.getAmqpTemplate().receiveAndConvert(this.queueName);
 		if (object == null) {
 			return null;
 		}

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

@@ -0,0 +1,76 @@
+/*
+ * 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.
+ */
+
+package org.springframework.integration.amqp.channel;
+
+import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.amqp.core.AmqpTemplate;
+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.listener.SimpleMessageListenerContainer;
+import org.springframework.integration.dispatcher.BroadcastingDispatcher;
+import org.springframework.integration.dispatcher.MessageDispatcher;
+
+/**
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel {
+
+	private volatile FanoutExchange exchange;
+
+
+	public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) {
+		super(channelName, container, amqpTemplate);
+	}
+
+
+	/**
+	 * Configure the FanoutExchange instance. If this is not provided, then a
+	 * FanoutExchange will be declared implicitly, and its name will be the same
+	 * as the channel name prefixed by "si.fanout.". In either case, an effectively
+	 * anonymous Queue will be declared automatically.
+	 */
+	public void setExchange(FanoutExchange exchange) {
+		this.exchange = exchange;
+	}
+
+	@Override
+	protected Queue initializeQueue(AmqpAdmin admin, String channelName) {
+		if (this.exchange == null) {
+			String exchangeName = "si.fanout." + channelName;
+			this.exchange = new FanoutExchange(exchangeName);
+		}
+		admin.declareExchange(this.exchange);
+		Queue queue = admin.declareQueue();
+		Binding binding = BindingBuilder.bind(queue).to(this.exchange);
+		admin.declareBinding(binding);
+		return queue;
+	}
+
+	@Override
+	protected MessageDispatcher createDispatcher() {
+		return new BroadcastingDispatcher();
+	}
+
+	@Override
+	protected String getExchangeName() {
+		return (this.exchange != null) ? this.exchange.getName() : "";
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java

@@ -22,6 +22,9 @@ import java.util.concurrent.Executor;
 import org.aopalliance.aop.Advice;
 
 import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.FanoutExchange;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
@@ -29,12 +32,14 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.config.AbstractFactoryBean;
 import org.springframework.context.Lifecycle;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
+import org.springframework.integration.amqp.channel.PointToPointSubscribableAmqpChannel;
 import org.springframework.integration.amqp.channel.PollableAmqpChannel;
-import org.springframework.integration.amqp.channel.SubscribableAmqpChannel;
+import org.springframework.integration.amqp.channel.PublishSubscribeAmqpChannel;
 import org.springframework.integration.channel.ChannelInterceptor;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.interceptor.TransactionAttribute;
@@ -42,10 +47,11 @@ import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ErrorHandler;
 import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * If point-to-point, we send to the default exchange with the routing key
- * equal to "si.[beanName]" and we declare that same Queue and register a listener
+ * equal to "[beanName]" and we declare that same Queue and register a listener
  * if message-driven or poll explicitly otherwise. If publish-subscribe, we declare
  * a FanoutExchange named "si.fanout.[beanName]" and we send to that without any
  * routing key, and on the receiving side, we create an anonymous Queue that is
@@ -62,10 +68,16 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	private final boolean messageDriven;
 
-	private final RabbitTemplate rabbitTemplate = new RabbitTemplate();
+	private final AmqpTemplate amqpTemplate = new RabbitTemplate();
 
 	private volatile SimpleMessageListenerContainer container;
 
+	private volatile AmqpAdmin amqpAdmin;
+
+	private volatile FanoutExchange exchange;
+
+	private volatile String queueName;
+
 	private volatile boolean autoStartup = true;
 
 	private volatile Advice[] adviceChain;
@@ -127,16 +139,55 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		this.interceptors = interceptors;
 	}
 
+	/**
+	 * This is an optional reference to an AmqpAdmin to use when
+	 * declaring a Queue implicitly for a PollableAmqpChannel. It
+	 * is not needed for the message-driven (Subscribable) channels
+	 * since those are able to create a RabbitAdmin instance using
+	 * the underlying listener container's ConnectionFactory.
+	 */
+	public void setAmqpAdmin(AmqpAdmin amqpAdmin) {
+		this.amqpAdmin = amqpAdmin;
+	}
+
+	/**
+	 * Set the FanoutExchange to use. This is only relevant for
+	 * publish-subscribe-channels, and even then if not provided,
+	 * a FanoutExchange will be implicitly created.
+	 */
+	public void setExchange(FanoutExchange exchange) {
+		this.exchange = exchange;
+	}
+
+	/**
+	 * Set the Queue name to use. This is only relevant for
+	 * point-to-point channels, even then if not provided,
+	 * a Queue will be implicitly created.
+	 */
+	public void setQueueName(String queueName) {
+		this.queueName = queueName;
+	}
+
 	/*
 	 * Template-only properties
 	 */
 
 	public void setEncoding(String encoding) {
-		this.rabbitTemplate.setEncoding(encoding);
+		if (this.amqpTemplate instanceof RabbitTemplate) {
+			((RabbitTemplate) this.amqpTemplate).setEncoding(encoding);
+		}
+		else if (logger.isInfoEnabled()) {
+			logger.info("AmqpTemplate is not a RabbitTemplate, so configured 'encoding' value will be ignored.");
+		}
 	}
 
 	public void setMessageConverter(MessageConverter messageConverter) {
-		this.rabbitTemplate.setMessageConverter(messageConverter);
+		if (this.amqpTemplate instanceof RabbitTemplate) {
+			((RabbitTemplate) this.amqpTemplate).setMessageConverter(messageConverter);
+		}
+		else if (logger.isInfoEnabled()) {
+			logger.info("AmqpTemplate is not a RabbitTemplate, so configured MessageConverter will be ignored.");
+		}
 	}
 
 	/*
@@ -145,17 +196,23 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	public void setChannelTransacted(boolean channelTransacted) {
 		this.channelTransacted = channelTransacted;
-		this.rabbitTemplate.setChannelTransacted(channelTransacted);
+		if (this.amqpTemplate instanceof RabbitTemplate) {
+			((RabbitTemplate) this.amqpTemplate).setChannelTransacted(channelTransacted);
+		}
 	}
 
 	public void setConnectionFactory(ConnectionFactory connectionFactory) {
 		this.connectionFactory = connectionFactory;
-		this.rabbitTemplate.setConnectionFactory(this.connectionFactory);
+		if (this.amqpTemplate instanceof RabbitTemplate) {
+			((RabbitTemplate) this.amqpTemplate).setConnectionFactory(this.connectionFactory);
+		}
 	}
 
 	public void setMessagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) {
-		this.rabbitTemplate.setMessagePropertiesConverter(messagePropertiesConverter);
 		this.messagePropertiesConverter = messagePropertiesConverter;
+		if (this.amqpTemplate instanceof RabbitTemplate) {
+			((RabbitTemplate) this.amqpTemplate).setMessagePropertiesConverter(messagePropertiesConverter);
+		}
 	}
 
 	/*
@@ -235,12 +292,37 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 	protected AbstractAmqpChannel createInstance() throws Exception {
 		if (this.messageDriven) {
 			this.container = this.createContainer();
-			this.channel = new SubscribableAmqpChannel(this.beanName, this.container, this.rabbitTemplate, this.isPubSub);
+			if (this.amqpTemplate instanceof InitializingBean) {
+				((InitializingBean) this.amqpTemplate).afterPropertiesSet();
+			}
+			if (this.isPubSub) {
+				PublishSubscribeAmqpChannel pubsub = new PublishSubscribeAmqpChannel(
+						this.beanName, this.container, this.amqpTemplate);
+				if (this.exchange != null) {
+					pubsub.setExchange(this.exchange);
+				}
+				this.channel = pubsub;
+			}
+			else {
+				PointToPointSubscribableAmqpChannel p2p = new PointToPointSubscribableAmqpChannel(
+						this.beanName, this.container, this.amqpTemplate);
+				if (StringUtils.hasText(this.queueName)) {
+					p2p.setQueueName(this.queueName);
+				}
+				this.channel = p2p;
+			}
 		}
 		else {
 			Assert.isTrue(!Boolean.TRUE.equals(this.isPubSub),
 					"An AMQP 'publish-subscribe-channel' must be message-driven.");
-			this.channel = new PollableAmqpChannel(this.beanName, this.rabbitTemplate);
+			PollableAmqpChannel pollable = new PollableAmqpChannel(this.beanName, this.amqpTemplate);
+			if (this.amqpAdmin != null) {
+				pollable.setAmqpAdmin(this.amqpAdmin);
+			}
+			if (StringUtils.hasText(this.queueName)) {
+				pollable.setQueueName(this.queueName);
+			}
+			this.channel = pollable;
 		}
 		if (!CollectionUtils.isEmpty(this.interceptors)) {
 			this.channel.setInterceptors(this.interceptors);

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java

@@ -53,16 +53,19 @@ public class AmqpChannelParser extends AbstractChannelParser {
 		}
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge-mode");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "advice-chain");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "amqp-admin");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "channel-transacted");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "concurrent-consumers");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "encoding");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-handler");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "exchange");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expose-listener-channel");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-properties-converter");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "prefetch-count");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-name");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "shutdown-timeout");

spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd

@@ -259,18 +259,66 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="queue-name" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	Provide an explicitly configured queue name. If this is not provided, then a Queue will be created
+	implicitly with the same name as the channel itself (the "id" of this element). If this channel is
+	not message-driven, the implicit creation will require that either an AmqpAdmin instance has been
+	provided via the "amqp-admin" attribute or that the configured AmqpTemplate is an instance of RabbitTemplate.
+	If the channel is message-driven, the AmqpAdmin will be created using the underlying listener container's
+	ConnectionFactory.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="amqp-admin" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	An AmqpAdmin instance to use when declaring a Queue implicitly. This is only needed if an explicit
+	"queue-name" is not provided and the channel is not message-driven. Even then, if the referenced
+	AmqpTemplate is an instance of RabbitTemplate, the AmqpAdmin can be constructed from that template's
+	ConnectionFactory.
+							</xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="org.springframework.amqp.core.AmqpAdmin"/>
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
 	</xsd:element>
 
-	<xsd:element name="publish-subscribe-channel" type="channelType">
+	<xsd:element name="publish-subscribe-channel">
 		<xsd:annotation>
 			<xsd:documentation>
 	Creates a publish-subscribe-channel that is backed by an AMQP FanoutExchange.
 	Always message-driven (subscribable).
 			</xsd:documentation>
 		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="channelType">
+					<xsd:attribute name="exchange" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	Reference to a FanoutExchange instance to which this channel should send Messages. If not provided,
+	a FanoutExchange will be declared with this channel's name prefixed by "si.fanout.".
+	A Queue will be declared automatically and bound to that exchange to handle the consumer role
+	of this channel. 
+							</xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="org.springframework.amqp.core.FanoutExchange"/>
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
 	</xsd:element>
 
 	<xsd:complexType name="channelType">