From 9c76eccfe3daf9dbf8a8292d707df1f9e37a4ee4 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 13 Dec 2017 11:17:09 -0500 Subject: [PATCH] Fix producer ConnectionFactory The `RabbitMessageChannelBinder` creates an internal distinct `ConnectionFactory` instance based on the `RabbitProperties` for the non-transactional producers to avoid dead locks when connection is blocked. In case of Cloud Connectors the `ConnectionFactory` bean is overridden, but not `RabbitProperties`. Therefore the original `ConnectionFactory` for consumers is good, cloud-based, but for producers it is still based on the `RabbitProperties` from Spring Boot, in most cases with default options. * Add one more `producerConnectionFactory` `@Bean` to the `CloudConnectors` configuration to obtain a distinct `ConnectionFactory` instance from the cloud provider * When we ara not in the cloud profile, create a fresh `ConnectionFactory` based on the `RabbitProperties` * Inject the extra `producerConnectionFactory` instance into the `RabbitMessageChannelBinder` instead of the internal non-stable solution **Cherry-pick to 1.3.x** --- .../rabbit/RabbitMessageChannelBinder.java | 101 +++------------ ...bbitMessageChannelBinderConfiguration.java | 122 ++++++++++++++++-- .../RabbitServiceAutoConfiguration.java | 31 ++++- .../binder/rabbit/RabbitTestBinder.java | 1 + .../integration/RabbitBinderModuleTests.java | 67 ++++++++-- 5 files changed, 211 insertions(+), 111 deletions(-) diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index d535fb105..b8600844e 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -30,7 +30,6 @@ import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory; -import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean; import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.support.BatchingStrategy; @@ -146,6 +145,15 @@ 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. @@ -180,14 +188,6 @@ public class RabbitMessageChannelBinder @Override public void onInit() throws Exception { super.onInit(); - - CachingConnectionFactory producerConnectionFactory = createProducerConnectionFactory(this.rabbitProperties); - producerConnectionFactory.setApplicationContext(getApplicationContext()); - getApplicationContext().addApplicationListener(producerConnectionFactory); - producerConnectionFactory.afterPropertiesSet(); - - this.producerConnectionFactory = producerConnectionFactory; - if (this.clustered) { String[] addresses = StringUtils.commaDelimitedListToStringArray(this.rabbitProperties.getAddresses()); @@ -205,74 +205,6 @@ public class RabbitMessageChannelBinder } } - /** - * @see org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration.RabbitConnectionFactoryCreator - */ - private CachingConnectionFactory createProducerConnectionFactory(RabbitProperties config) throws Exception { - com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory = null; - if (this.connectionFactory instanceof CachingConnectionFactory) { - rabbitConnectionFactory = ((CachingConnectionFactory) this.connectionFactory).getRabbitConnectionFactory(); - } - else { - RabbitConnectionFactoryBean factory = new RabbitConnectionFactoryBean(); - if (config.determineHost() != null) { - factory.setHost(config.determineHost()); - } - factory.setPort(config.determinePort()); - if (config.determineUsername() != null) { - factory.setUsername(config.determineUsername()); - } - if (config.determinePassword() != null) { - factory.setPassword(config.determinePassword()); - } - if (config.determineVirtualHost() != null) { - factory.setVirtualHost(config.determineVirtualHost()); - } - if (config.getRequestedHeartbeat() != null) { - factory.setRequestedHeartbeat((int)config.getRequestedHeartbeat().getSeconds()); - } - RabbitProperties.Ssl ssl = config.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()); - } - if (config.getConnectionTimeout() != null) { - factory.setConnectionTimeout((int)config.getConnectionTimeout().getSeconds()); - } - factory.afterPropertiesSet(); - - rabbitConnectionFactory = factory.getObject(); - } - - CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitConnectionFactory); - connectionFactory.setAddresses(config.determineAddresses()); - connectionFactory.setPublisherConfirms(config.isPublisherConfirms()); - connectionFactory.setPublisherReturns(config.isPublisherReturns()); - if (config.getCache().getChannel().getSize() != null) { - connectionFactory - .setChannelCacheSize(config.getCache().getChannel().getSize()); - } - if (config.getCache().getConnection().getMode() != null) { - connectionFactory - .setCacheMode(config.getCache().getConnection().getMode()); - } - if (config.getCache().getConnection().getSize() != null) { - connectionFactory.setConnectionCacheSize( - config.getCache().getConnection().getSize()); - } - if (config.getCache().getChannel().getCheckoutTimeout() != null) { - connectionFactory.setChannelCheckoutTimeout( - config.getCache().getChannel().getCheckoutTimeout()); - } - return connectionFactory; - } - @Override public void destroy() throws Exception { if (this.connectionFactory instanceof DisposableBean) { @@ -293,8 +225,7 @@ public class RabbitMessageChannelBinder @Override protected MessageHandler createProducerMessageHandler(final ProducerDestination producerDestination, - ExtendedProducerProperties producerProperties, MessageChannel errorChannel) - throws Exception { + ExtendedProducerProperties producerProperties, MessageChannel errorChannel) { Assert.state(!HeaderMode.embeddedHeaders.equals(producerProperties.getHeaderMode()), "the RabbitMQ binder does not support embedded headers since RabbitMQ supports headers natively"); String prefix = producerProperties.getExtension().getPrefix(); @@ -304,14 +235,14 @@ public class RabbitMessageChannelBinder buildRabbitTemplate(producerProperties.getExtension(), errorChannel != null)); endpoint.setExchangeName(producerDestination.getName()); RabbitProducerProperties extendedProperties = producerProperties.getExtension(); - boolean expresssionInterceptorNeeded = expresssionInterceptorNeeded(extendedProperties); + boolean expressionInterceptorNeeded = expressionInterceptorNeeded(extendedProperties); String routingKeyExpression = extendedProperties.getRoutingKeyExpression(); if (!producerProperties.isPartitioned()) { if (routingKeyExpression == null) { endpoint.setRoutingKey(destination); } else { - if (expresssionInterceptorNeeded) { + if (expressionInterceptorNeeded) { endpoint.setRoutingKeyExpressionString("headers['" + RabbitExpressionEvaluatingInterceptor.ROUTING_KEY_HEADER + "']"); } @@ -325,7 +256,7 @@ public class RabbitMessageChannelBinder endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(destination, false)); } else { - if (expresssionInterceptorNeeded) { + if (expressionInterceptorNeeded) { endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression("headers['" + RabbitExpressionEvaluatingInterceptor.ROUTING_KEY_HEADER + "']", true)); } @@ -336,7 +267,7 @@ public class RabbitMessageChannelBinder } } if (extendedProperties.getDelayExpression() != null) { - if (expresssionInterceptorNeeded) { + if (expressionInterceptorNeeded) { endpoint.setDelayExpressionString("headers['" + RabbitExpressionEvaluatingInterceptor.DELAY_HEADER + "']"); } @@ -368,14 +299,14 @@ public class RabbitMessageChannelBinder protected void postProcessOutputChannel(MessageChannel outputChannel, ExtendedProducerProperties producerProperties) { RabbitProducerProperties extendedProperties = producerProperties.getExtension(); - if (expresssionInterceptorNeeded(extendedProperties)) { + if (expressionInterceptorNeeded(extendedProperties)) { ((AbstractMessageChannel) outputChannel).addInterceptor(0, new RabbitExpressionEvaluatingInterceptor(extendedProperties.getRoutingKeyExpression(), extendedProperties.getDelayExpression(), getEvaluationContext())); } } - public boolean expresssionInterceptorNeeded(RabbitProducerProperties extendedProperties) { + public boolean expressionInterceptorNeeded(RabbitProducerProperties extendedProperties) { return extendedProperties.getRoutingKeyExpression() != null && extendedProperties.getRoutingKeyExpression().contains("payload") || (extendedProperties.getDelayExpression() != null 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 6077fbb23..b877991a3 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 @@ -17,10 +17,14 @@ package org.springframework.cloud.stream.binder.rabbit.config; 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.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; @@ -28,23 +32,26 @@ import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder import org.springframework.cloud.stream.binder.rabbit.properties.RabbitBinderConfigurationProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties; import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; - /** * Configuration class for RabbitMQ message channel binder. * * @author David Turanski * @author Vinicius Carvalho + * @author Artem Bilan */ @Configuration -@Import({PropertyPlaceholderAutoConfiguration.class}) -@EnableConfigurationProperties({RabbitBinderConfigurationProperties.class, RabbitExtendedBindingProperties.class}) +@Import({ PropertyPlaceholderAutoConfiguration.class }) +@EnableConfigurationProperties({ RabbitBinderConfigurationProperties.class, RabbitExtendedBindingProperties.class }) public class RabbitMessageChannelBinderConfiguration { + @Autowired + private ConfigurableApplicationContext applicationContext; @Autowired private ConnectionFactory rabbitConnectionFactory; @@ -59,17 +66,108 @@ public class RabbitMessageChannelBinderConfiguration { private RabbitExtendedBindingProperties rabbitExtendedBindingProperties; @Bean - RabbitMessageChannelBinder rabbitMessageChannelBinder() { - RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(rabbitConnectionFactory, rabbitProperties, - provisioningProvider()); - binder.setAdminAddresses(rabbitBinderConfigurationProperties.getAdminAddresses()); + RabbitMessageChannelBinder rabbitMessageChannelBinder( + @Qualifier("producerConnectionFactory") ObjectProvider producerConnectionFactory) + throws Exception { + + RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(this.rabbitConnectionFactory, + this.rabbitProperties, provisioningProvider()); + binder.setProducerConnectionFactory(obtainProducerConnectionFactory(producerConnectionFactory)); + binder.setAdminAddresses(this.rabbitBinderConfigurationProperties.getAdminAddresses()); binder.setCompressingPostProcessor(gZipPostProcessor()); binder.setDecompressingPostProcessor(deCompressingPostProcessor()); - binder.setNodes(rabbitBinderConfigurationProperties.getNodes()); - binder.setExtendedBindingProperties(rabbitExtendedBindingProperties); + binder.setNodes(this.rabbitBinderConfigurationProperties.getNodes()); + binder.setExtendedBindingProperties(this.rabbitExtendedBindingProperties); return binder; } + private ConnectionFactory obtainProducerConnectionFactory( + ObjectProvider connectionFactoryObjectProvider) throws Exception { + + ConnectionFactory connectionFactory = connectionFactoryObjectProvider.getIfAvailable(); + + if (connectionFactory != null) { + return connectionFactory; + } + else { + CachingConnectionFactory producerConnectionFactory = buildProducerConnectionFactory(); + producerConnectionFactory.setApplicationContext(this.applicationContext); + this.applicationContext.addApplicationListener(producerConnectionFactory); + producerConnectionFactory.afterPropertiesSet(); + + return producerConnectionFactory; + } + } + + /** + * @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(); + if (this.rabbitProperties.determineHost() != null) { + factory.setHost(this.rabbitProperties.determineHost()); + } + factory.setPort(this.rabbitProperties.determinePort()); + if (this.rabbitProperties.determineUsername() != null) { + factory.setUsername(this.rabbitProperties.determineUsername()); + } + if (this.rabbitProperties.determinePassword() != null) { + factory.setPassword(this.rabbitProperties.determinePassword()); + } + if (this.rabbitProperties.determineVirtualHost() != null) { + factory.setVirtualHost(this.rabbitProperties.determineVirtualHost()); + } + if (this.rabbitProperties.getRequestedHeartbeat() != null) { + factory.setRequestedHeartbeat((int)this.rabbitProperties.getRequestedHeartbeat().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()); + } + if (this.rabbitProperties.getConnectionTimeout() != null) { + factory.setConnectionTimeout((int)this.rabbitProperties.getConnectionTimeout().getSeconds()); + } + factory.afterPropertiesSet(); + + rabbitConnectionFactory = factory.getObject(); + } + + 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()); + } + return connectionFactory; + } + @Bean MessagePostProcessor deCompressingPostProcessor() { return new DelegatingDecompressingPostProcessor(); @@ -78,13 +176,13 @@ public class RabbitMessageChannelBinderConfiguration { @Bean MessagePostProcessor gZipPostProcessor() { GZipPostProcessor gZipPostProcessor = new GZipPostProcessor(); - gZipPostProcessor.setLevel(rabbitBinderConfigurationProperties.getCompressionLevel()); + gZipPostProcessor.setLevel(this.rabbitBinderConfigurationProperties.getCompressionLevel()); return gZipPostProcessor; } @Bean RabbitExchangeQueueProvisioner provisioningProvider() { - return new RabbitExchangeQueueProvisioner(rabbitConnectionFactory); + return new RabbitExchangeQueueProvisioner(this.rabbitConnectionFactory); } -} +} 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 1f56deaba..434d5a9f8 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -33,6 +33,7 @@ import org.springframework.cloud.stream.binder.Binder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.Primary; import org.springframework.context.annotation.Profile; /** @@ -45,6 +46,7 @@ import org.springframework.context.annotation.Profile; * @author Eric Bottard * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Artem Bilan */ @Configuration @ConditionalOnMissingBean(Binder.class) @@ -67,6 +69,7 @@ public class RabbitServiceAutoConfiguration { protected static class CloudConnectors { @Bean + @ConditionalOnMissingBean public Cloud cloud() { return new CloudFactory().getCloud(); } @@ -76,7 +79,8 @@ public class RabbitServiceAutoConfiguration { * 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. @@ -91,11 +95,22 @@ public class RabbitServiceAutoConfiguration { * @return the {@link ConnectionFactory} used by the binder. */ @Bean - @ConditionalOnMissingBean(ConnectionFactory.class) + @Primary ConnectionFactory rabbitConnectionFactory(Cloud cloud) { return cloud.getSingletonServiceConnector(ConnectionFactory.class, null); } + /** + * 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); + } + @Bean @ConditionalOnMissingBean(RabbitTemplate.class) RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { @@ -106,25 +121,29 @@ public class RabbitServiceAutoConfiguration { /** * Configuration to be used if * {@code spring.cloud.stream.overrideCloudConnectors} is set to {@code true}. - * Defers to Spring Boot Autoconfiguration. + * Defers to Spring Boot auto-configuration. */ @Configuration @ConditionalOnProperty("spring.cloud.stream.overrideCloudConnectors") @Import(RabbitAutoConfiguration.class) protected static class OverrideCloudConnectors { + } + } @Configuration @ConditionalOnMissingClass("org.springframework.cloud.Cloud") @Import(RabbitAutoConfiguration.class) protected static class NoCloudConnectors { + } + } /** * Configuration to be used when the cloud profile is not set. Defer to Spring Boot - * autoconfiguration. + * auto-configuration. */ @Configuration @Profile("!cloud") @@ -140,5 +159,7 @@ public class RabbitServiceAutoConfiguration { public HealthIndicator binderHealthIndicator(RabbitTemplate rabbitTemplate) { return new RabbitHealthIndicator(rabbitTemplate); } + } + } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java index 6a80b06c8..430c76232 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java @@ -63,6 +63,7 @@ public class RabbitTestBinder extends AbstractTestBinder binder = binderFactory.getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); @@ -195,10 +206,7 @@ public class RabbitBinderModuleTests { .run(params.toArray(new String[params.size()])); BinderFactory binderFactory = context.getBean(BinderFactory.class); @SuppressWarnings("unchecked") - Binder, - ExtendedProducerProperties> binder = - (Binder, - ExtendedProducerProperties>) binderFactory + Binder, ExtendedProducerProperties> binder = (Binder, ExtendedProducerProperties>) binderFactory .getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); @@ -231,6 +239,32 @@ public class RabbitBinderModuleTests { context.close(); } + @Test + public void testCloudProfile() { + this.context = new SpringApplicationBuilder(SimpleProcessor.class, MockCloudConfiguration.class) + .web(WebApplicationType.NONE) + .profiles("cloud") + .run(); + BinderFactory binderFactory = this.context.getBean(BinderFactory.class); + Binder binder = binderFactory.getBinder(null, MessageChannel.class); + assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); + DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); + ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor + .getPropertyValue("connectionFactory"); + ConnectionFactory connectionFactory = this.context.getBean(ConnectionFactory.class); + assertThat(binderConnectionFactory).isNotSameAs(connectionFactory); + + ConnectionFactory producerConnectionFactory = (ConnectionFactory) binderFieldAccessor + .getPropertyValue("producerConnectionFactory"); + assertThat(producerConnectionFactory).isNotSameAs(connectionFactory); + + assertThat(binderConnectionFactory).isNotSameAs(connectionFactory); + + Cloud cloud = this.context.getBean(Cloud.class); + + verify(cloud, times(2)).getSingletonServiceConnector(ConnectionFactory.class, null); + } + @EnableBinding(Processor.class) @SpringBootApplication public static class SimpleProcessor { @@ -246,4 +280,19 @@ public class RabbitBinderModuleTests { } + public static class MockCloudConfiguration { + + @Bean + public Cloud cloud() { + Cloud cloud = mock(Cloud.class); + + willReturn(mock(ConnectionFactory.class), mock(ConnectionFactory.class)) + .given(cloud) + .getSingletonServiceConnector(ConnectionFactory.class, null); + + return cloud; + } + + } + }