diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java b/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java index 79f6d01e..e31c3127 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java @@ -48,7 +48,7 @@ public class Address { */ public static final String AMQ_RABBITMQ_REPLY_TO = "amq.rabbitmq.reply-to"; - private static final Pattern pattern = Pattern.compile("^(?:.*://)?([^/]*)/?(.*)$"); + private static final Pattern ADDRESS_PATTERN = Pattern.compile("^(?:.*://)?([^/]*)/?(.*)$"); private final String exchangeName; @@ -77,7 +77,7 @@ public class Address { this.exchangeName = ""; } else { - Matcher matcher = pattern.matcher(address); + Matcher matcher = ADDRESS_PATTERN.matcher(address); boolean matchFound = matcher.find(); if (matchFound) { this.exchangeName = matcher.group(1); @@ -131,7 +131,8 @@ public class Address { @Override public int hashCode() { int result = this.exchangeName != null ? this.exchangeName.hashCode() : 0; - result = 31 * result + (this.routingKey != null ? this.routingKey.hashCode() : 0); + int prime = 31; // NOSONAR magic # + result = prime * result + (this.routingKey != null ? this.routingKey.hashCode() : 0); return result; } diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpTemplate.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpTemplate.java index 645b6cdc..10ffd47c 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpTemplate.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpTemplate.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. @@ -17,7 +17,6 @@ package org.springframework.amqp.core; import org.springframework.amqp.AmqpException; -import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.core.ParameterizedTypeReference; import org.springframework.lang.Nullable; @@ -26,8 +25,9 @@ import org.springframework.lang.Nullable; * * Provides synchronous send and receive methods. The {@link #convertAndSend(Object)} and * {@link #receiveAndConvert()} methods allow let you send and receive POJO objects. - * Implementations are expected to delegate to an instance of {@link MessageConverter} to - * perform conversion to and from AMQP byte[] payload type. + * Implementations are expected to delegate to an instance of + * {@link org.springframework.amqp.support.converter.MessageConverter} to perform + * conversion to and from AMQP byte[] payload type. * * @author Mark Pollack * @author Mark Fisher diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AnonymousQueue.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AnonymousQueue.java index 2abf3ec0..1ef1b3e6 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/AnonymousQueue.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/AnonymousQueue.java @@ -150,7 +150,7 @@ public class AnonymousQueue extends Queue { @Override public String generateName() { UUID uuid = UUID.randomUUID(); - ByteBuffer bb = ByteBuffer.wrap(new byte[16]); + ByteBuffer bb = ByteBuffer.wrap(new byte[16]); // NOSONAR - Magic # deprecated anyway bb.putLong(uuid.getMostSignificantBits()) .putLong(uuid.getLeastSignificantBits()); // Convert to base64 and remove trailing = diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/Base64UrlNamingStrategy.java b/spring-amqp/src/main/java/org/springframework/amqp/core/Base64UrlNamingStrategy.java index 087ca50c..593acca4 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/Base64UrlNamingStrategy.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/Base64UrlNamingStrategy.java @@ -34,6 +34,8 @@ import org.springframework.util.Base64Utils; */ public class Base64UrlNamingStrategy implements NamingStrategy { + private static final int SIXTEEN = 16; + /** * The default instance - using {@code spring.gen-} as the prefix. */ @@ -60,7 +62,7 @@ public class Base64UrlNamingStrategy implements NamingStrategy { @Override public String generateName() { UUID uuid = UUID.randomUUID(); - ByteBuffer bb = ByteBuffer.wrap(new byte[16]); + ByteBuffer bb = ByteBuffer.wrap(new byte[SIXTEEN]); bb.putLong(uuid.getMostSignificantBits()) .putLong(uuid.getLeastSignificantBits()); // Convert to base64 and remove trailing = diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java b/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java index c635594e..a1cf1b2a 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java @@ -35,6 +35,8 @@ import java.util.Map; */ public class MessageProperties implements Serializable { + private static final int INT_MASK = 32; + private static final long serialVersionUID = 1619000546531112290L; public static final String CONTENT_TYPE_BYTES = "application/octet-stream"; @@ -123,11 +125,11 @@ public class MessageProperties implements Serializable { private volatile long publishSequenceNumber; - private volatile transient Type inferredArgumentType; + private transient volatile Type inferredArgumentType; - private volatile transient Method targetMethod; + private transient volatile Method targetMethod; - private volatile transient Object targetBean; + private transient volatile Object targetBean; public void setHeader(String key, Object value) { this.headers.put(key, value); @@ -538,11 +540,11 @@ public class MessageProperties implements Serializable { result = prime * result + ((this.appId == null) ? 0 : this.appId.hashCode()); result = prime * result + ((this.clusterId == null) ? 0 : this.clusterId.hashCode()); result = prime * result + ((this.contentEncoding == null) ? 0 : this.contentEncoding.hashCode()); - result = prime * result + (int) (this.contentLength ^ (this.contentLength >>> 32)); + result = prime * result + (int) (this.contentLength ^ (this.contentLength >>> INT_MASK)); result = prime * result + ((this.contentType == null) ? 0 : this.contentType.hashCode()); result = prime * result + this.correlationId.hashCode(); result = prime * result + ((this.deliveryMode == null) ? 0 : this.deliveryMode.hashCode()); - result = prime * result + (int) (this.deliveryTag ^ (this.deliveryTag >>> 32)); + result = prime * result + (int) (this.deliveryTag ^ (this.deliveryTag >>> INT_MASK)); result = prime * result + ((this.expiration == null) ? 0 : this.expiration.hashCode()); result = prime * result + this.headers.hashCode(); result = prime * result + ((this.messageCount == null) ? 0 : this.messageCount.hashCode()); diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java b/spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java index 9638be92..601cf8be 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java @@ -27,7 +27,7 @@ import java.util.Map; */ public final class QueueBuilder extends AbstractBuilder { - private static final NamingStrategy namingStrategy = Base64UrlNamingStrategy.DEFAULT; + private static final NamingStrategy namingStrategy = Base64UrlNamingStrategy.DEFAULT; // NOSONAR lower case private final String name; diff --git a/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpClientInterceptor.java b/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpClientInterceptor.java index 99cf648f..c1afac9f 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpClientInterceptor.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpClientInterceptor.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. @@ -22,7 +22,6 @@ import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.springframework.amqp.core.AmqpTemplate; -import org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter; import org.springframework.remoting.RemoteProxyFailureException; import org.springframework.remoting.support.DefaultRemoteInvocationFactory; import org.springframework.remoting.support.RemoteAccessor; @@ -36,7 +35,7 @@ import org.springframework.remoting.support.RemoteInvocationResult; * @author David Bilge * @author Gary Russell * @since 1.2 - * @see AmqpInvokerServiceExporter + * @see org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter * @see AmqpProxyFactoryBean * @see org.springframework.remoting.RemoteAccessException */ diff --git a/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpProxyFactoryBean.java b/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpProxyFactoryBean.java index 3deb15c4..48a64195 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpProxyFactoryBean.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpProxyFactoryBean.java @@ -16,13 +16,10 @@ package org.springframework.amqp.remoting.client; -import org.springframework.amqp.core.AmqpTemplate; -import org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter; import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; -import org.springframework.remoting.rmi.RmiServiceExporter; /** * {@link FactoryBean} for AMQP proxies. Exposes the proxied service for use as a bean reference, using the specified @@ -32,8 +29,9 @@ import org.springframework.remoting.rmi.RmiServiceExporter; * This is intended for an "RMI-style" (i.e. synchroneous) usage of the AMQP protocol. Obviously, AMQP allows for a much * broader scope of execution styles, which are not the scope of the mechanism at hand. *

- * Calling a method on the proxy will cause an AMQP message being sent according to the configured {@link AmqpTemplate}. - * This can be received and answered by an {@link AmqpInvokerServiceExporter}. + * Calling a method on the proxy will cause an AMQP message being sent according to the configured + * {@link org.springframework.amqp.core.AmqpTemplate}. + * This can be received and answered by an {@link org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter}. * * @author David Bilge * @author Gary Russell @@ -41,7 +39,7 @@ import org.springframework.remoting.rmi.RmiServiceExporter; * @since 1.2 * @see #setServiceInterface * @see AmqpClientInterceptor - * @see RmiServiceExporter + * @see org.springframework.remoting.rmi.RmiServiceExporter * @see org.springframework.remoting.RemoteAccessException */ public class AmqpProxyFactoryBean extends AmqpClientInterceptor implements FactoryBean, BeanClassLoaderAware, diff --git a/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java b/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java index d71d7220..e4841367 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java @@ -22,7 +22,6 @@ import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.MessageProperties; -import org.springframework.amqp.remoting.client.AmqpProxyFactoryBean; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.remoting.support.RemoteInvocation; @@ -31,7 +30,7 @@ import org.springframework.remoting.support.RemoteInvocationResult; /** * This message listener exposes a plain java service via AMQP. Such services can be accessed via plain AMQP or via - * {@link AmqpProxyFactoryBean}. + * {@link org.springframework.amqp.remoting.client.AmqpProxyFactoryBean}. * * To configure this message listener so that it actually receives method calls via AMQP, it needs to be put into a * listener container. See {@link MessageListener}. diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/AmqpHeaders.java b/spring-amqp/src/main/java/org/springframework/amqp/support/AmqpHeaders.java index 470f7851..58ca7b7d 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/AmqpHeaders.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/AmqpHeaders.java @@ -65,7 +65,7 @@ public abstract class AmqpHeaders { public static final String RECEIVED_DELAY = PREFIX + "receivedDelay"; - public final static String RECEIVED_DELIVERY_MODE = PREFIX + "receivedDeliveryMode"; + public static final String RECEIVED_DELIVERY_MODE = PREFIX + "receivedDeliveryMode"; public static final String RECEIVED_EXCHANGE = PREFIX + "receivedExchange"; diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DelegatingDecompressingPostProcessor.java b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DelegatingDecompressingPostProcessor.java index 1e4e1c88..7af2bd4b 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DelegatingDecompressingPostProcessor.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DelegatingDecompressingPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-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,7 +22,6 @@ import java.util.Map; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; -import org.springframework.amqp.core.MessageProperties; import org.springframework.core.Ordered; /** @@ -69,7 +68,7 @@ public class DelegatingDecompressingPostProcessor implements MessagePostProcesso /** * Remove the decompressor for this encoding; content will not be decompressed even if the - * {@link MessageProperties#SPRING_AUTO_DECOMPRESS} header is true. + * {@link org.springframework.amqp.core.MessageProperties#SPRING_AUTO_DECOMPRESS} header is true. * @param contentEncoding the content encoding. * @return the decompressor if it was present. */ diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/GZipPostProcessor.java b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/GZipPostProcessor.java index fbb9aef8..9bf2b9a7 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/GZipPostProcessor.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/GZipPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-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. @@ -20,12 +20,10 @@ import java.io.IOException; import java.io.OutputStream; import java.util.zip.GZIPOutputStream; -import org.springframework.amqp.core.MessageProperties; - /** - * A post processor that uses a {@link GZIPOutputStream} to compress the - * message body. Sets {@link MessageProperties#SPRING_AUTO_DECOMPRESS} to true - * by default. + * A post processor that uses a {@link GZIPOutputStream} to compress the message body. + * Sets {@link org.springframework.amqp.core.MessageProperties#SPRING_AUTO_DECOMPRESS} to + * true by default. * * @author Gary Russell * @since 1.4.2 diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/ZipPostProcessor.java b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/ZipPostProcessor.java index 6bbf37a1..6e7ea3c5 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/ZipPostProcessor.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/ZipPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-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. @@ -21,11 +21,9 @@ import java.io.OutputStream; import java.util.zip.ZipEntry; import java.util.zip.ZipOutputStream; -import org.springframework.amqp.core.MessageProperties; - /** - * A post processor that uses a {@link ZipOutputStream} to compress the - * message body. Sets {@link MessageProperties#SPRING_AUTO_DECOMPRESS} to true + * A post processor that uses a {@link ZipOutputStream} to compress the message body. Sets + * {@link org.springframework.amqp.core.MessageProperties#SPRING_AUTO_DECOMPRESS} to true * by default. * * @author Gary Russell 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 9740a9a7..e9d21d9e 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 @@ -33,7 +33,6 @@ import java.util.concurrent.TimeoutException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.junit.Assume; -import org.junit.internal.AssumptionViolatedException; import org.junit.rules.TestWatcher; import org.junit.runner.Description; import org.junit.runners.model.Statement; @@ -77,10 +76,12 @@ import com.rabbitmq.http.client.Client; * * @since 1.7 * @see Assume - * @see AssumptionViolatedException + * @see org.junit.internal.AssumptionViolatedException */ public final class BrokerRunning extends TestWatcher { + private static final int SIXTEEN = 16; + public static final String BROKER_ADMIN_URI = "RABBITMQ_TEST_ADMIN_URI"; public static final String BROKER_HOSTNAME = "RABBITMQ_TEST_HOSTNAME"; @@ -101,15 +102,15 @@ public final class BrokerRunning extends TestWatcher { private static final String GUEST = "guest"; - private static final Log logger = LogFactory.getLog(BrokerRunning.class); + private static final Log logger = LogFactory.getLog(BrokerRunning.class); // NOSONAR - lower case // Static so that we only test once on failure: speeds up test suite - private static final Map brokerOnline = new HashMap(); + private static final Map brokerOnline = new HashMap(); // NOSONAR - lower case // Static so that we only test once on failure - private static final Map brokerOffline = new HashMap(); + private static final Map brokerOffline = new HashMap(); // NOSONAR - lower case - private static final Map environmentOverrides = new HashMap<>(); + private static final Map environmentOverrides = new HashMap<>(); // NOSONAR - lower case private final boolean assumeOnline; @@ -461,7 +462,7 @@ public final class BrokerRunning extends TestWatcher { */ public String generateId() { UUID uuid = UUID.randomUUID(); - ByteBuffer bb = ByteBuffer.wrap(new byte[16]); + ByteBuffer bb = ByteBuffer.wrap(new byte[SIXTEEN]); bb.putLong(uuid.getMostSignificantBits()) .putLong(uuid.getLeastSignificantBits()); return "SpringBrokerRunning." + Base64Utils.encodeToUrlSafeString(bb.array()).replaceAll("=", ""); diff --git a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/LongRunningIntegrationTest.java b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/LongRunningIntegrationTest.java index e665ed31..2d1b4f7d 100644 --- a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/LongRunningIntegrationTest.java +++ b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/LongRunningIntegrationTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2017 the original author or authors. + * Copyright 2013-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. @@ -34,7 +34,7 @@ import org.junit.runners.model.Statement; */ public class LongRunningIntegrationTest extends TestWatcher { - private final static Log logger = LogFactory.getLog(LongRunningIntegrationTest.class); + private static final Log logger = LogFactory.getLog(LongRunningIntegrationTest.class); // NOSONAR - lower case public static final String RUN_LONG_INTEGRATION_TESTS = "RUN_LONG_INTEGRATION_TESTS"; diff --git a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCondition.java b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCondition.java index 1a9d0a35..5d3e8db0 100644 --- a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCondition.java +++ b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCondition.java @@ -50,7 +50,7 @@ public class RabbitAvailableCondition implements ExecutionCondition, AfterAllCal private static final ConditionEvaluationResult ENABLED = ConditionEvaluationResult.enabled( "@RabbitAvailable is not present"); - private static final ThreadLocal brokerRunningHolder = new ThreadLocal<>(); + private static final ThreadLocal brokerRunningHolder = new ThreadLocal<>(); // NOSONAR - lower case @Override public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) { diff --git a/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java b/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java index 55a8755c..b5d3015e 100644 --- a/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java +++ b/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java @@ -141,8 +141,9 @@ public class TestRabbitTemplate extends RabbitTemplate implements ApplicationCon AbstractAdaptableMessageListener adapter = (AbstractAdaptableMessageListener) listener; willAnswer(i -> { Envelope envelope = new Envelope(1, false, "", REPLY_QUEUE); - reply.set(MessageBuilder.withBody(i.getArgument(4)) - .andProperties(getMessagePropertiesConverter().toMessageProperties(i.getArgument(3), envelope, + reply.set(MessageBuilder.withBody(i.getArgument(4)) // NOSONAR magic # + .andProperties(getMessagePropertiesConverter() + .toMessageProperties(i.getArgument(3), envelope, // NOSONAR magic # adapter.getEncoding())) .build()); return null; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java index 6511443d..f2da037a 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java @@ -32,7 +32,6 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint; -import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener; import org.springframework.amqp.support.ConsumerTagStrategy; import org.springframework.amqp.support.converter.MessageConverter; @@ -161,7 +160,7 @@ public abstract class AbstractRabbitListenerContainerFactoryThis should be the default for most users and a good transition paths - * for those that are used to build such container definition manually. + *

+ * This should be the default for most users and a good transition paths for those that + * are used to build such container definition manually. * * @author Stephane Nicoll * @author Gary Russell diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerEndpoint.java index 01d238f8..8b63766b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-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. @@ -20,11 +20,11 @@ package org.springframework.amqp.rabbit.config; import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.rabbit.listener.AbstractRabbitListenerEndpoint; import org.springframework.amqp.rabbit.listener.MessageListenerContainer; -import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint; /** - * A {@link RabbitListenerEndpoint} simply providing the {@link MessageListener} to - * invoke to process an incoming message for this endpoint. + * A {@link org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint} simply + * providing the {@link MessageListener} to invoke to process an incoming message for this + * endpoint. * * @author Stephane Nicoll * @since 1.4 diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index b0e9ab81..09340b54 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -108,15 +108,15 @@ public class CachingConnectionFactory extends AbstractConnectionFactory /** * Create a unique ID for the pool. */ - private static final AtomicInteger threadPoolId = new AtomicInteger(); + private static final AtomicInteger threadPoolId = new AtomicInteger(); // NOSONAR lower case - private static final Set txStarts = new HashSet<>(Arrays.asList("basicPublish", "basicAck", + private static final Set txStarts = new HashSet<>(Arrays.asList("basicPublish", "basicAck", // NOSONAR "basicNack", "basicReject")); - private static final Set ackMethods = new HashSet<>(Arrays.asList("basicAck", + private static final Set ackMethods = new HashSet<>(Arrays.asList("basicAck", // NOSONAR "basicNack", "basicReject")); - private static final Set txEnds = new HashSet<>(Arrays.asList("txCommit", "txRollback")); + private static final Set txEnds = new HashSet<>(Arrays.asList("txCommit", "txRollback")); // NOSONAR private final ChannelCachingConnectionProxy connection = new ChannelCachingConnectionProxy(null); @@ -471,10 +471,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory public void shutdownCompleted(ShutdownSignalException cause) { this.closeExceptionLogger.log(logger, "Channel shutdown", cause); int protocolClassId = cause.getReason().protocolClassId(); - if (protocolClassId == 20) { + if (protocolClassId == RabbitUtils.CHANNEL_PROTOCOL_CLASS_ID_20) { getChannelListener().onShutDown(cause); } - else if (protocolClassId == 10) { + else if (protocolClassId == RabbitUtils.CONNECTION_PROTOCOL_CLASS_ID_10) { getConnectionListener().onShutDown(cause); } @@ -489,10 +489,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory ChannelProxy channel = null; if (connection.isOpen()) { channel = findOpenChannel(channelList, channel); - if (channel != null) { - if (logger.isTraceEnabled()) { - logger.trace("Found cached Rabbit Channel: " + channel.toString()); - } + if (channel != null && logger.isTraceEnabled()) { + logger.trace("Found cached Rabbit Channel: " + channel.toString()); } } if (channel == null) { @@ -662,10 +660,9 @@ public class CachingConnectionFactory extends AbstractConnectionFactory logger.error("Could not configure the channel to receive publisher confirms", e); } } - if (this.publisherConfirms || this.publisherReturns) { - if (!(channel instanceof PublisherCallbackChannelImpl)) { - channel = new PublisherCallbackChannelImpl(channel, getChannelsExecutor()); - } + if ((this.publisherConfirms || this.publisherReturns) + && !(channel instanceof PublisherCallbackChannelImpl)) { + channel = new PublisherCallbackChannelImpl(channel, getChannelsExecutor()); } if (channel != null) { channel.addShutdownListener(this); @@ -996,6 +993,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory private final class CachedChannelInvocationHandler implements InvocationHandler { + private static final int ASYNC_CLOSE_TIMEOUT = 5_000; + private final ChannelCachingConnectionProxy theConnection; private final LinkedList channelList; // NOSONAR addLast() @@ -1278,10 +1277,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory executorService.execute(() -> { try { if (CachingConnectionFactory.this.publisherConfirms) { - channel.waitForConfirmsOrDie(5000); + channel.waitForConfirmsOrDie(ASYNC_CLOSE_TIMEOUT); } else { - Thread.sleep(5000); + Thread.sleep(ASYNC_CLOSE_TIMEOUT); } } catch (InterruptedException e1) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ClosingRecoveryListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ClosingRecoveryListener.java index e025d9c2..9da6d3a5 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ClosingRecoveryListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ClosingRecoveryListener.java @@ -39,11 +39,12 @@ import com.rabbitmq.client.impl.recovery.AutorecoveringChannel; */ public final class ClosingRecoveryListener implements RecoveryListener { - private static final Log logger = LogFactory.getLog(ClosingRecoveryListener.class); + private static final Log logger = LogFactory.getLog(ClosingRecoveryListener.class); // NOSONAR - lower case private static final RecoveryListener INSTANCE = new ClosingRecoveryListener(); - private static final ConcurrentMap hasListener = new ConcurrentHashMap<>(); + private static final ConcurrentMap hasListener // NOSONAR - lower case + = new ConcurrentHashMap<>(); private ClosingRecoveryListener() { super(); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConsumerChannelRegistry.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConsumerChannelRegistry.java index 2718f826..92f4ffc6 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConsumerChannelRegistry.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConsumerChannelRegistry.java @@ -36,9 +36,10 @@ import com.rabbitmq.client.Channel; */ public final class ConsumerChannelRegistry { - private static final Log logger = LogFactory.getLog(ConsumerChannelRegistry.class); + private static final Log logger = LogFactory.getLog(ConsumerChannelRegistry.class); // NOSONAR - lower case - private static final ThreadLocal consumerChannel = new ThreadLocal(); + private static final ThreadLocal consumerChannel // NOSONAR - lower case + = new ThreadLocal(); private ConsumerChannelRegistry() { super(); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java index c0633d93..e87d6152 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java @@ -92,7 +92,8 @@ import com.rabbitmq.client.impl.recovery.AutorecoveringChannel; public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, ConfirmListener, ReturnListener, ShutdownListener { - private static final MessagePropertiesConverter converter = new DefaultMessagePropertiesConverter(); + private static final MessagePropertiesConverter converter // NOSONAR - lower case + = new DefaultMessagePropertiesConverter(); private final Log logger = LogFactory.getLog(this.getClass()); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java index cc2a0712..28ce47d0 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java @@ -27,8 +27,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.amqp.AmqpException; import org.springframework.amqp.AmqpIOException; -import org.springframework.amqp.rabbit.core.RabbitTemplate; -import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager; import org.springframework.lang.Nullable; import org.springframework.transaction.support.ResourceHolderSupport; import org.springframework.util.Assert; @@ -49,14 +47,12 @@ import com.rabbitmq.client.Channel; * @author Dave Syer * @author Gary Russell * - * @see RabbitTransactionManager - * @see RabbitTemplate + * @see org.springframework.amqp.rabbit.transaction.RabbitTransactionManager + * @see org.springframework.amqp.rabbit.core.RabbitTemplate */ public class RabbitResourceHolder extends ResourceHolderSupport { - private static final Log logger = LogFactory.getLog(RabbitResourceHolder.class); - - private final boolean frozen = false; + private static final Log logger = LogFactory.getLog(RabbitResourceHolder.class); // NOSONAR - lower case private final List connections = new LinkedList<>(); @@ -87,8 +83,9 @@ public class RabbitResourceHolder extends ResourceHolderSupport { this.releaseAfterCompletion = releaseAfterCompletion; } + @Deprecated public final boolean isFrozen() { - return this.frozen; + return false; } /** @@ -112,7 +109,6 @@ public class RabbitResourceHolder extends ResourceHolderSupport { } public final void addConnection(Connection connection) { - Assert.isTrue(!this.frozen, "Cannot add Connection because RabbitResourceHolder is frozen"); Assert.notNull(connection, "Connection must not be null"); if (!this.connections.contains(connection)) { this.connections.add(connection); @@ -124,7 +120,6 @@ public class RabbitResourceHolder extends ResourceHolderSupport { } public final void addChannel(Channel channel, @Nullable Connection connection) { - Assert.isTrue(!this.frozen, "Cannot add Channel because RabbitResourceHolder is frozen"); Assert.notNull(channel, "Channel must not be null"); if (!this.channels.contains(channel)) { this.channels.add(channel); 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 d4f04e42..500b6fc3 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 @@ -42,9 +42,44 @@ import com.rabbitmq.client.impl.recovery.AutorecoveringChannel; */ public abstract class RabbitUtils { - private static final Log logger = LogFactory.getLog(RabbitUtils.class); + /** + * AMQP declare method. + */ + public static final int DECLARE_METHOD_ID_10 = 10; - private static final ThreadLocal physicalCloseRequired = new ThreadLocal<>(); + /** + * AMQP consume method. + */ + public static final int CONSUME_METHOD_ID_20 = 20; + + /** + * AMQP exchange class id. + */ + public static final int EXCHANGE_CLASS_ID_40 = 40; + + /** + * AMQP queue class id. + */ + public static final int QUEUE_CLASS_ID_50 = 50; + + /** + * AMQP basic class id. + */ + public static final int BASIC_CLASS_ID_60 = 60; + + /** + * AMQP Connection protocol class id. + */ + public static final int CONNECTION_PROTOCOL_CLASS_ID_10 = 10; + + /** + * AMQP Channel protocol class id. + */ + public static final int CHANNEL_PROTOCOL_CLASS_ID_20 = 20; + + private static final Log logger = LogFactory.getLog(RabbitUtils.class); // NOSONAR - lower case + + private static final ThreadLocal physicalCloseRequired = new ThreadLocal<>(); // NOSONAR - lower case /** * Close the given RabbitMQ Connection and ignore any thrown exception. This is useful for typical @@ -235,9 +270,9 @@ public abstract class RabbitUtils { Method shutdownReason = sig.getReason(); return shutdownReason instanceof AMQP.Channel.Close // NOSONAR boolean complexity && AMQP.NOT_FOUND == ((AMQP.Channel.Close) shutdownReason).getReplyCode() - && ((((AMQP.Channel.Close) shutdownReason).getClassId() == 40 // exchange - || ((AMQP.Channel.Close) shutdownReason).getClassId() == 50) // queue - && ((AMQP.Channel.Close) shutdownReason).getMethodId() == 10); // declare + && ((((AMQP.Channel.Close) shutdownReason).getClassId() == EXCHANGE_CLASS_ID_40 + || ((AMQP.Channel.Close) shutdownReason).getClassId() == QUEUE_CLASS_ID_50) + && ((AMQP.Channel.Close) shutdownReason).getMethodId() == DECLARE_METHOD_ID_10); } /** @@ -251,8 +286,8 @@ public abstract class RabbitUtils { Method shutdownReason = sig.getReason(); return shutdownReason instanceof AMQP.Channel.Close // NOSONAR boolean complexity && AMQP.ACCESS_REFUSED == ((AMQP.Channel.Close) shutdownReason).getReplyCode() - && ((AMQP.Channel.Close) shutdownReason).getClassId() == 60 // basic - && ((AMQP.Channel.Close) shutdownReason).getMethodId() == 20 // consume + && ((AMQP.Channel.Close) shutdownReason).getClassId() == BASIC_CLASS_ID_60 + && ((AMQP.Channel.Close) shutdownReason).getMethodId() == CONSUME_METHOD_ID_20 && ((AMQP.Channel.Close) shutdownReason).getReplyText().contains("exclusive"); } @@ -281,8 +316,8 @@ public abstract class RabbitUtils { Method shutdownReason = sig.getReason(); return shutdownReason instanceof AMQP.Channel.Close && AMQP.PRECONDITION_FAILED == ((AMQP.Channel.Close) shutdownReason).getReplyCode() - && ((AMQP.Channel.Close) shutdownReason).getClassId() == 50 // queue - && ((AMQP.Channel.Close) shutdownReason).getMethodId() == 10; // declare + && ((AMQP.Channel.Close) shutdownReason).getClassId() == QUEUE_CLASS_ID_50 + && ((AMQP.Channel.Close) shutdownReason).getMethodId() == DECLARE_METHOD_ID_10; } } @@ -311,8 +346,8 @@ public abstract class RabbitUtils { Method shutdownReason = sig.getReason(); return shutdownReason instanceof AMQP.Connection.Close && AMQP.COMMAND_INVALID == ((AMQP.Connection.Close) shutdownReason).getReplyCode() - && ((AMQP.Connection.Close) shutdownReason).getClassId() == 40 // exchange - && ((AMQP.Connection.Close) shutdownReason).getMethodId() == 10; // declare + && ((AMQP.Connection.Close) shutdownReason).getClassId() == EXCHANGE_CLASS_ID_40 + && ((AMQP.Connection.Close) shutdownReason).getMethodId() == DECLARE_METHOD_ID_10; } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BrokerEventListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BrokerEventListener.java index 9059c45b..2887520a 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BrokerEventListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BrokerEventListener.java @@ -56,7 +56,7 @@ import org.springframework.util.ObjectUtils; public class BrokerEventListener implements MessageListener, ApplicationEventPublisherAware, ConnectionListener, SmartLifecycle { - private static final Log logger = LogFactory.getLog(BrokerEventListener.class); + private static final Log logger = LogFactory.getLog(BrokerEventListener.class); // NOSONAR - lower case private final AbstractMessageListenerContainer container; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java index aef5639d..3f1135a1 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java @@ -78,6 +78,14 @@ import com.rabbitmq.client.Channel; public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, ApplicationEventPublisherAware, BeanNameAware, InitializingBean { + private static final int DECLARE_MAX_ATTEMPTS = 5; + + private static final int DECLARE_INITIAL_RETRY_INTERVAL = 1000; + + private static final int DECLARE_MAX_RETRY_INTERVAL = 5000; + + private static final double DECLARE_RETRY_MULTIPLIER = 2.0; + /** * The default exchange name. */ @@ -478,11 +486,11 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat if (this.retryTemplate == null && !this.retryDisabled) { this.retryTemplate = new RetryTemplate(); - this.retryTemplate.setRetryPolicy(new SimpleRetryPolicy(5)); + this.retryTemplate.setRetryPolicy(new SimpleRetryPolicy(DECLARE_MAX_ATTEMPTS)); ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy(); - backOffPolicy.setInitialInterval(1000); - backOffPolicy.setMultiplier(2.0); - backOffPolicy.setMaxInterval(5000); + backOffPolicy.setInitialInterval(DECLARE_INITIAL_RETRY_INTERVAL); + backOffPolicy.setMultiplier(DECLARE_RETRY_MULTIPLIER); + backOffPolicy.setMaxInterval(DECLARE_MAX_RETRY_INTERVAL); this.retryTemplate.setBackOffPolicy(backOffPolicy); } if (this.connectionFactory instanceof CachingConnectionFactory && diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitOperations.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitOperations.java index d0c3d751..0b45f10c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitOperations.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitOperations.java @@ -429,7 +429,7 @@ public interface RabbitOperations extends AmqpTemplate { * @since 2.0 */ @FunctionalInterface - public interface OperationsCallback { + interface OperationsCallback { /** * Execute any number of operations using a dedicated 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 44bba10c..dae66dc3 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 @@ -1947,10 +1947,10 @@ public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count/co (RetryCallback) context -> doExecute(action, connectionFactory), (RecoveryCallback) this.recoveryCallback); } + catch (RuntimeException e) { // NOSONAR catch and rethrow needed to avoid next catch + throw e; + } catch (Exception e) { - if (e instanceof RuntimeException) { - throw (RuntimeException) e; - } throw RabbitExceptionTranslator.convertRabbitAccessException(e); } } @@ -2580,7 +2580,7 @@ public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count/co * Adds {@link #toString()} to the {@link DefaultConsumer}. * @since 2.0 */ - protected static abstract class TemplateConsumer extends DefaultConsumer { + protected abstract static class TemplateConsumer extends DefaultConsumer { public TemplateConsumer(Channel channel) { super(channel); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/SimpleBatchingStrategy.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/SimpleBatchingStrategy.java index 5d8ca295..801f3e56 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/SimpleBatchingStrategy.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/SimpleBatchingStrategy.java @@ -79,7 +79,7 @@ public class SimpleBatchingStrategy implements BatchingStrategy { else { this.routingKey = routingKey; } - int bufferUse = 4 + message.getBody().length; + int bufferUse = Integer.BYTES + message.getBody().length; MessageBatch batch = null; if (this.messages.size() > 0 && this.currentSize + bufferUse > this.bufferLimit) { batch = doReleaseBatch(); 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 7f8b5302..da7afc17 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 @@ -1169,12 +1169,9 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor this.taskExecutor = new SimpleAsyncTaskExecutor(this.getBeanName() + "-"); this.taskExecutorSet = true; } - if (this.transactionManager != null) { - if (!isChannelTransacted()) { - logger.debug("The 'channelTransacted' is coerced to 'true', when 'transactionManager' is provided"); - setChannelTransacted(true); - } - + if (this.transactionManager != null && !isChannelTransacted()) { + logger.debug("The 'channelTransacted' is coerced to 'true', when 'transactionManager' is provided"); + setChannelTransacted(true); } this.initialized = true; } @@ -1646,12 +1643,10 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor this.amqpAdmin = admins.values().iterator().next(); } else { - if (isAutoDeclare() || isMismatchedQueuesFatal()) { - if (logger.isDebugEnabled()) { - logger.debug("For 'autoDeclare' and 'mismatchedQueuesFatal' to work, there must be exactly one " - + "AmqpAdmin in the context or you must inject one into this container; found: " - + admins.size() + " for container " + this.toString()); - } + if ((isAutoDeclare() || isMismatchedQueuesFatal()) && this.logger.isDebugEnabled()) { + logger.debug("For 'autoDeclare' and 'mismatchedQueuesFatal' to work, there must be exactly one " + + "AmqpAdmin in the context or you must inject one into this container; found: " + + admins.size() + " for container " + this.toString()); } if (isMismatchedQueuesFatal()) { throw new IllegalStateException("When 'mismatchedQueuesFatal' is 'true', there must be exactly " diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java index a7f46143..d774eb29 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java @@ -25,8 +25,6 @@ import java.util.Map; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.Queue; -import org.springframework.amqp.rabbit.config.SimpleRabbitListenerEndpoint; -import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -49,7 +47,7 @@ import org.springframework.util.Assert; * @since 1.4 * * @see MethodRabbitListenerEndpoint - * @see SimpleRabbitListenerEndpoint + * @see org.springframework.amqp.rabbit.config.SimpleRabbitListenerEndpoint */ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEndpoint, BeanFactoryAware { @@ -210,8 +208,8 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn } /** - * Set the {@link RabbitAdmin} instance to use. - * @param admin the {@link RabbitAdmin} instance. + * Set the {@link AmqpAdmin} instance to use. + * @param admin the {@link AmqpAdmin} instance. */ public void setAdmin(AmqpAdmin admin) { this.admin = admin; 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 92223766..2845d25a 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 @@ -88,6 +88,10 @@ import com.rabbitmq.utility.Utility; */ public class BlockingQueueConsumer { + private static final int DEFAULT_DECLARATION_RETRIES = 3; + + private static final int DEFAULT_RETRY_DECLARATION_INTERVAL = 60000; + private static Log logger = LogFactory.getLog(BlockingQueueConsumer.class); private final BlockingQueue queue; @@ -133,12 +137,12 @@ public class BlockingQueueConsumer { private final Set missingQueues = Collections.synchronizedSet(new HashSet()); - private long retryDeclarationInterval = 60000; + private long retryDeclarationInterval = DEFAULT_RETRY_DECLARATION_INTERVAL; private long failedDeclarationRetryInterval = AbstractMessageListenerContainer.DEFAULT_FAILED_DECLARATION_RETRY_INTERVAL; - private int declarationRetries = 3; + private int declarationRetries = DEFAULT_DECLARATION_RETRIES; private long lastRetryDeclaration; @@ -794,11 +798,9 @@ public class BlockingQueueConsumer { boolean ackRequired = !this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual(); - if (ackRequired) { - if (!this.transactional || isLocallyTransacted) { - long deliveryTag = new ArrayList(this.deliveryTags).get(this.deliveryTags.size() - 1); - this.channel.basicAck(deliveryTag, true); - } + if (ackRequired && (!this.transactional || isLocallyTransacted)) { + long deliveryTag = new ArrayList(this.deliveryTags).get(this.deliveryTags.size() - 1); + this.channel.basicAck(deliveryTag, true); } if (isLocallyTransacted) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java index a6f8dfd1..08540b8e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java @@ -28,8 +28,6 @@ import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException; import org.springframework.amqp.support.converter.MessageConversionException; import org.springframework.messaging.MessagingException; -import org.springframework.messaging.handler.annotation.support.MethodArgumentNotValidException; -import org.springframework.messaging.handler.annotation.support.MethodArgumentTypeMismatchException; import org.springframework.messaging.handler.invocation.MethodArgumentResolutionException; import org.springframework.util.ErrorHandler; @@ -43,7 +41,8 @@ import org.springframework.util.ErrorHandler; * The default strategy will do this if the exception is a * {@link ListenerExecutionFailedException} with a cause of {@link MessageConversionException}, * {@link org.springframework.messaging.converter.MessageConversionException}, - * {@link MethodArgumentNotValidException}, {@link MethodArgumentTypeMismatchException}, + * {@link org.springframework.messaging.handler.annotation.support.MethodArgumentNotValidException}, + * {@link org.springframework.messaging.handler.annotation.support.MethodArgumentTypeMismatchException}, * {@link NoSuchMethodException} or {@link ClassCastException}. *

* The exception will not be wrapped if the {@code cause} chain already contains an 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 10fcc685..b76fd537 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 @@ -92,6 +92,8 @@ import com.rabbitmq.client.ShutdownSignalException; */ public class DirectMessageListenerContainer extends AbstractMessageListenerContainer { + private static final int START_WAIT_TIME = 60; + private static final int DEFAULT_MONITOR_INTERVAL = 10_000; private static final int DEFAULT_ACK_TIMEOUT = 20_000; @@ -340,7 +342,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta private void checkStartState() { if (!this.isRunning()) { try { - Assert.state(this.startedLatch.await(60, TimeUnit.SECONDS), + Assert.state(this.startedLatch.await(START_WAIT_TIME, TimeUnit.SECONDS), "Container is not started - cannot adjust queues"); } catch (InterruptedException e) { @@ -458,11 +460,10 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } private void checkIdle(long idleEventInterval, long now) { - if (idleEventInterval > 0) { - if (now - getLastReceive() > idleEventInterval && now - this.lastAlertAt > idleEventInterval) { - publishIdleContainerEvent(now - getLastReceive()); - this.lastAlertAt = now; - } + if (idleEventInterval > 0 + && now - getLastReceive() > idleEventInterval && now - this.lastAlertAt > idleEventInterval) { + publishIdleContainerEvent(now - getLastReceive()); + this.lastAlertAt = now; } } @@ -663,7 +664,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } catch (Exception e) { addConsumerToRestart(new SimpleConsumer(null, null, queue)); - throw e instanceof AmqpConnectException + throw e instanceof AmqpConnectException // NOSONAR exception type check ? (AmqpConnectException) e : new AmqpConnectException(e); } @@ -920,6 +921,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) { + MessageProperties messageProperties = getMessagePropertiesConverter().toMessageProperties(properties, envelope, "UTF-8"); messageProperties.setConsumerTag(consumerTag); @@ -934,13 +936,14 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta try { executeListenerInTransaction(message, deliveryTag); } - catch (Throwable e) { // NOSONAR - errors are rethrown - if (e instanceof WrappedTransactionException) { - if (e.getCause() instanceof Error) { - throw (Error) e.getCause(); - } + catch (WrappedTransactionException e) { + if (e.getCause() instanceof Error) { + throw (Error) e.getCause(); } } + catch (Exception e) { + // empty + } finally { if (this.isRabbitTxManager) { ConsumerChannelRegistry.unRegisterConsumerChannel(); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java index 3d4c0517..1ff71db4 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java @@ -205,13 +205,11 @@ public class DirectReplyToMessageListenerContainer extends DirectMessageListener public void releaseConsumerFor(ChannelHolder channelHolder, boolean cancelConsumer, @Nullable String message) { synchronized (this.consumersMonitor) { SimpleConsumer consumer = this.inUseConsumerChannels.get(channelHolder.getChannel()); - if (consumer != null) { - if (consumer.getEpoch() == channelHolder.getConsumerEpoch()) { - this.inUseConsumerChannels.remove(channelHolder.getChannel()); - if (cancelConsumer) { - Assert.isTrue(message != null, "A 'message' is required when 'cancelConsumer' is 'true'"); - consumer.cancelConsumer("Consumer " + this + " canceled due to " + message); - } + if (consumer != null && consumer.getEpoch() == channelHolder.getConsumerEpoch()) { + this.inUseConsumerChannels.remove(channelHolder.getChannel()); + if (cancelConsumer) { + Assert.isTrue(message != null, "A 'message' is required when 'cancelConsumer' is 'true'"); + consumer.cancelConsumer("Consumer " + this + " canceled due to " + message); } } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerContainerIdleEvent.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerContainerIdleEvent.java index 25c9f39f..8f8c12fc 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerContainerIdleEvent.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerContainerIdleEvent.java @@ -16,6 +16,7 @@ package org.springframework.amqp.rabbit.listener; +import java.time.Duration; import java.util.Arrays; import java.util.List; @@ -75,7 +76,7 @@ public class ListenerContainerIdleEvent extends AmqpEvent { @Override public String toString() { return "ListenerContainerIdleEvent [idleTime=" - + ((float) this.idleTime / 1000) + "s, listenerId=" + this.listenerId + + Duration.ofMillis(this.idleTime) + ", listenerId=" + this.listenerId + ", container=" + getSource() + "]"; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MultiMethodRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MultiMethodRabbitListenerEndpoint.java index 62143d0b..259be9ce 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MultiMethodRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MultiMethodRabbitListenerEndpoint.java @@ -37,8 +37,6 @@ public class MultiMethodRabbitListenerEndpoint extends MethodRabbitListenerEndpo private final Method defaultMethod; - private DelegatingInvocableHandler delegatingHandler; - /** * Construct an instance for the provided methods and bean. * @param methods the methods. @@ -73,9 +71,8 @@ public class MultiMethodRabbitListenerEndpoint extends MethodRabbitListenerEndpo defaultHandler = handler; } } - this.delegatingHandler = new DelegatingInvocableHandler(invocableHandlerMethods, defaultHandler, - getBean(), getResolver(), getBeanExpressionContext()); - return new HandlerAdapter(this.delegatingHandler); + return new HandlerAdapter(new DelegatingInvocableHandler(invocableHandlerMethods, defaultHandler, + getBean(), getResolver(), getBeanExpressionContext())); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerContainerFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerContainerFactory.java index 6d87412f..ba972056 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerContainerFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerContainerFactory.java @@ -16,7 +16,6 @@ package org.springframework.amqp.rabbit.listener; -import org.springframework.amqp.core.MessageListener; import org.springframework.lang.Nullable; /** @@ -39,8 +38,9 @@ public interface RabbitListenerContainerFactoryBy default, {@link DefaultMessageHandlerMethodFactory} is used and it - * can be configured further to support additional method arguments - * or to customize conversion and validation support. See - * {@link DefaultMessageHandlerMethodFactory} javadoc for more details. + *

+ * By default, + * {@link org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory} + * is used and it can be configured further to support additional method arguments or + * to customize conversion and validation support. See + * {@link org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory} + * javadoc for more details. * @param rabbitHandlerMethodFactory the {@link MessageHandlerMethodFactory} instance. */ public void setMessageHandlerMethodFactory(MessageHandlerMethodFactory rabbitHandlerMethodFactory) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 5003905f..0aec2933 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.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. @@ -39,7 +39,6 @@ import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.AmqpIllegalStateException; import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.ImmediateAcknowledgeAmqpException; -import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -60,7 +59,6 @@ import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.Assert; -import org.springframework.util.backoff.BackOff; import org.springframework.util.backoff.BackOffExecution; import com.rabbitmq.client.Channel; @@ -79,6 +77,8 @@ import com.rabbitmq.client.ShutdownSignalException; */ public class SimpleMessageListenerContainer extends AbstractMessageListenerContainer { + private static final int RECOVERY_LOOP_WAIT_TIME = 200; + private static final long DEFAULT_START_CONSUMER_MIN_INTERVAL = 10000; private static final long DEFAULT_STOP_CONSUMER_MIN_INTERVAL = 60000; @@ -313,9 +313,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } /** - * Tells the container how many messages to process in a single transaction (if the channel is transactional). For - * best results it should be less than or equal to {@link #setPrefetchCount(int) the prefetch count}. Also affects - * how often acks are sent when using {@link AcknowledgeMode#AUTO} - one ack per txSize. Default is 1. + * Tells the container how many messages to process in a single transaction (if the + * channel is transactional). For best results it should be less than or equal to + * {@link #setPrefetchCount(int) the prefetch count}. Also affects how often acks are + * sent when using {@link org.springframework.amqp.core.AcknowledgeMode#AUTO} - one + * ack per txSize. Default is 1. * @param txSize the transaction size */ public void setTxSize(int txSize) { @@ -917,12 +919,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } - /** - * Wait for a period determined by the {@link #setRecoveryInterval(long) recoveryInterval} - * or {@link #setRecoveryBackOff(BackOff)} to give the container a - * chance to recover from consumer startup failure, e.g. if the broker is down. - * @param backOffExecution the BackOffExecution to get the {@code recoveryInterval} - */ protected void handleStartupFailure(BackOffExecution backOffExecution) { long recoveryInterval = backOffExecution.nextBackOff(); if (BackOffExecution.STOP == recoveryInterval) { @@ -940,7 +936,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } long timeout = System.currentTimeMillis() + recoveryInterval; while (isActive() && System.currentTimeMillis() < timeout) { - Thread.sleep(200); + Thread.sleep(RECOVERY_LOOP_WAIT_TIME); } } catch (InterruptedException e) { @@ -975,6 +971,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private final class AsyncMessageProcessingConsumer implements Runnable { + private static final int ABORT_EVENT_WAIT_SECONDS = 5; + private final BlockingQueueConsumer consumer; private final CountDownLatch start; @@ -1261,7 +1259,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta ListenerContainerConsumerFailedEvent event = null; do { try { - event = SimpleMessageListenerContainer.this.abortEvents.poll(5, TimeUnit.SECONDS); + event = SimpleMessageListenerContainer.this.abortEvents.poll(ABORT_EVENT_WAIT_SECONDS, + TimeUnit.SECONDS); if (event != null) { SimpleMessageListenerContainer.this.publishConsumerFailedEvent( event.getReason(), event.isFatal(), event.getThrowable()); 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 index cb793d00..4a6ff7a4 100644 --- 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 @@ -27,7 +27,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Address; import org.springframework.amqp.core.Message; -import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; @@ -56,8 +55,8 @@ import com.rabbitmq.client.Channel; import reactor.core.publisher.Mono; /** - * An abstract {@link MessageListener} adapter providing the necessary infrastructure - * to extract the payload of a {@link Message}. + * An abstract {@link org.springframework.amqp.core.MessageListener} adapter providing the + * necessary infrastructure to extract the payload of a {@link Message}. * * @author Stephane Nicoll * @author Gary Russell @@ -77,7 +76,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe private static final ParserContext PARSER_CONTEXT = new TemplateParserContext("!{", "}"); - private static final boolean monoPresent = + private static final boolean monoPresent = // NOSONAR - lower case ClassUtils.isPresent("reactor.core.publisher.Mono", ChannelAwareMessageListener.class.getClassLoader());; /** Logger available to subclasses. */ diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java index 81a4753c..9c8d4e3d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java @@ -39,7 +39,6 @@ import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.handler.annotation.Header; -import org.springframework.messaging.handler.annotation.Payload; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.util.Assert; @@ -47,8 +46,9 @@ import org.springframework.util.Assert; /** * Delegates to an {@link InvocableHandlerMethod} based on the message payload type. - * Matches a single, non-annotated parameter or one that is annotated with {@link Payload}. - * Matches must be unambiguous. + * Matches a single, non-annotated parameter or one that is annotated with + * {@link org.springframework.messaging.handler.annotation.Payload}. Matches must be + * unambiguous. * * @author Gary Russell * @author Artem Bilan @@ -221,24 +221,22 @@ public class DelegatingInvocableHandler { // Single param; no annotation or not @Header if (parameterAnnotations.length == 1) { MethodParameter methodParameter = new MethodParameter(method, 0); - if (methodParameter.getParameterAnnotations().length == 0 - || !methodParameter.hasParameterAnnotation(Header.class)) { - if (methodParameter.getParameterType().isAssignableFrom(payloadClass)) { - return true; - } + if ((methodParameter.getParameterAnnotations().length == 0 + || !methodParameter.hasParameterAnnotation(Header.class)) + && methodParameter.getParameterType().isAssignableFrom(payloadClass)) { + return true; } } boolean foundCandidate = false; for (int i = 0; i < parameterAnnotations.length; i++) { MethodParameter methodParameter = new MethodParameter(method, i); - if (methodParameter.getParameterAnnotations().length == 0 - || !methodParameter.hasParameterAnnotation(Header.class)) { - if (methodParameter.getParameterType().isAssignableFrom(payloadClass)) { - if (foundCandidate) { - throw new AmqpException("Ambiguous payload parameter for " + method.toGenericString()); - } - foundCandidate = true; + if ((methodParameter.getParameterAnnotations().length == 0 + || !methodParameter.hasParameterAnnotation(Header.class)) + && methodParameter.getParameterType().isAssignableFrom(payloadClass)) { + if (foundCandidate) { + throw new AmqpException("Ambiguous payload parameter for " + method.toGenericString()); } + foundCandidate = true; } } return foundCandidate; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java index 0f268f70..3f95f92b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java @@ -30,7 +30,6 @@ import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException; import org.springframework.amqp.support.converter.MessageConverter; -import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.util.Assert; import org.springframework.util.MethodInvoker; import org.springframework.util.ObjectUtils; @@ -47,7 +46,8 @@ import com.rabbitmq.client.Channel; * By default, the content of incoming Rabbit messages gets extracted before being passed into the target listener * method, to let the target method operate on message content types such as String or byte array instead of the raw * {@link Message}. Message type conversion is delegated to a Spring AMQ {@link MessageConverter}. By default, a - * {@link SimpleMessageConverter} will be used. (If you do not want such automatic message conversion taking place, then + * {@link org.springframework.amqp.support.converter.SimpleMessageConverter} will be used. + * (If you do not want such automatic message conversion taking place, then * be sure to set the {@link #setMessageConverter MessageConverter} to null.) * *

diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/FatalListenerExecutionException.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/FatalListenerExecutionException.java index dd189f02..3a8afe1b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/FatalListenerExecutionException.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/FatalListenerExecutionException.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. @@ -17,13 +17,13 @@ package org.springframework.amqp.rabbit.listener.exception; import org.springframework.amqp.AmqpException; -import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; /** - * Exception to be thrown when the execution of a listener method failed unrecoverably. + * Exception to be thrown when the execution of a listener method failed with an + * irrecoverable problem. * * @author Dave Syer - * @see MessageListenerAdapter + * @see org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter */ @SuppressWarnings("serial") public class FatalListenerExecutionException extends AmqpException { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/FatalListenerStartupException.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/FatalListenerStartupException.java index ed253cf1..a6d5616d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/FatalListenerStartupException.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/FatalListenerStartupException.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. @@ -17,13 +17,12 @@ package org.springframework.amqp.rabbit.listener.exception; import org.springframework.amqp.AmqpException; -import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; /** * Exception to be thrown when the execution of a listener method failed on startup. * * @author Dave Syer - * @see MessageListenerAdapter + * @see org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter */ @SuppressWarnings("serial") public class FatalListenerStartupException extends AmqpException { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/ListenerExecutionFailedException.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/ListenerExecutionFailedException.java index 96aad54b..8e2ea0c2 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/ListenerExecutionFailedException.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/ListenerExecutionFailedException.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. @@ -18,7 +18,6 @@ package org.springframework.amqp.rabbit.listener.exception; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Message; -import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; /** @@ -27,7 +26,7 @@ import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; * @author Juergen Hoeller * @author Gary Russell * - * @see MessageListenerAdapter + * @see org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter */ @SuppressWarnings("serial") public class ListenerExecutionFailedException extends AmqpException { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java index 8f948919..b7435152 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java @@ -354,7 +354,7 @@ public class AmqpAppender extends AbstractAppender { AmqpAppender.this.events.add(event); } - }, (long) (Math.pow(retries, Math.log(retries)) * 1000)); + }, (long) (Math.pow(retries, Math.log(retries)) * 1000)); // NOSONAR - magic # } else { getHandler().error("Could not send log message " + logEvent.getMessage() @@ -435,6 +435,8 @@ public class AmqpAppender extends AbstractAppender { */ protected static class AmqpManager extends AbstractManager { + private static final int DEFAULT_MAX_SENDER_RETRIES = 30; + /** * True to send events on separate threads. */ @@ -473,7 +475,7 @@ public class AmqpAppender extends AbstractAppender { /** * How many times to retry sending a message if the broker is unavailable or there is some other error. */ - private int maxSenderRetries = 30; + private int maxSenderRetries = DEFAULT_MAX_SENDER_RETRIES; /** * RabbitMQ ConnectionFactory. diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java index 8352d7d0..cc34c20f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java @@ -95,6 +95,8 @@ import com.rabbitmq.client.ConnectionFactory; */ public class AmqpAppender extends AppenderBase { + private static final int DEFAULT_MAX_SENDER_RETRIES = 30; + /** * Key name for the application id (if there is one set via the appender config) in the message properties. */ @@ -165,7 +167,7 @@ public class AmqpAppender extends AppenderBase { /** * How many times to retry sending a message if the broker is unavailable or there is some other error. */ - private int maxSenderRetries = 30; + private int maxSenderRetries = DEFAULT_MAX_SENDER_RETRIES; /** * Retries are delayed like: N ^ log(N), where N is the retry number. @@ -891,7 +893,7 @@ public class AmqpAppender extends AppenderBase { AmqpAppender.this.events.add(event); } - }, (long) (Math.pow(retries, Math.log(retries)) * 1000)); + }, (long) (Math.pow(retries, Math.log(retries)) * 1000)); // NOSONAR magic # } else { addError("Could not send log message " + logEvent.getMessage() 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 e03355a0..ada1d6fe 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 @@ -16,7 +16,6 @@ package org.springframework.amqp.rabbit.support; -import java.io.DataInputStream; import java.math.BigDecimal; import java.util.ArrayList; import java.util.Collections; @@ -75,7 +74,7 @@ public class DefaultMessagePropertiesConverter implements MessagePropertiesConve * Use this constructor with 'true' to restore pre-1.6 behavior. * @param longStringLimit the limit. * @param convertLongLongStrings {@link LongString} when false, - * {@link DataInputStream} when true. + * {@link java.io.DataInputStream} when true. * @since 1.6 */ public DefaultMessagePropertiesConverter(int longStringLimit, boolean convertLongLongStrings) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ExpressionFactoryBean.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ExpressionFactoryBean.java index 0c7d6849..48ca70bc 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ExpressionFactoryBean.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ExpressionFactoryBean.java @@ -32,7 +32,7 @@ import org.springframework.util.Assert; */ public class ExpressionFactoryBean extends AbstractFactoryBean { - private final static ExpressionParser DEFAULT_PARSER = new SpelExpressionParser(); + private static final ExpressionParser DEFAULT_PARSER = new SpelExpressionParser(); private final String expressionString; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ListenerContainerAware.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ListenerContainerAware.java index f74c11b7..a6227f97 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ListenerContainerAware.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ListenerContainerAware.java @@ -18,10 +18,8 @@ package org.springframework.amqp.rabbit.support; import java.util.Collection; -import org.springframework.amqp.core.MessageListener; - /** - * {@link MessageListener}s that also implement this + * {@link org.springframework.amqp.core.MessageListener}s that also implement this * interface can have configuration verified during initialization. * * @author Gary Russell diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java index a33d1ef8..d2ff32c8 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java @@ -17,11 +17,9 @@ package org.springframework.amqp.rabbit.transaction; import org.springframework.amqp.AmqpException; -import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils; import org.springframework.amqp.rabbit.connection.RabbitResourceHolder; -import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.InitializingBean; import org.springframework.transaction.CannotCreateTransactionException; import org.springframework.transaction.InvalidIsolationLevelException; @@ -33,8 +31,6 @@ import org.springframework.transaction.support.SmartTransactionObject; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; -import com.rabbitmq.client.Connection; - /** * {@link org.springframework.transaction.PlatformTransactionManager} implementation for a single Rabbit * {@link ConnectionFactory}. Binds a Rabbit Channel from the specified ConnectionFactory to the thread, potentially @@ -48,11 +44,14 @@ import com.rabbitmq.client.Connection; *

* Application code is required to retrieve the transactional Rabbit resources via * {@link ConnectionFactoryUtils#getTransactionalResourceHolder(ConnectionFactory, boolean)} instead of a standard - * {@link Connection#createChannel()} call with subsequent Channel creation. Spring's {@link RabbitTemplate} will + * {@link org.springframework.amqp.rabbit.connection.Connection#createChannel(boolean)} call with subsequent + * Channel creation. Spring's + * {@link org.springframework.amqp.rabbit.core.RabbitTemplate} will * autodetect a thread-bound Channel and automatically participate in it. * *

- * The use of {@link CachingConnectionFactory} as a target for this transaction manager is strongly recommended. + * The use of {@link org.springframework.amqp.rabbit.connection.CachingConnectionFactory} + * as a target for this transaction manager is strongly recommended. * CachingConnectionFactory uses a single Rabbit Connection for all Rabbit access in order to avoid the overhead of * repeated Connection creation, as well as maintaining a cache of Channels. Each transaction will then share the same * Rabbit Connection, while still using its own individual Rabbit Channel. diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index 4114a703..9cfd7ff0 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -948,7 +948,7 @@ public class RabbitTemplateIntegrationTests { }); RabbitTemplate template = createSendAndReceiveRabbitTemplate(this.connectionFactory); String result = (String) template.convertSendAndReceive("", ROUTE, "message"); - assertEquals("message", received.get(1000, TimeUnit.MILLISECONDS)); + assertEquals("message", received.get(10_000, TimeUnit.MILLISECONDS)); assertEquals("message", result); // Message was consumed so nothing left on queue result = (String) template.receiveAndConvert(ROUTE); @@ -987,7 +987,7 @@ public class RabbitTemplateIntegrationTests { throw new AmqpException("unexpected failure in test", e); } }); - assertEquals("MESSAGE", received.get(1000, TimeUnit.MILLISECONDS)); + assertEquals("MESSAGE", received.get(10_000, TimeUnit.MILLISECONDS)); assertEquals("MESSAGE", result); // Message was consumed so nothing left on queue result = (String) template.receiveAndConvert(); @@ -1023,7 +1023,7 @@ public class RabbitTemplateIntegrationTests { throw new AmqpException("unexpected failure in test", e); } }); - assertEquals("MESSAGE", received.get(1000, TimeUnit.MILLISECONDS)); + assertEquals("MESSAGE", received.get(10_000, TimeUnit.MILLISECONDS)); assertEquals("MESSAGE", result); // Message was consumed so nothing left on queue result = (String) template.receiveAndConvert(ROUTE); @@ -1059,7 +1059,7 @@ public class RabbitTemplateIntegrationTests { throw new AmqpException("unexpected failure in test", e); } }); - assertEquals("MESSAGE", received.get(1000, TimeUnit.MILLISECONDS)); + assertEquals("MESSAGE", received.get(10_000, TimeUnit.MILLISECONDS)); assertEquals("MESSAGE", result); // Message was consumed so nothing left on queue result = (String) template.receiveAndConvert(ROUTE);