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

Add ServiceConnectorConfig injection support

Browse Source 
The SC-Connectors `ServiceConnector` can be customized via
`ServiceConnectorConfig`.

* Allow end-user to specify such a bean and support its injection in the
`rabbitConnectionFactory` and `producerConnectionFactory` bean definitions

**Cherry-pick to 1.3.x**

Add `spring-cloud-spring-service-connector` dependency to be able to
request from the application context only specific `RabbitConnectionFactoryConfig`.
Use `ObjectProvider.getIfUnique()` for the case when we have not a single
`RabbitConnectionFactoryConfig` bean in the ctx

* Configure Cloud CCF via Boot's `RabbitProperties`
* Obtain from the Cloud only one CF service
This commit is contained in:
Artem Bilan
2017-12-19 10:01:30 -05:00
committed by Gary Russell
parent 7f4580fba1
commit 7cf4ccca25
4 changed files with 71 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-cloud-stream-binder-rabbit/pom.xml
View File

@@ -38,6 +38,12 @@
<artifactId>spring-cloud-connectors-core</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-spring-service-connector</artifactId>
<optional>true</optional>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>

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

@@ -24,9 +24,7 @@ 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.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -68,13 +66,10 @@ public class RabbitMessageChannelBinderConfiguration {
private RabbitExtendedBindingProperties rabbitExtendedBindingProperties;
@Bean
RabbitMessageChannelBinder rabbitMessageChannelBinder(
@Qualifier("producerConnectionFactory") ObjectProvider<ConnectionFactory> producerConnectionFactory)
throws Exception {
RabbitMessageChannelBinder rabbitMessageChannelBinder() throws Exception {
RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(this.rabbitConnectionFactory,
this.rabbitProperties, provisioningProvider());
binder.setProducerConnectionFactory(obtainProducerConnectionFactory(producerConnectionFactory));
binder.setProducerConnectionFactory(buildProducerConnectionFactory());
binder.setAdminAddresses(this.rabbitBinderConfigurationProperties.getAdminAddresses());
binder.setCompressingPostProcessor(gZipPostProcessor());
binder.setDecompressingPostProcessor(deCompressingPostProcessor());
@@ -83,25 +78,14 @@ public class RabbitMessageChannelBinderConfiguration {
return binder;
}
private ConnectionFactory obtainProducerConnectionFactory(ObjectProvider<ConnectionFactory> connectionFactoryObjectProvider) {
return connectionFactoryObjectProvider.getIfAvailable(() -> {
try {
return buildProducerConnectionFactory();
}
catch (Exception e) {
throw new IllegalStateException(e);
}
});
}
/**
* @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();
rabbitConnectionFactory = ((CachingConnectionFactory) this.rabbitConnectionFactory)
.getRabbitConnectionFactory();
}
else {
RabbitConnectionFactoryBean factory = new RabbitConnectionFactoryBean();
@@ -122,9 +106,9 @@ public class RabbitMessageChannelBinderConfiguration {
if (vHost != null) {
factory.setVirtualHost(vHost);
}
Duration requestedHeartbeatDuration = this.rabbitProperties.getRequestedHeartbeat();
Duration requestedHeartbeatDuration = this.rabbitProperties.getRequestedHeartbeat();
if (requestedHeartbeatDuration != null) {
factory.setRequestedHeartbeat((int)requestedHeartbeatDuration.getSeconds());
factory.setRequestedHeartbeat((int) requestedHeartbeatDuration.getSeconds());
}
RabbitProperties.Ssl ssl = this.rabbitProperties.getSsl();
if (ssl.isEnabled()) {
@@ -139,7 +123,7 @@ public class RabbitMessageChannelBinderConfiguration {
}
Duration connectionTimeoutDuration = this.rabbitProperties.getConnectionTimeout();
if (connectionTimeoutDuration != null) {
factory.setConnectionTimeout((int)connectionTimeoutDuration.getSeconds());
factory.setConnectionTimeout((int) connectionTimeoutDuration.getSeconds());
}
factory.afterPropertiesSet();
@@ -147,26 +131,10 @@ public class RabbitMessageChannelBinderConfiguration {
}
CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitConnectionFactory);
connectionFactory.setAddresses(this.rabbitProperties.determineAddresses());
connectionFactory.setPublisherConfirms(this.rabbitProperties.isPublisherConfirms());
connectionFactory.setPublisherReturns(this.rabbitProperties.isPublisherReturns());
if (this.rabbitProperties.getCache().getChannel().getSize() != null) {
connectionFactory.setChannelCacheSize(this.rabbitProperties.getCache().getChannel().getSize());
}
if (this.rabbitProperties.getCache().getConnection().getMode() != null) {
connectionFactory.setCacheMode(this.rabbitProperties.getCache().getConnection().getMode());
}
if (this.rabbitProperties.getCache().getConnection().getSize() != null) {
connectionFactory.setConnectionCacheSize(
this.rabbitProperties.getCache().getConnection().getSize());
}
if (this.rabbitProperties.getCache().getChannel().getCheckoutTimeout() != null) {
connectionFactory.setChannelCheckoutTimeout(
this.rabbitProperties.getCache().getChannel().getCheckoutTimeout());
}
connectionFactory.setApplicationContext(this.applicationContext);
this.applicationContext.addApplicationListener(connectionFactory);
connectionFactory.afterPropertiesSet();
RabbitServiceAutoConfiguration.configureCachingConnectionFactory(connectionFactory, this.applicationContext,
this.rabbitProperties);
return connectionFactory;
}

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

@@ -16,8 +16,10 @@
package org.springframework.cloud.stream.binder.rabbit.config;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.actuate.amqp.RabbitHealthIndicator;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
@@ -29,7 +31,9 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.Cloud;
import org.springframework.cloud.CloudFactory;
import org.springframework.cloud.service.messaging.RabbitConnectionFactoryConfig;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
@@ -50,7 +54,8 @@ import org.springframework.context.annotation.Profile;
*/
@Configuration
@ConditionalOnMissingBean(Binder.class)
@Import({RabbitMessageChannelBinderConfiguration.class, RabbitServiceAutoConfiguration.RabbitHealthIndicatorConfiguration.class})
@Import({ RabbitMessageChannelBinderConfiguration.class,
RabbitServiceAutoConfiguration.RabbitHealthIndicatorConfiguration.class })
public class RabbitServiceAutoConfiguration {
/**
@@ -61,8 +66,8 @@ public class RabbitServiceAutoConfiguration {
protected static class CloudProfile {
/**
* Configuration to be used when the cloud profile is set, and Cloud Connectors
* are found on the classpath.
* Configuration to be used when the cloud profile is set, and Cloud Connectors are found
* on the classpath.
*/
@Configuration
@ConditionalOnClass(Cloud.class)
@@ -75,12 +80,11 @@ public class RabbitServiceAutoConfiguration {
}
/**
* Active only if {@code spring.cloud.stream.overrideCloudConnectors} is not
* set to {@code true}.
* Active only if {@code spring.cloud.stream.overrideCloudConnectors} is not set to
* {@code true}.
*/
@Configuration
@ConditionalOnProperty(value = "spring.cloud.stream.overrideCloudConnectors",
havingValue = "false", matchIfMissing = true)
@ConditionalOnProperty(value = "spring.cloud.stream.overrideCloudConnectors", havingValue = "false", matchIfMissing = true)
// Required to parse Rabbit properties which are passed to the binder for
// clustering. We need to enable it here explicitly as the default Rabbit
// configuration is not triggered.
@@ -88,27 +92,26 @@ public class RabbitServiceAutoConfiguration {
protected static class UseCloudConnectors {
/**
* Creates a {@link ConnectionFactory} using the singleton service
* connector.
*
* Creates a {@link ConnectionFactory} using the singleton service connector.
* @param cloud {@link Cloud} instance to be used for accessing services.
* @param connectorConfigObjectProvider the {@link ObjectProvider} for the
* {@link RabbitConnectionFactoryConfig}.
* @return the {@link ConnectionFactory} used by the binder.
*/
@Bean
@Primary
ConnectionFactory rabbitConnectionFactory(Cloud cloud) {
return cloud.getSingletonServiceConnector(ConnectionFactory.class, null);
}
ConnectionFactory rabbitConnectionFactory(Cloud cloud,
ObjectProvider<RabbitConnectionFactoryConfig> connectorConfigObjectProvider,
ConfigurableApplicationContext applicationContext,
RabbitProperties rabbitProperties) throws Exception {
/**
* Creates a {@link ConnectionFactory} for non-transactional producers
* using the singleton service connector.
* @param cloud {@link Cloud} instance to be used for accessing services.
* @return the {@link ConnectionFactory} used by the binder for non-transactional producers.
*/
@Bean
ConnectionFactory producerConnectionFactory(Cloud cloud) {
return cloud.getSingletonServiceConnector(ConnectionFactory.class, null);
ConnectionFactory connectionFactory = cloud.getSingletonServiceConnector(ConnectionFactory.class,
connectorConfigObjectProvider.getIfUnique());
configureCachingConnectionFactory((CachingConnectionFactory) connectionFactory,
applicationContext, rabbitProperties);
return connectionFactory;
}
@Bean
@@ -119,9 +122,8 @@ public class RabbitServiceAutoConfiguration {
}
/**
* Configuration to be used if
* {@code spring.cloud.stream.overrideCloudConnectors} is set to {@code true}.
* Defers to Spring Boot auto-configuration.
* Configuration to be used if {@code spring.cloud.stream.overrideCloudConnectors} is set
* to {@code true}. Defers to Spring Boot auto-configuration.
*/
@Configuration
@ConditionalOnProperty("spring.cloud.stream.overrideCloudConnectors")
@@ -162,4 +164,29 @@ public class RabbitServiceAutoConfiguration {
}
static void configureCachingConnectionFactory(CachingConnectionFactory connectionFactory,
ConfigurableApplicationContext applicationContext, RabbitProperties rabbitProperties) throws Exception {
connectionFactory.setAddresses(rabbitProperties.determineAddresses());
connectionFactory.setPublisherConfirms(rabbitProperties.isPublisherConfirms());
connectionFactory.setPublisherReturns(rabbitProperties.isPublisherReturns());
if (rabbitProperties.getCache().getChannel().getSize() != null) {
connectionFactory.setChannelCacheSize(rabbitProperties.getCache().getChannel().getSize());
}
if (rabbitProperties.getCache().getConnection().getMode() != null) {
connectionFactory.setCacheMode(rabbitProperties.getCache().getConnection().getMode());
}
if (rabbitProperties.getCache().getConnection().getSize() != null) {
connectionFactory.setConnectionCacheSize(
rabbitProperties.getCache().getConnection().getSize());
}
if (rabbitProperties.getCache().getChannel().getCheckoutTimeout() != null) {
connectionFactory.setChannelCheckoutTimeout(
rabbitProperties.getCache().getChannel().getCheckoutTimeout());
}
connectionFactory.setApplicationContext(applicationContext);
applicationContext.addApplicationListener(connectionFactory);
connectionFactory.afterPropertiesSet();
}
}

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

@@ -19,7 +19,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.times;
import static org.mockito.Mockito.verify;
import java.util.ArrayList;
@@ -262,7 +261,7 @@ public class RabbitBinderModuleTests {
Cloud cloud = this.context.getBean(Cloud.class);
verify(cloud, times(2)).getSingletonServiceConnector(ConnectionFactory.class, null);
verify(cloud).getSingletonServiceConnector(ConnectionFactory.class, null);
}
@EnableBinding(Processor.class)
@@ -286,7 +285,7 @@ public class RabbitBinderModuleTests {
public Cloud cloud() {
Cloud cloud = mock(Cloud.class);
willReturn(mock(ConnectionFactory.class), mock(ConnectionFactory.class))
willReturn(new CachingConnectionFactory())
.given(cloud)
.getSingletonServiceConnector(ConnectionFactory.class, null);