From 4f619600d2254392b3e2975834465a4922b7bc37 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 1 May 2017 16:18:59 -0400 Subject: [PATCH 1/4] GH-9041: Spring AMQP: Support Both Container Types Resolves: https://github.com/spring-projects/spring-boot/issues/9041 Spring AMQP 2.0 introduced a new container type. Add support for auto configuration - select container type and separate discrete properties. --- .../RabbitAnnotationDrivenConfiguration.java | 24 ++- ...itListenerContainerFactoryConfigurer.java} | 52 +++++-- .../autoconfigure/amqp/RabbitProperties.java | 145 +++++++++++++----- .../amqp/RabbitAutoConfigurationTests.java | 44 +++++- .../appendix-application-properties.adoc | 8 +- 5 files changed, 208 insertions(+), 65 deletions(-) rename spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/{SimpleRabbitListenerContainerFactoryConfigurer.java => RabbitListenerContainerFactoryConfigurer.java} (68%) diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAnnotationDrivenConfiguration.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAnnotationDrivenConfiguration.java index 18c107ca2d78..1efc4e89fc5d 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAnnotationDrivenConfiguration.java +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAnnotationDrivenConfiguration.java @@ -17,12 +17,15 @@ package org.springframework.boot.autoconfigure.amqp; import org.springframework.amqp.rabbit.annotation.EnableRabbit; +import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.config.RabbitListenerConfigUtils; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.retry.MessageRecoverer; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.beans.factory.ObjectProvider; +import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ContainerType; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.context.annotation.Bean; @@ -55,8 +58,8 @@ class RabbitAnnotationDrivenConfiguration { @Bean @ConditionalOnMissingBean - public SimpleRabbitListenerContainerFactoryConfigurer rabbitListenerContainerFactoryConfigurer() { - SimpleRabbitListenerContainerFactoryConfigurer configurer = new SimpleRabbitListenerContainerFactoryConfigurer(); + public RabbitListenerContainerFactoryConfigurer rabbitListenerContainerFactoryConfigurer() { + RabbitListenerContainerFactoryConfigurer configurer = new RabbitListenerContainerFactoryConfigurer(); configurer.setMessageConverter(this.messageConverter.getIfUnique()); configurer.setMessageRecoverer(this.messageRecoverer.getIfUnique()); configurer.setRabbitProperties(this.properties); @@ -65,12 +68,19 @@ public SimpleRabbitListenerContainerFactoryConfigurer rabbitListenerContainerFac @Bean @ConditionalOnMissingBean(name = "rabbitListenerContainerFactory") - public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory( - SimpleRabbitListenerContainerFactoryConfigurer configurer, + public AbstractRabbitListenerContainerFactory rabbitListenerContainerFactory( + RabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) { - SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); - configurer.configure(factory, connectionFactory); - return factory; + if (this.properties.getListener().getContainerType().equals(ContainerType.SIMPLE)) { + SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); + configurer.configure(factory, connectionFactory); + return factory; + } + else { + DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory(); + configurer.configure(factory, connectionFactory); + return factory; + } } @EnableRabbit diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/SimpleRabbitListenerContainerFactoryConfigurer.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitListenerContainerFactoryConfigurer.java similarity index 68% rename from spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/SimpleRabbitListenerContainerFactoryConfigurer.java rename to spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitListenerContainerFactoryConfigurer.java index 3b90e9904272..d9454b07fac4 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/SimpleRabbitListenerContainerFactoryConfigurer.java +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitListenerContainerFactoryConfigurer.java @@ -16,6 +16,8 @@ package org.springframework.boot.autoconfigure.amqp; +import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -33,7 +35,7 @@ * @author Gary Russell * @since 1.3.3 */ -public final class SimpleRabbitListenerContainerFactoryConfigurer { +public final class RabbitListenerContainerFactoryConfigurer { private MessageConverter messageConverter; @@ -67,7 +69,7 @@ void setRabbitProperties(RabbitProperties rabbitProperties) { } /** - * Configure the specified rabbit listener container factory. The factory can be + * Configure the specified simple listener container factory. The factory can be * further tuned and default settings can be overridden. * @param factory the {@link SimpleRabbitListenerContainerFactory} instance to * configure @@ -75,6 +77,43 @@ void setRabbitProperties(RabbitProperties rabbitProperties) { */ public void configure(SimpleRabbitListenerContainerFactory factory, ConnectionFactory connectionFactory) { + configureCommon(factory, connectionFactory); + RabbitProperties.Listener listenerConfig = this.rabbitProperties.getListener(); + if (listenerConfig.getSimple().getConcurrency() != null) { + factory.setConcurrentConsumers(listenerConfig.getSimple().getConcurrency()); + } + if (listenerConfig.getSimple().getMaxConcurrency() != null) { + factory.setMaxConcurrentConsumers(listenerConfig.getSimple().getMaxConcurrency()); + } + if (listenerConfig.getSimple().getTransactionSize() != null) { + factory.setTxSize(listenerConfig.getSimple().getTransactionSize()); + } + } + + /** + * Configure the specified direct listener container factory. The factory can be + * further tuned and default settings can be overridden. + * @param factory the {@link DirectRabbitListenerContainerFactory} instance to + * configure + * @param connectionFactory the {@link ConnectionFactory} to use + */ + public void configure(DirectRabbitListenerContainerFactory factory, + ConnectionFactory connectionFactory) { + configureCommon(factory, connectionFactory); + RabbitProperties.Listener listenerConfig = this.rabbitProperties.getListener(); + if (listenerConfig.getDirect().getConsumersPerQueue() != null) { + factory.setConsumersPerQueue(listenerConfig.getDirect().getConsumersPerQueue()); + } + } + + /** + * Configure the common properties on the specified rabbit listener container factory. + * @param factory the {@link AbstractRabbitListenerContainerFactory} instance to + * configure + * @param connectionFactory the {@link ConnectionFactory} to use + */ + private void configureCommon(AbstractRabbitListenerContainerFactory factory, + ConnectionFactory connectionFactory) { Assert.notNull(factory, "Factory must not be null"); Assert.notNull(connectionFactory, "ConnectionFactory must not be null"); factory.setConnectionFactory(connectionFactory); @@ -86,18 +125,9 @@ public void configure(SimpleRabbitListenerContainerFactory factory, if (listenerConfig.getAcknowledgeMode() != null) { factory.setAcknowledgeMode(listenerConfig.getAcknowledgeMode()); } - if (listenerConfig.getConcurrency() != null) { - factory.setConcurrentConsumers(listenerConfig.getConcurrency()); - } - if (listenerConfig.getMaxConcurrency() != null) { - factory.setMaxConcurrentConsumers(listenerConfig.getMaxConcurrency()); - } if (listenerConfig.getPrefetch() != null) { factory.setPrefetchCount(listenerConfig.getPrefetch()); } - if (listenerConfig.getTransactionSize() != null) { - factory.setTxSize(listenerConfig.getTransactionSize()); - } if (listenerConfig.getDefaultRequeueRejected() != null) { factory.setDefaultRequeueRejected(listenerConfig.getDefaultRequeueRejected()); } diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java index 6a9a4a7abfda..37c5b32822c8 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java @@ -39,6 +39,20 @@ @ConfigurationProperties(prefix = "spring.rabbitmq") public class RabbitProperties { + public enum ContainerType { + + /** + * SimpleMessageListenerContainer. + */ + SIMPLE, + + /** + * DirectMessageListenerContainer + */ + DIRECT + + } + /** * RabbitMQ host. */ @@ -466,6 +480,11 @@ public void setSize(Integer size) { public static class Listener { + /** + * The container type. + */ + private ContainerType containerType = ContainerType.SIMPLE; + /** * Start the container automatically on startup. */ @@ -476,28 +495,12 @@ public static class Listener { */ private AcknowledgeMode acknowledgeMode; - /** - * Minimum number of consumers. - */ - private Integer concurrency; - - /** - * Maximum number of consumers. - */ - private Integer maxConcurrency; - /** * Number of messages to be handled in a single request. It should be greater than * or equal to the transaction size (if used). */ private Integer prefetch; - /** - * Number of messages to be processed in a transaction. For best results it should - * be less than or equal to the prefetch count. - */ - private Integer transactionSize; - /** * Whether rejected deliveries are requeued by default; default true. */ @@ -514,6 +517,20 @@ public static class Listener { @NestedConfigurationProperty private final ListenerRetry retry = new ListenerRetry(); + @NestedConfigurationProperty + private final SimpleContainer simple = new SimpleContainer(); + + @NestedConfigurationProperty + private final DirectContainer direct = new DirectContainer(); + + public ContainerType getContainerType() { + return this.containerType; + } + + public void setContainerType(ContainerType containerType) { + this.containerType = containerType; + } + public boolean isAutoStartup() { return this.autoStartup; } @@ -530,6 +547,67 @@ public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) { this.acknowledgeMode = acknowledgeMode; } + public Integer getPrefetch() { + return this.prefetch; + } + + public void setPrefetch(Integer prefetch) { + this.prefetch = prefetch; + } + + public Boolean getDefaultRequeueRejected() { + return this.defaultRequeueRejected; + } + + public void setDefaultRequeueRejected(Boolean defaultRequeueRejected) { + this.defaultRequeueRejected = defaultRequeueRejected; + } + + public Long getIdleEventInterval() { + return this.idleEventInterval; + } + + public void setIdleEventInterval(Long idleEventInterval) { + this.idleEventInterval = idleEventInterval; + } + + public ListenerRetry getRetry() { + return this.retry; + } + + public SimpleContainer getSimple() { + return this.simple; + } + + public DirectContainer getDirect() { + return this.direct; + } + + } + + /** + * SMLC properties + * @since 2.0 + */ + public static class SimpleContainer { + + /** + * Minimum number of listener invoker threads. + */ + private Integer concurrency; + + /** + * Maximum number of listener invoker threads. + */ + private Integer maxConcurrency; + + /** + * Number of messages to be processed in a transaction; number of messages + * between acks. For best results it should + * be less than or equal to the prefetch count. + */ + private Integer transactionSize; + public Integer getConcurrency() { return this.concurrency; } @@ -546,14 +624,6 @@ public void setMaxConcurrency(Integer maxConcurrency) { this.maxConcurrency = maxConcurrency; } - public Integer getPrefetch() { - return this.prefetch; - } - - public void setPrefetch(Integer prefetch) { - this.prefetch = prefetch; - } - public Integer getTransactionSize() { return this.transactionSize; } @@ -562,24 +632,25 @@ public void setTransactionSize(Integer transactionSize) { this.transactionSize = transactionSize; } - public Boolean getDefaultRequeueRejected() { - return this.defaultRequeueRejected; - } + } - public void setDefaultRequeueRejected(Boolean defaultRequeueRejected) { - this.defaultRequeueRejected = defaultRequeueRejected; - } + /** + * DMLC properties + * @since 2.0 + */ + public static class DirectContainer { - public Long getIdleEventInterval() { - return this.idleEventInterval; - } + /** + * The number of consumers per queue. + */ + private Integer consumersPerQueue; - public void setIdleEventInterval(Long idleEventInterval) { - this.idleEventInterval = idleEventInterval; + public Integer getConsumersPerQueue() { + return this.consumersPerQueue; } - public ListenerRetry getRetry() { - return this.retry; + public void setConsumersPerQueue(Integer consumersPerQueue) { + this.consumersPerQueue = consumersPerQueue; } } diff --git a/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java b/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java index 7db7b272bbdb..bc159ecc8311 100644 --- a/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java +++ b/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java @@ -20,6 +20,7 @@ import javax.net.ssl.SSLSocketFactory; import com.rabbitmq.client.Address; + import org.aopalliance.aop.Advice; import org.junit.After; import org.junit.Rule; @@ -30,6 +31,7 @@ import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.EnableRabbit; +import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.config.RabbitListenerConfigUtils; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; @@ -293,7 +295,7 @@ public void testRabbitListenerContainerFactoryBackOff() { } @Test - public void testRabbitListenerContainerFactoryWithCustomSettings() { + public void testSimpleRabbitListenerContainerFactoryWithCustomSettings() { load(new Class[] { MessageConvertersConfiguration.class, MessageRecoverersConfiguration.class }, "spring.rabbitmq.listener.retry.enabled:true", @@ -303,23 +305,51 @@ public void testRabbitListenerContainerFactoryWithCustomSettings() { "spring.rabbitmq.listener.retry.maxInterval:5000", "spring.rabbitmq.listener.autoStartup:false", "spring.rabbitmq.listener.acknowledgeMode:manual", - "spring.rabbitmq.listener.concurrency:5", - "spring.rabbitmq.listener.maxConcurrency:10", + "spring.rabbitmq.listener.simple.concurrency:5", + "spring.rabbitmq.listener.simple.maxConcurrency:10", "spring.rabbitmq.listener.prefetch:40", "spring.rabbitmq.listener.defaultRequeueRejected:false", "spring.rabbitmq.listener.idleEventInterval:5", - "spring.rabbitmq.listener.transactionSize:20"); + "spring.rabbitmq.listener.simple.transactionSize:20"); SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory = this.context .getBean("rabbitListenerContainerFactory", SimpleRabbitListenerContainerFactory.class); DirectFieldAccessor dfa = new DirectFieldAccessor(rabbitListenerContainerFactory); + assertThat(dfa.getPropertyValue("concurrentConsumers")).isEqualTo(5); + assertThat(dfa.getPropertyValue("maxConcurrentConsumers")).isEqualTo(10); + assertThat(dfa.getPropertyValue("txSize")).isEqualTo(20); + checkCommonProps(dfa); + } + + @Test + public void testDirectRabbitListenerContainerFactoryWithCustomSettings() { + load(new Class[] { MessageConvertersConfiguration.class, + MessageRecoverersConfiguration.class }, + "spring.rabbitmq.listener.container-type:direct", + "spring.rabbitmq.listener.retry.enabled:true", + "spring.rabbitmq.listener.retry.maxAttempts:4", + "spring.rabbitmq.listener.retry.initialInterval:2000", + "spring.rabbitmq.listener.retry.multiplier:1.5", + "spring.rabbitmq.listener.retry.maxInterval:5000", + "spring.rabbitmq.listener.autoStartup:false", + "spring.rabbitmq.listener.acknowledgeMode:manual", + "spring.rabbitmq.listener.direct.consumers-per-queue:5", + "spring.rabbitmq.listener.prefetch:40", + "spring.rabbitmq.listener.defaultRequeueRejected:false", + "spring.rabbitmq.listener.idleEventInterval:5"); + DirectRabbitListenerContainerFactory rabbitListenerContainerFactory = this.context + .getBean("rabbitListenerContainerFactory", + DirectRabbitListenerContainerFactory.class); + DirectFieldAccessor dfa = new DirectFieldAccessor(rabbitListenerContainerFactory); + assertThat(dfa.getPropertyValue("consumersPerQueue")).isEqualTo(5); + checkCommonProps(dfa); + } + + private void checkCommonProps(DirectFieldAccessor dfa) { assertThat(dfa.getPropertyValue("autoStartup")).isEqualTo(Boolean.FALSE); assertThat(dfa.getPropertyValue("acknowledgeMode")) .isEqualTo(AcknowledgeMode.MANUAL); - assertThat(dfa.getPropertyValue("concurrentConsumers")).isEqualTo(5); - assertThat(dfa.getPropertyValue("maxConcurrentConsumers")).isEqualTo(10); assertThat(dfa.getPropertyValue("prefetchCount")).isEqualTo(40); - assertThat(dfa.getPropertyValue("txSize")).isEqualTo(20); assertThat(dfa.getPropertyValue("messageConverter")) .isSameAs(this.context.getBean("myMessageConverter")); assertThat(dfa.getPropertyValue("defaultRequeueRejected")) diff --git a/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc b/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc index 658f787f655a..314827439b57 100644 --- a/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc +++ b/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc @@ -972,10 +972,10 @@ content into your application; rather pick only the properties that you need. spring.rabbitmq.host=localhost # RabbitMQ host. spring.rabbitmq.listener.acknowledge-mode= # Acknowledge mode of container. spring.rabbitmq.listener.auto-startup=true # Start the container automatically on startup. - spring.rabbitmq.listener.concurrency= # Minimum number of consumers. + spring.rabbitmq.listener.container-type= # The listener container type `simple` or `direct`; default `simple`. spring.rabbitmq.listener.default-requeue-rejected= # Whether or not to requeue delivery failures; default `true`. + spring.rabbitmq.listener.direct.consumers-per-queue= # The number of Consumers per queue for a `direct` container. spring.rabbitmq.listener.idle-event-interval= # How often idle container events should be published in milliseconds. - spring.rabbitmq.listener.max-concurrency= # Maximum number of consumers. spring.rabbitmq.listener.prefetch= # Number of messages to be handled in a single request. It should be greater than or equal to the transaction size (if used). spring.rabbitmq.listener.retry.enabled=false # Whether or not publishing retries are enabled. spring.rabbitmq.listener.retry.initial-interval=1000 # Interval between the first and second attempt to deliver a message. @@ -983,7 +983,9 @@ content into your application; rather pick only the properties that you need. spring.rabbitmq.listener.retry.max-interval=10000 # Maximum interval between attempts. spring.rabbitmq.listener.retry.multiplier=1.0 # A multiplier to apply to the previous delivery retry interval. spring.rabbitmq.listener.retry.stateless=true # Whether or not retry is stateless or stateful. - spring.rabbitmq.listener.transaction-size= # Number of messages to be processed in a transaction. For best results it should be less than or equal to the prefetch count. + spring.rabbitmq.listener.simple.concurrency= # Minimum number of listener invoker threads for a `simple` container. + spring.rabbitmq.listener.simple.max-concurrency= # Maximum number of listener invoker for a `simple` container. + spring.rabbitmq.listener.simple.transaction-size= # Number of messages to be processed in a transaction; number of messages between acks for a `simple` container. For best results it should be less than or equal to the prefetch count. spring.rabbitmq.password= # Login to authenticate against the broker. spring.rabbitmq.port=5672 # RabbitMQ port. spring.rabbitmq.publisher-confirms=false # Enable publisher confirms. From ae82c3601bfc8e89bd409c025371cc2f62043abf Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 2 May 2017 13:26:08 -0400 Subject: [PATCH 2/4] Polishing - PR Comments - Remove listener type hierarcy in properties - Add configurer class hierarchy and conditional beans --- ...itListenerContainerFactoryConfigurer.java} | 68 +++----- ...bitListenerContainerFactoryConfigurer.java | 39 +++++ .../RabbitAnnotationDrivenConfiguration.java | 78 ++++++--- .../autoconfigure/amqp/RabbitProperties.java | 150 +++++++----------- ...bitListenerContainerFactoryConfigurer.java | 46 ++++++ .../amqp/RabbitAutoConfigurationTests.java | 10 +- .../appendix-application-properties.adoc | 10 +- 7 files changed, 232 insertions(+), 169 deletions(-) rename spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/{RabbitListenerContainerFactoryConfigurer.java => AbstractRabbitListenerContainerFactoryConfigurer.java} (63%) create mode 100644 spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/DirectRabbitListenerContainerFactoryConfigurer.java create mode 100644 spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/SimpleRabbitListenerContainerFactoryConfigurer.java diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitListenerContainerFactoryConfigurer.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/AbstractRabbitListenerContainerFactoryConfigurer.java similarity index 63% rename from spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitListenerContainerFactoryConfigurer.java rename to spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/AbstractRabbitListenerContainerFactoryConfigurer.java index d9454b07fac4..05a50ea6ce63 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitListenerContainerFactoryConfigurer.java +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/AbstractRabbitListenerContainerFactoryConfigurer.java @@ -17,9 +17,7 @@ package org.springframework.boot.autoconfigure.amqp; import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory; -import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder; -import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; import org.springframework.amqp.rabbit.retry.MessageRecoverer; @@ -31,11 +29,14 @@ /** * Configure {@link RabbitListenerContainerFactory} with sensible defaults. * - * @author Stephane Nicoll + * @param the container factory type. + * * @author Gary Russell - * @since 1.3.3 + * @since 2.0 + * */ -public final class RabbitListenerContainerFactoryConfigurer { +public abstract class AbstractRabbitListenerContainerFactoryConfigurer< + T extends AbstractRabbitListenerContainerFactory> { private MessageConverter messageConverter; @@ -48,7 +49,7 @@ public final class RabbitListenerContainerFactoryConfigurer { * converter should be used. * @param messageConverter the {@link MessageConverter} */ - void setMessageConverter(MessageConverter messageConverter) { + protected void setMessageConverter(MessageConverter messageConverter) { this.messageConverter = messageConverter; } @@ -56,7 +57,7 @@ void setMessageConverter(MessageConverter messageConverter) { * Set the {@link MessageRecoverer} to use or {@code null} to rely on the default. * @param messageRecoverer the {@link MessageRecoverer} */ - void setMessageRecoverer(MessageRecoverer messageRecoverer) { + protected void setMessageRecoverer(MessageRecoverer messageRecoverer) { this.messageRecoverer = messageRecoverer; } @@ -64,56 +65,18 @@ void setMessageRecoverer(MessageRecoverer messageRecoverer) { * Set the {@link RabbitProperties} to use. * @param rabbitProperties the {@link RabbitProperties} */ - void setRabbitProperties(RabbitProperties rabbitProperties) { + protected void setRabbitProperties(RabbitProperties rabbitProperties) { this.rabbitProperties = rabbitProperties; } /** - * Configure the specified simple listener container factory. The factory can be - * further tuned and default settings can be overridden. - * @param factory the {@link SimpleRabbitListenerContainerFactory} instance to - * configure - * @param connectionFactory the {@link ConnectionFactory} to use - */ - public void configure(SimpleRabbitListenerContainerFactory factory, - ConnectionFactory connectionFactory) { - configureCommon(factory, connectionFactory); - RabbitProperties.Listener listenerConfig = this.rabbitProperties.getListener(); - if (listenerConfig.getSimple().getConcurrency() != null) { - factory.setConcurrentConsumers(listenerConfig.getSimple().getConcurrency()); - } - if (listenerConfig.getSimple().getMaxConcurrency() != null) { - factory.setMaxConcurrentConsumers(listenerConfig.getSimple().getMaxConcurrency()); - } - if (listenerConfig.getSimple().getTransactionSize() != null) { - factory.setTxSize(listenerConfig.getSimple().getTransactionSize()); - } - } - - /** - * Configure the specified direct listener container factory. The factory can be + * Configure the specified rabbit listener container factory. The factory can be * further tuned and default settings can be overridden. - * @param factory the {@link DirectRabbitListenerContainerFactory} instance to - * configure - * @param connectionFactory the {@link ConnectionFactory} to use - */ - public void configure(DirectRabbitListenerContainerFactory factory, - ConnectionFactory connectionFactory) { - configureCommon(factory, connectionFactory); - RabbitProperties.Listener listenerConfig = this.rabbitProperties.getListener(); - if (listenerConfig.getDirect().getConsumersPerQueue() != null) { - factory.setConsumersPerQueue(listenerConfig.getDirect().getConsumersPerQueue()); - } - } - - /** - * Configure the common properties on the specified rabbit listener container factory. * @param factory the {@link AbstractRabbitListenerContainerFactory} instance to * configure * @param connectionFactory the {@link ConnectionFactory} to use */ - private void configureCommon(AbstractRabbitListenerContainerFactory factory, - ConnectionFactory connectionFactory) { + public final void configure(T factory, ConnectionFactory connectionFactory) { Assert.notNull(factory, "Factory must not be null"); Assert.notNull(connectionFactory, "ConnectionFactory must not be null"); factory.setConnectionFactory(connectionFactory); @@ -147,7 +110,14 @@ private void configureCommon(AbstractRabbitListenerContainerFactory factory, builder.recoverer(recoverer); factory.setAdviceChain(builder.build()); } - + configure(factory, this.rabbitProperties); } + /** + * Perform factory-specific configuration. + * + * @param factory the factory + */ + protected abstract void configure(T factory, RabbitProperties rabbitProperties); + } diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/DirectRabbitListenerContainerFactoryConfigurer.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/DirectRabbitListenerContainerFactoryConfigurer.java new file mode 100644 index 000000000000..4a7026210bc7 --- /dev/null +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/DirectRabbitListenerContainerFactoryConfigurer.java @@ -0,0 +1,39 @@ +/* + * Copyright 2012-2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * 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.boot.autoconfigure.amqp; + +import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; + +/** + * Configure {@link RabbitListenerContainerFactory} with sensible defaults. + * + * @author Gary Russell + * @since 2.0 + */ +public final class DirectRabbitListenerContainerFactoryConfigurer + extends AbstractRabbitListenerContainerFactoryConfigurer { + + @Override + protected void configure(DirectRabbitListenerContainerFactory factory, RabbitProperties rabbitProperties) { + RabbitProperties.Listener listenerConfig = rabbitProperties.getListener(); + if (listenerConfig.getConsumersPerQueue() != null) { + factory.setConsumersPerQueue(listenerConfig.getConsumersPerQueue()); + } + } + +} diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAnnotationDrivenConfiguration.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAnnotationDrivenConfiguration.java index 1efc4e89fc5d..b75045460184 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAnnotationDrivenConfiguration.java +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAnnotationDrivenConfiguration.java @@ -17,7 +17,6 @@ package org.springframework.boot.autoconfigure.amqp; import org.springframework.amqp.rabbit.annotation.EnableRabbit; -import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.config.RabbitListenerConfigUtils; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; @@ -25,9 +24,9 @@ import org.springframework.amqp.rabbit.retry.MessageRecoverer; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.beans.factory.ObjectProvider; -import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ContainerType; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -42,11 +41,11 @@ @ConditionalOnClass(EnableRabbit.class) class RabbitAnnotationDrivenConfiguration { - private final ObjectProvider messageConverter; + protected final ObjectProvider messageConverter; - private final ObjectProvider messageRecoverer; + protected final ObjectProvider messageRecoverer; - private final RabbitProperties properties; + protected final RabbitProperties properties; RabbitAnnotationDrivenConfiguration(ObjectProvider messageConverter, ObjectProvider messageRecoverer, @@ -56,27 +55,64 @@ class RabbitAnnotationDrivenConfiguration { this.properties = properties; } - @Bean - @ConditionalOnMissingBean - public RabbitListenerContainerFactoryConfigurer rabbitListenerContainerFactoryConfigurer() { - RabbitListenerContainerFactoryConfigurer configurer = new RabbitListenerContainerFactoryConfigurer(); - configurer.setMessageConverter(this.messageConverter.getIfUnique()); - configurer.setMessageRecoverer(this.messageRecoverer.getIfUnique()); - configurer.setRabbitProperties(this.properties); - return configurer; - } - @Bean - @ConditionalOnMissingBean(name = "rabbitListenerContainerFactory") - public AbstractRabbitListenerContainerFactory rabbitListenerContainerFactory( - RabbitListenerContainerFactoryConfigurer configurer, - ConnectionFactory connectionFactory) { - if (this.properties.getListener().getContainerType().equals(ContainerType.SIMPLE)) { + @Configuration + @ConditionalOnProperty(prefix = "spring.rabbitmq", name = "listener.type", havingValue = "simple", + matchIfMissing = true) + public static class SimpleContainerConfiguration extends RabbitAnnotationDrivenConfiguration { + + SimpleContainerConfiguration(ObjectProvider messageConverter, + ObjectProvider messageRecoverer, RabbitProperties properties) { + super(messageConverter, messageRecoverer, properties); + } + + @Bean + @ConditionalOnMissingBean + public SimpleRabbitListenerContainerFactoryConfigurer rabbitListenerContainerFactoryConfigurer() { + SimpleRabbitListenerContainerFactoryConfigurer configurer = + new SimpleRabbitListenerContainerFactoryConfigurer(); + configurer.setMessageConverter(this.messageConverter.getIfUnique()); + configurer.setMessageRecoverer(this.messageRecoverer.getIfUnique()); + configurer.setRabbitProperties(this.properties); + return configurer; + } + + @Bean + @ConditionalOnMissingBean(name = "rabbitListenerContainerFactory") + public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory( + SimpleRabbitListenerContainerFactoryConfigurer configurer, + ConnectionFactory connectionFactory) { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); configurer.configure(factory, connectionFactory); return factory; } - else { + } + + @Configuration + @ConditionalOnProperty(prefix = "spring.rabbitmq", name = "listener.type", havingValue = "direct") + public static class DirectContainerConfiguration extends RabbitAnnotationDrivenConfiguration { + + DirectContainerConfiguration(ObjectProvider messageConverter, + ObjectProvider messageRecoverer, RabbitProperties properties) { + super(messageConverter, messageRecoverer, properties); + } + + @Bean + @ConditionalOnMissingBean + public DirectRabbitListenerContainerFactoryConfigurer rabbitListenerContainerFactoryConfigurer() { + DirectRabbitListenerContainerFactoryConfigurer configurer = + new DirectRabbitListenerContainerFactoryConfigurer(); + configurer.setMessageConverter(this.messageConverter.getIfUnique()); + configurer.setMessageRecoverer(this.messageRecoverer.getIfUnique()); + configurer.setRabbitProperties(this.properties); + return configurer; + } + + @Bean + @ConditionalOnMissingBean(name = "rabbitListenerContainerFactory") + public DirectRabbitListenerContainerFactory rabbitListenerContainerFactory( + DirectRabbitListenerContainerFactoryConfigurer configurer, + ConnectionFactory connectionFactory) { DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory(); configurer.configure(factory, connectionFactory); return factory; diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java index 37c5b32822c8..cfc94067bdf7 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java @@ -42,12 +42,14 @@ public class RabbitProperties { public enum ContainerType { /** - * SimpleMessageListenerContainer. + * SimpleMessageListenerContainer - legacy container where the RabbitMQ consumer + * dispatches messages to an invoker thread. */ SIMPLE, /** - * DirectMessageListenerContainer + * DirectMessageListenerContainer - container where the listener is invoked + * directly on the RabbitMQ consumer thread. */ DIRECT @@ -481,9 +483,9 @@ public void setSize(Integer size) { public static class Listener { /** - * The container type. + * Container type. */ - private ContainerType containerType = ContainerType.SIMPLE; + private ContainerType type = ContainerType.SIMPLE; /** * Start the container automatically on startup. @@ -495,12 +497,34 @@ public static class Listener { */ private AcknowledgeMode acknowledgeMode; + /** + * Minimum number of listener invoker threads - applies only to simple containers. + */ + private Integer concurrency; + + /** + * Maximum number of listener invoker threads - applies only to simple containers. + */ + private Integer maxConcurrency; + + /** + * Number of RabbitMQ consumers per queue - applies only to direct containers. + */ + private Integer consumersPerQueue; + /** * Number of messages to be handled in a single request. It should be greater than * or equal to the transaction size (if used). */ private Integer prefetch; + /** + * Number of messages to be processed in a transaction; number of messages between + * acks. For best results it should be less than or equal to the prefetch count - + * applies only to simple containers. + */ + private Integer transactionSize; + /** * Whether rejected deliveries are requeued by default; default true. */ @@ -517,18 +541,12 @@ public static class Listener { @NestedConfigurationProperty private final ListenerRetry retry = new ListenerRetry(); - @NestedConfigurationProperty - private final SimpleContainer simple = new SimpleContainer(); - - @NestedConfigurationProperty - private final DirectContainer direct = new DirectContainer(); - - public ContainerType getContainerType() { - return this.containerType; + public ContainerType getType() { + return this.type; } - public void setContainerType(ContainerType containerType) { - this.containerType = containerType; + public void setType(ContainerType containerType) { + this.type = containerType; } public boolean isAutoStartup() { @@ -547,67 +565,6 @@ public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) { this.acknowledgeMode = acknowledgeMode; } - public Integer getPrefetch() { - return this.prefetch; - } - - public void setPrefetch(Integer prefetch) { - this.prefetch = prefetch; - } - - public Boolean getDefaultRequeueRejected() { - return this.defaultRequeueRejected; - } - - public void setDefaultRequeueRejected(Boolean defaultRequeueRejected) { - this.defaultRequeueRejected = defaultRequeueRejected; - } - - public Long getIdleEventInterval() { - return this.idleEventInterval; - } - - public void setIdleEventInterval(Long idleEventInterval) { - this.idleEventInterval = idleEventInterval; - } - - public ListenerRetry getRetry() { - return this.retry; - } - - public SimpleContainer getSimple() { - return this.simple; - } - - public DirectContainer getDirect() { - return this.direct; - } - - } - - /** - * SMLC properties - * @since 2.0 - */ - public static class SimpleContainer { - - /** - * Minimum number of listener invoker threads. - */ - private Integer concurrency; - - /** - * Maximum number of listener invoker threads. - */ - private Integer maxConcurrency; - - /** - * Number of messages to be processed in a transaction; number of messages - * between acks. For best results it should - * be less than or equal to the prefetch count. - */ - private Integer transactionSize; - public Integer getConcurrency() { return this.concurrency; } @@ -624,6 +581,22 @@ public void setMaxConcurrency(Integer maxConcurrency) { this.maxConcurrency = maxConcurrency; } + public Integer getConsumersPerQueue() { + return this.consumersPerQueue; + } + + public void setConsumersPerQueue(Integer consumersPerQueue) { + this.consumersPerQueue = consumersPerQueue; + } + + public Integer getPrefetch() { + return this.prefetch; + } + + public void setPrefetch(Integer prefetch) { + this.prefetch = prefetch; + } + public Integer getTransactionSize() { return this.transactionSize; } @@ -632,25 +605,24 @@ public void setTransactionSize(Integer transactionSize) { this.transactionSize = transactionSize; } - } + public Boolean getDefaultRequeueRejected() { + return this.defaultRequeueRejected; + } - /** - * DMLC properties - * @since 2.0 - */ - public static class DirectContainer { + public void setDefaultRequeueRejected(Boolean defaultRequeueRejected) { + this.defaultRequeueRejected = defaultRequeueRejected; + } - /** - * The number of consumers per queue. - */ - private Integer consumersPerQueue; + public Long getIdleEventInterval() { + return this.idleEventInterval; + } - public Integer getConsumersPerQueue() { - return this.consumersPerQueue; + public void setIdleEventInterval(Long idleEventInterval) { + this.idleEventInterval = idleEventInterval; } - public void setConsumersPerQueue(Integer consumersPerQueue) { - this.consumersPerQueue = consumersPerQueue; + public ListenerRetry getRetry() { + return this.retry; } } diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/SimpleRabbitListenerContainerFactoryConfigurer.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/SimpleRabbitListenerContainerFactoryConfigurer.java new file mode 100644 index 000000000000..8d2845c38b37 --- /dev/null +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/SimpleRabbitListenerContainerFactoryConfigurer.java @@ -0,0 +1,46 @@ +/* + * Copyright 2012-2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * 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.boot.autoconfigure.amqp; + +import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; + +/** + * Configure {@link RabbitListenerContainerFactory} with sensible defaults. + * + * @author Stephane Nicoll + * @author Gary Russell + * @since 1.3.3 + */ +public final class SimpleRabbitListenerContainerFactoryConfigurer + extends AbstractRabbitListenerContainerFactoryConfigurer { + + @Override + protected void configure(SimpleRabbitListenerContainerFactory factory, RabbitProperties rabbitProperties) { + RabbitProperties.Listener listenerConfig = rabbitProperties.getListener(); + if (listenerConfig.getConcurrency() != null) { + factory.setConcurrentConsumers(listenerConfig.getConcurrency()); + } + if (listenerConfig.getMaxConcurrency() != null) { + factory.setMaxConcurrentConsumers(listenerConfig.getMaxConcurrency()); + } + if (listenerConfig.getTransactionSize() != null) { + factory.setTxSize(listenerConfig.getTransactionSize()); + } + } + +} diff --git a/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java b/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java index bc159ecc8311..6d80320f6f32 100644 --- a/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java +++ b/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java @@ -305,12 +305,12 @@ public void testSimpleRabbitListenerContainerFactoryWithCustomSettings() { "spring.rabbitmq.listener.retry.maxInterval:5000", "spring.rabbitmq.listener.autoStartup:false", "spring.rabbitmq.listener.acknowledgeMode:manual", - "spring.rabbitmq.listener.simple.concurrency:5", - "spring.rabbitmq.listener.simple.maxConcurrency:10", + "spring.rabbitmq.listener.concurrency:5", + "spring.rabbitmq.listener.maxConcurrency:10", "spring.rabbitmq.listener.prefetch:40", "spring.rabbitmq.listener.defaultRequeueRejected:false", "spring.rabbitmq.listener.idleEventInterval:5", - "spring.rabbitmq.listener.simple.transactionSize:20"); + "spring.rabbitmq.listener.transactionSize:20"); SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory = this.context .getBean("rabbitListenerContainerFactory", SimpleRabbitListenerContainerFactory.class); @@ -325,7 +325,7 @@ public void testSimpleRabbitListenerContainerFactoryWithCustomSettings() { public void testDirectRabbitListenerContainerFactoryWithCustomSettings() { load(new Class[] { MessageConvertersConfiguration.class, MessageRecoverersConfiguration.class }, - "spring.rabbitmq.listener.container-type:direct", + "spring.rabbitmq.listener.type:direct", "spring.rabbitmq.listener.retry.enabled:true", "spring.rabbitmq.listener.retry.maxAttempts:4", "spring.rabbitmq.listener.retry.initialInterval:2000", @@ -333,7 +333,7 @@ public void testDirectRabbitListenerContainerFactoryWithCustomSettings() { "spring.rabbitmq.listener.retry.maxInterval:5000", "spring.rabbitmq.listener.autoStartup:false", "spring.rabbitmq.listener.acknowledgeMode:manual", - "spring.rabbitmq.listener.direct.consumers-per-queue:5", + "spring.rabbitmq.listener.consumers-per-queue:5", "spring.rabbitmq.listener.prefetch:40", "spring.rabbitmq.listener.defaultRequeueRejected:false", "spring.rabbitmq.listener.idleEventInterval:5"); diff --git a/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc b/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc index 314827439b57..f3dedb2469fa 100644 --- a/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc +++ b/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc @@ -972,10 +972,11 @@ content into your application; rather pick only the properties that you need. spring.rabbitmq.host=localhost # RabbitMQ host. spring.rabbitmq.listener.acknowledge-mode= # Acknowledge mode of container. spring.rabbitmq.listener.auto-startup=true # Start the container automatically on startup. - spring.rabbitmq.listener.container-type= # The listener container type `simple` or `direct`; default `simple`. - spring.rabbitmq.listener.default-requeue-rejected= # Whether or not to requeue delivery failures; default `true`. + spring.rabbitmq.listener.concurrency= # Minimum number of listener invoker threads for a `simple` container. spring.rabbitmq.listener.direct.consumers-per-queue= # The number of Consumers per queue for a `direct` container. + spring.rabbitmq.listener.default-requeue-rejected= # Whether or not to requeue delivery failures; default `true`. spring.rabbitmq.listener.idle-event-interval= # How often idle container events should be published in milliseconds. + spring.rabbitmq.listener.max-concurrency= # Maximum number of listener invoker for a `simple` container. spring.rabbitmq.listener.prefetch= # Number of messages to be handled in a single request. It should be greater than or equal to the transaction size (if used). spring.rabbitmq.listener.retry.enabled=false # Whether or not publishing retries are enabled. spring.rabbitmq.listener.retry.initial-interval=1000 # Interval between the first and second attempt to deliver a message. @@ -983,9 +984,8 @@ content into your application; rather pick only the properties that you need. spring.rabbitmq.listener.retry.max-interval=10000 # Maximum interval between attempts. spring.rabbitmq.listener.retry.multiplier=1.0 # A multiplier to apply to the previous delivery retry interval. spring.rabbitmq.listener.retry.stateless=true # Whether or not retry is stateless or stateful. - spring.rabbitmq.listener.simple.concurrency= # Minimum number of listener invoker threads for a `simple` container. - spring.rabbitmq.listener.simple.max-concurrency= # Maximum number of listener invoker for a `simple` container. - spring.rabbitmq.listener.simple.transaction-size= # Number of messages to be processed in a transaction; number of messages between acks for a `simple` container. For best results it should be less than or equal to the prefetch count. + spring.rabbitmq.listener.transaction-size= Number of messages to be processed in a transaction; number of messages between acks. For best results it should be less than or equal to the prefetch count - applies only to `simple` containers. + spring.rabbitmq.listener.type= # The listener container type `simple` or `direct`; default `simple`. spring.rabbitmq.password= # Login to authenticate against the broker. spring.rabbitmq.port=5672 # RabbitMQ port. spring.rabbitmq.publisher-confirms=false # Enable publisher confirms. From fbcc855b84fb63128bc1d4913498dbf856496144 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 2 May 2017 13:30:35 -0400 Subject: [PATCH 3/4] Fix property name in doc --- .../src/main/asciidoc/appendix-application-properties.adoc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc b/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc index f3dedb2469fa..3ae2319f19a3 100644 --- a/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc +++ b/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc @@ -973,7 +973,7 @@ content into your application; rather pick only the properties that you need. spring.rabbitmq.listener.acknowledge-mode= # Acknowledge mode of container. spring.rabbitmq.listener.auto-startup=true # Start the container automatically on startup. spring.rabbitmq.listener.concurrency= # Minimum number of listener invoker threads for a `simple` container. - spring.rabbitmq.listener.direct.consumers-per-queue= # The number of Consumers per queue for a `direct` container. + spring.rabbitmq.listener.consumers-per-queue= # The number of Consumers per queue for a `direct` container. spring.rabbitmq.listener.default-requeue-rejected= # Whether or not to requeue delivery failures; default `true`. spring.rabbitmq.listener.idle-event-interval= # How often idle container events should be published in milliseconds. spring.rabbitmq.listener.max-concurrency= # Maximum number of listener invoker for a `simple` container. From 907c8e62dd7018913768486403ba2ca0a08becbb Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 2 May 2017 15:13:58 -0400 Subject: [PATCH 4/4] checkstyle fixes --- .../amqp/AbstractRabbitListenerContainerFactoryConfigurer.java | 3 ++- .../boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java | 1 - 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/AbstractRabbitListenerContainerFactoryConfigurer.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/AbstractRabbitListenerContainerFactoryConfigurer.java index 05a50ea6ce63..fc2baffa3c3e 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/AbstractRabbitListenerContainerFactoryConfigurer.java +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/AbstractRabbitListenerContainerFactoryConfigurer.java @@ -116,7 +116,8 @@ public final void configure(T factory, ConnectionFactory connectionFactory) { /** * Perform factory-specific configuration. * - * @param factory the factory + * @param factory the factory. + * @param rabbitProperties the properties. */ protected abstract void configure(T factory, RabbitProperties rabbitProperties); diff --git a/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java b/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java index 6d80320f6f32..613ecbe8ee25 100644 --- a/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java +++ b/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java @@ -20,7 +20,6 @@ import javax.net.ssl.SSLSocketFactory; import com.rabbitmq.client.Address; - import org.aopalliance.aop.Advice; import org.junit.After; import org.junit.Rule;