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

GH-130: Remove producerConnectionFactory

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/130

Remove the creation of a separate `producerConnectionFactory`; use the `RabbitTemplate`'s
built-in 'usePublisherConnection` instead.

Fix `Duration` retry properties.
This commit is contained in:
Gary Russell
2018-02-27 11:34:20 -05:00
committed by Artem Bilan
parent 306796c8d9
commit f7446deddb
5 changed files with 30 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -129,8 +129,6 @@ public class RabbitMessageChannelBinder
private ConnectionFactory connectionFactory;
private ConnectionFactory producerConnectionFactory;
private MessagePostProcessor decompressingPostProcessor = new DelegatingDecompressingPostProcessor();
private MessagePostProcessor compressingPostProcessor = new GZipPostProcessor();
@@ -152,15 +150,6 @@ public class RabbitMessageChannelBinder
this.rabbitProperties = rabbitProperties;
}
/**
* Specify a distinct {@link ConnectionFactory} for the non-transactional producers to avoid dead locks
* on blocked connections.
* @param producerConnectionFactory the ConnectionFactory to use for non-transactional producers.
*/
public void setProducerConnectionFactory(ConnectionFactory producerConnectionFactory) {
this.producerConnectionFactory = producerConnectionFactory;
}
/**
* Set a {@link MessagePostProcessor} to decompress messages. Defaults to a
* {@link DelegatingDecompressingPostProcessor} with its default delegates.
@@ -219,7 +208,6 @@ public class RabbitMessageChannelBinder
if (this.destroyConnectionFactory) {
((DisposableBean) this.connectionFactory).destroy();
}
((DisposableBean) this.producerConnectionFactory).destroy();
}
}
@@ -439,7 +427,11 @@ public class RabbitMessageChannelBinder
return new MessageHandler() {
private final RabbitTemplate template = new RabbitTemplate(
RabbitMessageChannelBinder.this.producerConnectionFactory);
RabbitMessageChannelBinder.this.connectionFactory);
{
this.template.setUsePublisherConnection(true);
}
private final String exchange = deadLetterExchangeName(properties.getExtension());
@@ -584,12 +576,8 @@ public class RabbitMessageChannelBinder
rabbitTemplate = new RabbitTemplate();
}
rabbitTemplate.setChannelTransacted(properties.isTransacted());
if (rabbitTemplate.isChannelTransacted()) {
rabbitTemplate.setConnectionFactory(this.connectionFactory);
}
else {
rabbitTemplate.setConnectionFactory(this.producerConnectionFactory);
}
rabbitTemplate.setConnectionFactory(this.connectionFactory);
rabbitTemplate.setUsePublisherConnection(true);
if (properties.isCompress()) {
rabbitTemplate.setBeforePublishPostProcessors(this.compressingPostProcessor);
}
@@ -598,9 +586,9 @@ public class RabbitMessageChannelBinder
Retry retry = rabbitProperties.getTemplate().getRetry();
RetryPolicy retryPolicy = new SimpleRetryPolicy(retry.getMaxAttempts());
ExponentialBackOffPolicy backOff = new ExponentialBackOffPolicy();
backOff.setInitialInterval(retry.getInitialInterval());
backOff.setInitialInterval(retry.getInitialInterval().toMillis());
backOff.setMultiplier(retry.getMultiplier());
backOff.setMaxInterval(retry.getMaxInterval());
backOff.setMaxInterval(retry.getMaxInterval().toMillis());
RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setRetryPolicy(retryPolicy);
retryTemplate.setBackOffPolicy(backOff);

68
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2017 the original author or authors.
* Copyright 2015-2018 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.
@@ -16,12 +16,8 @@
package org.springframework.cloud.stream.binder.rabbit.config;
import java.time.Duration;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean;
import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -44,6 +40,7 @@ import org.springframework.context.annotation.Import;
* @author Vinicius Carvalho
* @author Artem Bilan
* @author Oleg Zhurakousky
* @author Gary Russell
*/
@Configuration
@Import({ PropertyPlaceholderAutoConfiguration.class })
@@ -69,7 +66,6 @@ public class RabbitMessageChannelBinderConfiguration {
RabbitMessageChannelBinder rabbitMessageChannelBinder() throws Exception {
RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(this.rabbitConnectionFactory,
this.rabbitProperties, provisioningProvider());
binder.setProducerConnectionFactory(buildProducerConnectionFactory());
binder.setAdminAddresses(this.rabbitBinderConfigurationProperties.getAdminAddresses());
binder.setCompressingPostProcessor(gZipPostProcessor());
binder.setDecompressingPostProcessor(deCompressingPostProcessor());
@@ -78,66 +74,6 @@ public class RabbitMessageChannelBinderConfiguration {
return binder;
}
/**
* @see org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration.RabbitConnectionFactoryCreator
*/
private CachingConnectionFactory buildProducerConnectionFactory() throws Exception {
com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory;
if (this.rabbitConnectionFactory instanceof CachingConnectionFactory) {
rabbitConnectionFactory = ((CachingConnectionFactory) this.rabbitConnectionFactory)
.getRabbitConnectionFactory();
}
else {
RabbitConnectionFactoryBean factory = new RabbitConnectionFactoryBean();
String host = this.rabbitProperties.determineHost();
if (host != null) {
factory.setHost(host);
}
factory.setPort(this.rabbitProperties.determinePort());
String user = this.rabbitProperties.determineUsername();
if (user != null) {
factory.setUsername(user);
}
String password = this.rabbitProperties.determinePassword();
if (password != null) {
factory.setPassword(password);
}
String vHost = this.rabbitProperties.determineVirtualHost();
if (vHost != null) {
factory.setVirtualHost(vHost);
}
Duration requestedHeartbeatDuration = this.rabbitProperties.getRequestedHeartbeat();
if (requestedHeartbeatDuration != null) {
factory.setRequestedHeartbeat((int) requestedHeartbeatDuration.getSeconds());
}
RabbitProperties.Ssl ssl = this.rabbitProperties.getSsl();
if (ssl.isEnabled()) {
factory.setUseSSL(true);
if (ssl.getAlgorithm() != null) {
factory.setSslAlgorithm(ssl.getAlgorithm());
}
factory.setKeyStore(ssl.getKeyStore());
factory.setKeyStorePassphrase(ssl.getKeyStorePassword());
factory.setTrustStore(ssl.getTrustStore());
factory.setTrustStorePassphrase(ssl.getTrustStorePassword());
}
Duration connectionTimeoutDuration = this.rabbitProperties.getConnectionTimeout();
if (connectionTimeoutDuration != null) {
factory.setConnectionTimeout((int) connectionTimeoutDuration.getSeconds());
}
factory.afterPropertiesSet();
rabbitConnectionFactory = factory.getObject();
}
CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitConnectionFactory);
RabbitServiceAutoConfiguration.configureCachingConnectionFactory(connectionFactory, this.applicationContext,
this.rabbitProperties);
return connectionFactory;
}
@Bean
MessagePostProcessor deCompressingPostProcessor() {
return new DelegatingDecompressingPostProcessor();

5
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2017 the original author or authors.
* Copyright 2015-2018 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.
@@ -52,6 +52,7 @@ import org.springframework.util.StringUtils;
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Artem Bilan
* @author Gary Russell
*/
@Configuration
@ConditionalOnMissingBean(Binder.class)
@@ -186,7 +187,7 @@ public class RabbitServiceAutoConfiguration {
}
if (rabbitProperties.getCache().getChannel().getCheckoutTimeout() != null) {
connectionFactory.setChannelCheckoutTimeout(
rabbitProperties.getCache().getChannel().getCheckoutTimeout());
rabbitProperties.getCache().getChannel().getCheckoutTimeout().toMillis());
}
connectionFactory.setApplicationContext(applicationContext);
applicationContext.addApplicationListener(connectionFactory);

1
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java
View File

@@ -67,7 +67,6 @@ public class RabbitTestBinder extends
public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitMessageChannelBinder binder) {
this.applicationContext = new AnnotationConfigApplicationContext(Config.class);
binder.setApplicationContext(this.applicationContext);
binder.setProducerConnectionFactory(connectionFactory);
this.setPollableConsumerBinder(binder);
this.rabbitAdmin = new RabbitAdmin(connectionFactory);
}

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

@@ -16,11 +16,6 @@
package org.springframework.cloud.stream.binder.rabbit.integration;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.BDDMockito.willReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -59,10 +54,16 @@ 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.messaging.support.GenericMessage;
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;
import static org.mockito.BDDMockito.willReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
/**
* @author Marius Bogoevici
* @author Gary Russell
@@ -106,10 +107,6 @@ public class RabbitBinderModuleTests {
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory).isSameAs(connectionFactory);
ConnectionFactory producerConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("producerConnectionFactory");
assertThat(producerConnectionFactory).isNotSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
@@ -119,6 +116,16 @@ public class RabbitBinderModuleTests {
.getPropertyValue("indicators");
assertThat(healthIndicators).containsKey(("rabbit"));
assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo((Status.UP));
ConnectionFactory publisherConnectionFactory = binderConnectionFactory.getPublisherConnectionFactory();
assertThat(TestUtils.getPropertyValue(publisherConnectionFactory, "connection.target")).isNull();
DirectChannel checkPf = new DirectChannel();
Binding<MessageChannel> binding = ((RabbitMessageChannelBinder) binder).bindProducer("checkPF", checkPf,
new ExtendedProducerProperties<>(
new RabbitProducerProperties()));
checkPf.send(new GenericMessage<>("foo".getBytes()));
binding.unbind();
assertThat(TestUtils.getPropertyValue(publisherConnectionFactory, "connection.target")).isNotNull();
}
@Test
@@ -254,16 +261,8 @@ public class RabbitBinderModuleTests {
assertThat(binderConnectionFactory).isNotSameAs(connectionFactory);
ConnectionFactory producerConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("producerConnectionFactory");
assertThat(producerConnectionFactory).isNotSameAs(connectionFactory);
assertThat(binderConnectionFactory).isNotSameAs(producerConnectionFactory);
assertThat(TestUtils.getPropertyValue(connectionFactory, "addresses")).isNotNull();
assertThat(TestUtils.getPropertyValue(binderConnectionFactory, "addresses")).isNull();
assertThat(TestUtils.getPropertyValue(producerConnectionFactory, "addresses")).isNull();
Cloud cloud = this.context.getBean(Cloud.class);