diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractJavaTypeMapper.java b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractJavaTypeMapper.java index 49dca2ad..989bdab9 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractJavaTypeMapper.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractJavaTypeMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -22,6 +22,7 @@ import java.util.Map; import org.springframework.amqp.core.MessageProperties; import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.lang.Nullable; import org.springframework.util.ClassUtils; import com.fasterxml.jackson.databind.JavaType; @@ -93,6 +94,7 @@ public abstract class AbstractJavaTypeMapper implements BeanClassLoaderAware { return classId; } + @Nullable protected String retrieveHeaderAsString(MessageProperties properties, String headerName) { Map headers = properties.getHeaders(); Object classIdFieldNameValue = headers.get(headerName); diff --git a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java index d6562a15..2220fcd1 100644 --- a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java +++ b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java @@ -392,7 +392,7 @@ public final class BrokerRunning extends TestWatcher { } public void isUp() throws Exception { - Connection connection = getConnectionFactory().newConnection(); + Connection connection = getConnectionFactory().newConnection(); // NOSONAR - closeResources() Channel channel = null; try { channel = createQueues(connection); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PendingConfirm.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PendingConfirm.java index fc6300ce..5519d22f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PendingConfirm.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PendingConfirm.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -16,6 +16,8 @@ package org.springframework.amqp.rabbit.connection; +import org.springframework.lang.Nullable; + /** * Instances of this object track pending publisher confirms. * The timestamp allows the pending confirmation to be @@ -37,7 +39,7 @@ public class PendingConfirm { * @param correlationData The correlation data. * @param timestamp The timestamp. */ - public PendingConfirm(CorrelationData correlationData, long timestamp) { + public PendingConfirm(@Nullable CorrelationData correlationData, long timestamp) { this.correlationData = correlationData; this.timestamp = timestamp; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java index 2a779b48..abcc62da 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java @@ -24,6 +24,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; import com.rabbitmq.client.AMQP; @@ -50,7 +51,7 @@ public abstract class RabbitUtils { * finally blocks in manual RabbitMQ code. * @param connection the RabbitMQ Connection to close (may be null) */ - public static void closeConnection(Connection connection) { + public static void closeConnection(@Nullable Connection connection) { if (connection != null) { try { connection.close(); @@ -69,7 +70,7 @@ public abstract class RabbitUtils { * blocks in manual RabbitMQ code. * @param channel the RabbitMQ Channel to close (may be null) */ - public static void closeChannel(Channel channel) { + public static void closeChannel(@Nullable Channel channel) { if (channel != null) { try { channel.close(); 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 index 31cfa43e..d92475da 100644 --- 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 @@ -1382,18 +1382,21 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, @Override public Object convertSendAndReceive(final String routingKey, final Object message, @Nullable CorrelationData correlationData) throws AmqpException { + return convertSendAndReceive(this.exchange, routingKey, message, null, correlationData); } @Override public Object convertSendAndReceive(final String exchange, final String routingKey, final Object message) throws AmqpException { + return convertSendAndReceive(exchange, routingKey, message, (CorrelationData) null); } @Override public Object convertSendAndReceive(final String exchange, final String routingKey, final Object message, - CorrelationData correlationData) throws AmqpException { + @Nullable CorrelationData correlationData) throws AmqpException { + return convertSendAndReceive(exchange, routingKey, message, null, correlationData); } @@ -1405,19 +1408,21 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, @Override public Object convertSendAndReceive(final Object message, final MessagePostProcessor messagePostProcessor, - CorrelationData correlationData) throws AmqpException { + @Nullable CorrelationData correlationData) throws AmqpException { + return convertSendAndReceive(this.exchange, this.routingKey, message, messagePostProcessor, correlationData); } @Override public Object convertSendAndReceive(final String routingKey, final Object message, final MessagePostProcessor messagePostProcessor) throws AmqpException { + return convertSendAndReceive(routingKey, message, messagePostProcessor, null); } @Override public Object convertSendAndReceive(final String routingKey, final Object message, final MessagePostProcessor messagePostProcessor, - CorrelationData correlationData) throws AmqpException { + @Nullable CorrelationData correlationData) throws AmqpException { return convertSendAndReceive(this.exchange, routingKey, message, messagePostProcessor, correlationData); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index 3e0168bc..98168d3c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -73,6 +73,7 @@ import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.lang.Nullable; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.transaction.interceptor.TransactionAttribute; @@ -632,6 +633,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * @since 1.6.9 * @see #setLookupKeyQualifier(String) */ + @Nullable protected String getRoutingLookupKey() { return super.getConnectionFactory() instanceof RoutingConnectionFactory ? this.lookupKeyQualifier + "[" + this.queues.stream() @@ -646,6 +648,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * @return the {@link RoutingConnectionFactory} or null. * @since 1.6.9 */ + @Nullable protected RoutingConnectionFactory getRoutingConnectionFactory() { return super.getConnectionFactory() instanceof RoutingConnectionFactory ? (RoutingConnectionFactory) super.getConnectionFactory() @@ -1602,7 +1605,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor return e; } - protected void publishConsumerFailedEvent(String reason, boolean fatal, Throwable t) { + protected void publishConsumerFailedEvent(String reason, boolean fatal, @Nullable Throwable t) { if (this.applicationEventPublisher != null) { this.applicationEventPublisher .publishEvent(t == null ? new ListenerContainerConsumerTerminatedEvent(this, reason) : @@ -1665,7 +1668,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor } else { try { - Connection connection = getConnectionFactory().createConnection(); + Connection connection = getConnectionFactory().createConnection(); // NOSONAR if (connection != null) { connection.close(); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index e140bfad..409ff5cc 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -60,6 +60,7 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator; import org.springframework.amqp.support.ConsumerTagStrategy; import org.springframework.context.ApplicationEventPublisher; +import org.springframework.lang.Nullable; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.ObjectUtils; import org.springframework.util.backoff.BackOffExecution; @@ -217,7 +218,7 @@ public class BlockingQueueConsumer { MessagePropertiesConverter messagePropertiesConverter, ActiveObjectCounter activeObjectCounter, AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, boolean defaultRequeueRejected, - Map consumerArgs, String... queues) { + @Nullable Map consumerArgs, String... queues) { this(connectionFactory, messagePropertiesConverter, activeObjectCounter, acknowledgeMode, transactional, prefetchCount, defaultRequeueRejected, consumerArgs, false, queues); } @@ -443,6 +444,7 @@ public class BlockingQueueConsumer { * @return A message built from the contents. * @throws InterruptedException if the thread is interrupted. */ + @Nullable private Message handle(Delivery delivery) throws InterruptedException { if ((delivery == null && this.shutdown != null)) { throw this.shutdown; @@ -489,6 +491,7 @@ public class BlockingQueueConsumer { * @throws InterruptedException if an interrupt is received while waiting * @throws ShutdownSignalException if the connection is shut down while waiting */ + @Nullable public Message nextMessage(long timeout) throws InterruptedException, ShutdownSignalException { if (logger.isTraceEnabled()) { logger.trace("Retrieving delivery for " + this); 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 index de809cec..04869bcc 100644 --- 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 @@ -512,7 +512,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta this.logger.error("Error creating consumer; retrying in " + nextBackOff, e); doShutdown(); try { - Thread.sleep(nextBackOff); + Thread.sleep(nextBackOff); // NOSONAR } catch (InterruptedException e1) { Thread.currentThread().interrupt(); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerContainerConsumerFailedEvent.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerContainerConsumerFailedEvent.java index 7ec60567..dcb67b3e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerContainerConsumerFailedEvent.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerContainerConsumerFailedEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2018 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. @@ -17,6 +17,7 @@ package org.springframework.amqp.rabbit.listener; import org.springframework.amqp.event.AmqpEvent; +import org.springframework.lang.Nullable; /** * Published when a listener consumer fails. @@ -43,7 +44,7 @@ public class ListenerContainerConsumerFailedEvent extends AmqpEvent { * @param fatal true if the startup failure was fatal (will not be retried). */ public ListenerContainerConsumerFailedEvent(Object source, String reason, - Throwable throwable, boolean fatal) { + @Nullable Throwable throwable, boolean fatal) { super(source); this.reason = reason; this.fatal = fatal; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java index 687eb766..69b2e77b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java @@ -24,6 +24,7 @@ import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListener import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; @@ -35,6 +36,7 @@ import org.springframework.util.Assert; * * @author Stephane Nicoll * @author Artem Bilan + * @author Gary Russell * * @since 1.4 */ @@ -155,6 +157,7 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint return new MessagingMessageListenerAdapter(this.bean, this.method, this.returnExceptions, this.errorHandler); } + @Nullable private String getDefaultReplyToAddress() { Method method = getMethod(); if (method != null) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java index a2e6f6e6..0d8ba763 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -27,6 +27,7 @@ import java.util.Map; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.MessageProperties; +import org.springframework.lang.Nullable; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; @@ -180,7 +181,8 @@ public class DefaultMessagePropertiesConverter implements MessagePropertiesConve * @param value the value. * @return the converted value. */ - private Object convertHeaderValueIfNecessary(Object value) { + @Nullable + private Object convertHeaderValueIfNecessary(@Nullable Object value) { boolean valid = (value instanceof String) || (value instanceof byte[]) || (value instanceof Boolean) || (value instanceof LongString) || (value instanceof Integer) || (value instanceof Long) || (value instanceof Float) || (value instanceof Double) || (value instanceof BigDecimal)