SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-69: Honor Boot Retry Properties in Producers

Browse Source 
Fixes #69

Add a `RetryTemplate` to producer `RabbitTemplate`s if `spring.rabbitmq.template.retry.enabled=true`.
This commit is contained in:
Gary Russell
2017-07-13 16:26:31 -04:00
committed by Artem Bilan
parent 4f1d975c10
commit 58df5c12e8
2 changed files with 59 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -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;
}

36
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java
View File

@@ -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<String> 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<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>,
ExtendedProducerProperties<RabbitProducerProperties>> binder =
(Binder<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>,
ExtendedProducerProperties<RabbitProducerProperties>>) 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<MessageChannel> 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)