GH-95: Add producerConnFactory into Rabbit Binder

Fixes spring-cloud/spring-cloud-stream-binder-rabbit#95

* To avoid blocked connection dead lock on consumers add
`ConnectionFactory` clone in the `RabbitMessageChannelBinder` for
non-transactional producers
* Add `DisposableBean` implementation into the `RabbitMessageChannelBinder`
to `destroy` all the spawned internal connection factories including
`LocalizedQueueConnectionFactory`

Address some PR comments:

* Copy `RabbitConnectionFactory` from the injected `CCF`
* Don't wrap `producerCF` into the `LocalizedQueueCF`
* Polishing code style

Make Latebinding test transactional

So the same connection factory is used.

There is a race condition in the proxy.

spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc

@@ -44,7 +44,7 @@ If retry is enabled (`maxAttempts > 1`) failed messages will be delivered to the
 If retry is disabled (`maxAttempts = 1`), you should set `requeueRejected` to `false` (default) so that a failed message will be routed to the DLQ, instead of being requeued.
 In addition, `republishToDlq` causes the binder to publish a failed message to the DLQ (instead of rejecting it); this enables additional information to be added to the message in headers, such as the stack trace in the `x-exception-stacktrace` header.
 This option does not need retry enabled; you can republish a failed message after just one attempt.
-Starting with _version 1.2_, you can configure the delivery mode of republished messsages; see property `republishDeliveryMode`.
+Starting with _version 1.2_, you can configure the delivery mode of republished messages; see property `republishDeliveryMode`.
 
 IMPORTANT: Setting `requeueRejected` to `true` will cause the message to be requeued and redelivered continually, which is likely not what you want unless the failure issue is transient.
 In general, it's better to enable retry within the binder by setting `maxAttempts` to greater than one, or set `republishToDlq` to `true`.
@@ -56,9 +56,11 @@ Some options are described in <<rabbit-dlq-processing>>.
 
 [NOTE]
 ====
-When *multiple* RabbitMQ binders are used in a Spring Cloud Stream application, it is important to disable 'RabbitAutoConfiguration' to avoid the same configuration from RabbitAutoConfiguration being applied to the two binders.
+When *multiple* RabbitMQ binders are used in a Spring Cloud Stream application, it is important to disable 'RabbitAutoConfiguration' to avoid the same configuration from `RabbitAutoConfiguration` being applied to the two binders.
 ====
 
+Starting with _version 1.3_, the `RabbitMessageChannelBinder` creates an internal `ConnectionFactory` copy for the non-transactional producers to avoid dead locks on consumers when shared, cached connections are blocked because of https://www.rabbitmq.com/memory.html[Memory Alarm] on Broker.
+
 == Configuration Options
 
 This section contains settings specific to the RabbitMQ Binder and bound channels.

spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java

@@ -30,6 +30,7 @@ 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;
@@ -42,7 +43,7 @@ import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
 import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
-import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties.Retry;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
@@ -82,7 +83,9 @@ import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Envelope;
 
 /**
- * A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by RabbitMQ.
+ * A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by
+ * RabbitMQ.
+ *
  * @author Mark Fisher
  * @author Gary Russell
  * @author Jennifer Hickey
@@ -90,11 +93,13 @@ import com.rabbitmq.client.Envelope;
  * @author Ilayaperumal Gopinathan
  * @author David Turanski
  * @author Marius Bogoevici
+ * @author Artem Bilan
  */
 public class RabbitMessageChannelBinder
 		extends AbstractMessageChannelBinder<ExtendedConsumerProperties<RabbitConsumerProperties>,
 		ExtendedProducerProperties<RabbitProducerProperties>, RabbitExchangeQueueProvisioner>
-		implements ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties> {
+		implements ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties>,
+		DisposableBean {
 
 	private static final AmqpMessageHeaderErrorMessageStrategy errorMessageStrategy =
 			new AmqpMessageHeaderErrorMessageStrategy();
@@ -109,12 +114,15 @@ public class RabbitMessageChannelBinder
 					properties.setDeliveryMode(null);
 					return properties;
 				}
+
 			};
 
 	private final RabbitProperties rabbitProperties;
 
 	private ConnectionFactory connectionFactory;
 
+	private ConnectionFactory producerConnectionFactory;
+
 	private MessagePostProcessor decompressingPostProcessor = new DelegatingDecompressingPostProcessor();
 
 	private MessagePostProcessor compressingPostProcessor = new GZipPostProcessor();
@@ -128,7 +136,7 @@ public class RabbitMessageChannelBinder
 	private RabbitExtendedBindingProperties extendedBindingProperties = new RabbitExtendedBindingProperties();
 
 	public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties,
-										RabbitExchangeQueueProvisioner provisioningProvider) {
+			RabbitExchangeQueueProvisioner provisioningProvider) {
 		super(true, new String[0], provisioningProvider);
 		Assert.notNull(connectionFactory, "connectionFactory must not be null");
 		Assert.notNull(rabbitProperties, "rabbitProperties must not be null");
@@ -146,8 +154,8 @@ public class RabbitMessageChannelBinder
 	}
 
 	/**
-	 * Set a {@link org.springframework.amqp.core.MessagePostProcessor} to compress messages. Defaults to a
-	 * {@link org.springframework.amqp.support.postprocessor.GZipPostProcessor}.
+	 * Set a {@link org.springframework.amqp.core.MessagePostProcessor} to compress messages.
+	 * Defaults to a {@link org.springframework.amqp.support.postprocessor.GZipPostProcessor}.
 	 * @param compressingPostProcessor the post processor.
 	 */
 	public void setCompressingPostProcessor(MessagePostProcessor compressingPostProcessor) {
@@ -168,12 +176,23 @@ public class RabbitMessageChannelBinder
 	}
 
 	@Override
-	public void onInit() {
+	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());
+
 			Assert.state(addresses.length == this.adminAddresses.length
-							&& addresses.length == this.nodes.length,
+					&& addresses.length == this.nodes.length,
 					"'addresses', 'adminAddresses', and 'nodes' properties must have equal length");
+
 			this.connectionFactory = new LocalizedQueueConnectionFactory(this.connectionFactory, addresses,
 					this.adminAddresses, this.nodes, this.rabbitProperties.getVirtualHost(),
 					this.rabbitProperties.getUsername(), this.rabbitProperties.getPassword(),
@@ -184,6 +203,82 @@ 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(config.getRequestedHeartbeat());
+			}
+			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(config.getConnectionTimeout());
+			}
+			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) {
+			((DisposableBean) this.connectionFactory).destroy();
+			((DisposableBean) this.producerConnectionFactory).destroy();
+		}
+	}
+
 	@Override
 	public RabbitConsumerProperties getExtendedConsumerProperties(String channelName) {
 		return this.extendedBindingProperties.getExtendedConsumerProperties(channelName);
@@ -197,7 +292,7 @@ public class RabbitMessageChannelBinder
 	@Override
 	protected MessageHandler createProducerMessageHandler(final ProducerDestination producerDestination,
 			ExtendedProducerProperties<RabbitProducerProperties> producerProperties, MessageChannel errorChannel)
-					throws Exception {
+			throws Exception {
 		String prefix = producerProperties.getExtension().getPrefix();
 		String exchangeName = producerDestination.getName();
 		String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length());
@@ -274,7 +369,7 @@ public class RabbitMessageChannelBinder
 
 	@Override
 	protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group,
-													ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
+			ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
 		String destination = consumerDestination.getName();
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(
 				this.connectionFactory);
@@ -330,7 +425,7 @@ public class RabbitMessageChannelBinder
 			return new MessageHandler() {
 
 				private final RabbitTemplate template = new RabbitTemplate(
-						RabbitMessageChannelBinder.this.connectionFactory);
+						RabbitMessageChannelBinder.this.producerConnectionFactory);
 
 				private final String exchange = deadLetterExchangeName(properties.getExtension());
 
@@ -423,13 +518,6 @@ public class RabbitMessageChannelBinder
 	}
 
 	private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, boolean mandatory) {
-		RabbitProperties rabbitProperties = null;
-		try {
-			rabbitProperties = getApplicationContext().getBean(RabbitProperties.class);
-		}
-		catch (NoSuchBeanDefinitionException e) {
-			logger.debug("No RabbitProperties in context; no producer retry will be configured");
-		}
 		RabbitTemplate rabbitTemplate;
 		if (properties.isBatchingEnabled()) {
 			BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(
@@ -443,11 +531,16 @@ public class RabbitMessageChannelBinder
 		else {
 			rabbitTemplate = new RabbitTemplate();
 		}
-		rabbitTemplate.setConnectionFactory(this.connectionFactory);
+		rabbitTemplate.setChannelTransacted(properties.isTransacted());
+		if (rabbitTemplate.isChannelTransacted()) {
+			rabbitTemplate.setConnectionFactory(this.connectionFactory);
+		}
+		else {
+			rabbitTemplate.setConnectionFactory(this.producerConnectionFactory);
+		}
 		if (properties.isCompress()) {
 			rabbitTemplate.setBeforePublishPostProcessors(this.compressingPostProcessor);
 		}
-		rabbitTemplate.setChannelTransacted(properties.isTransacted());
 		rabbitTemplate.setMandatory(mandatory); // returned messages
 		if (rabbitProperties != null && rabbitProperties.getTemplate().getRetry().isEnabled()) {
 			Retry retry = rabbitProperties.getTemplate().getRetry();

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

@@ -43,6 +43,7 @@ import org.springframework.amqp.core.ExchangeTypes;
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitManagementTemplate;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -120,10 +121,13 @@ public class RabbitBinderTests extends
 
 	@Override
 	protected RabbitTestBinder getBinder() {
-		if (testBinder == null) {
-			testBinder = new RabbitTestBinder(rabbitAvailableRule.getResource(), new RabbitProperties());
+		if (this.testBinder == null) {
+			RabbitProperties rabbitProperties = new RabbitProperties();
+			rabbitProperties.setPublisherConfirms(true);
+			rabbitProperties.setPublisherReturns(true);
+			this.testBinder = new RabbitTestBinder(rabbitAvailableRule.getResource(), rabbitProperties);
 		}
-		return testBinder;
+		return this.testBinder;
 	}
 
 	@Override
@@ -155,6 +159,20 @@ public class RabbitBinderTests extends
 				createProducerProperties());
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel,
 				createConsumerProperties());
+
+		ConnectionFactory producerConnectionFactory =
+				TestUtils.getPropertyValue(producerBinding, "lifecycle.amqpTemplate.connectionFactory",
+						ConnectionFactory.class);
+
+		ConnectionFactory consumerConnectionFactory =
+				TestUtils.getPropertyValue(consumerBinding, "lifecycle.messageListenerContainer.connectionFactory",
+						ConnectionFactory.class);
+
+		assertThat(producerConnectionFactory).isNotSameAs(consumerConnectionFactory);
+
+		assertThat(producerConnectionFactory.createConnection())
+				.isNotEqualTo(consumerConnectionFactory.createConnection());
+
 		Message<?> message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar")
 				.build();
 		final CountDownLatch latch = new CountDownLatch(3);
@@ -496,8 +514,15 @@ public class RabbitBinderTests extends
 
 		BindingProperties producerBindingProperties = createProducerBindingProperties(producerProperties);
 		DirectChannel channel = createBindableChannel("output", producerBindingProperties);
-		producerBinding = binder.bindProducer("props.0", channel,
-				producerProperties);
+		producerBinding = binder.bindProducer("props.0", channel, producerProperties);
+
+		ConnectionFactory producerConnectionFactory =
+				TestUtils.getPropertyValue(producerBinding, "lifecycle.amqpTemplate.connectionFactory",
+						ConnectionFactory.class);
+
+		assertThat(this.rabbitAvailableRule.getResource())
+				.isSameAs(producerConnectionFactory);
+
 		endpoint = extractEndpoint(producerBinding);
 		assertThat(getEndpointRouting(endpoint))
 				.isEqualTo("'props.0-' + headers['" + BinderHeaders.PARTITION_HEADER + "']");
@@ -726,12 +751,12 @@ public class RabbitBinderTests extends
 	}
 
 	@Test
-	public void testAutoBindDLQPartionedConsumerFirstWithRepublishNoRetry() throws Exception {
+	public void testAutoBindDLQPartitionedConsumerFirstWithRepublishNoRetry() throws Exception {
 		testAutoBindDLQPartionedConsumerFirstWithRepublishGuts(false);
 	}
 
 	@Test
-	public void testAutoBindDLQPartionedConsumerFirstWithRepublishWithRetry() throws Exception {
+	public void testAutoBindDLQPartitionedConsumerFirstWithRepublishWithRetry() throws Exception {
 		testAutoBindDLQPartionedConsumerFirstWithRepublishGuts(true);
 	}
 
@@ -1076,6 +1101,7 @@ public class RabbitBinderTests extends
 		ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.getExtension().setPrefix("latebinder.");
 		producerProperties.getExtension().setAutoBindDlq(true);
+		producerProperties.getExtension().setTransacted(true);
 
 		MessageChannel moduleOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties));
 		Binding<MessageChannel> late0ProducerBinding = binder.bindProducer("late.0", moduleOutputChannel, producerProperties);
@@ -1135,7 +1161,7 @@ public class RabbitBinderTests extends
 		proxy.start();
 
 		moduleOutputChannel.send(new GenericMessage<>("foo"));
-		Message<?> message = moduleInputChannel.receive(10000);
+		Message<?> message = moduleInputChannel.receive(20000);
 		assertThat(message).isNotNull();
 		assertThat(message.getPayload()).isNotNull();