diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java index 5ead17b6c9..901ad60c3f 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java +++ b/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); } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java index 100c0f2d73..3965b24900 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java +++ b/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); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java index 31425d07ee..361ddfbf65 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java +++ b/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); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java index fa62c21321..21b3681b60 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java +++ b/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"; diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java index d6105b62c6..44956bfaa9 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java +++ b/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 + + + + + + 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. @@ -915,6 +925,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA The timeout for each attempt by a consumer to receive the next message. + Not allowed when 'consumers-per-queue' is set. @@ -980,6 +991,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA 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. @@ -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. diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml index 434a326f84..2633d4b12f 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml @@ -18,7 +18,7 @@ + template-channel-transacted="true" consumers-per-queue="2" /> diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java index a0759b5586..3f5df6374a 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java +++ b/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 diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests-context.xml index 74595b4a0b..ca3b3c4294 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests-context.xml +++ b/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"/> + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java index a338866822..c852b2ebcc 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java +++ b/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 diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 3c3d8dd677..c3e7327fe6 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/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] ---- @@ -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 `` 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] ---- @@ -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 <> 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. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index a5d5f6e072..23e49b1cd6 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -60,6 +60,8 @@ See <> for more information. The AMQP outbound endpoints now support setting a delay expression for when using the RabbitMQ Delayed Message Exchange plugin. See <> for more information. +The inbound endpoints now support the Spring AMQP `DirectMessageListenerContainer`. +See <> for more information. ==== HTTP Changes