diff --git a/spring-cloud-stream-binder-rabbit/pom.xml b/spring-cloud-stream-binder-rabbit/pom.xml index 7e141f698..034201032 100644 --- a/spring-cloud-stream-binder-rabbit/pom.xml +++ b/spring-cloud-stream-binder-rabbit/pom.xml @@ -38,6 +38,12 @@ spring-cloud-connectors-core true + + org.springframework.cloud + spring-cloud-spring-service-connector + true + provided + org.springframework.boot spring-boot-starter-amqp diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index 86479a230..e80415adb 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -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 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 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; } diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java index 434d5a9f8..576387c4d 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java @@ -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 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(); + } + } 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 1c7cc4fae..7b896e86c 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 @@ -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);