Merge pull request #111 from markfisher/INT-1878

* INT-1878:
  initial implementation of AMQP channels

docs/src/reference/docbook/amqp.xml

@@ -18,6 +18,10 @@
 			<listitem>Outbound Gateway</listitem>
 	    </itemizedlist>
     </para>
+	<para>
+		Spring Integration also provides a point-to-point Message Channel as well as a
+		publish/subscribe Message Channel backed by AMQP Exchanges and Queues.
+	</para>
     <para>
         In order to provide AMQP support, Spring Integration relies on Spring AMQP 
         (<ulink url="http://www.springsource.org/spring-amqp">http://www.springsource.org/spring-amqp</ulink>) 
@@ -27,7 +31,7 @@
     </para>
     <para>
         Whereas the provided AMQP Channel Adapters are intended for unidirectional 
-        Messaging (send or receive), only, Spring Integration also provides inbound 
+        Messaging (send or receive) only, Spring Integration also provides inbound 
         and outbound AMQP Gateways for request/reply operations. 
     </para>
     <tip>
@@ -250,48 +254,48 @@
 
     <para>A configuration sample for an AMQP Outbound Channel Adapter is shown 
           below with all available parameters.</para>
-    <programlisting language="xml"><![CDATA[<int-amqp:outbound-channel-adapter id="outboundAmqp"]]><co              id="amqp-inbound-channel-adapter-xml-1-co" linkends="amqp-inbound-channel-adapter-xml-1" /><![CDATA[
-                               channel="outboundChannel"]]><co      id="amqp-inbound-channel-adapter-xml-2-co" linkends="amqp-inbound-channel-adapter-xml-2" /><![CDATA[
-                               amqp-template="myAmqpTemplate"]]><co id="amqp-inbound-channel-adapter-xml-3-co" linkends="amqp-inbound-channel-adapter-xml-3" /><![CDATA[
-                               exchange-name=""]]><co               id="amqp-inbound-channel-adapter-xml-4-co" linkends="amqp-inbound-channel-adapter-xml-4" /><![CDATA[
-                               order="1"]]><co                      id="amqp-inbound-channel-adapter-xml-5-co" linkends="amqp-inbound-channel-adapter-xml-5" /><![CDATA[
-                               routing-key=""]]><co                 id="amqp-inbound-channel-adapter-xml-6-co" linkends="amqp-inbound-channel-adapter-xml-6" /><![CDATA[
-                               routing-key-expression=""]]><co      id="amqp-inbound-channel-adapter-xml-7-co" linkends="amqp-inbound-channel-adapter-xml-7" /><![CDATA[/>]]>
+    <programlisting language="xml"><![CDATA[<int-amqp:outbound-channel-adapter id="outboundAmqp"]]><co              id="amqp-outbound-channel-adapter-xml-1-co" linkends="amqp-outbound-channel-adapter-xml-1" /><![CDATA[
+                               channel="outboundChannel"]]><co      id="amqp-outbound-channel-adapter-xml-2-co" linkends="amqp-outbound-channel-adapter-xml-2" /><![CDATA[
+                               amqp-template="myAmqpTemplate"]]><co id="amqp-outbound-channel-adapter-xml-3-co" linkends="amqp-outbound-channel-adapter-xml-3" /><![CDATA[
+                               exchange-name=""]]><co               id="amqp-outbound-channel-adapter-xml-4-co" linkends="amqp-outbound-channel-adapter-xml-4" /><![CDATA[
+                               order="1"]]><co                      id="amqp-outbound-channel-adapter-xml-5-co" linkends="amqp-outbound-channel-adapter-xml-5" /><![CDATA[
+                               routing-key=""]]><co                 id="amqp-outbound-channel-adapter-xml-6-co" linkends="amqp-outbound-channel-adapter-xml-6" /><![CDATA[
+                               routing-key-expression=""]]><co      id="amqp-outbound-channel-adapter-xml-7-co" linkends="amqp-outbound-channel-adapter-xml-7" /><![CDATA[/>]]>
     </programlisting>
     <para>
         <calloutlist>
-	        <callout arearefs="amqp-inbound-channel-adapter-xml-1-co" id="amqp-inbound-channel-adapter-xml-1">
+	        <callout arearefs="amqp-outbound-channel-adapter-xml-1-co" id="amqp-outbound-channel-adapter-xml-1">
 	            <para>Unique ID for this adapter.
 	            <emphasis>Optional</emphasis>.</para>
 	        </callout>
-            <callout arearefs="amqp-inbound-channel-adapter-xml-2-co" id="amqp-inbound-channel-adapter-xml-2">
+            <callout arearefs="amqp-outbound-channel-adapter-xml-2-co" id="amqp-outbound-channel-adapter-xml-2">
                 <para>Message Channel to which Messages should be sent 
 					  in order to have them converted and published to an 
 					  AMQP Exchange.
                 <emphasis>Required</emphasis>.</para>
             </callout>
-            <callout arearefs="amqp-inbound-channel-adapter-xml-3-co" id="amqp-inbound-channel-adapter-xml-3">
+            <callout arearefs="amqp-outbound-channel-adapter-xml-3-co" id="amqp-outbound-channel-adapter-xml-3">
                 <para>Bean Reference to the configured AMQP Template
                 <emphasis>Optional (Defaults to "amqpTemplate")</emphasis>.</para>
             </callout>
-            <callout arearefs="amqp-inbound-channel-adapter-xml-4-co" id="amqp-inbound-channel-adapter-xml-4">
+            <callout arearefs="amqp-outbound-channel-adapter-xml-4-co" id="amqp-outbound-channel-adapter-xml-4">
                 <para>The name of the AMQP Exchange to which Messages 
 					  should be sent. If not provided, Messages will be sent 
 					  to the default, no-name Exchange.
                 <emphasis>Optional</emphasis>.</para>
             </callout>
-            <callout arearefs="amqp-inbound-channel-adapter-xml-5-co" id="amqp-inbound-channel-adapter-xml-5">
+            <callout arearefs="amqp-outbound-channel-adapter-xml-5-co" id="amqp-outbound-channel-adapter-xml-5">
                 <para>The order for this consumer when multiple 
 					  consumers are registered thereby enabling load-
 					  balancing and/or failover.
                 <emphasis>Optional (Defaults to Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE])</emphasis>.</para>
             </callout>
-            <callout arearefs="amqp-inbound-channel-adapter-xml-6-co" id="amqp-inbound-channel-adapter-xml-6">
+            <callout arearefs="amqp-outbound-channel-adapter-xml-6-co" id="amqp-outbound-channel-adapter-xml-6">
                 <para>The fixed routing-key to use when sending Messages. By
                       default, this will be an empty String.
                 <emphasis>Optional</emphasis>.</para>
             </callout>
-            <callout arearefs="amqp-inbound-channel-adapter-xml-7-co" id="amqp-inbound-channel-adapter-xml-7">
+            <callout arearefs="amqp-outbound-channel-adapter-xml-7-co" id="amqp-outbound-channel-adapter-xml-7">
                 <para>The routing-key to use when sending Messages 
 					  evaluated as an expression on the message (e.g. 
 					  'payload.key'). By default, this will be an empty String.
@@ -419,15 +423,30 @@
     <title>AMQP Backed Message Channels</title>
 
     <para>
-        This feature is not currently available, yet, but it is planned for the 
-        Spring Integration 2.1 M2 release. In order to follow progress, please 
-        visit the Spring Integration issue tracker at 
-        <ulink url="https://jira.springsource.org/browse/INT-1878">
-            https://jira.springsource.org/browse/INT-1878
-        </ulink>
+        There are two Message Channel implementations available. One is point-to-point, and the other is publish/subscribe.
+		Both of these channels provide a wide range of configuration attributes for the underlying AmqpTemplate and
+		SimpleMessageListenerContainer as you have seen on the Channel Adapters and Gateways. However, the examples we'll
+		show here are going to have minimal configuration. Explore the XML schema to view the available attributes.
     </para>
-    
+	<para>
+		A point-to-point channel would look like this:
+		<programlisting language="xml"><![CDATA[<int-amqp:channel id="p2pChannel"/>]]></programlisting>
+		Under the covers a Queue named "si.p2pChannel" would be declared, and this channel will send to
+		that Queue (technically by sending to the no-name Direct Exchange with a routing key that matches this Queue's name).
+		This channel will also register a consumer on that Queue. If for some reason, you want the Queue to be "pollable"
+		instead of message-driven, then simply provide the "message-driven" flag with a value of false:
+		<programlisting language="xml"><![CDATA[<int-amqp:channel id="p2pPollableChannel"  message-driven="false"/>]]></programlisting>
+	</para>
+	<para>
+    	A publish/subscribe channel would look like this:
+		<programlisting language="xml"><![CDATA[<int-amqp:publish-subscribe-channel id="pubSubChannel"/>]]></programlisting>
+		Under the covers a Fanout Exchange named "si.fanout.pubSubChannel" would be declared, and this channel will send
+		to that Fanout Exchange. This channel will also declare a server-named exclusive, autodelete, non-durable Queue
+		and bind that to the Fanout Exchange while registering a consumer on that Queue to receive Messages. There is no
+		"pollable" option for a publish-subscribe-channel; it must be message-driven.
+	</para>
   </section>
+
   <section>
     <title>AMQP Samples</title>
     <para>

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

@@ -0,0 +1,49 @@
+/*
+ * 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.AmqpTemplate;
+import org.springframework.integration.Message;
+import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.util.Assert;
+
+/**
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
+
+	private final AmqpTemplate amqpTemplate;
+
+
+	AbstractAmqpChannel(AmqpTemplate amqpTemplate) {
+		Assert.notNull(amqpTemplate, "amqpTemplate must not be null");
+		this.amqpTemplate = amqpTemplate;
+	}
+
+
+	AmqpTemplate getAmqpTemplate() {
+		return this.amqpTemplate;
+	}
+
+	@Override
+	protected boolean doSend(Message<?> message, long timeout) {
+		this.amqpTemplate.convertAndSend(message);
+		return true;
+	}
+
+}

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

@@ -0,0 +1,85 @@
+/*
+ * 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.AmqpTemplate;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.integration.Message;
+import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.util.Assert;
+
+/**
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public class PollableAmqpChannel extends AbstractAmqpChannel implements PollableChannel {
+
+	private final String channelName;
+
+
+	public PollableAmqpChannel(String channelName, AmqpTemplate amqpTemplate) {
+		super(amqpTemplate);
+		Assert.hasText(channelName, "channel name must not be empty");
+		this.channelName = channelName;
+	}
+
+
+	@Override
+	protected void onInit() throws Exception {
+		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());
+		String queueName = "si." + this.channelName;
+		Queue queue = new Queue(queueName);
+		admin.declareQueue(queue);
+		rabbitTemplate.setRoutingKey(queueName);
+		rabbitTemplate.setQueue(queueName);
+	}
+
+	public Message<?> receive() {
+		if (!this.getInterceptors().preReceive(this)) {
+ 			return null;
+ 		}
+		Object object = this.getAmqpTemplate().receiveAndConvert();
+		if (object == null) {
+			return null;
+		}
+		Message<?> replyMessage = null;
+		if (object instanceof Message<?>) {
+			replyMessage = (Message<?>) object;
+		}
+		else {
+			replyMessage = MessageBuilder.withPayload(object).build();
+		}
+		return this.getInterceptors().postReceive(replyMessage, this) ;
+	}
+
+	public Message<?> receive(long timeout) {
+		if (logger.isInfoEnabled()) {
+			logger.info("Calling receive with a timeout value on PollableAmqpChannel. " +
+					"The timeout will be ignored since no receive timeout is supported.");
+		}
+		return this.receive();
+	}
+
+}

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

@@ -0,0 +1,196 @@
+/*
+ * 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.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+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.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;
+
+/**
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public class SubscribableAmqpChannel 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) {
+		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;
+	}
+
+
+	public boolean subscribe(MessageHandler handler) {
+		return this.dispatcher.addHandler(handler);
+	}
+
+	public boolean unsubscribe(MessageHandler handler) {
+		return this.dispatcher.removeHandler(handler);
+	}
+
+	@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.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;
+		}
+	}
+
+
+	private static class DispatchingMessageListener implements MessageListener {
+
+		private final Log logger = LogFactory.getLog(this.getClass());
+
+		private final RabbitTemplate rabbitTemplate;
+
+		private final MessageDispatcher dispatcher;
+
+
+		private DispatchingMessageListener(RabbitTemplate rabbitTemplate, MessageDispatcher dispatcher) {
+			this.rabbitTemplate = rabbitTemplate;
+			this.dispatcher = dispatcher;
+		}
+
+
+		public void onMessage(org.springframework.amqp.core.Message message) {
+			try {
+				Object converted = this.rabbitTemplate.getMessageConverter().fromMessage(message);
+				if (converted != null) {
+					Message<?> messageToSend = (converted instanceof Message<?>) ? (Message<?>) converted
+							: MessageBuilder.withPayload(converted).build();
+					this.dispatcher.dispatch(messageToSend);
+				}
+				else if (this.logger.isWarnEnabled()) {
+					logger.warn("MessageConverter returned null, no Message to dispatch");
+				}
+			}
+			catch (Exception e) {
+				throw new MessagingException("Failure occured in AMQP listener while attempting to convert and dispatch Message.", e);
+			}
+		}
+	}
+
+
+	/*
+	 * SmartLifecycle implementation (delegates to the MessageListener container)
+	 */
+
+	public boolean isAutoStartup() {
+		return (this.container != null) ? this.container.isAutoStartup() : false;
+	}
+
+	public int getPhase() {
+		return (this.container != null) ? this.container.getPhase() : 0;
+	}
+
+	public boolean isRunning() {
+		return (this.container != null) ? this.container.isRunning() : false;
+	}
+
+	public void start() {
+		if (this.container != null) {
+			this.container.start();
+		}
+	}
+
+	public void stop() {
+		if (this.container != null) {
+			this.container.stop();
+		}
+	}
+
+	public void stop(Runnable callback) {
+		if (this.container != null) {
+			this.container.stop(callback);
+		}
+	}
+
+	public void destroy() throws Exception {
+		if (this.container != null) {
+			this.container.destroy();
+		}
+	}
+
+}

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

@@ -0,0 +1,350 @@
+/*
+ * 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.config;
+
+import java.util.List;
+import java.util.concurrent.Executor;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+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.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.PollableAmqpChannel;
+import org.springframework.integration.amqp.channel.SubscribableAmqpChannel;
+import org.springframework.integration.channel.ChannelInterceptor;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.TransactionAttribute;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.ErrorHandler;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * 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
+ * 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
+ * bound to that exchange.
+ * 
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChannel> implements SmartLifecycle, DisposableBean, BeanNameAware {
+
+	private volatile AbstractAmqpChannel channel;
+
+	private volatile List<ChannelInterceptor> interceptors;
+
+	private final boolean messageDriven;
+
+	private final RabbitTemplate rabbitTemplate = new RabbitTemplate();
+
+	private volatile SimpleMessageListenerContainer container;
+
+	private volatile boolean autoStartup = true;
+
+	private volatile Advice[] adviceChain;
+
+	private volatile Integer concurrentConsumers;
+
+	private volatile ConnectionFactory connectionFactory;
+
+	private volatile MessagePropertiesConverter messagePropertiesConverter;
+
+	private volatile ErrorHandler errorHandler;
+
+	private volatile Boolean exposeListenerChannel;
+
+	private volatile Integer phase;
+
+	private volatile Integer prefetchCount;
+
+	private volatile Boolean isPubSub;
+
+	private volatile Long receiveTimeout;
+
+	private volatile Long recoveryInterval;
+
+	private volatile Long shutdownTimeout;
+
+	private volatile String beanName;
+
+	private volatile AcknowledgeMode acknowledgeMode;
+
+	/**
+	 * This value differs from the container implementations' default (which is false).
+	 */
+	private volatile boolean channelTransacted = true;
+
+	private volatile Executor taskExecutor;
+
+	private volatile PlatformTransactionManager transactionManager;
+
+	private volatile TransactionAttribute transactionAttribute;
+
+	private volatile Integer txSize;
+
+
+	public AmqpChannelFactoryBean() {
+		this(true);
+	}
+
+	public AmqpChannelFactoryBean(boolean messageDriven) {
+		this.messageDriven = messageDriven;
+	}
+
+
+	public void setBeanName(String name) {
+		this.beanName = name;
+	}
+
+	public void setInterceptors(List<ChannelInterceptor> interceptors) {
+		this.interceptors = interceptors;
+	}
+
+	/*
+	 * Template-only properties
+	 */
+
+	public void setEncoding(String encoding) {
+		this.rabbitTemplate.setEncoding(encoding);
+	}
+
+	public void setMessageConverter(MessageConverter messageConverter) {
+		this.rabbitTemplate.setMessageConverter(messageConverter);
+	}
+
+	/*
+	 * Template and Container properties
+	 */
+
+	public void setChannelTransacted(boolean channelTransacted) {
+		this.channelTransacted = channelTransacted;
+		this.rabbitTemplate.setChannelTransacted(channelTransacted);
+	}
+
+	public void setConnectionFactory(ConnectionFactory connectionFactory) {
+		this.connectionFactory = connectionFactory;
+		this.rabbitTemplate.setConnectionFactory(this.connectionFactory);
+	}
+
+	public void setMessagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) {
+		this.rabbitTemplate.setMessagePropertiesConverter(messagePropertiesConverter);
+		this.messagePropertiesConverter = messagePropertiesConverter;
+	}
+
+	/*
+	 * Container-only properties
+	 */
+
+	public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) {
+		this.acknowledgeMode = acknowledgeMode;
+	}
+
+	public void setAdviceChain(Advice[] adviceChain) {
+		this.adviceChain = adviceChain;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	public void setConcurrentConsumers(int concurrentConsumers) {
+		this.concurrentConsumers = concurrentConsumers;
+	}
+
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
+	public void setExposeListenerChannel(boolean exposeListenerChannel) {
+		this.exposeListenerChannel = exposeListenerChannel;
+	}
+
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	public void setPrefetchCount(int prefetchCount) {
+		this.prefetchCount = prefetchCount;
+	}
+
+	public void setPubSub(boolean pubSub) {
+		this.isPubSub = pubSub;
+	}
+
+	public void setReceiveTimeout(long receiveTimeout) {
+		this.receiveTimeout = receiveTimeout;
+	}
+
+	public void setRecoveryInterval(long recoveryInterval) {
+		this.recoveryInterval = recoveryInterval;
+	}
+
+	public void setShutdownTimeout(long shutdownTimeout) {
+		this.shutdownTimeout = shutdownTimeout;
+	}
+
+	public void setTaskExecutor(Executor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+	}
+
+	public void setTransactionAttribute(TransactionAttribute transactionAttribute) {
+		this.transactionAttribute = transactionAttribute;
+	}
+
+	public void setTransactionManager(PlatformTransactionManager transactionManager) {
+		this.transactionManager = transactionManager;
+	}
+
+	public void setTxSize(int txSize) {
+		this.txSize = this.txSize;
+	}
+
+	@Override
+	public Class<?> getObjectType() {
+		return (this.channel != null) ? this.channel.getClass() : AbstractAmqpChannel.class;
+	}
+
+	@Override
+	protected AbstractAmqpChannel createInstance() throws Exception {
+		if (this.messageDriven) {
+			this.container = this.createContainer();
+			this.channel = new SubscribableAmqpChannel(this.beanName, this.container, this.rabbitTemplate, this.isPubSub);
+		}
+		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);
+		}
+		if (!CollectionUtils.isEmpty(this.interceptors)) {
+			this.channel.setInterceptors(this.interceptors);
+		}
+		this.channel.afterPropertiesSet();
+		this.channel.setBeanName(this.beanName);
+		return this.channel;
+	}
+
+	private SimpleMessageListenerContainer createContainer() throws Exception {
+		//if (!messageDriven) TODO: no container attributes would apply if not message-driven
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
+		if (this.acknowledgeMode != null) {
+			container.setAcknowledgeMode(this.acknowledgeMode);
+		}
+		if (!ObjectUtils.isEmpty(this.adviceChain)) {
+			container.setAdviceChain(this.adviceChain);
+		}
+		container.setAutoStartup(this.autoStartup);
+		container.setChannelTransacted(this.channelTransacted);
+		if (this.concurrentConsumers != null) {
+			container.setConcurrentConsumers(this.concurrentConsumers);
+		}
+		container.setConnectionFactory(this.connectionFactory);
+		if (this.errorHandler != null) {
+			container.setErrorHandler(this.errorHandler);
+		}
+		if (this.exposeListenerChannel != null) {
+			container.setExposeListenerChannel(this.exposeListenerChannel);
+		}
+		if (this.messagePropertiesConverter != null) {
+			container.setMessagePropertiesConverter(this.messagePropertiesConverter);
+		}
+		if (this.phase != null) {
+			container.setPhase(this.phase);
+		}
+		if (this.prefetchCount != null) {
+			container.setPrefetchCount(this.prefetchCount);
+		}
+		if (this.receiveTimeout != null) {
+			container.setReceiveTimeout(this.receiveTimeout);
+		}
+		if (this.recoveryInterval != null) {
+			container.setRecoveryInterval(this.recoveryInterval);
+		}
+		if (this.shutdownTimeout != null) {
+			container.setShutdownTimeout(this.shutdownTimeout);
+		}
+		if (this.taskExecutor != null) {
+			container.setTaskExecutor(this.taskExecutor);
+		}
+		if (this.transactionAttribute != null) {
+			container.setTransactionAttribute(this.transactionAttribute);
+		}
+		if (this.transactionManager != null) {
+			container.setTransactionManager(this.transactionManager);
+		}
+		if (this.txSize != null) {
+			container.setTxSize(this.txSize);
+		}
+		return container;
+	}
+
+	/*
+	 * SmartLifecycle implementation (delegates to the created channel if message-driven)
+	 */
+
+	public boolean isAutoStartup() {
+		return (this.channel instanceof SmartLifecycle) ?
+				((SmartLifecycle) this.channel).isAutoStartup() : false;
+	}
+
+	public int getPhase() {
+		return (this.channel instanceof SmartLifecycle) ?
+				((SmartLifecycle) this.channel).getPhase() : 0;
+	}
+
+	public boolean isRunning() {
+		return (this.channel instanceof Lifecycle) ?
+				((Lifecycle) this.channel).isRunning() : false;
+	}
+
+	public void start() {
+		if (this.channel instanceof Lifecycle) {
+			((Lifecycle) this.channel).start();
+		}
+	}
+
+	public void stop() {
+		if (this.channel instanceof Lifecycle) {
+			((Lifecycle) this.channel).stop();
+		}
+	}
+
+	public void stop(Runnable callback) {
+		if (this.channel instanceof SmartLifecycle) {
+			((SmartLifecycle) this.channel).stop(callback);
+		}
+	}
+
+	protected void destroyInstance(AbstractAmqpChannel instance) throws Exception {
+		if (instance instanceof DisposableBean) {
+			((DisposableBean) this.channel).destroy();
+		}
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.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.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractChannelParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the 'channel' and 'publish-subscribe-channel' elements of the
+ * Spring Integration AMQP namespace.
+ * 
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public class AmqpChannelParser extends AbstractChannelParser {
+
+	@Override
+	protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(AmqpChannelFactoryBean.class);
+		String messageDriven = element.getAttribute("message-driven");
+		if (StringUtils.hasText(messageDriven)) {
+			builder.addConstructorArgValue(messageDriven);
+		}
+		String connectionFactory = element.getAttribute("connection-factory");
+		if (!StringUtils.hasText(connectionFactory)) {
+			connectionFactory = "rabbitConnectionFactory";
+		}
+		builder.addPropertyReference("connectionFactory", connectionFactory);
+		if ("channel".equals(element.getLocalName())) {
+			builder.addPropertyValue("pubSub", false);
+		}
+		else if ("publish-subscribe-channel".equals(element.getLocalName())) {
+			builder.addPropertyValue("pubSub", true);
+		}
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge-mode");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "advice-chain");
+		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.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, "receive-timeout");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "shutdown-timeout");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-attribute");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-manager");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "tx-size");
+		return builder;
+	}
+
+}

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

@@ -27,6 +27,8 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
 public class AmqpNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 
 	public void init() {
+		this.registerBeanDefinitionParser("channel", new AmqpChannelParser());
+		this.registerBeanDefinitionParser("publish-subscribe-channel", new AmqpChannelParser());
 		this.registerBeanDefinitionParser("inbound-channel-adapter", new AmqpInboundChannelAdapterParser());
 		this.registerBeanDefinitionParser("inbound-gateway", new AmqpInboundGatewayParser());
 		this.registerBeanDefinitionParser("outbound-channel-adapter", new AmqpOutboundChannelAdapterParser());

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

@@ -242,6 +242,60 @@
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="channel">
+		<xsd:annotation>
+			<xsd:documentation>
+	Creates a point-to-point channel that is backed by an AMQP Queue.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="channelType">
+					<xsd:attribute name="message-driven" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	Indicate whether this channel should be message-driven (subscribable) or not (pollable).
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="publish-subscribe-channel" type="channelType">
+		<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:element>
+
+	<xsd:complexType name="channelType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Base type for 'channel' and 'publish-subscribe-channel'.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:attribute name="id" type="xsd:string" use="required">
+			<xsd:annotation>
+				<xsd:documentation>
+	Unique ID for this Message Channel.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="auto-startup" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+	Flag to indicate whether this Message Channel should start automatically.
+	This only applies to a message-driven channel. Default is true.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attributeGroup ref="containerAndTemplateAttributes"/>
+	</xsd:complexType>
+
 	<xsd:complexType name="inboundType">
 		<xsd:annotation>
 			<xsd:documentation>
@@ -255,58 +309,7 @@
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="acknowledge-mode">
-			<xsd:annotation>
-				<xsd:documentation>
-	Acknowledge Mode for the MessageListenerContainer.
-				</xsd:documentation>
-			</xsd:annotation>
-			<xsd:simpleType>
-				<xsd:restriction base="xsd:string">
-					<xsd:enumeration value="AUTO"/>
-					<xsd:enumeration value="MANUAL"/>
-					<xsd:enumeration value="NONE"/>
-				</xsd:restriction>
-			</xsd:simpleType>
-		</xsd:attribute>
-		<xsd:attribute name="advice-chain" type="xsd:string">
-			<xsd:annotation>
-				<xsd:appinfo>
-					<tool:annotation kind="ref">
-						<tool:expected-type type="java.util.Collection" />
-					</tool:annotation>
-					<xsd:documentation>
-	Array of AOP Advice instances to be applied to the MessageListener.
-					</xsd:documentation>
-				</xsd:appinfo>
-			</xsd:annotation>
-		</xsd:attribute>
-		<xsd:attribute name="channel-transacted" type="xsd:string">
-			<xsd:annotation>
-				<xsd:documentation>
-	Flag to indicate that AMQP channels created by this component will be transactional.
-				</xsd:documentation>
-			</xsd:annotation>
-		</xsd:attribute>
-		<xsd:attribute name="concurrent-consumers" type="xsd:string">
-			<xsd:annotation>
-				<xsd:documentation>
-	Specify the number of concurrent consumers to create. Default is 1.
-	Raising the number of concurrent consumers is recommended in order to scale the consumption of messages coming in
-	from a queue. However, note that any ordering guarantees are lost once multiple consumers are registered. In
-	general, stick with 1 consumer for low-volume queues.
-				</xsd:documentation>
-			</xsd:annotation>
-		</xsd:attribute>
-		<xsd:attribute name="connection-factory" type="xsd:string">
-			<xsd:annotation>
-				<xsd:appinfo>
-					<tool:annotation kind="ref">
-						<tool:expected-type type="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>
-					</tool:annotation>
-				</xsd:appinfo>
-			</xsd:annotation>
-		</xsd:attribute>
+		<xsd:attributeGroup ref="containerAttributes"/>
 		<xsd:attribute name="error-channel" use="optional" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation>
@@ -319,28 +322,18 @@
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="error-handler" use="optional" type="xsd:string">
+		<xsd:attribute name="message-converter" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation>
-	ErrorHandler to be configured on the underlying MessageListener container.
+	MessageConverter to use when receiving AMQP Messages.
 				</xsd:documentation>
 				<xsd:appinfo>
 					<tool:annotation kind="ref">
-						<tool:expected-type type="org.springframework.util.ErrorHandler"/>
+						<tool:expected-type type="org.springframework.amqp.support.converter.MessageConverter"/>
 					</tool:annotation>
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="expose-listener-channel" type="xsd:string">
-			<xsd:annotation>
-				<xsd:appinfo>
-					<xsd:documentation>
-	Set whether to expose the listener Rabbit Channel to a registered ChannelAwareMessageListener as well as
-	to RabbitTemplate calls.
-					</xsd:documentation>
-				</xsd:appinfo>
-			</xsd:annotation>
-		</xsd:attribute>
 		<xsd:attribute name="header-mapper" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation>
@@ -369,14 +362,47 @@
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="message-converter" type="xsd:string">
+		<xsd:attribute name="queue-names" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation>
-	MessageConverter to use when receiving AMQP Messages.
+	Names of the AMQP Queues from which Messages should be consumed (comma-separated list).
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:attributeGroup name="containerAttributes">
+		<xsd:attributeGroup ref="containerOnlyAttributes"/>
+		<xsd:attributeGroup ref="containerAndTemplateSharedAttributes"/>
+	</xsd:attributeGroup>
+
+	<xsd:attributeGroup name="templateAttributes">
+		<xsd:attributeGroup ref="templateOnlyAttributes"/>
+		<xsd:attributeGroup ref="containerAndTemplateSharedAttributes"/>
+	</xsd:attributeGroup>
+
+	<xsd:attributeGroup name="containerAndTemplateAttributes">
+		<xsd:attributeGroup ref="containerOnlyAttributes"/>
+		<xsd:attributeGroup ref="templateOnlyAttributes"/>
+		<xsd:attributeGroup ref="containerAndTemplateSharedAttributes"/>
+	</xsd:attributeGroup>
+
+	<xsd:attributeGroup name="containerAndTemplateSharedAttributes">
+		<xsd:attribute name="channel-transacted" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+	Flag to indicate that channels created by this component will be transactional.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="connection-factory" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+	Reference to the Rabbit ConnectionFactory to be used by this component.	
 				</xsd:documentation>
 				<xsd:appinfo>
 					<tool:annotation kind="ref">
-						<tool:expected-type type="org.springframework.amqp.support.converter.MessageConverter"/>
+						<tool:expected-type type="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>
 					</tool:annotation>
 				</xsd:appinfo>
 			</xsd:annotation>
@@ -393,6 +419,105 @@
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
+	</xsd:attributeGroup>
+
+	<xsd:attributeGroup name="templateOnlyAttributes">
+		<xsd:annotation>
+			<xsd:documentation>
+	Attributes for a RabbitTemplate. This does not include the exchange, queue, or routingKey properties
+	since those may or may not be exposed for configuration depending on what type of component uses this
+	attribute group. This group also does not include any of the properties that are shared with the
+	SimpleMessageListenerContainer, such as channelTransacted, connectionFactory, and messagePropertiesConverter.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:attribute name="encoding" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+	The encoding to use when converting between byte arrays and Strings in message properties.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="message-converter" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+	Reference to a MessageConverter to be used by this RabbitTemplate.
+				</xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.amqp.support.converter.MessageConverter"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
+
+	<xsd:attributeGroup name="containerOnlyAttributes">
+		<xsd:annotation>
+			<xsd:documentation>
+	Attributes for a SimpleMesssageListenerContainer's properties other than queues, queueNames, messageListener, and 
+	autoStartup which may or may not be exposed for configuration depending on what type of component uses this attribute group.
+	This group also does not include any of the properties that are shared with RabbitTemplate, such as channelTransacted,
+	connectionFactory, and messsagePropertiesConverter.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:attribute name="acknowledge-mode">
+			<xsd:annotation>
+				<xsd:documentation>
+	Acknowledge Mode for the MessageListenerContainer.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:restriction base="xsd:string">
+					<xsd:enumeration value="AUTO"/>
+					<xsd:enumeration value="MANUAL"/>
+					<xsd:enumeration value="NONE"/>
+				</xsd:restriction>
+			</xsd:simpleType>
+		</xsd:attribute>
+		<xsd:attribute name="advice-chain" type="xsd:string">
+			<xsd:annotation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="java.util.Collection" />
+					</tool:annotation>
+					<xsd:documentation>
+	Array of AOP Advice instances to be applied to the MessageListener.
+					</xsd:documentation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="concurrent-consumers" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+	Specify the number of concurrent consumers to create. Default is 1.
+	Raising the number of concurrent consumers is recommended in order to scale the consumption of messages coming in
+	from a queue. However, note that any ordering guarantees are lost once multiple consumers are registered. In
+	general, stick with 1 consumer for low-volume queues.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="error-handler" use="optional" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+	ErrorHandler to be configured on the underlying MessageListener container.
+				</xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.util.ErrorHandler"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="expose-listener-channel" type="xsd:string">
+			<xsd:annotation>
+				<xsd:appinfo>
+					<xsd:documentation>
+	Set whether to expose the listener Rabbit Channel to a registered ChannelAwareMessageListener as well as
+	to RabbitTemplate calls.
+					</xsd:documentation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="phase" type="xsd:string">
 			<xsd:annotation>
 				<xsd:appinfo>
@@ -412,13 +537,6 @@
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="queue-names" type="xsd:string">
-			<xsd:annotation>
-				<xsd:documentation>
-	Names of the AMQP Queues from which Messages should be consumed (comma-separated list).
-				</xsd:documentation>
-			</xsd:annotation>
-		</xsd:attribute>
 		<xsd:attribute name="receive-timeout" type="xsd:string">
 			<xsd:annotation>
 				<xsd:appinfo>
@@ -446,16 +564,6 @@
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>	
-		<xsd:attribute name="tx-size" type="xsd:string">
-			<xsd:annotation>
-				<xsd:appinfo>
-					<xsd:documentation>
-	How many messages to process in a single transaction (if the channel is transactional). For best results it should be
-	less than or equal to the prefetch count.
-					</xsd:documentation>
-				</xsd:appinfo>
-			</xsd:annotation>
-		</xsd:attribute>
 		<xsd:attribute name="task-executor" type="xsd:string">
 			<xsd:annotation>
 				<xsd:appinfo>
@@ -493,6 +601,16 @@
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
-	</xsd:complexType>
+		<xsd:attribute name="tx-size" type="xsd:string">
+			<xsd:annotation>
+				<xsd:appinfo>
+					<xsd:documentation>
+	How many messages to process in a single transaction (if the channel is transactional). For best results it should be
+	less than or equal to the prefetch count.
+					</xsd:documentation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
 
 </xsd:schema>

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

@@ -0,0 +1,28 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:amqp="http://www.springframework.org/schema/integration/amqp"
+	xmlns:console="http://www.springframework.org/schema/integration/stream"
+	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-2.1.xsd
+		http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
+		http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<console:stdin-channel-adapter channel="rabbitChannel">
+		<poller fixed-delay="1000" max-messages-per-poll="1" />
+	</console:stdin-channel-adapter>
+
+	<amqp:channel id="rabbitChannel" />
+
+	<logging-channel-adapter channel="rabbitChannel" expression="'1:' + payload" logger-name="test.p2p.1"/>
+
+	<logging-channel-adapter channel="rabbitChannel" expression="'2:' + payload" logger-name="test.p2p.2"/>
+
+	<logging-channel-adapter channel="rabbitChannel" expression="'3:' + payload" logger-name="test.p2p.3"/>
+
+	<rabbit:connection-factory id="rabbitConnectionFactory" />
+
+</beans:beans>

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample.java

@@ -0,0 +1,31 @@
+/*
+ * 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.config;
+
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+/**
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public class ChannelSample {
+
+	public static void main(String[] args) {
+		new ClassPathXmlApplicationContext("ChannelSample-context.xml", ChannelSample.class);
+	}
+
+}

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

@@ -0,0 +1,28 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:amqp="http://www.springframework.org/schema/integration/amqp"
+	xmlns:console="http://www.springframework.org/schema/integration/stream"
+	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-2.1.xsd
+		http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
+		http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<console:stdin-channel-adapter channel="rabbitChannel">
+		<poller fixed-delay="1000" max-messages-per-poll="1" />
+	</console:stdin-channel-adapter>
+
+	<amqp:publish-subscribe-channel id="rabbitChannel" />
+
+	<logging-channel-adapter channel="rabbitChannel" expression="'1:' + payload" logger-name="test.pubsub.1"/>
+
+	<logging-channel-adapter channel="rabbitChannel" expression="'2:' + payload" logger-name="test.pubsub.2"/>
+
+	<logging-channel-adapter channel="rabbitChannel" expression="'3:' + payload" logger-name="test.pubsub.3"/>
+
+	<rabbit:connection-factory id="rabbitConnectionFactory" />
+
+</beans:beans>

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample.java

@@ -0,0 +1,31 @@
+/*
+ * 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.config;
+
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+/**
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public class PubSubChannelSample {
+
+	public static void main(String[] args) {
+		new ClassPathXmlApplicationContext("PubSubChannelSample-context.xml", PubSubChannelSample.class);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2009 the original author or authors.
+ * 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.
@@ -25,6 +25,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.integration.core.MessageHandler;
+import org.springframework.util.Assert;
 
 /**
  * Base class for {@link MessageDispatcher} implementations.
@@ -61,6 +62,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
 	 * @return the result of {@link Set#add(Object)}
 	 */
 	public boolean addHandler(MessageHandler handler) {
+		Assert.notNull(handler, "handler must not be null");
 		return this.handlers.add(handler);
 	}
 
@@ -70,6 +72,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
 	 * @return the result of {@link Set#remove(Object)}
 	 */
 	public boolean removeHandler(MessageHandler handler) {
+		Assert.notNull(handler, "handler must not be null");
 		return this.handlers.remove(handler);
 	}