Commit e236b716 authored by Stephane Nicoll's avatar Stephane Nicoll

Move RabbitMQ Simple listener support to dedicated namespace

Move `spring.rabbitmq.listener.*` to `spring.rabbitmq.listener.simple.*`
in preparation for Spring AMQP 2.0 that supports different container
types.

Closes gh-9108
See gh-9055
parent 3d836f77
...@@ -22,6 +22,7 @@ import java.util.List; ...@@ -22,6 +22,7 @@ import java.util.List;
import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.DeprecatedConfigurationProperty;
import org.springframework.boot.context.properties.NestedConfigurationProperty; import org.springframework.boot.context.properties.NestedConfigurationProperty;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils; import org.springframework.util.StringUtils;
...@@ -466,6 +467,111 @@ public class RabbitProperties { ...@@ -466,6 +467,111 @@ public class RabbitProperties {
public static class Listener { public static class Listener {
@NestedConfigurationProperty
private final AmqpContainer simple = new AmqpContainer();
@DeprecatedConfigurationProperty(replacement = "spring.rabbitmq.listener.simple.auto-startup")
@Deprecated
public boolean isAutoStartup() {
return getSimple().isAutoStartup();
}
@Deprecated
public void setAutoStartup(boolean autoStartup) {
getSimple().setAutoStartup(autoStartup);
}
@DeprecatedConfigurationProperty(replacement = "spring.rabbitmq.listener.simple.acknowledge-mode")
@Deprecated
public AcknowledgeMode getAcknowledgeMode() {
return getSimple().getAcknowledgeMode();
}
@Deprecated
public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) {
getSimple().setAcknowledgeMode(acknowledgeMode);
}
@DeprecatedConfigurationProperty(replacement = "spring.rabbitmq.listener.simple.concurrency")
@Deprecated
public Integer getConcurrency() {
return getSimple().getConcurrency();
}
@Deprecated
public void setConcurrency(Integer concurrency) {
getSimple().setConcurrency(concurrency);
}
@DeprecatedConfigurationProperty(replacement = "spring.rabbitmq.listener.simple.max-concurrency")
@Deprecated
public Integer getMaxConcurrency() {
return getSimple().getMaxConcurrency();
}
@Deprecated
public void setMaxConcurrency(Integer maxConcurrency) {
getSimple().setMaxConcurrency(maxConcurrency);
}
@DeprecatedConfigurationProperty(replacement = "spring.rabbitmq.listener.simple.prefetch")
@Deprecated
public Integer getPrefetch() {
return getSimple().getPrefetch();
}
@Deprecated
public void setPrefetch(Integer prefetch) {
getSimple().setPrefetch(prefetch);
}
@DeprecatedConfigurationProperty(replacement = "spring.rabbitmq.listener.simple.transaction-size")
@Deprecated
public Integer getTransactionSize() {
return getSimple().getTransactionSize();
}
@Deprecated
public void setTransactionSize(Integer transactionSize) {
getSimple().setTransactionSize(transactionSize);
}
@DeprecatedConfigurationProperty(replacement = "spring.rabbitmq.listener.simple.default-requeue-rejected")
@Deprecated
public Boolean getDefaultRequeueRejected() {
return getSimple().getDefaultRequeueRejected();
}
@Deprecated
public void setDefaultRequeueRejected(Boolean defaultRequeueRejected) {
getSimple().setDefaultRequeueRejected(defaultRequeueRejected);
}
@DeprecatedConfigurationProperty(replacement = "spring.rabbitmq.listener.simple.idle-event-interval")
@Deprecated
public Long getIdleEventInterval() {
return getSimple().getIdleEventInterval();
}
@Deprecated
public void setIdleEventInterval(Long idleEventInterval) {
getSimple().setIdleEventInterval(idleEventInterval);
}
@DeprecatedConfigurationProperty(replacement = "spring.rabbitmq.listener.simple.retry")
@Deprecated
public ListenerRetry getRetry() {
return getSimple().getRetry();
}
public AmqpContainer getSimple() {
return this.simple;
}
}
public static class AmqpContainer {
/** /**
* Start the container automatically on startup. * Start the container automatically on startup.
*/ */
......
...@@ -81,30 +81,31 @@ public final class SimpleRabbitListenerContainerFactoryConfigurer { ...@@ -81,30 +81,31 @@ public final class SimpleRabbitListenerContainerFactoryConfigurer {
if (this.messageConverter != null) { if (this.messageConverter != null) {
factory.setMessageConverter(this.messageConverter); factory.setMessageConverter(this.messageConverter);
} }
RabbitProperties.Listener listenerConfig = this.rabbitProperties.getListener(); RabbitProperties.AmqpContainer config = this.rabbitProperties.getListener()
factory.setAutoStartup(listenerConfig.isAutoStartup()); .getSimple();
if (listenerConfig.getAcknowledgeMode() != null) { factory.setAutoStartup(config.isAutoStartup());
factory.setAcknowledgeMode(listenerConfig.getAcknowledgeMode()); if (config.getAcknowledgeMode() != null) {
factory.setAcknowledgeMode(config.getAcknowledgeMode());
} }
if (listenerConfig.getConcurrency() != null) { if (config.getConcurrency() != null) {
factory.setConcurrentConsumers(listenerConfig.getConcurrency()); factory.setConcurrentConsumers(config.getConcurrency());
} }
if (listenerConfig.getMaxConcurrency() != null) { if (config.getMaxConcurrency() != null) {
factory.setMaxConcurrentConsumers(listenerConfig.getMaxConcurrency()); factory.setMaxConcurrentConsumers(config.getMaxConcurrency());
} }
if (listenerConfig.getPrefetch() != null) { if (config.getPrefetch() != null) {
factory.setPrefetchCount(listenerConfig.getPrefetch()); factory.setPrefetchCount(config.getPrefetch());
} }
if (listenerConfig.getTransactionSize() != null) { if (config.getTransactionSize() != null) {
factory.setTxSize(listenerConfig.getTransactionSize()); factory.setTxSize(config.getTransactionSize());
} }
if (listenerConfig.getDefaultRequeueRejected() != null) { if (config.getDefaultRequeueRejected() != null) {
factory.setDefaultRequeueRejected(listenerConfig.getDefaultRequeueRejected()); factory.setDefaultRequeueRejected(config.getDefaultRequeueRejected());
} }
if (listenerConfig.getIdleEventInterval() != null) { if (config.getIdleEventInterval() != null) {
factory.setIdleEventInterval(listenerConfig.getIdleEventInterval()); factory.setIdleEventInterval(config.getIdleEventInterval());
} }
ListenerRetry retryConfig = listenerConfig.getRetry(); ListenerRetry retryConfig = config.getRetry();
if (retryConfig.isEnabled()) { if (retryConfig.isEnabled()) {
RetryInterceptorBuilder<?> builder = (retryConfig.isStateless() RetryInterceptorBuilder<?> builder = (retryConfig.isStateless()
? RetryInterceptorBuilder.stateless() ? RetryInterceptorBuilder.stateless()
......
...@@ -351,6 +351,72 @@ ...@@ -351,6 +351,72 @@
"description": "Create an AmqpAdmin bean.", "description": "Create an AmqpAdmin bean.",
"defaultValue": true "defaultValue": true
}, },
{
"name": "spring.rabbitmq.listener.retry.enabled",
"type": "java.lang.Boolean",
"description": "Whether or not publishing retries are enabled.",
"sourceType": "org.springframework.boot.autoconfigure.amqp.RabbitProperties$ListenerRetry",
"defaultValue": false,
"deprecated": true,
"deprecation": {
"replacement": "spring.rabbitmq.listener.simple.retry.enabled"
}
},
{
"name": "spring.rabbitmq.listener.retry.initial-interval",
"type": "java.lang.Long",
"description": "Interval between the first and second attempt to publish or deliver a message.",
"sourceType": "org.springframework.boot.autoconfigure.amqp.RabbitProperties$ListenerRetry",
"defaultValue": 1000,
"deprecated": true,
"deprecation": {
"replacement": "spring.rabbitmq.listener.simple.retry.initial-interval"
}
},
{
"name": "spring.rabbitmq.listener.retry.max-attempts",
"type": "java.lang.Integer",
"description": "Maximum number of attempts to publish or deliver a message.",
"sourceType": "org.springframework.boot.autoconfigure.amqp.RabbitProperties$ListenerRetry",
"defaultValue": 3,
"deprecated": true,
"deprecation": {
"replacement": "spring.rabbitmq.listener.simple.retry.max-attempts"
}
},
{
"name": "spring.rabbitmq.listener.retry.max-interval",
"type": "java.lang.Long",
"description": "Maximum interval between attempts.",
"sourceType": "org.springframework.boot.autoconfigure.amqp.RabbitProperties$ListenerRetry",
"defaultValue": 10000,
"deprecated": true,
"deprecation": {
"replacement": "spring.rabbitmq.listener.simple.retry.max-interval"
}
},
{
"name": "spring.rabbitmq.listener.retry.multiplier",
"type": "java.lang.Double",
"description": "A multiplier to apply to the previous retry interval.",
"sourceType": "org.springframework.boot.autoconfigure.amqp.RabbitProperties$ListenerRetry",
"defaultValue": 1,
"deprecated": true,
"deprecation": {
"replacement": "spring.rabbitmq.listener.simple.retry.multiplier"
}
},
{
"name": "spring.rabbitmq.listener.retry.stateless",
"type": "java.lang.Boolean",
"description": "Whether or not retries are stateless or stateful.",
"sourceType": "org.springframework.boot.autoconfigure.amqp.RabbitProperties$ListenerRetry",
"defaultValue": true,
"deprecated": true,
"deprecation": {
"replacement": "spring.rabbitmq.listener.simple.retry.stateless"
}
},
{ {
"name": "spring.session.hazelcast.flush-mode", "name": "spring.session.hazelcast.flush-mode",
"defaultValue": "on-save" "defaultValue": "on-save"
......
...@@ -293,9 +293,9 @@ public class RabbitAutoConfigurationTests { ...@@ -293,9 +293,9 @@ public class RabbitAutoConfigurationTests {
} }
@Test @Test
public void testRabbitListenerContainerFactoryWithCustomSettings() { @Deprecated
load(new Class<?>[] { MessageConvertersConfiguration.class, public void testSimpleRabbitListenerContainerFactoryWithCustomDeprecatedSettings() {
MessageRecoverersConfiguration.class }, testSimpleRabbitListenerContainerFactoryWithCustomSettings(
"spring.rabbitmq.listener.retry.enabled:true", "spring.rabbitmq.listener.retry.enabled:true",
"spring.rabbitmq.listener.retry.maxAttempts:4", "spring.rabbitmq.listener.retry.maxAttempts:4",
"spring.rabbitmq.listener.retry.initialInterval:2000", "spring.rabbitmq.listener.retry.initialInterval:2000",
...@@ -309,10 +309,37 @@ public class RabbitAutoConfigurationTests { ...@@ -309,10 +309,37 @@ public class RabbitAutoConfigurationTests {
"spring.rabbitmq.listener.defaultRequeueRejected:false", "spring.rabbitmq.listener.defaultRequeueRejected:false",
"spring.rabbitmq.listener.idleEventInterval:5", "spring.rabbitmq.listener.idleEventInterval:5",
"spring.rabbitmq.listener.transactionSize:20"); "spring.rabbitmq.listener.transactionSize:20");
}
@Test
public void testSimpleRabbitListenerContainerFactoryWithCustomSettings() {
testSimpleRabbitListenerContainerFactoryWithCustomSettings(
"spring.rabbitmq.listener.simple.retry.enabled:true",
"spring.rabbitmq.listener.simple.retry.maxAttempts:4",
"spring.rabbitmq.listener.simple.retry.initialInterval:2000",
"spring.rabbitmq.listener.simple.retry.multiplier:1.5",
"spring.rabbitmq.listener.simple.retry.maxInterval:5000",
"spring.rabbitmq.listener.simple.autoStartup:false",
"spring.rabbitmq.listener.simple.acknowledgeMode:manual",
"spring.rabbitmq.listener.simple.concurrency:5",
"spring.rabbitmq.listener.simple.maxConcurrency:10",
"spring.rabbitmq.listener.simple.prefetch:40",
"spring.rabbitmq.listener.simple.defaultRequeueRejected:false",
"spring.rabbitmq.listener.simple.idleEventInterval:5",
"spring.rabbitmq.listener.simple.transactionSize:20");
}
private void testSimpleRabbitListenerContainerFactoryWithCustomSettings(String... environment) {
load(new Class<?>[] { MessageConvertersConfiguration.class,
MessageRecoverersConfiguration.class }, environment);
SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory = this.context SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory = this.context
.getBean("rabbitListenerContainerFactory", .getBean("rabbitListenerContainerFactory",
SimpleRabbitListenerContainerFactory.class); SimpleRabbitListenerContainerFactory.class);
DirectFieldAccessor dfa = new DirectFieldAccessor(rabbitListenerContainerFactory); DirectFieldAccessor dfa = new DirectFieldAccessor(rabbitListenerContainerFactory);
checkCommonProps(dfa);
}
private void checkCommonProps(DirectFieldAccessor dfa) {
assertThat(dfa.getPropertyValue("autoStartup")).isEqualTo(Boolean.FALSE); assertThat(dfa.getPropertyValue("autoStartup")).isEqualTo(Boolean.FALSE);
assertThat(dfa.getPropertyValue("acknowledgeMode")) assertThat(dfa.getPropertyValue("acknowledgeMode"))
.isEqualTo(AcknowledgeMode.MANUAL); .isEqualTo(AcknowledgeMode.MANUAL);
......
...@@ -936,20 +936,20 @@ content into your application; rather pick only the properties that you need. ...@@ -936,20 +936,20 @@ content into your application; rather pick only the properties that you need.
spring.rabbitmq.connection-timeout= # Connection timeout, in milliseconds; zero for infinite. spring.rabbitmq.connection-timeout= # Connection timeout, in milliseconds; zero for infinite.
spring.rabbitmq.dynamic=true # Create an AmqpAdmin bean. spring.rabbitmq.dynamic=true # Create an AmqpAdmin bean.
spring.rabbitmq.host=localhost # RabbitMQ host. spring.rabbitmq.host=localhost # RabbitMQ host.
spring.rabbitmq.listener.acknowledge-mode= # Acknowledge mode of container. spring.rabbitmq.listener.simple.acknowledge-mode= # Acknowledge mode of container.
spring.rabbitmq.listener.auto-startup=true # Start the container automatically on startup. spring.rabbitmq.listener.simple.auto-startup=true # Start the container automatically on startup.
spring.rabbitmq.listener.concurrency= # Minimum number of consumers. spring.rabbitmq.listener.simple.concurrency= # Minimum number of consumers.
spring.rabbitmq.listener.default-requeue-rejected= # Whether or not to requeue delivery failures; default `true`. spring.rabbitmq.listener.simple.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.simple.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.simple.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.simple.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.simple.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. spring.rabbitmq.listener.simple.retry.initial-interval=1000 # Interval between the first and second attempt to deliver a message.
spring.rabbitmq.listener.retry.max-attempts=3 # Maximum number of attempts to deliver a message. spring.rabbitmq.listener.simple.retry.max-attempts=3 # Maximum number of attempts to deliver a message.
spring.rabbitmq.listener.retry.max-interval=10000 # Maximum interval between attempts. spring.rabbitmq.listener.simple.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.simple.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.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.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.password= # Login to authenticate against the broker. spring.rabbitmq.password= # Login to authenticate against the broker.
spring.rabbitmq.port=5672 # RabbitMQ port. spring.rabbitmq.port=5672 # RabbitMQ port.
spring.rabbitmq.publisher-confirms=false # Enable publisher confirms. spring.rabbitmq.publisher-confirms=false # Enable publisher confirms.
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment