From 58df5c12e83e961f8338327aac253e3e0dfb4e7e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 13 Jul 2017 16:26:31 -0400 Subject: [PATCH] GH-69: Honor Boot Retry Properties in Producers Fixes #69 Add a `RetryTemplate` to producer `RabbitTemplate`s if `spring.rabbitmq.template.retry.enabled=true`. --- .../rabbit/RabbitMessageChannelBinder.java | 25 +++++++++++++ .../integration/RabbitBinderModuleTests.java | 36 +++++++++++++++++-- 2 files changed, 59 insertions(+), 2 deletions(-) diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 998dd4ec7..424e11641 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -41,7 +41,9 @@ import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor; import org.springframework.amqp.support.postprocessor.GZipPostProcessor; +import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; +import org.springframework.boot.autoconfigure.amqp.RabbitProperties.Retry; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; @@ -66,6 +68,10 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ErrorMessage; +import org.springframework.retry.RetryPolicy; +import org.springframework.retry.backoff.ExponentialBackOffPolicy; +import org.springframework.retry.policy.SimpleRetryPolicy; +import org.springframework.retry.support.RetryTemplate; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -384,6 +390,13 @@ public class RabbitMessageChannelBinder } private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties) { + RabbitProperties rabbitProperties = null; + try { + rabbitProperties = getApplicationContext().getBean(RabbitProperties.class); + } + catch (NoSuchBeanDefinitionException e) { + logger.debug("No RabbitProperties in context; no producer retry will be configured"); + } RabbitTemplate rabbitTemplate; if (properties.isBatchingEnabled()) { BatchingStrategy batchingStrategy = new SimpleBatchingStrategy( @@ -402,6 +415,18 @@ public class RabbitMessageChannelBinder rabbitTemplate.setBeforePublishPostProcessors(this.compressingPostProcessor); } rabbitTemplate.setChannelTransacted(properties.isTransacted()); + if (rabbitProperties != null && rabbitProperties.getTemplate().getRetry().isEnabled()) { + Retry retry = rabbitProperties.getTemplate().getRetry(); + RetryPolicy retryPolicy = new SimpleRetryPolicy(retry.getMaxAttempts()); + ExponentialBackOffPolicy backOff = new ExponentialBackOffPolicy(); + backOff.setInitialInterval(retry.getInitialInterval()); + backOff.setMultiplier(retry.getMultiplier()); + backOff.setMaxInterval(retry.getMaxInterval()); + RetryTemplate retryTemplate = new RetryTemplate(); + retryTemplate.setRetryPolicy(retryPolicy); + retryTemplate.setBackOffPolicy(backOff); + rabbitTemplate.setRetryTemplate(retryTemplate); + } rabbitTemplate.afterPropertiesSet(); return rabbitTemplate; } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index 9f45816d9..e67b5a8cf 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -40,13 +40,21 @@ import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.messaging.Processor; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.integration.channel.DirectChannel; import org.springframework.messaging.MessageChannel; +import org.springframework.retry.backoff.ExponentialBackOffPolicy; +import org.springframework.retry.policy.SimpleRetryPolicy; +import org.springframework.retry.support.RetryTemplate; import static org.assertj.core.api.Assertions.assertThat; @@ -165,16 +173,26 @@ public class RabbitBinderModuleTests { } @Test - public void testParentConnectionFactoryNotInheritedByCustomizedBinders() { + public void testParentConnectionFactoryNotInheritedByCustomizedBindersAndProducerRetryBootProperties() { List params = new ArrayList<>(); params.add("--spring.cloud.stream.input.binder=custom"); params.add("--spring.cloud.stream.output.binder=custom"); params.add("--spring.cloud.stream.binders.custom.type=rabbit"); params.add("--spring.cloud.stream.binders.custom.environment.foo=bar"); params.add("--server.port=0"); + params.add("--spring.rabbitmq.template.retry.enabled=true"); + params.add("--spring.rabbitmq.template.retry.maxAttempts=2"); + params.add("--spring.rabbitmq.template.retry.initial-interval=1000"); + params.add("--spring.rabbitmq.template.retry.multiplier=1.1"); + params.add("--spring.rabbitmq.template.retry.max-interval=3000"); context = SpringApplication.run(SimpleProcessor.class, params.toArray(new String[params.size()])); BinderFactory binderFactory = context.getBean(BinderFactory.class); - Binder binder = binderFactory.getBinder(null, MessageChannel.class); + @SuppressWarnings("unchecked") + Binder, + ExtendedProducerProperties> binder = + (Binder, + ExtendedProducerProperties>) binderFactory + .getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor @@ -190,6 +208,20 @@ public class RabbitBinderModuleTests { .getPropertyValue("indicators"); assertThat(healthIndicators).containsKey("custom"); assertThat(healthIndicators.get("custom").health().getStatus()).isEqualTo(Status.UP); + Binding binding = binder.bindProducer("foo", new DirectChannel(), + new ExtendedProducerProperties<>(new RabbitProducerProperties())); + RetryTemplate template = TestUtils.getPropertyValue(binding, "lifecycle.amqpTemplate.retryTemplate", + RetryTemplate.class); + assertThat(template).isNotNull(); + SimpleRetryPolicy retryPolicy = TestUtils.getPropertyValue(template, "retryPolicy", SimpleRetryPolicy.class); + ExponentialBackOffPolicy backOff = TestUtils.getPropertyValue(template, "backOffPolicy", + ExponentialBackOffPolicy.class); + assertThat(retryPolicy.getMaxAttempts()).isEqualTo(2); + assertThat(backOff.getInitialInterval()).isEqualTo(1000L); + assertThat(backOff.getMultiplier()).isEqualTo(1.1); + assertThat(backOff.getMaxInterval()).isEqualTo(3000L); + binding.unbind(); + context.close(); } @EnableBinding(Processor.class)