Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Sonar Fixes

Browse Source 
- all minors

* Remaining issues - mostly imports for Javadocs only.

* Final final

* Polising - bogus chars in comment

* Polishing idle time.
This commit is contained in:
Gary Russell
2018-12-23 12:27:04 -05:00
committed by Artem Bilan
parent 24521d8937
commit 6ba27ee97d
56 changed files with 279 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-amqp/src/main/java/org/springframework/amqp/core/Address.java
View File

@@ -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;
}

8
spring-amqp/src/main/java/org/springframework/amqp/core/AmqpTemplate.java
View File

@@ -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

2
spring-amqp/src/main/java/org/springframework/amqp/core/AnonymousQueue.java
View File

@@ -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 =

4
spring-amqp/src/main/java/org/springframework/amqp/core/Base64UrlNamingStrategy.java
View File

@@ -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 =

12
spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java
View File

@@ -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());

2
spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java
View File

@@ -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;

5
spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpClientInterceptor.java
View File

@@ -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
*/

10
spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpProxyFactoryBean.java
View File

@@ -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.
* <p>
* 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<Object>, BeanClassLoaderAware,

3
spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java
View File

@@ -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}.

2
spring-amqp/src/main/java/org/springframework/amqp/support/AmqpHeaders.java
View File

@@ -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";

5
spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DelegatingDecompressingPostProcessor.java
View File

@@ -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.
*/

10
spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/GZipPostProcessor.java
View File

@@ -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

8
spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/ZipPostProcessor.java
View File

@@ -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

15
spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java
View File

@@ -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<Integer, Boolean> brokerOnline = new HashMap<Integer, Boolean>();
private static final Map<Integer, Boolean> brokerOnline = new HashMap<Integer, Boolean>(); // NOSONAR - lower case
// Static so that we only test once on failure
private static final Map<Integer, Boolean> brokerOffline = new HashMap<Integer, Boolean>();
private static final Map<Integer, Boolean> brokerOffline = new HashMap<Integer, Boolean>(); // NOSONAR - lower case
private static final Map<String, String> environmentOverrides = new HashMap<>();
private static final Map<String, String> 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("=", "");

4
spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/LongRunningIntegrationTest.java
View File

@@ -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";

2
spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCondition.java
View File

@@ -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<BrokerRunning> brokerRunningHolder = new ThreadLocal<>();
private static final ThreadLocal<BrokerRunning> brokerRunningHolder = new ThreadLocal<>(); // NOSONAR - lower case
@Override
public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) {

5
spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java
View File

@@ -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;

21
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java
View File

@@ -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 AbstractRabbitListenerContainerFactory<C extends AbstractM
/**
* @param taskExecutor the {@link Executor} to use.
* @see SimpleMessageListenerContainer#setTaskExecutor
* @see AbstractMessageListenerContainer#setTaskExecutor
*/
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
@@ -169,7 +168,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
/**
* @param transactionManager the {@link PlatformTransactionManager} to use.
* @see SimpleMessageListenerContainer#setTransactionManager
* @see AbstractMessageListenerContainer#setTransactionManager
*/
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
@@ -177,7 +176,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
/**
* @param prefetch the prefetch count
* @see SimpleMessageListenerContainer#setPrefetchCount(int)
* @see AbstractMessageListenerContainer#setPrefetchCount(int)
*/
public void setPrefetchCount(Integer prefetch) {
this.prefetchCount = prefetch;
@@ -185,7 +184,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
/**
* @param requeueRejected true to reject by default.
* @see SimpleMessageListenerContainer#setDefaultRequeueRejected
* @see AbstractMessageListenerContainer#setDefaultRequeueRejected
*/
public void setDefaultRequeueRejected(Boolean requeueRejected) {
this.defaultRequeueRejected = requeueRejected;
@@ -202,7 +201,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
/**
* @param adviceChain the advice chain to set.
* @see SimpleMessageListenerContainer#setAdviceChain
* @see AbstractMessageListenerContainer#setAdviceChain
*/
public void setAdviceChain(Advice... adviceChain) {
this.adviceChain = adviceChain == null ? null : Arrays.copyOf(adviceChain, adviceChain.length);
@@ -210,7 +209,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
/**
* @param recoveryInterval The recovery interval.
* @see SimpleMessageListenerContainer#setRecoveryInterval
* @see AbstractMessageListenerContainer#setRecoveryInterval
*/
public void setRecoveryInterval(Long recoveryInterval) {
this.recoveryBackOff = new FixedBackOff(recoveryInterval, FixedBackOff.UNLIMITED_ATTEMPTS);
@@ -219,7 +218,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
/**
* @param recoveryBackOff The BackOff to recover.
* @since 1.5
* @see SimpleMessageListenerContainer#setRecoveryBackOff(BackOff)
* @see AbstractMessageListenerContainer#setRecoveryBackOff(BackOff)
*/
public void setRecoveryBackOff(BackOff recoveryBackOff) {
this.recoveryBackOff = recoveryBackOff;
@@ -227,7 +226,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
/**
* @param missingQueuesFatal the missingQueuesFatal to set.
* @see SimpleMessageListenerContainer#setMissingQueuesFatal
* @see AbstractMessageListenerContainer#setMissingQueuesFatal
*/
public void setMissingQueuesFatal(Boolean missingQueuesFatal) {
this.missingQueuesFatal = missingQueuesFatal;
@@ -236,7 +235,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
/**
* @param mismatchedQueuesFatal the mismatchedQueuesFatal to set.
* @since 1.6
* @see SimpleMessageListenerContainer#setMismatchedQueuesFatal(boolean)
* @see AbstractMessageListenerContainer#setMismatchedQueuesFatal(boolean)
*/
public void setMismatchedQueuesFatal(Boolean mismatchedQueuesFatal) {
this.mismatchedQueuesFatal = mismatchedQueuesFatal;
@@ -244,7 +243,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
/**
* @param consumerTagStrategy the consumerTagStrategy to set
* @see SimpleMessageListenerContainer#setConsumerTagStrategy(ConsumerTagStrategy)
* @see AbstractMessageListenerContainer#setConsumerTagStrategy(ConsumerTagStrategy)
*/
public void setConsumerTagStrategy(ConsumerTagStrategy consumerTagStrategy) {
this.consumerTagStrategy = consumerTagStrategy;

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/DirectRabbitListenerContainerFactory.java
View File

@@ -17,13 +17,12 @@
package org.springframework.amqp.rabbit.config;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
import org.springframework.scheduling.TaskScheduler;
/**
* A {@link RabbitListenerContainerFactory} implementation to build a regular
* {@link DirectMessageListenerContainer}.
* A {@link org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory}
* implementation to build a regular {@link DirectMessageListenerContainer}.
*
* @author Gary Russell
* @since 2.0

10
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java
View File

@@ -16,16 +16,16 @@
package org.springframework.amqp.rabbit.config;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
/**
* A {@link RabbitListenerContainerFactory} implementation to build a regular
* {@link SimpleMessageListenerContainer}.
* A {@link org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory}
* implementation to build a regular {@link SimpleMessageListenerContainer}.
*
* <p>This should be the default for most users and a good transition paths
* for those that are used to build such container definition manually.
* <p>
* 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

8
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerEndpoint.java
View File

@@ -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

31
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
View File

@@ -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<String> txStarts = new HashSet<>(Arrays.asList("basicPublish", "basicAck",
private static final Set<String> txStarts = new HashSet<>(Arrays.asList("basicPublish", "basicAck", // NOSONAR
"basicNack", "basicReject"));
private static final Set<String> ackMethods = new HashSet<>(Arrays.asList("basicAck",
private static final Set<String> ackMethods = new HashSet<>(Arrays.asList("basicAck", // NOSONAR
"basicNack", "basicReject"));
private static final Set<String> txEnds = new HashSet<>(Arrays.asList("txCommit", "txRollback"));
private static final Set<String> 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<ChannelProxy> 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) {

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ClosingRecoveryListener.java
View File

@@ -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<AutorecoveringChannel, Boolean> hasListener = new ConcurrentHashMap<>();
private static final ConcurrentMap<AutorecoveringChannel, Boolean> hasListener // NOSONAR - lower case
= new ConcurrentHashMap<>();
private ClosingRecoveryListener() {
super();

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConsumerChannelRegistry.java
View File

@@ -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<ChannelHolder> consumerChannel = new ThreadLocal<ChannelHolder>();
private static final ThreadLocal<ChannelHolder> consumerChannel // NOSONAR - lower case
= new ThreadLocal<ChannelHolder>();
private ConsumerChannelRegistry() {
super();

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java
View File

@@ -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());

15
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java
View File

@@ -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<Connection> 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);

57
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java
View File

@@ -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<Boolean> 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<Boolean> 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;
}
}

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BrokerEventListener.java
View File

@@ -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;

16
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
View File

@@ -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 &&

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitOperations.java
View File

@@ -429,7 +429,7 @@ public interface RabbitOperations extends AmqpTemplate {
* @since 2.0
*/
@FunctionalInterface
public interface OperationsCallback<T> {
interface OperationsCallback<T> {
/**
* Execute any number of operations using a dedicated

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

@@ -1947,10 +1947,10 @@ public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count/co
(RetryCallback<T, Exception>) context -> doExecute(action, connectionFactory),
(RecoveryCallback<T>) 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);

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/SimpleBatchingStrategy.java
View File

@@ -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();

19
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
View File

@@ -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 "

8
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java
View File

@@ -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;

16
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
View File

@@ -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<Delivery> queue;
@@ -133,12 +137,12 @@ public class BlockingQueueConsumer {
private final Set<String> missingQueues = Collections.synchronizedSet(new HashSet<String>());
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<Long>(this.deliveryTags).get(this.deliveryTags.size() - 1);
this.channel.basicAck(deliveryTag, true);
}
if (ackRequired && (!this.transactional || isLocallyTransacted)) {
long deliveryTag = new ArrayList<Long>(this.deliveryTags).get(this.deliveryTags.size() - 1);
this.channel.basicAck(deliveryTag, true);
}
if (isLocallyTransacted) {

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java
View File

@@ -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}.
* <p>
* The exception will not be wrapped if the {@code cause} chain already contains an

27
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java
View File

@@ -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();

12
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java
View File

@@ -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);
}
}
}

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerContainerIdleEvent.java
View File

@@ -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() + "]";
}

7
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MultiMethodRabbitListenerEndpoint.java
View File

@@ -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()));
}
}

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerContainerFactory.java
View File

@@ -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 RabbitListenerContainerFactory<C extends MessageListenerContain
C createListenerContainer(@Nullable RabbitListenerEndpoint endpoint);
/**
* Create a {@link MessageListenerContainer} with no {@link MessageListener}
* or queues; the listener must be added later before the container is started.
* Create a {@link MessageListenerContainer} with no
* {@link org.springframework.amqp.core.MessageListener} or queues; the listener must
* be added later before the container is started.
* @return the created container.
* @since 2.1.
*/

12
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistrar.java
View File

@@ -23,7 +23,6 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.lang.Nullable;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
import org.springframework.util.Assert;
@@ -75,10 +74,13 @@ public class RabbitListenerEndpointRegistrar implements BeanFactoryAware, Initia
/**
* Set the {@link MessageHandlerMethodFactory} to use to configure the message
* listener responsible to serve an endpoint detected by this processor.
* <p>By 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.
* <p>
* 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) {

27
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

@@ -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());

7
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java
View File

@@ -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. */

28
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java
View File

@@ -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;

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java
View File

@@ -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 <code>null</code>.)
*
* <p>

8
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/FatalListenerExecutionException.java
View File

@@ -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 {

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/FatalListenerStartupException.java
View File

@@ -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 {

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/exception/ListenerExecutionFailedException.java
View File

@@ -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 {

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java
View File

@@ -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.

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java
View File

@@ -95,6 +95,8 @@ import com.rabbitmq.client.ConnectionFactory;
*/
public class AmqpAppender extends AppenderBase<ILoggingEvent> {
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<ILoggingEvent> {
/**
* 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<ILoggingEvent> {
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()

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java
View File

@@ -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) {

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ExpressionFactoryBean.java
View File

@@ -32,7 +32,7 @@ import org.springframework.util.Assert;
*/
public class ExpressionFactoryBean extends AbstractFactoryBean<Expression> {
private final static ExpressionParser DEFAULT_PARSER = new SpelExpressionParser();
private static final ExpressionParser DEFAULT_PARSER = new SpelExpressionParser();
private final String expressionString;

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ListenerContainerAware.java
View File

@@ -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

11
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java
View File

@@ -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;
* <p>
* 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.
*
* <p>
* <b>The use of {@link CachingConnectionFactory} as a target for this transaction manager is strongly recommended.</b>
* <b>The use of {@link org.springframework.amqp.rabbit.connection.CachingConnectionFactory}
* as a target for this transaction manager is strongly recommended.</b>
* 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.

8
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
View File

@@ -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);