Fix Sonar issues

Author: garyrussell

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

@@ -1157,51 +1157,45 @@ public <T> T receiveAndConvert(String queueName, long timeoutMillis, Parameteriz
 	}
 
 	@Override
-	@Nullable
 	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback) throws AmqpException {
 		return receiveAndReply(this.getRequiredQueue(), callback);
 	}
 
 	@Override
 	@SuppressWarnings(UNCHECKED)
-	@Nullable
 	public <R, S> boolean receiveAndReply(final String queueName, ReceiveAndReplyCallback<R, S> callback) throws AmqpException {
 		return receiveAndReply(queueName, callback, (ReplyToAddressCallback<S>) this.defaultReplyToAddressCallback);
 
 	}
 
 	@Override
-	@Nullable
 	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback, final String exchange, final String routingKey)
 			throws AmqpException {
 		return receiveAndReply(this.getRequiredQueue(), callback, exchange, routingKey);
 	}
 
 	@Override
-	@Nullable
 	public <R, S> boolean receiveAndReply(final String queueName, ReceiveAndReplyCallback<R, S> callback, final String replyExchange,
 			final String replyRoutingKey) throws AmqpException {
 		return receiveAndReply(queueName, callback,
 				(request, reply) -> new Address(replyExchange, replyRoutingKey));
 	}
 
 	@Override
-	@Nullable
 	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback, ReplyToAddressCallback<S> replyToAddressCallback)
 			throws AmqpException {
 		return receiveAndReply(this.getRequiredQueue(), callback, replyToAddressCallback);
 	}
 
 	@Override
-	@Nullable
 	public <R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback,
 			ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException {
 		return doReceiveAndReply(queueName, callback, replyToAddressCallback);
 	}
 
-	@Nullable
 	private <R, S> boolean doReceiveAndReply(final String queueName, final ReceiveAndReplyCallback<R, S> callback,
 			final ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException {
+
 		return execute(channel -> {
 			Message receiveMessage = receiveForReply(queueName, channel);
 			if (receiveMessage != null) {
@@ -2006,17 +2000,7 @@ private <T> T doExecute(ChannelCallback<T> action, ConnectionFactory connectionF
 			invokeScope = true;
 		}
 		try {
-			if (this.confirmsOrReturnsCapable == null) {
-				determineConfirmsReturnsCapability(connectionFactory);
-			}
-			if (this.confirmsOrReturnsCapable) {
-				addListener(channel);
-			}
-			if (logger.isDebugEnabled()) {
-				logger.debug(
-						"Executing callback " + action.getClass().getSimpleName() + " on RabbitMQ Channel: " + channel);
-			}
-			return action.doInRabbit(channel);
+			return invokeAction(action, connectionFactory, channel);
 		}
 		catch (Exception ex) {
 			if (isChannelLocallyTransacted(channel)) {
@@ -2025,18 +2009,40 @@ private <T> T doExecute(ChannelCallback<T> action, ConnectionFactory connectionF
 			throw convertRabbitAccessException(ex);
 		}
 		finally {
-			if (!invokeScope) {
-				if (resourceHolder != null) {
-					ConnectionFactoryUtils.releaseResources(resourceHolder);
-				}
-				else {
-					RabbitUtils.closeChannel(channel);
-					RabbitUtils.closeConnection(connection);
-				}
+			cleanUpAfterAction(channel, invokeScope, resourceHolder, connection);
+		}
+	}
+
+	private void cleanUpAfterAction(Channel channel, boolean invokeScope, RabbitResourceHolder resourceHolder,
+			Connection connection) {
+
+		if (!invokeScope) {
+			if (resourceHolder != null) {
+				ConnectionFactoryUtils.releaseResources(resourceHolder);
+			}
+			else {
+				RabbitUtils.closeChannel(channel);
+				RabbitUtils.closeConnection(connection);
 			}
 		}
 	}
 
+	private <T> T invokeAction(ChannelCallback<T> action, ConnectionFactory connectionFactory, Channel channel)
+			throws Exception {
+
+		if (this.confirmsOrReturnsCapable == null) {
+			determineConfirmsReturnsCapability(connectionFactory);
+		}
+		if (this.confirmsOrReturnsCapable) {
+			addListener(channel);
+		}
+		if (logger.isDebugEnabled()) {
+			logger.debug(
+					"Executing callback " + action.getClass().getSimpleName() + " on RabbitMQ Channel: " + channel);
+		}
+		return action.doInRabbit(channel);
+	}
+
 	@Override
 	@Nullable
 	public <T> T invoke(OperationsCallback<T> action, @Nullable com.rabbitmq.client.ConfirmCallback acks,