Fix new Sonar smells

artembilan · artembilan · commit 638d10479439 · 2019-07-02T10:18:09.000-04:00
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
@@ -148,7 +148,7 @@
  */
 public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count
 		implements BeanFactoryAware, RabbitOperations, MessageListener,
-			ListenerContainerAware, PublisherCallbackChannel.Listener, Lifecycle, BeanNameAware {
+		ListenerContainerAware, PublisherCallbackChannel.Listener, Lifecycle, BeanNameAware {
 
 	private static final String UNCHECKED = "unchecked";
 
@@ -894,9 +894,9 @@ protected void doStart() {
 	public final void stop() {
 		synchronized (this.directReplyToContainers) {
 			this.directReplyToContainers.values()
-				.stream()
-				.filter(AbstractMessageListenerContainer::isRunning)
-				.forEach(AbstractMessageListenerContainer::stop);
+					.stream()
+					.filter(AbstractMessageListenerContainer::isRunning)
+					.forEach(AbstractMessageListenerContainer::stop);
 			this.directReplyToContainers.clear();
 		}
 		doStop();
@@ -958,15 +958,13 @@ protected boolean useDirectReplyTo() {
 				while (cause != null && !(cause instanceof ShutdownSignalException)) {
 					cause = cause.getCause();
 				}
-				if (cause instanceof ShutdownSignalException) {
-					if (RabbitUtils.isPassiveDeclarationChannelClose((ShutdownSignalException) cause)) {
-						if (logger.isWarnEnabled()) {
-							logger.warn("Broker does not support fast replies via 'amq.rabbitmq.reply-to', temporary "
-									+ "queues will be used: " + cause.getMessage() + ".");
-						}
-						this.replyAddress = null;
-						return false;
+				if (cause != null && RabbitUtils.isPassiveDeclarationChannelClose((ShutdownSignalException) cause)) {
+					if (logger.isWarnEnabled()) {
+						logger.warn("Broker does not support fast replies via 'amq.rabbitmq.reply-to', temporary "
+								+ "queues will be used: " + cause.getMessage() + ".");
 					}
+					this.replyAddress = null;
+					return false;
 				}
 				if (logger.isDebugEnabled()) {
 					logger.debug("IO error, deferring directReplyTo detection: " + ex.toString());
@@ -1001,7 +999,7 @@ public void send(final String exchange, final String routingKey,
 					(RabbitTemplate.this.returnCallback != null
 							|| (correlationData != null && StringUtils.hasText(correlationData.getId())))
 							&& RabbitTemplate.this.mandatoryExpression.getValue(
-									RabbitTemplate.this.evaluationContext, message, Boolean.class),
+							RabbitTemplate.this.evaluationContext, message, Boolean.class),
 					correlationData);
 			return null;
 		}, obtainTargetConnectionFactory(this.sendConnectionFactorySelectorExpression, message));
@@ -1262,21 +1260,24 @@ public <R, S> boolean receiveAndReply(final String queueName, ReceiveAndReplyCal
 	}
 
 	@Override
-	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback, final String exchange, final String routingKey)
-			throws AmqpException {
+	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
-	public <R, S> boolean receiveAndReply(final String queueName, ReceiveAndReplyCallback<R, S> callback, final String replyExchange,
-			final String replyRoutingKey) throws AmqpException {
+	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
-	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback, ReplyToAddressCallback<S> replyToAddressCallback)
-			throws AmqpException {
+	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback,
+			ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException {
+
 		return receiveAndReply(this.getRequiredQueue(), callback, replyToAddressCallback);
 	}
 
@@ -1290,18 +1291,17 @@ private <R, S> boolean doReceiveAndReply(final String queueName, final ReceiveAn
 			final ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException {
 
 		Boolean result = execute(channel -> {
-					Message receiveMessage = receiveForReply(queueName, channel);
-					if (receiveMessage != null) {
-						return sendReply(callback, replyToAddressCallback, channel, receiveMessage);
-					}
-					return false;
-				}, obtainTargetConnectionFactory(this.receiveConnectionFactorySelectorExpression, queueName));
+			Message receiveMessage = receiveForReply(queueName, channel);
+			if (receiveMessage != null) {
+				return sendReply(callback, replyToAddressCallback, channel, receiveMessage);
+			}
+			return false;
+		}, obtainTargetConnectionFactory(this.receiveConnectionFactorySelectorExpression, queueName));
 		return result == null ? false : result;
 	}
 
 	@Nullable
 	private Message receiveForReply(final String queueName, Channel channel) throws IOException {
-
 		boolean channelTransacted = isChannelTransacted();
 		boolean channelLocallyTransacted = isChannelLocallyTransacted(channel);
 		Message receiveMessage = null;
@@ -1358,7 +1358,7 @@ private Delivery consumeDelivery(Channel channel, String queueName, long timeout
 		DefaultConsumer consumer = null;
 		try {
 			consumer = createConsumer(queueName, channel, future,
-						timeoutMillis < 0 ? DEFAULT_CONSUME_TIMEOUT : timeoutMillis);
+					timeoutMillis < 0 ? DEFAULT_CONSUME_TIMEOUT : timeoutMillis);
 			if (timeoutMillis < 0) {
 				delivery = future.get();
 			}
@@ -1404,7 +1404,7 @@ else if (logger.isDebugEnabled()) {
 	@SuppressWarnings(UNCHECKED)
 	private <R, S> boolean sendReply(final ReceiveAndReplyCallback<R, S> callback,
 			final ReplyToAddressCallback<S> replyToAddressCallback, Channel channel, Message receiveMessage)
-					throws IOException {
+			throws IOException {
 
 		Object receive = receiveMessage;
 		if (!(ReceiveAndReplyMessageCallback.class.isAssignableFrom(callback.getClass()))) {
@@ -1583,8 +1583,10 @@ public Object convertSendAndReceive(final String routingKey, final Object messag
 
 	@Override
 	@Nullable
-	public Object convertSendAndReceive(final String routingKey, final Object message, final MessagePostProcessor messagePostProcessor,
-			@Nullable CorrelationData correlationData) throws AmqpException {
+	public Object convertSendAndReceive(final String routingKey, final Object message,
+			final MessagePostProcessor messagePostProcessor, @Nullable CorrelationData correlationData)
+			throws AmqpException {
+
 		return convertSendAndReceive(this.exchange, routingKey, message, messagePostProcessor, correlationData);
 	}
 
@@ -1637,7 +1639,7 @@ public <T> T convertSendAndReceiveAsType(final String routingKey, final Object m
 	@Nullable
 	public <T> T convertSendAndReceiveAsType(final String routingKey, final Object message,
 			@Nullable CorrelationData correlationData, ParameterizedTypeReference<T> responseType)
-					throws AmqpException {
+			throws AmqpException {
 
 		return convertSendAndReceiveAsType(this.exchange, routingKey, message, null, correlationData, responseType);
 	}
@@ -1664,7 +1666,7 @@ public <T> T convertSendAndReceiveAsType(final Object message,
 	public <T> T convertSendAndReceiveAsType(final Object message,
 			@Nullable final MessagePostProcessor messagePostProcessor,
 			@Nullable CorrelationData correlationData, ParameterizedTypeReference<T> responseType)
-					throws AmqpException {
+			throws AmqpException {
 
 		return convertSendAndReceiveAsType(this.exchange, this.routingKey, message, messagePostProcessor,
 				correlationData, responseType);
@@ -1674,7 +1676,7 @@ public <T> T convertSendAndReceiveAsType(final Object message,
 	@Nullable
 	public <T> T convertSendAndReceiveAsType(final String routingKey, final Object message,
 			@Nullable final MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<T> responseType)
-					throws AmqpException {
+			throws AmqpException {
 
 		return convertSendAndReceiveAsType(routingKey, message, messagePostProcessor, null, responseType);
 	}
@@ -1684,6 +1686,7 @@ public <T> T convertSendAndReceiveAsType(final String routingKey, final Object m
 	public <T> T convertSendAndReceiveAsType(final String routingKey, final Object message,
 			@Nullable final MessagePostProcessor messagePostProcessor, @Nullable CorrelationData correlationData,
 			ParameterizedTypeReference<T> responseType) throws AmqpException {
+
 		return convertSendAndReceiveAsType(this.exchange, routingKey, message, messagePostProcessor, correlationData,
 				responseType);
 	}
@@ -1693,6 +1696,7 @@ public <T> T convertSendAndReceiveAsType(final String routingKey, final Object m
 	public <T> T convertSendAndReceiveAsType(final String exchange, final String routingKey, final Object message,
 			final MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<T> responseType)
 			throws AmqpException {
+
 		return convertSendAndReceiveAsType(exchange, routingKey, message, messagePostProcessor, null, responseType);
 	}
 
@@ -1754,6 +1758,7 @@ protected Message convertMessageIfNecessary(final Object object) {
 	@Nullable
 	protected Message doSendAndReceive(final String exchange, final String routingKey, final Message message,
 			@Nullable CorrelationData correlationData) {
+
 		if (!this.evaluatedFastReplyTo) {
 			synchronized (this) {
 				if (!this.evaluatedFastReplyTo) {
@@ -1776,13 +1781,15 @@ else if (this.replyAddress == null || this.usingFastReplyTo) {
 	@Nullable
 	protected Message doSendAndReceiveWithTemporary(final String exchange, final String routingKey,
 			final Message message, final CorrelationData correlationData) {
+
 		return execute(channel -> {
 			final PendingReply pendingReply = new PendingReply();
 			String messageTag = String.valueOf(RabbitTemplate.this.messageTagProvider.incrementAndGet());
 			RabbitTemplate.this.replyHolder.putIfAbsent(messageTag, pendingReply);
 
 			Assert.isNull(message.getMessageProperties().getReplyTo(),
-					"Send-and-receive methods can only be used if the Message does not already have a replyTo property.");
+					"Send-and-receive methods can only be used " +
+							"if the Message does not already have a replyTo property.");
 			String replyTo;
 			if (RabbitTemplate.this.usingFastReplyTo) {
 				replyTo = Address.AMQ_RABBITMQ_REPLY_TO;
@@ -2065,8 +2072,9 @@ private <T> T doExecute(ChannelCallback<T> action, ConnectionFactory connectionF
 		Connection connection = null; // NOSONAR (close)
 		if (channel == null) {
 			if (isChannelTransacted()) {
-				resourceHolder = ConnectionFactoryUtils.
-					getTransactionalResourceHolder(connectionFactory, true, this.usePublisherConnection);
+				resourceHolder =
+						ConnectionFactoryUtils.getTransactionalResourceHolder(connectionFactory,
+								true, this.usePublisherConnection);
 				channel = resourceHolder.getChannel();
 				if (channel == null) {
 					ConnectionFactoryUtils.releaseResources(resourceHolder);
@@ -2266,8 +2274,7 @@ public void determineConfirmsReturnsCapability(ConnectionFactory connectionFacto
 	 * @throws IOException If thrown by RabbitMQ API methods.
 	 */
 	public void doSend(Channel channel, String exchangeArg, String routingKeyArg, Message message, // NOSONAR complexity
-			boolean mandatory, @Nullable CorrelationData correlationData)
-					throws IOException {
+			boolean mandatory, @Nullable CorrelationData correlationData) throws IOException {
 
 		String exch = exchangeArg;
 		String rKey = routingKeyArg;
@@ -2353,7 +2360,8 @@ private Message buildMessageFromDelivery(Delivery delivery) {
 	}
 
 	private Message buildMessageFromResponse(GetResponse response) {
-		return buildMessage(response.getEnvelope(), response.getProps(), response.getBody(), response.getMessageCount());
+		return buildMessage(response.getEnvelope(), response.getProps(), response.getBody(),
+				response.getMessageCount());
 	}
 
 	private Message buildMessage(Envelope envelope, BasicProperties properties, byte[] body, int msgCount) {
@@ -2414,7 +2422,8 @@ private Address getReplyToAddress(Message request) throws AmqpException {
 			if (this.exchange == null) {
 				throw new AmqpException(
 						"Cannot determine ReplyTo message property value: "
-								+ "Request message does not contain reply-to property, and no default Exchange was set.");
+								+ "Request message does not contain reply-to property, " +
+								"and no default Exchange was set.");
 			}
 			replyTo = new Address(this.exchange, this.routingKey);
 		}
@@ -2448,7 +2457,8 @@ public void addListener(Channel channel) {
 	@Override
 	public void handleConfirm(PendingConfirm pendingConfirm, boolean ack) {
 		if (this.confirmCallback != null) {
-			this.confirmCallback.confirm(pendingConfirm.getCorrelationData(), ack, pendingConfirm.getCause()); // NOSONAR never null
+			this.confirmCallback
+					.confirm(pendingConfirm.getCorrelationData(), ack, pendingConfirm.getCause()); // NOSONAR never null
 		}
 	}
 
@@ -2569,7 +2579,7 @@ private void restoreProperties(Message message, PendingReply pendingReply) {
 
 	private DefaultConsumer createConsumer(final String queueName, Channel channel,
 			CompletableFuture<Delivery> future, long timeoutMillis) throws IOException, TimeoutException,
-				InterruptedException {
+			InterruptedException {
 
 		channel.basicQos(1);
 		final CountDownLatch latch = new CountDownLatch(1);
@@ -2717,8 +2727,8 @@ public interface ReturnCallback {
 		 * @param exchange the exchange.
 		 * @param routingKey the routing key.
 		 */
-		void returnedMessage(Message message, int replyCode, String replyText,
-				String exchange, String routingKey);
+		void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey);
+
 	}
 
 }
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java
@@ -720,7 +720,7 @@ private SimpleConsumer consume(String queue, Connection connection) {
 	}
 
 	@Nullable
-	private SimpleConsumer handleConsumeException(String queue, SimpleConsumer consumerArg, Exception e) {
+	private SimpleConsumer handleConsumeException(String queue, @Nullable SimpleConsumer consumerArg, Exception e) {
 
 		SimpleConsumer consumer = consumerArg;
 		if (e.getCause() instanceof ShutdownSignalException
@@ -732,7 +732,7 @@ private SimpleConsumer handleConsumeException(String queue, SimpleConsumer consu
 		else if (e.getCause() instanceof ShutdownSignalException
 				&& RabbitUtils.isPassiveDeclarationChannelClose((ShutdownSignalException) e.getCause())) {
 			this.logger.error("Queue not present, scheduling consumer "
-				+ (consumer == null ? "for queue " + queue : consumer) + " for restart", e);
+					+ (consumer == null ? "for queue " + queue : consumer) + " for restart", e);
 		}
 		else if (this.logger.isWarnEnabled()) {
 			this.logger.warn("basicConsume failed, scheduling consumer "
@@ -755,7 +755,8 @@ protected void doShutdown() {
 		boolean waitForConsumers = false;
 		synchronized (this.consumersMonitor) {
 			if (this.started || this.aborted) {
-				// Copy in the same order to avoid ConcurrentModificationException during remove in the cancelConsumer().
+				// Copy in the same order to avoid ConcurrentModificationException during remove in the
+				// cancelConsumer().
 				canceledConsumers = new LinkedList<>(this.consumers);
 				actualShutDown(canceledConsumers);
 				waitForConsumers = true;
@@ -879,8 +880,6 @@ final class SimpleConsumer extends DefaultConsumer {
 
 		private long lastAck = System.currentTimeMillis();
 
-		private long lastDeliveryComplete = this.lastAck;
-
 		private long latestDeferredDeliveryTag;
 
 		private volatile String consumerTag;
@@ -893,7 +892,7 @@ final class SimpleConsumer extends DefaultConsumer {
 
 		private volatile boolean ackFailed;
 
-		SimpleConsumer(Connection connection, Channel channel, String queue) {
+		SimpleConsumer(@Nullable Connection connection, @Nullable Channel channel, String queue) {
 			super(channel);
 			this.connection = connection;
 			this.queue = queue;
@@ -1094,8 +1093,7 @@ private void handleAck(long deliveryTag, boolean channelLocallyTransacted) {
 						synchronized (this) {
 							this.latestDeferredDeliveryTag = deliveryTag;
 							this.pendingAcks++;
-							this.lastDeliveryComplete = System.currentTimeMillis();
-							ackIfNecessary(this.lastDeliveryComplete);
+							ackIfNecessary(this.lastAck);
 						}
 					}
 					else if (!isChannelTransacted() || isLocallyTransacted) {
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java
@@ -598,7 +598,7 @@ public Object getResult() {
 
 	}
 
-	private static class MonoHandler {
+	private static class MonoHandler { // NOSONAR - pointless to name it ..Utils|Helper
 
 		static boolean isMono(Object result) {
 			return result instanceof Mono;

Original file line number	Diff line number	Diff line change
`@@ -598,7 +598,7 @@ public Object getResult() {`
`598`	`598`
`599`	`599`	`}`
`600`	`600`
`601`		`- private static class MonoHandler {`
	`601`	`+ private static class MonoHandler { // NOSONAR - pointless to name it ..Utils\|Helper`
`602`	`602`
`603`	`603`	`static boolean isMono(Object result) {`
`604`	`604`	`return result instanceof Mono;`