INT-4136: Support DirectMessageListenerContainer

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

Fix a couple typos

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

@@ -27,7 +27,7 @@ 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.SimpleMessageListenerContainer;
+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;
@@ -57,7 +57,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 
 	private final String channelName;
 
-	private final SimpleMessageListenerContainer container;
+	private final AbstractMessageListenerContainer container;
 
 	private volatile AbstractDispatcher dispatcher;
 
@@ -77,7 +77,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 	 * @param amqpTemplate the template.
 	 * @see #setExtractPayload(boolean)
 	 */
-	protected AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container,
+	protected AbstractSubscribableAmqpChannel(String channelName, AbstractMessageListenerContainer container,
 			AmqpTemplate amqpTemplate) {
 		this(channelName, container, amqpTemplate, false);
 	}
@@ -93,7 +93,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 	 * @see #setExtractPayload(boolean)
 	 * @since 4.3
 	 */
-	protected AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container,
+	protected AbstractSubscribableAmqpChannel(String channelName, AbstractMessageListenerContainer container,
 			AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) {
 		this(channelName, container, amqpTemplate, false, outboundMapper, inboundMapper);
 	}
@@ -108,7 +108,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 	 * @see #setExtractPayload(boolean)
 	 */
 	protected AbstractSubscribableAmqpChannel(String channelName,
-			SimpleMessageListenerContainer container,
+			AbstractMessageListenerContainer container,
 			AmqpTemplate amqpTemplate, boolean isPubSub) {
 		super(amqpTemplate);
 		Assert.notNull(container, "container must not be null");
@@ -133,7 +133,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 	 * @since 4.3
 	 */
 	protected AbstractSubscribableAmqpChannel(String channelName,
-			SimpleMessageListenerContainer container,
+			AbstractMessageListenerContainer container,
 			AmqpTemplate amqpTemplate, boolean isPubSub,
 			AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) {
 		super(amqpTemplate, outboundMapper, inboundMapper);
@@ -312,7 +312,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 				}
 			}
 			catch (Exception e) {
-				throw new MessagingException("Failure occured in AMQP listener " +
+				throw new MessagingException("Failure occurred in AMQP listener " +
 						"while attempting to convert and dispatch Message.", e);
 			}
 		}

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

@@ -19,7 +19,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.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.dispatcher.AbstractDispatcher;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
@@ -43,7 +43,7 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq
 	 * @param amqpTemplate the template.
 	 * @see #setExtractPayload(boolean)
 	 */
-	public PointToPointSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container,
+	public PointToPointSubscribableAmqpChannel(String channelName, AbstractMessageListenerContainer container,
 			AmqpTemplate amqpTemplate) {
 		super(channelName, container, amqpTemplate);
 	}
@@ -60,7 +60,7 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq
 	 * @see #setExtractPayload(boolean)
 	 * @since 4.3
 	 */
-	public PointToPointSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container,
+	public PointToPointSubscribableAmqpChannel(String channelName, AbstractMessageListenerContainer container,
 			AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) {
 		super(channelName, container, amqpTemplate, outboundMapper, inboundMapper);
 	}

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

@@ -27,7 +27,7 @@ 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.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.dispatcher.AbstractDispatcher;
 import org.springframework.integration.dispatcher.BroadcastingDispatcher;
@@ -55,7 +55,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
 	 * @param amqpTemplate the template.
 	 * @see #setExtractPayload(boolean)
 	 */
-	public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container,
+	public PublishSubscribeAmqpChannel(String channelName, AbstractMessageListenerContainer container,
 			AmqpTemplate amqpTemplate) {
 		super(channelName, container, amqpTemplate, true);
 	}
@@ -71,7 +71,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
 	 * @see #setExtractPayload(boolean)
 	 * @since 4.3
 	 */
-	public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container,
+	public PublishSubscribeAmqpChannel(String channelName, AbstractMessageListenerContainer container,
 			AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) {
 		super(channelName, container, amqpTemplate, true, outboundMapper, inboundMapper);
 	}

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

@@ -22,6 +22,8 @@ import java.util.List;
 
 import org.w3c.dom.Element;
 
+import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
@@ -46,6 +48,7 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
 		"acknowledge-mode",
 		"channel-transacted",
 		"concurrent-consumers",
+		"consumers-per-queue",
 		"expose-listener-channel",
 		"phase",
 		"prefetch-count",
@@ -124,8 +127,26 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
 			parserContext.getReaderContext().error("If no 'listener-container' reference is provided, " +
 					"the 'queue-names' attribute is required.", element);
 		}
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
-				"org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer");
+		String consumersPerQueue = element.getAttribute("consumers-per-queue");
+		BeanDefinitionBuilder builder;
+		if (StringUtils.hasText(consumersPerQueue)) {
+			builder = BeanDefinitionBuilder.genericBeanDefinition(DirectMessageListenerContainer.class);
+			if (StringUtils.hasText(element.getAttribute("concurrent-consumers"))) {
+				parserContext.getReaderContext().error("'consumers-per-queue' and 'concurrent-consumers' are mutually "
+						+ "exclusive", element);
+			}
+			if (StringUtils.hasText(element.getAttribute("tx-size"))) {
+				parserContext.getReaderContext().error("'tx-size' is not allowed with 'consumers-per-queue'", element);
+			}
+			if (StringUtils.hasText(element.getAttribute("receive-timeout"))) {
+				parserContext.getReaderContext().error("'receive-timeout' is not allowed with 'consumers-per-queue'",
+						element);
+			}
+			builder.addPropertyValue("consumersPerQueue", consumersPerQueue);
+		}
+		else {
+			builder = BeanDefinitionBuilder.genericBeanDefinition(SimpleMessageListenerContainer.class);
+		}
 		String connectionFactoryRef = element.getAttribute("connection-factory");
 		if (!StringUtils.hasText(connectionFactoryRef)) {
 			connectionFactoryRef = "rabbitConnectionFactory";

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

@@ -29,6 +29,8 @@ import org.springframework.amqp.core.FanoutExchange;
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.amqp.support.converter.MessageConverter;
@@ -89,6 +91,8 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	private volatile Integer concurrentConsumers;
 
+	private volatile Integer consumersPerQueue;
+
 	private volatile ConnectionFactory connectionFactory;
 
 	private volatile MessagePropertiesConverter messagePropertiesConverter;
@@ -261,6 +265,10 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		this.concurrentConsumers = concurrentConsumers;
 	}
 
+	public void setConsumersPerQueue(Integer consumersPerQueue) {
+		this.consumersPerQueue = consumersPerQueue;
+	}
+
 	public void setErrorHandler(ErrorHandler errorHandler) {
 		this.errorHandler = errorHandler;
 	}
@@ -341,7 +349,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 	@Override
 	protected AbstractAmqpChannel createInstance() throws Exception {
 		if (this.messageDriven) {
-			SimpleMessageListenerContainer container = this.createContainer();
+			AbstractMessageListenerContainer container = this.createContainer();
 			if (this.amqpTemplate instanceof InitializingBean) {
 				((InitializingBean) this.amqpTemplate).afterPropertiesSet();
 			}
@@ -397,8 +405,26 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		return this.channel;
 	}
 
-	private SimpleMessageListenerContainer createContainer() throws Exception {
-		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
+	private AbstractMessageListenerContainer createContainer() throws Exception {
+		AbstractMessageListenerContainer container;
+		if (this.consumersPerQueue == null) {
+			SimpleMessageListenerContainer smlc = new SimpleMessageListenerContainer();
+			if (this.concurrentConsumers != null) {
+				smlc.setConcurrentConsumers(this.concurrentConsumers);
+			}
+			if (this.receiveTimeout != null) {
+				smlc.setReceiveTimeout(this.receiveTimeout);
+			}
+			if (this.txSize != null) {
+				smlc.setTxSize(this.txSize);
+			}
+			container = smlc;
+		}
+		else {
+			DirectMessageListenerContainer dmlc = new DirectMessageListenerContainer();
+			dmlc.setConsumersPerQueue(this.consumersPerQueue);
+			container = dmlc;
+		}
 		if (this.acknowledgeMode != null) {
 			container.setAcknowledgeMode(this.acknowledgeMode);
 		}
@@ -407,9 +433,6 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		}
 		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);
@@ -426,9 +449,6 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		if (this.prefetchCount != null) {
 			container.setPrefetchCount(this.prefetchCount);
 		}
-		if (this.receiveTimeout != null) {
-			container.setReceiveTimeout(this.receiveTimeout);
-		}
 		if (this.recoveryInterval != null) {
 			container.setRecoveryInterval(this.recoveryInterval);
 		}
@@ -444,9 +464,6 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		if (this.transactionManager != null) {
 			container.setTransactionManager(this.transactionManager);
 		}
-		if (this.txSize != null) {
-			container.setTxSize(this.txSize);
-		}
 		if (this.missingQueuesFatal != null) {
 			container.setMissingQueuesFatal(this.missingQueuesFatal);
 		}

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

@@ -50,6 +50,21 @@ public class AmqpChannelParser extends AbstractChannelParser {
 
 		builder.addPropertyValue("pubSub", "publish-subscribe-channel".equals(element.getLocalName()));
 
+		String consumersPerQueue = element.getAttribute("consumers-per-queue");
+		if (StringUtils.hasText(consumersPerQueue)) {
+			if (StringUtils.hasText(element.getAttribute("concurrent-consumers"))) {
+				parserContext.getReaderContext().error("'consumers-per-queue' and 'concurrent-consumers' are mutually "
+						+ "exclusive", element);
+			}
+			if (StringUtils.hasText(element.getAttribute("tx-size"))) {
+				parserContext.getReaderContext().error("'tx-size' is not allowed with 'consumers-per-queue'", element);
+			}
+			if (StringUtils.hasText(element.getAttribute("receive-timeout"))) {
+				parserContext.getReaderContext().error("'receive-timeout' is not allowed with 'consumers-per-queue'",
+						element);
+			}
+			builder.addPropertyValue("consumersPerQueue", consumersPerQueue);
+		}
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-subscribers");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge-mode");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "advice-chain");

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

@@ -874,7 +874,17 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
 	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.
+	general, use 1 consumer for low-volume queues. Mutually exclusive with 'consumers-per-queue'.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="consumers-per-queue" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+	Specify the number of consumers to create for each queue.
+	Setting this attribute creates a 'DirectMessageListenerContainer' instead of the default
+	'SimpleMessageListenerContainer'. Refer to the Spring AMQP reference documentation for
+	more information about these containers.
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
@@ -915,6 +925,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
 				<xsd:appinfo>
 					<xsd:documentation>
 	The timeout for each attempt by a consumer to receive the next message.
+	Not allowed when 'consumers-per-queue' is set.
 					</xsd:documentation>
 				</xsd:appinfo>
 			</xsd:annotation>
@@ -980,6 +991,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
 					<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.
+	Not allowed when 'consumers-per-queue' is set.
 					</xsd:documentation>
 				</xsd:appinfo>
 			</xsd:annotation>
@@ -991,7 +1003,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
 	If 'true', and none of the queues are available on the broker, the container will throw a fatal exception during
 	startup and will stop if the queues are deleted when the container is running (after making 3 attempts to
 	passively declare the queues). If false, the container will not throw an exception and go into recovery mode,
-	attempting to restart according to the 'recovery-interval'. Default 'true'.
+	attempting to restart according to the 'recovery-interval'. Default 'true' unless 'consumers-per-queue' is set.
 					</xsd:documentation>
 				</xsd:appinfo>
 			</xsd:annotation>

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

@@ -18,7 +18,7 @@
 	<bean id="rabbitConnectionFactory" class="org.springframework.integration.amqp.StubRabbitConnectionFactory"/>
 
 	<amqp:channel id="channelWithSubscriberLimit" max-subscribers="1" missing-queues-fatal="false"
-			template-channel-transacted="true"/>
+			template-channel-transacted="true" consumers-per-queue="2" />
 
 	<amqp:publish-subscribe-channel id="pubSub" />
 

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.amqp.config;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
@@ -30,6 +31,8 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
@@ -80,6 +83,7 @@ public class AmqpChannelParserTests {
 		assertTrue(TestUtils.getPropertyValue(channel, "container.missingQueuesFatal", Boolean.class));
 		assertFalse(TestUtils.getPropertyValue(channel, "container.transactional", Boolean.class));
 		assertFalse(TestUtils.getPropertyValue(channel, "amqpTemplate.transactional", Boolean.class));
+		assertThat(TestUtils.getPropertyValue(channel, "container"), instanceOf(SimpleMessageListenerContainer.class));
 	}
 
 	@Test
@@ -91,6 +95,8 @@ public class AmqpChannelParserTests {
 		assertFalse(TestUtils.getPropertyValue(channel, "container.transactional", Boolean.class));
 		assertTrue(TestUtils.getPropertyValue(channel, "amqpTemplate.transactional", Boolean.class));
 		assertFalse(TestUtils.getPropertyValue(channel, "extractPayload", Boolean.class));
+		assertThat(TestUtils.getPropertyValue(channel, "container"), instanceOf(DirectMessageListenerContainer.class));
+		assertEquals(2, TestUtils.getPropertyValue(channel, "container.consumersPerQueue"));
 	}
 
 	@Test

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

@@ -43,6 +43,9 @@
 		channel="requestChannel" queue-names="inboundchanneladapter.test.2"
 		auto-startup="false" phase="123"/>
 
+	<amqp:inbound-channel-adapter id="dmlc" queue-names="inboundchanneladapter.test.2" consumers-per-queue="2"
+		auto-startup="false" />
+
 	<int:channel id="requestChannel">
 		<int:queue/>
 	</int:channel>

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

@@ -16,10 +16,12 @@
 
 package org.springframework.integration.amqp.config;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import org.junit.Test;
@@ -30,6 +32,8 @@ import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
@@ -66,6 +70,18 @@ public class AmqpInboundChannelAdapterParserTests {
 		assertEquals(Boolean.TRUE, TestUtils.getPropertyValue(adapter, "autoStartup"));
 		assertEquals(Integer.MAX_VALUE / 2, TestUtils.getPropertyValue(adapter, "phase"));
 		assertTrue(TestUtils.getPropertyValue(adapter, "messageListenerContainer.missingQueuesFatal", Boolean.class));
+		assertThat(TestUtils.getPropertyValue(adapter, "messageListenerContainer"),
+				instanceOf(SimpleMessageListenerContainer.class));
+	}
+
+	@Test
+	public void verifyDMCC() {
+		Object adapter = context.getBean("dmlc.adapter");
+		assertEquals(AmqpInboundChannelAdapter.class, adapter.getClass());
+		assertFalse(TestUtils.getPropertyValue(adapter, "messageListenerContainer.missingQueuesFatal", Boolean.class));
+		assertThat(TestUtils.getPropertyValue(adapter, "messageListenerContainer"),
+				instanceOf(DirectMessageListenerContainer.class));
+		assertEquals(2, TestUtils.getPropertyValue(adapter, "messageListenerContainer.consumersPerQueue"));
 	}
 
 	@Test

src/reference/asciidoc/amqp.adoc

@@ -32,7 +32,7 @@ It provides much more in-depth information regarding Spring's integration with A
 
 A configuration sample for an AMQP Inbound Channel Adapter is shown below.
 
-[source,xml]
+[source, xml]
 ----
 <int-amqp:inbound-channel-adapter
                                   id="inboundAmqp" <1>
@@ -59,8 +59,8 @@ A configuration sample for an AMQP Inbound Channel Adapter is shown below.
                                   task-executor="" <22>
                                   transaction-attribute="" <23>
                                   transaction-manager="" <24>
-                                  tx-size="" /> <25>
-
+                                  tx-size="" <25>
+                                  consumers-per-queue /> <26>
 ----
 
 <1> Unique ID for this adapter.
@@ -92,7 +92,8 @@ _Optional (Defaults to false)_.
 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.
+In general, use 1 consumer for low-volume queues.
+Not allowed when 'consumers-per-queue' is set.
 _Optional_.
 
 
@@ -122,7 +123,7 @@ The values in this list can also be simple patterns to be matched against the he
 "\*" or "foo*, bar" or "*foo").
 
 
-<13> Reference to the `SimpleMessageListenerContainer` to use for receiving AMQP Messages.
+<13> Reference to the `AbstractMessageListenerContainer` to use for receiving AMQP Messages.
 If this attribute is provided, then no other attribute related to the listener container configuration should be provided.
 In other words, by setting this reference, you must take full responsibility of the listener container configuration.
 The only exception is the MessageListener itself.
@@ -138,7 +139,7 @@ _Optional_.
 _Optional_.
 
 
-<16> Specify the phase in which the underlying `SimpleMessageListenerContainer` should be started and stopped.
+<16> Specify the phase in which the underlying `AbstractMessageListenerContainer` should be started and stopped.
 The startup order proceeds from lowest to highest, and the shutdown order is the reverse of that.
 By default this value is Integer.MAX_VALUE meaning that this container starts as late as possible and stops as soon as possible.
 _Optional_.
@@ -154,7 +155,7 @@ _Optional (Defaults to 1)_.
 _Optional (Defaults to 1000)_.
 
 
-<19> Specifies the interval between recovery attempts of the underlying `SimpleMessageListenerContainer` (in milliseconds)
+<19> Specifies the interval between recovery attempts of the underlying `AbstractMessageListenerContainer` (in milliseconds)
 ._Optional (Defaults to 5000)_.
 
 
@@ -163,24 +164,24 @@ If false, the container will not throw an exception and go into recovery mode, a
 _Optional (Defaults to `true`)_.
 
 
-<21> The time to wait for workers in milliseconds after the underlying `SimpleMessageListenerContainer` is stopped, and before the AMQP connection is forced closed.
+<21> The time to wait for workers in milliseconds after the underlying `AbstractMessageListenerContainer` is stopped, and before the AMQP connection is forced closed.
 If any workers are active when the shutdown signal comes they will be allowed to finish processing as long as they can finish within this timeout.
 Otherwise the connection is closed and messages remain unacked (if the channel is transactional).
 _Optional (Defaults to 5000)_.
 
 
-<22> By default, the underlying `SimpleMessageListenerContainer` uses a SimpleAsyncTaskExecutor implementation, that fires up a new Thread for each task, executing it asynchronously.
+<22> By default, the underlying `AbstractMessageListenerContainer` uses a SimpleAsyncTaskExecutor implementation, that fires up a new Thread for each task, executing it asynchronously.
 By default, the number of concurrent threads is unlimited.
 *NOTE:* This implementation does not reuse threads.
 Consider a thread-pooling TaskExecutor implementation as an alternative.
 _Optional (Defaults to SimpleAsyncTaskExecutor)_.
 
 
-<23> By default the underlying `SimpleMessageListenerContainer` creates a new instance of the DefaultTransactionAttribute (takes the EJB approach to rolling back on runtime, but not checked exceptions.
+<23> By default the underlying `AbstractMessageListenerContainer` creates a new instance of the DefaultTransactionAttribute (takes the EJB approach to rolling back on runtime, but not checked exceptions.
 _Optional (Defaults to DefaultTransactionAttribute)_.
 
 
-<24> Sets a Bean reference to an external `PlatformTransactionManager` on the underlying SimpleMessageListenerContainer.
+<24> Sets a Bean reference to an external `PlatformTransactionManager` on the underlying AbstractMessageListenerContainer.
 The transaction manager works in conjunction with the "channel-transacted" attribute.
 If there is already a transaction in progress when the framework is sending or receiving a message, and the channelTransacted flag is true, then the commit or rollback of the messaging transaction will be deferred until the end of the current transaction.
 If the channelTransacted flag is false, then no transaction semantics apply to the messaging operation (it is auto-acked).
@@ -191,8 +192,12 @@ _Optional_.
 
 <25> Tells the `SimpleMessageListenerContainer` 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 set "prefetch-count".
+Not allowed when 'consumers-per-queue' is set.
 _Optional (Defaults to 1)_.
 
+<26> Indicates that the underlying listener container should be a `DirectMessageListenerContainer` instead of the default `SimpleMessageListenerContainer`.
+Refer to the Spring AMQP Reference Manual for more information.
+
 
 [NOTE]
 .container
@@ -215,8 +220,8 @@ For this reason, you must define the container using a normal Spring `<bean/>` d
 
 IMPORTANT: Even though the Spring Integration JMS and AMQP support is very similar, important differences exist.
 The JMS Inbound Channel Adapter is using a `JmsDestinationPollingSource` under the covers and expects a configured Poller.
-The AMQP Inbound Channel Adapter on the other side uses a`SimpleMessageListenerContainer` and is message driven.
-In that regard it is more similar to the JMS Message Driven Channel Adapter.
+The AMQP Inbound Channel Adapter uses an `AbstractMessageListenerContainer` and is message driven.
+In that regard, it is more similar to the JMS Message Driven Channel Adapter.
 
 ==== Configuring with Java Configuration
 
@@ -301,7 +306,7 @@ public class AmqpJavaApplication {
 
 The inbound gateway supports all the attributes on the inbound channel adapter (except 'channel' is replaced by 'request-channel'), plus some additional attributes:
 
-[source,xml]
+[source, xml]
 ----
 <int-amqp:inbound-gateway
                           id="inboundGateway" <1>
@@ -328,21 +333,19 @@ _Required_.
 _Optional_.
 By default only standard AMQP properties (e.g.
 `contentType`) will be copied to and from Spring Integration `MessageHeaders`.
-Any user-defined headers within the AMQP`MessageProperties` will NOT be copied to or from an AMQP Message by the default `DefaultAmqpHeaderMapper`.
+Any user-defined headers within the AMQP `MessageProperties` will NOT be copied to or from an AMQP Message by the default `DefaultAmqpHeaderMapper`.
 Not allowed if 'request-header-names' or 'reply-header-names' is provided.
 
 
 <4> Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the `MessageHeaders`.
 This can only be provided if the 'header-mapper' reference is not provided.
-The values in this list can also be simple patterns to be matched against the header names (e.g.
-"\*" or "foo*, bar" or "*foo").
+The values in this list can also be simple patterns to be matched against the header names (e.g. `"\*"` or `"foo*, bar"` or `"*foo"`).
 
 
 <5> Comma-separated list of names of `MessageHeaders` to be mapped into the AMQP Message Properties of the AMQP reply message.
 All standard Headers (e.g., `contentType`) will be mapped to AMQP Message Properties while user-defined headers will be mapped to the 'headers' property.
 This can only be provided if the 'header-mapper' reference is not provided.
-The values in this list can also be simple patterns to be matched against the header names (e.g.
-"\*" or "foo*, bar" or "*foo").
+The values in this list can also be simple patterns to be matched against the header names (e.g. `"\*"` or `"foo*, bar"` or `"*foo"`).
 
 
 <6> Message Channel where reply Messages will be expected.
@@ -369,6 +372,7 @@ See the note in <<amqp-inbound-channel-adapter>> about configuring the `listener
 ==== Configuring with Java Configuration
 
 The following Spring Boot application provides an example of configuring the inbound gateway using Java configuration:
+
 [source, java]
 ----
 @SpringBootApplication
@@ -594,8 +598,7 @@ _Optional_.
 
 <15> Comma-separated list of names of AMQP Headers to be mapped from the `MessageHeaders` to the AMQP Message.
 Not allowed if the 'header-mapper' reference is provided.
-The values in this list can also be simple patterns to be matched against the header names (e.g.
-"\*" or "foo*, bar" or "*foo").
+The values in this list can also be simple patterns to be matched against the header names (e.g. `"\*"` or `"foo*, bar"` or `"*foo"`).
 
 
 <16> When set to `false`, the endpoint will attempt to connect to the broker during application context initialization.

src/reference/asciidoc/whats-new.adoc

@@ -60,6 +60,8 @@ See <<barrier>> for more information.
 
 The AMQP outbound endpoints now support setting a delay expression for when using the RabbitMQ Delayed Message Exchange plugin.
 See <<amqp-delay>> for more information.
+The inbound endpoints now support the Spring AMQP `DirectMessageListenerContainer`.
+See <<amqp-inbound-channel-adapter>> for more information.
 
 ==== HTTP Changes