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**

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<RabbitProducerProperties> producerProperties, MessageChannel errorChannel)
-			throws Exception {
+			ExtendedProducerProperties<RabbitProducerProperties> 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<RabbitProducerProperties> 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

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<ConnectionFactory> 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<ConnectionFactory> 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);
 	}
-}
 
+}

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);
 		}
+
 	}
+
 }

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<RabbitMessageChannelBin
 	public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitMessageChannelBinder binder) {
 		this.applicationContext = new AnnotationConfigApplicationContext(Config.class);
 		binder.setApplicationContext(this.applicationContext);
+		binder.setProducerConnectionFactory(connectionFactory);
 		this.setBinder(binder);
 		this.rabbitAdmin = new RabbitAdmin(connectionFactory);
 	}

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

@@ -16,6 +16,12 @@
 
 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;
 import java.util.List;
 import java.util.Map;
@@ -37,6 +43,7 @@ import org.springframework.boot.actuate.health.HealthIndicator;
 import org.springframework.boot.actuate.health.Status;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.Cloud;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.BinderFactory;
@@ -57,11 +64,10 @@ 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;
-
 /**
  * @author Marius Bogoevici
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class RabbitBinderModuleTests {
 
@@ -70,7 +76,7 @@ public class RabbitBinderModuleTests {
 
 	private ConfigurableApplicationContext context;
 
-	public static final ConnectionFactory MOCK_CONNECTION_FACTORY = Mockito.mock(ConnectionFactory.class,
+	public static final ConnectionFactory MOCK_CONNECTION_FACTORY = mock(ConnectionFactory.class,
 			Mockito.RETURNS_MOCKS);
 
 	@After
@@ -100,6 +106,11 @@ public class RabbitBinderModuleTests {
 		assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class);
 		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);
@@ -117,8 +128,8 @@ public class RabbitBinderModuleTests {
 		context = new SpringApplicationBuilder(SimpleProcessor.class)
 				.web(WebApplicationType.NONE)
 				.run("--server.port=0",
-					"--spring.cloud.stream.rabbit.bindings.input.consumer.transacted=true",
-					"--spring.cloud.stream.rabbit.bindings.output.producer.transacted=true");
+						"--spring.cloud.stream.rabbit.bindings.input.consumer.transacted=true",
+						"--spring.cloud.stream.rabbit.bindings.output.producer.transacted=true");
 		BinderFactory binderFactory = context.getBean(BinderFactory.class);
 		Binder<?, ?, ?> 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<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>,
-								ExtendedProducerProperties<RabbitProducerProperties>> binder =
-			(Binder<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>,
-					ExtendedProducerProperties<RabbitProducerProperties>>) binderFactory
+		Binder<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> binder = (Binder<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>>) 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;
+		}
+
+	}
+
 }