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/RabbitBinderConfiguration.java similarity index 95% rename from spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java rename to spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfiguration.java index 02fe668c1..9d2efd80f 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/RabbitBinderConfiguration.java @@ -23,7 +23,6 @@ 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; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; @@ -54,12 +53,13 @@ import org.springframework.util.StringUtils; * @author Ilayaperumal Gopinathan * @author Artem Bilan * @author Gary Russell + * @author Chris Bono */ @Configuration @ConditionalOnMissingBean(Binder.class) @Import({ RabbitMessageChannelBinderConfiguration.class, - RabbitServiceAutoConfiguration.RabbitHealthIndicatorConfiguration.class }) -public abstract class RabbitServiceAutoConfiguration { + RabbitBinderConfiguration.RabbitHealthIndicatorConfiguration.class }) +public abstract class RabbitBinderConfiguration { static void configureCachingConnectionFactory( CachingConnectionFactory connectionFactory, @@ -172,7 +172,7 @@ public abstract class RabbitServiceAutoConfiguration { */ @Configuration @ConditionalOnProperty("spring.cloud.stream.override-cloud-connectors") - @Import(RabbitAutoConfiguration.class) + @Import(RabbitConfiguration.class) protected static class OverrideCloudConnectors { } @@ -181,7 +181,7 @@ public abstract class RabbitServiceAutoConfiguration { @Configuration @ConditionalOnMissingClass("org.springframework.cloud.Cloud") - @Import(RabbitAutoConfiguration.class) + @Import(RabbitConfiguration.class) protected static class NoCloudConnectors { } @@ -194,7 +194,7 @@ public abstract class RabbitServiceAutoConfiguration { */ @Configuration @Profile("!cloud") - @Import(RabbitAutoConfiguration.class) + @Import(RabbitConfiguration.class) protected static class NoCloudProfile { } diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitConfiguration.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitConfiguration.java new file mode 100644 index 000000000..1d9744677 --- /dev/null +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitConfiguration.java @@ -0,0 +1,115 @@ +/* + * Copyright 2015-2021 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.rabbit.config; + +import java.util.stream.Collectors; + +import com.rabbitmq.client.impl.CredentialsProvider; +import com.rabbitmq.client.impl.CredentialsRefreshService; + +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; +import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean; +import org.springframework.amqp.rabbit.core.RabbitOperations; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.boot.autoconfigure.amqp.CachingConnectionFactoryConfigurer; +import org.springframework.boot.autoconfigure.amqp.ConnectionFactoryCustomizer; +import org.springframework.boot.autoconfigure.amqp.RabbitConnectionFactoryBeanConfigurer; +import org.springframework.boot.autoconfigure.amqp.RabbitProperties; +import org.springframework.boot.autoconfigure.amqp.RabbitRetryTemplateCustomizer; +import org.springframework.boot.autoconfigure.amqp.RabbitTemplateConfigurer; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.io.ResourceLoader; + +/** + * Configuration for {@link RabbitTemplate} and {@link CachingConnectionFactory}. + * + * @author Chris Bono + * @since 3.2 + */ +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(RabbitProperties.class) +public class RabbitConfiguration { + + @Bean + @ConditionalOnMissingBean + RabbitConnectionFactoryBeanConfigurer rabbitConnectionFactoryBeanConfigurer(RabbitProperties properties, + ResourceLoader resourceLoader, ObjectProvider credentialsProvider, + ObjectProvider credentialsRefreshService) { + RabbitConnectionFactoryBeanConfigurer configurer = new RabbitConnectionFactoryBeanConfigurer(resourceLoader, properties); + configurer.setCredentialsProvider(credentialsProvider.getIfUnique()); + configurer.setCredentialsRefreshService(credentialsRefreshService.getIfUnique()); + return configurer; + } + + @Bean + @ConditionalOnMissingBean + CachingConnectionFactoryConfigurer rabbitConnectionFactoryConfigurer(RabbitProperties rabbitProperties, + ObjectProvider connectionNameStrategy) { + CachingConnectionFactoryConfigurer configurer = new CachingConnectionFactoryConfigurer(rabbitProperties); + configurer.setConnectionNameStrategy(connectionNameStrategy.getIfUnique()); + return configurer; + } + + @Bean + @ConditionalOnMissingBean(ConnectionFactory.class) + CachingConnectionFactory rabbitConnectionFactory( + RabbitConnectionFactoryBeanConfigurer rabbitConnectionFactoryBeanConfigurer, + CachingConnectionFactoryConfigurer rabbitCachingConnectionFactoryConfigurer, + ObjectProvider connectionFactoryCustomizers) throws Exception { + + RabbitConnectionFactoryBean connectionFactoryBean = new RabbitConnectionFactoryBean(); + rabbitConnectionFactoryBeanConfigurer.configure(connectionFactoryBean); + connectionFactoryBean.afterPropertiesSet(); + com.rabbitmq.client.ConnectionFactory connectionFactory = connectionFactoryBean.getObject(); + connectionFactoryCustomizers.orderedStream() + .forEach((customizer) -> customizer.customize(connectionFactory)); + + CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory); + rabbitCachingConnectionFactoryConfigurer.configure(cachingConnectionFactory); + + return cachingConnectionFactory; + } + + @Bean + @ConditionalOnMissingBean + public RabbitTemplateConfigurer rabbitTemplateConfigurer(RabbitProperties properties, + ObjectProvider messageConverter, + ObjectProvider retryTemplateCustomizers) { + RabbitTemplateConfigurer configurer = new RabbitTemplateConfigurer(properties); + configurer.setMessageConverter(messageConverter.getIfUnique()); + configurer.setRetryTemplateCustomizers(retryTemplateCustomizers.orderedStream().collect(Collectors.toList())); + return configurer; + } + + @Bean + @ConditionalOnSingleCandidate(ConnectionFactory.class) + @ConditionalOnMissingBean(RabbitOperations.class) + public RabbitTemplate rabbitTemplate(RabbitTemplateConfigurer configurer, ConnectionFactory connectionFactory) { + RabbitTemplate template = new RabbitTemplate(); + configurer.configure(template, connectionFactory); + return template; + } + +} diff --git a/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders b/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders index 7a65c4682..78f94972b 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders +++ b/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders @@ -1,2 +1,2 @@ rabbit:\ -org.springframework.cloud.stream.binder.rabbit.config.RabbitServiceAutoConfiguration +org.springframework.cloud.stream.binder.rabbit.config.RabbitBinderConfiguration