AMQP-633: Ensure Resources are Closed

JIRA: https://jira.spring.io/browse/AMQP-633

Close connection for non-transactional template.

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java

@@ -1374,37 +1374,38 @@ private <T> T doExecute(ChannelCallback<T> action, ConnectionFactory connectionF
 		Assert.notNull(action, "Callback object must not be null");
 		Channel channel;
 		RabbitResourceHolder resourceHolder = null;
+		Connection connection = null;
 		if (isChannelTransacted()) {
 			resourceHolder = ConnectionFactoryUtils.getTransactionalResourceHolder(connectionFactory, true);
 			channel = resourceHolder.getChannel();
 			if (channel == null) {
+				ConnectionFactoryUtils.releaseResources(resourceHolder);
 				throw new IllegalStateException("Resource holder returned a null channel");
 			}
 		}
 		else {
-			Connection connection = connectionFactory.createConnection();
+			connection = connectionFactory.createConnection(); // NOSONAR - RabbitUtils
 			if (connection == null) {
 				throw new IllegalStateException("Connection factory returned a null connection");
 			}
-			channel = connection.createChannel(false);
-			if (channel == null) {
-				throw new IllegalStateException("Connection returned a null channel");
-			}
-		}
-		if (this.confirmsOrReturnsCapable == null) {
-			if (connectionFactory instanceof PublisherCallbackChannelConnectionFactory) {
-				PublisherCallbackChannelConnectionFactory pcccf =
-						(PublisherCallbackChannelConnectionFactory) connectionFactory;
-				this.confirmsOrReturnsCapable = pcccf.isPublisherConfirms() || pcccf.isPublisherReturns();
+			try {
+				channel = connection.createChannel(false);
+				if (channel == null) {
+					throw new IllegalStateException("Connection returned a null channel");
+				}
 			}
-			else {
-				this.confirmsOrReturnsCapable = Boolean.FALSE;
+			catch (RuntimeException e) {
+				RabbitUtils.closeConnection(connection);
+				throw e;
 			}
 		}
-		if (this.confirmsOrReturnsCapable) {
-			addListener(channel);
-		}
 		try {
+			if (this.confirmsOrReturnsCapable == null) {
+				determineConfirmsReturnsCapability(connectionFactory);
+			}
+			if (this.confirmsOrReturnsCapable) {
+				addListener(channel);
+			}
 			if (logger.isDebugEnabled()) {
 				logger.debug("Executing callback on RabbitMQ Channel: " + channel);
 			}
@@ -1422,10 +1423,22 @@ private <T> T doExecute(ChannelCallback<T> action, ConnectionFactory connectionF
 			}
 			else {
 				RabbitUtils.closeChannel(channel);
+				RabbitUtils.closeConnection(connection);
 			}
 		}
 	}
 
+	public void determineConfirmsReturnsCapability(ConnectionFactory connectionFactory) {
+		if (connectionFactory instanceof PublisherCallbackChannelConnectionFactory) {
+			PublisherCallbackChannelConnectionFactory pcccf =
+					(PublisherCallbackChannelConnectionFactory) connectionFactory;
+			this.confirmsOrReturnsCapable = pcccf.isPublisherConfirms() || pcccf.isPublisherReturns();
+		}
+		else {
+			this.confirmsOrReturnsCapable = Boolean.FALSE;
+		}
+	}
+
 	/**
 	 * Send the given message to the specified exchange.
 	 *