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

checkstyle FinalClass

Browse Source 
checkstyle Import Rules

checkstyle HideUtilityCtor

checkstyle InnerTypeLast

checkstyle Annotation Rules

checkstyle Block Rules

checkstyle InnerAssignment

checkstyle Boolean Rules

checkstyle Remaining Code Rules

checkstyle ImportOrder

checkstyle Misc Rules and Trailing Whitespace

checkstyle GenericWhitespace

checkstyle ParenPad

checkstyle WhiteSpaceAfter Script

checkstyle WhiteSpaceAfter

checkstyle WhiteSpaceAround Script

checkstyle WhiteSpaceAround
This commit is contained in:
Gary Russell
2016-04-04 13:56:48 -04:00
committed by Artem Bilan
parent 7e6da05443
commit b89756eec7
133 changed files with 1054 additions and 756 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -65,6 +65,8 @@ subprojects { subproject ->
apply from: "${rootDir}/src/checkstyle/fixHeaders.gradle"
apply from: "${rootDir}/src/checkstyle/fixThis.gradle"
apply from: "${rootDir}/src/checkstyle/fixModifiers.gradle"
apply from: "${rootDir}/src/checkstyle/fixFinal.gradle"
apply from: "${rootDir}/src/checkstyle/fixWhiteAround.gradle"
if (project.hasProperty('platformVersion')) {
apply plugin: 'spring-io'

6
spring-amqp/src/main/java/org/springframework/amqp/core/AddressUtils.java
View File

@@ -22,7 +22,7 @@ package org.springframework.amqp.core;
*
*/
@Deprecated
public class AddressUtils {
public final class AddressUtils {
/**
* @deprecated Use the constant in {@link Address#AMQ_RABBITMQ_REPLY_TO}.
@@ -30,6 +30,10 @@ public class AddressUtils {
@Deprecated
public static final String AMQ_RABBITMQ_REPLY_TO = Address.AMQ_RABBITMQ_REPLY_TO;
private AddressUtils() {
super();
}
/**
* Decodes the reply-to {@link Address} into exchange/key.
*

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

@@ -44,7 +44,7 @@ public class AnonymousQueue extends Queue {
* Construct a queue with a UUID name with the supplied arguments.
* @param arguments the arguments.
*/
public AnonymousQueue(Map<String,Object> arguments) {
public AnonymousQueue(Map<String, Object> arguments) {
super(UUID.randomUUID().toString(), false, true, true, arguments);
}

49
spring-amqp/src/main/java/org/springframework/amqp/core/BindingBuilder.java
View File

@@ -29,9 +29,14 @@ import org.springframework.util.Assert;
* @author Mark Pollack
* @author Mark Fisher
* @author Dave Syer
* @author Gary Russell
*/
public final class BindingBuilder {
private BindingBuilder() {
super();
}
public static DestinationConfigurer bind(Queue queue) {
return new DestinationConfigurer(queue.getName(), DestinationType.QUEUE);
}
@@ -40,7 +45,15 @@ public final class BindingBuilder {
return new DestinationConfigurer(exchange.getName(), DestinationType.EXCHANGE);
}
public static class DestinationConfigurer {
private static Map<String, Object> createMapForKeys(String... keys) {
Map<String, Object> map = new HashMap<String, Object>();
for (String key : keys) {
map.put(key, null);
}
return map;
}
public static final class DestinationConfigurer {
protected final String name;
protected final DestinationType type;
@@ -71,7 +84,7 @@ public final class BindingBuilder {
}
}
public static class HeadersExchangeMapConfigurer {
public static final class HeadersExchangeMapConfigurer {
protected final DestinationConfigurer destination;
@@ -102,7 +115,7 @@ public final class BindingBuilder {
return new HeadersExchangeMapBindingCreator(headerValues, true);
}
public class HeadersExchangeSingleValueBindingCreator {
public final class HeadersExchangeSingleValueBindingCreator {
private final String key;
@@ -126,7 +139,7 @@ public final class BindingBuilder {
}
}
public class HeadersExchangeKeysBindingCreator {
public final class HeadersExchangeKeysBindingCreator {
private final Map<String, Object> headerMap;
@@ -143,7 +156,7 @@ public final class BindingBuilder {
}
}
public class HeadersExchangeMapBindingCreator {
public final class HeadersExchangeMapBindingCreator {
private final Map<String, Object> headerMap;
@@ -173,7 +186,7 @@ public final class BindingBuilder {
}
}
public static class TopicExchangeRoutingKeyConfigurer extends AbstractRoutingKeyConfigurer<TopicExchange> {
public static final class TopicExchangeRoutingKeyConfigurer extends AbstractRoutingKeyConfigurer<TopicExchange> {
private TopicExchangeRoutingKeyConfigurer(DestinationConfigurer destination, TopicExchange exchange) {
super(destination, exchange.getName());
@@ -181,16 +194,16 @@ public final class BindingBuilder {
public Binding with(String routingKey) {
return new Binding(destination.name, destination.type, exchange, routingKey,
Collections.<String, Object> emptyMap());
Collections.<String, Object>emptyMap());
}
public Binding with(Enum<?> routingKeyEnum) {
return new Binding(destination.name, destination.type, exchange, routingKeyEnum.toString(),
Collections.<String, Object> emptyMap());
Collections.<String, Object>emptyMap());
}
}
public static class GenericExchangeRoutingKeyConfigurer extends AbstractRoutingKeyConfigurer<TopicExchange> {
public static final class GenericExchangeRoutingKeyConfigurer extends AbstractRoutingKeyConfigurer<TopicExchange> {
private GenericExchangeRoutingKeyConfigurer(DestinationConfigurer destination, Exchange exchange) {
super(destination, exchange.getName());
@@ -223,12 +236,12 @@ public final class BindingBuilder {
public Binding noargs() {
return new Binding(this.configurer.destination.name, this.configurer.destination.type, this.configurer.exchange,
this.routingKey, Collections.<String, Object> emptyMap());
this.routingKey, Collections.<String, Object>emptyMap());
}
}
public static class DirectExchangeRoutingKeyConfigurer extends AbstractRoutingKeyConfigurer<DirectExchange> {
public static final class DirectExchangeRoutingKeyConfigurer extends AbstractRoutingKeyConfigurer<DirectExchange> {
private DirectExchangeRoutingKeyConfigurer(DestinationConfigurer destination, DirectExchange exchange) {
super(destination, exchange.getName());
@@ -236,26 +249,18 @@ public final class BindingBuilder {
public Binding with(String routingKey) {
return new Binding(destination.name, destination.type, exchange, routingKey,
Collections.<String, Object> emptyMap());
Collections.<String, Object>emptyMap());
}
public Binding with(Enum<?> routingKeyEnum) {
return new Binding(destination.name, destination.type, exchange, routingKeyEnum.toString(),
Collections.<String, Object> emptyMap());
Collections.<String, Object>emptyMap());
}
public Binding withQueueName() {
return new Binding(destination.name, destination.type, exchange, destination.name,
Collections.<String, Object> emptyMap());
Collections.<String, Object>emptyMap());
}
}
private static Map<String, Object> createMapForKeys(String... keys) {
Map<String, Object> map = new HashMap<String, Object>();
for (String key : keys) {
map.put(key, null);
}
return map;
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2016 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.
@@ -40,7 +40,7 @@ public class DirectExchange extends AbstractExchange {
super(name, durable, autoDelete);
}
public DirectExchange(String name, boolean durable, boolean autoDelete, Map<String,Object> arguments) {
public DirectExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
super(name, durable, autoDelete, arguments);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2016 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.
@@ -36,7 +36,7 @@ public class FanoutExchange extends AbstractExchange {
super(name, durable, autoDelete);
}
public FanoutExchange(String name, boolean durable, boolean autoDelete, Map<String,Object> arguments) {
public FanoutExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
super(name, durable, autoDelete, arguments);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2016 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.
@@ -32,7 +32,7 @@ public class HeadersExchange extends AbstractExchange {
super(name, durable, autoDelete);
}
public HeadersExchange(String name, boolean durable, boolean autoDelete, Map<String,Object> arguments) {
public HeadersExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
super(name, durable, autoDelete, arguments);
}

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

@@ -45,13 +45,13 @@ public class Message implements Serializable {
private final byte[] body;
public Message(byte[] body, MessageProperties messageProperties) {//NOSONAR
this.body = body;//NOSONAR
public Message(byte[] body, MessageProperties messageProperties) { //NOSONAR
this.body = body; //NOSONAR
this.messageProperties = messageProperties;
}
public byte[] getBody() {
return this.body;//NOSONAR
return this.body; //NOSONAR
}
public MessageProperties getMessageProperties() {
@@ -90,7 +90,7 @@ public class Message implements Serializable {
// ignore
}
// Comes out as '[B@....b' (so harmless)
return this.body.toString()+"(byte["+this.body.length+"])";//NOSONAR
return this.body.toString() + "(byte[" + this.body.length + "])"; //NOSONAR
}
@Override

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

@@ -91,7 +91,7 @@ public final class MessageBuilder extends MessageBuilderSupport<Message> {
return new MessageBuilder(Arrays.copyOf(body, body.length), message.getMessageProperties());
}
private MessageBuilder(byte[] body) {//NOSONAR
private MessageBuilder(byte[] body) { //NOSONAR
this.body = body;
}
@@ -99,7 +99,7 @@ public final class MessageBuilder extends MessageBuilderSupport<Message> {
this(message.getBody(), message.getMessageProperties());
}
private MessageBuilder(byte[] body, MessageProperties properties) {//NOSONAR
private MessageBuilder(byte[] body, MessageProperties properties) { //NOSONAR
this.body = body;
this.copyProperties(properties);
}

21
spring-amqp/src/main/java/org/springframework/amqp/core/MessageDeliveryMode.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2016 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.
@@ -23,6 +23,7 @@ package org.springframework.amqp.core;
* passing into AMQP APIs.
*
* @author Mark Pollack
* @author Gary Russell
*
*/
public enum MessageDeliveryMode {
@@ -31,30 +32,24 @@ public enum MessageDeliveryMode {
public static int toInt(MessageDeliveryMode mode) {
switch (mode) {
case NON_PERSISTENT: {
case NON_PERSISTENT:
return 1;
}
case PERSISTENT: {
case PERSISTENT:
return 2;
}
default: {
default:
return -1;
}
}
}
public static MessageDeliveryMode fromInt(int modeAsNumber) {
switch (modeAsNumber) {
case 1: {
case 1:
return NON_PERSISTENT;
}
case 2: {
case 2:
return PERSISTENT;
}
default: {
default:
return null;
}
}
}
}

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

@@ -132,12 +132,12 @@ public class MessageProperties implements Serializable {
}
public void setTimestamp(Date timestamp) {
this.timestamp = timestamp;//NOSONAR
this.timestamp = timestamp; //NOSONAR
}
// NOTE qpid java timestamp is long, presumably can convert to Date.
public Date getTimestamp() {
return this.timestamp;//NOSONAR
return this.timestamp; //NOSONAR
}
// NOTE Not forward compatible with qpid 1.0 .NET
@@ -191,12 +191,12 @@ public class MessageProperties implements Serializable {
return this.type;
}
public void setCorrelationId(byte[] correlationId) {//NOSONAR
this.correlationId = correlationId;//NOSONAR
public void setCorrelationId(byte[] correlationId) { //NOSONAR
this.correlationId = correlationId; //NOSONAR
}
public byte[] getCorrelationId() {
return this.correlationId;//NOSONAR
return this.correlationId; //NOSONAR
}
public String getCorrelationIdString() {

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

@@ -23,7 +23,7 @@ package org.springframework.amqp.core;
* @since 1.3
*
*/
public class MessagePropertiesBuilder extends MessageBuilderSupport<MessageProperties> {
public final class MessagePropertiesBuilder extends MessageBuilderSupport<MessageProperties> {
/**
* Returns a builder with an initial set of properties.

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2016 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.
@@ -36,7 +36,7 @@ public class TopicExchange extends AbstractExchange {
super(name, durable, autoDelete);
}
public TopicExchange(String name, boolean durable, boolean autoDelete, Map<String,Object> arguments) {
public TopicExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
super(name, durable, autoDelete, arguments);
}

4
spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractMessageConverter.java
View File

@@ -50,12 +50,12 @@ public abstract class AbstractMessageConverter implements MessageConverter {
@Override
public final Message toMessage(Object object, MessageProperties messageProperties)
throws MessageConversionException {
if (messageProperties==null) {
if (messageProperties == null) {
messageProperties = new MessageProperties();
}
Message message = createMessage(object, messageProperties);
messageProperties = message.getMessageProperties();
if (this.createMessageIds && messageProperties.getMessageId()==null) {
if (this.createMessageIds && messageProperties.getMessageId() == null) {
messageProperties.setMessageId(UUID.randomUUID().toString());
}
return message;

6
spring-amqp/src/main/java/org/springframework/amqp/support/converter/DefaultClassMapper.java
View File

@@ -89,10 +89,12 @@ public class DefaultClassMapper implements ClassMapper, InitializingBean {
}
try {
return ClassUtils.forName(classId, getClass().getClassLoader());
} catch (ClassNotFoundException e) {
}
catch (ClassNotFoundException e) {
throw new MessageConversionException(
"failed to resolve class name [" + classId + "]", e);
} catch (LinkageError e) {
}
catch (LinkageError e) {
throw new MessageConversionException(
"failed to resolve class name [" + classId + "]", e);
}

3
spring-amqp/src/main/java/org/springframework/amqp/support/converter/SerializerMessageConverter.java
View File

@@ -189,7 +189,8 @@ public class SerializerMessageConverter extends WhiteListDeserializingMessageCon
if (object instanceof String) {
try {
bytes = ((String) object).getBytes(this.defaultCharset);
} catch (UnsupportedEncodingException e) {
}
catch (UnsupportedEncodingException e) {
throw new MessageConversionException("failed to convert Message content", e);
}
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);

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

@@ -82,7 +82,7 @@ public abstract class AbstractDecompressingPostProcessor implements MessagePostP
public Message postProcessMessage(Message message) throws AmqpException {
Object autoDecompress = message.getMessageProperties().getHeaders()
.get(MessageProperties.SPRING_AUTO_DECOMPRESS);
if (this.alwaysDecompress || (autoDecompress instanceof Boolean && ((Boolean)autoDecompress))) {
if (this.alwaysDecompress || (autoDecompress instanceof Boolean && ((Boolean) autoDecompress))) {
ByteArrayInputStream zipped = new ByteArrayInputStream(message.getBody());
try {
InputStream unzipper = getDecompressorStream(zipped);

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

@@ -56,6 +56,6 @@ public final class MessagePostProcessorUtils {
return sorted;
}
private MessagePostProcessorUtils() {}
private MessagePostProcessorUtils() { }
}

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

@@ -53,7 +53,7 @@ public class ZipPostProcessor extends AbstractDeflaterPostProcessor {
return "zip";
}
private static class SettableLevelZipOutputStream extends ZipOutputStream {
private static final class SettableLevelZipOutputStream extends ZipOutputStream {
private SettableLevelZipOutputStream(OutputStream zipped, int level) {
super(zipped);

12
spring-amqp/src/main/java/org/springframework/amqp/utils/SerializationUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2010 the original author or authors.
* Copyright 2006-2016 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.
@@ -26,8 +26,13 @@ import java.io.ObjectOutputStream;
* Static utility to help with serialization.
*
* @author Dave Syer
* @author Gary Russell
*/
public class SerializationUtils {
public final class SerializationUtils {
private SerializationUtils() {
super();
}
/**
* Serialize the object provided.
@@ -59,7 +64,8 @@ public class SerializationUtils {
}
try {
return deserialize(new ObjectInputStream(new ByteArrayInputStream(bytes)));
} catch (IOException e) {
}
catch (IOException e) {
throw new IllegalArgumentException("Could not deserialize object", e);
}
}

6
spring-amqp/src/main/java/org/springframework/amqp/utils/test/TestUtils.java
View File

@@ -27,7 +27,11 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @since 1.2
*/
public class TestUtils {
public final class TestUtils {
private TestUtils() {
super();
}
/**
* Uses nested {@link DirectFieldAccessor}s to obtain a property using dotted notation to traverse fields; e.g.

2
spring-amqp/src/test/java/org/springframework/amqp/core/BindingBuilderTests.java
View File

@@ -67,7 +67,7 @@ public class BindingBuilderTests {
bind(new Queue("q")).//
to(new CustomExchange("f")).//
with("r").//
and(Collections.<String, Object> singletonMap("k", new Object()));
and(Collections.<String, Object>singletonMap("k", new Object()));
assertNotNull(binding);
}

2
spring-amqp/src/test/java/org/springframework/amqp/core/MessagePropertiesTests.java
View File

@@ -16,7 +16,7 @@
package org.springframework.amqp.core;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import org.junit.Test;

16
spring-amqp/src/test/java/org/springframework/amqp/core/build/MessageBuilderTests.java
View File

@@ -223,8 +223,8 @@ public class MessageBuilderTests {
.setDeliveryTag(2L)
.setExpiration("expiration")
.setHeader("foo", "bar")
.copyHeaders(Collections. <String, Object> singletonMap("qux", "fiz"))
.copyHeaders(Collections. <String, Object> singletonMap("baz", "fuz"))
.copyHeaders(Collections.<String, Object>singletonMap("qux", "fiz"))
.copyHeaders(Collections.<String, Object>singletonMap("baz", "fuz"))
.setMessageCount(3)
.setMessageId("messageId")
.setPriority(4)
@@ -248,8 +248,8 @@ public class MessageBuilderTests {
.setDeliveryTagIfAbsent(20L)
.setExpirationIfAbsent("EXPIRATION")
.setHeaderIfAbsent("foo", "BAR")
.copyHeadersIfAbsent(Collections. <String, Object> singletonMap("qux", "FIZ"))
.copyHeadersIfAbsent(Collections. <String, Object> singletonMap("baz", "FUZ"))
.copyHeadersIfAbsent(Collections.<String, Object>singletonMap("qux", "FIZ"))
.copyHeadersIfAbsent(Collections.<String, Object>singletonMap("baz", "FUZ"))
.setMessageCountIfAbsent(30)
.setMessageIdIfAbsent("MESSAGEID")
.setPriorityIfAbsentOrDefault(40)
@@ -273,8 +273,8 @@ public class MessageBuilderTests {
.setDeliveryTag(2L)
.setExpiration("expiration")
.setHeader("foo", "bar")
.copyHeaders(Collections. <String, Object> singletonMap("qux", "fiz"))
.copyHeaders(Collections. <String, Object> singletonMap("baz", "fuz"))
.copyHeaders(Collections.<String, Object>singletonMap("qux", "fiz"))
.copyHeaders(Collections.<String, Object>singletonMap("baz", "fuz"))
.setMessageCount(3)
.setMessageId("messageId")
.setPriority(4)
@@ -298,8 +298,8 @@ public class MessageBuilderTests {
.setDeliveryTagIfAbsent(20L)
.setExpirationIfAbsent("EXPIRATION")
.setHeaderIfAbsent("foo", "BAR")
.copyHeadersIfAbsent(Collections. <String, Object> singletonMap("qux", "FIZ"))
.copyHeadersIfAbsent(Collections. <String, Object> singletonMap("baz", "FUZ"))
.copyHeadersIfAbsent(Collections.<String, Object>singletonMap("qux", "FIZ"))
.copyHeadersIfAbsent(Collections.<String, Object>singletonMap("baz", "FUZ"))
.setMessageCountIfAbsent(30)
.setMessageIdIfAbsent("MESSAGEID")
.setPriorityIfAbsentOrDefault(40)

6
spring-amqp/src/test/java/org/springframework/amqp/support/converter/Jackson2JsonMessageConverterTests.java
View File

@@ -219,7 +219,7 @@ public class Jackson2JsonMessageConverterTests {
byte[] bytes = "[ {\"name\" : \"foo\" } ]".getBytes();
MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType("application/json");
messageProperties.setInferredArgumentType((new ParameterizedTypeReference<List<Foo>>() {}).getType());
messageProperties.setInferredArgumentType((new ParameterizedTypeReference<List<Foo>>() { }).getType());
Message message = new Message(bytes, messageProperties);
Object foo = this.converter.fromMessage(message);
assertThat(foo, instanceOf(List.class));
@@ -232,7 +232,7 @@ public class Jackson2JsonMessageConverterTests {
MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType("application/json");
messageProperties.setInferredArgumentType(
(new ParameterizedTypeReference<Map<String, List<Bar>>>() {}).getType());
(new ParameterizedTypeReference<Map<String, List<Bar>>>() { }).getType());
Message message = new Message(bytes, messageProperties);
Object foo = this.converter.fromMessage(message);
assertThat(foo, instanceOf(LinkedHashMap.class));
@@ -250,7 +250,7 @@ public class Jackson2JsonMessageConverterTests {
MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType("application/json");
messageProperties.setInferredArgumentType(
(new ParameterizedTypeReference<Map<String, Map<String, Bar>>>() {}).getType());
(new ParameterizedTypeReference<Map<String, Map<String, Bar>>>() { }).getType());
Message message = new Message(bytes, messageProperties);
Object foo = this.converter.fromMessage(message);
assertThat(foo, instanceOf(LinkedHashMap.class));

2
spring-amqp/src/test/java/org/springframework/amqp/support/converter/JsonMessageConverterTests.java
View File

@@ -53,7 +53,7 @@ public class JsonMessageConverterTests {
private JsonMessageConverter jsonConverterWithDefaultType;
@Before
public void before(){
public void before() {
converter = new JsonMessageConverter();
trade = new SimpleTrade();
trade.setAccountName("Acct1");

7
spring-amqp/src/test/java/org/springframework/amqp/support/converter/MessagingMessageConverterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,6 +16,9 @@
package org.springframework.amqp.support.converter;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
@@ -24,10 +27,10 @@ import org.springframework.amqp.core.MessageProperties;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.junit.Assert.*;
/**
* @author Stephane Nicoll
* @author Gary Russell
*/
public class MessagingMessageConverterTests {

18
spring-amqp/src/test/java/org/springframework/amqp/support/converter/SimpleTrade.java
View File

@@ -135,7 +135,8 @@ public class SimpleTrade {
if (other.accountName != null) {
return false;
}
} else if (!accountName.equals(other.accountName)) {
}
else if (!accountName.equals(other.accountName)) {
return false;
}
if (buyRequest != other.buyRequest) {
@@ -145,14 +146,16 @@ public class SimpleTrade {
if (other.orderType != null) {
return false;
}
} else if (!orderType.equals(other.orderType)) {
}
else if (!orderType.equals(other.orderType)) {
return false;
}
if (price == null) {
if (other.price != null) {
return false;
}
} else if (!price.equals(other.price)) {
}
else if (!price.equals(other.price)) {
return false;
}
if (quantity != other.quantity) {
@@ -162,21 +165,24 @@ public class SimpleTrade {
if (other.requestId != null) {
return false;
}
} else if (!requestId.equals(other.requestId)) {
}
else if (!requestId.equals(other.requestId)) {
return false;
}
if (ticker == null) {
if (other.ticker != null) {
return false;
}
} else if (!ticker.equals(other.ticker)) {
}
else if (!ticker.equals(other.ticker)) {
return false;
}
if (userName == null) {
if (other.userName != null) {
return false;
}
} else if (!userName.equals(other.userName)) {
}
else if (!userName.equals(other.userName)) {
return false;
}
return true;

4
spring-rabbit-test/src/test/java/org/springframework/amqp/rabbit/test/ExampleRabbitListenerCaptureTest.java
View File

@@ -162,12 +162,12 @@ public class ExampleRabbitListenerCaptureTest {
private boolean failed;
@RabbitListener(id="foo", queues="#{queue1.name}")
@RabbitListener(id = "foo", queues = "#{queue1.name}")
public String foo(String foo) {
return foo.toUpperCase();
}
@RabbitListener(id="bar", queues="#{queue2.name}")
@RabbitListener(id = "bar", queues = "#{queue2.name}")
public void foo(@Payload String foo, @Header("amqp_receivedRoutingKey") String rk) {
if (!failed && foo.equals("ex")) {
failed = true;

4
spring-rabbit-test/src/test/java/org/springframework/amqp/rabbit/test/ExampleRabbitListenerSpyAndCaptureTest.java
View File

@@ -181,12 +181,12 @@ public class ExampleRabbitListenerSpyAndCaptureTest {
private boolean failed;
@RabbitListener(id="foo", queues="#{queue1.name}")
@RabbitListener(id = "foo", queues = "#{queue1.name}")
public String foo(String foo) {
return foo.toUpperCase();
}
@RabbitListener(id="bar", queues="#{queue2.name}")
@RabbitListener(id = "bar", queues = "#{queue2.name}")
public void foo(@Payload String foo, @Header("amqp_receivedRoutingKey") String rk) {
if (!failed && foo.equals("ex")) {
failed = true;

4
spring-rabbit-test/src/test/java/org/springframework/amqp/rabbit/test/ExampleRabbitListenerSpyTest.java
View File

@@ -142,12 +142,12 @@ public class ExampleRabbitListenerSpyTest {
public static class Listener {
@RabbitListener(id="foo", queues="#{queue1.name}")
@RabbitListener(id = "foo", queues = "#{queue1.name}")
public String foo(String foo) {
return foo.toUpperCase();
}
@RabbitListener(id="bar", queues="#{queue2.name}")
@RabbitListener(id = "bar", queues = "#{queue2.name}")
public void foo(@Payload String foo, @Header("amqp_receivedRoutingKey") String rk) {
}

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/Exchange.java
View File

@@ -16,9 +16,8 @@
package org.springframework.amqp.rabbit.annotation;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.amqp.core.ExchangeTypes;
@@ -31,7 +30,7 @@ import org.springframework.amqp.core.ExchangeTypes;
*
*/
@Target({})
@Retention(RUNTIME)
@Retention(RetentionPolicy.RUNTIME)
public @interface Exchange {
/**

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/Queue.java
View File

@@ -16,9 +16,8 @@
package org.springframework.amqp.rabbit.annotation;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
@@ -29,7 +28,7 @@ import java.lang.annotation.Target;
*
*/
@Target({})
@Retention(RUNTIME)
@Retention(RetentionPolicy.RUNTIME)
public @interface Queue {
/**

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/BindingFactoryBean.java
View File

@@ -74,7 +74,8 @@ public class BindingFactoryBean implements FactoryBean<Binding> {
if (this.destinationQueue != null) {
destination = this.destinationQueue.getName();
destinationType = DestinationType.QUEUE;
} else {
}
else {
destination = this.destinationExchange.getName();
destinationType = DestinationType.EXCHANGE;
}

11
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/NamespaceUtils.java
View File

@@ -143,7 +143,8 @@ public abstract class NamespaceUtils {
String value = element.getAttribute(attributeName);
if (StringUtils.hasText(value)) {
builder.addConstructorArgValue(new TypedStringValue(value));
} else {
}
else {
builder.addConstructorArgValue(defaultValue);
}
}
@@ -268,7 +269,7 @@ public abstract class NamespaceUtils {
}
String ref = element.getAttribute(REF_ATTRIBUTE);
Assert.isTrue(!StringUtils.hasText(ref) || innerComponentDefinition == null,//NOSONAR
Assert.isTrue(!StringUtils.hasText(ref) || innerComponentDefinition == null, //NOSONAR
"Ambiguous definition. Inner bean "
+ (innerComponentDefinition == null ? innerComponentDefinition : innerComponentDefinition
.getBeanDefinition().getBeanClassName()) + " declaration and \"ref\" " + ref
@@ -307,12 +308,12 @@ public abstract class NamespaceUtils {
boolean hasAttributeValue = StringUtils.hasText(valueElementValue);
boolean hasAttributeExpression = StringUtils.hasText(expressionElementValue);
if (hasAttributeValue && hasAttributeExpression){
if (hasAttributeValue && hasAttributeExpression) {
parserContext.getReaderContext().error("Only one of '" + valueElementName + "' or '"
+ expressionElementName + "' is allowed", element);
}
if (oneRequired && (!hasAttributeValue && !hasAttributeExpression)){
if (oneRequired && (!hasAttributeValue && !hasAttributeExpression)) {
parserContext.getReaderContext().error("One of '" + valueElementName + "' or '"
+ expressionElementName + "' is required", element);
}
@@ -333,7 +334,7 @@ public abstract class NamespaceUtils {
String expressionElementValue = element.getAttribute(expressionElementName);
if (StringUtils.hasText(expressionElementValue)){
if (StringUtils.hasText(expressionElementValue)) {
BeanDefinitionBuilder expressionDefBuilder =
BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class);
expressionDefBuilder.addConstructorArgValue(expressionElementValue);

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueArgumentsParser.java
View File

@@ -18,11 +18,12 @@ package org.springframework.amqp.rabbit.config;
import java.util.Map;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.MapFactoryBean;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.w3c.dom.Element;
/**
* @author Gary Russell

19
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceUtils.java
View File

@@ -33,7 +33,7 @@ import org.springframework.util.StringUtils;
* @since 1.0.1
*
*/
public class RabbitNamespaceUtils {
public final class RabbitNamespaceUtils {
private static final String CONNECTION_FACTORY_ATTRIBUTE = "connection-factory";
@@ -99,6 +99,11 @@ public class RabbitNamespaceUtils {
private static final String IDLE_EVENT_INTERVAL = "idle-event-interval";
private RabbitNamespaceUtils() {
super();
}
public static BeanDefinition parseContainer(Element containerEle, ParserContext parserContext) {
RootBeanDefinition containerDef = new RootBeanDefinition(SimpleMessageListenerContainer.class);
containerDef.setSource(parserContext.extractSource(containerEle));
@@ -276,17 +281,21 @@ public class RabbitNamespaceUtils {
if (StringUtils.hasText(acknowledge)) {
if (ACKNOWLEDGE_AUTO.equals(acknowledge)) {
acknowledgeMode = AcknowledgeMode.AUTO;
} else if (ACKNOWLEDGE_MANUAL.equals(acknowledge)) {
}
else if (ACKNOWLEDGE_MANUAL.equals(acknowledge)) {
acknowledgeMode = AcknowledgeMode.MANUAL;
} else if (ACKNOWLEDGE_NONE.equals(acknowledge)) {
}
else if (ACKNOWLEDGE_NONE.equals(acknowledge)) {
acknowledgeMode = AcknowledgeMode.NONE;
} else {
}
else {
parserContext.getReaderContext().error(
"Invalid listener container 'acknowledge' setting [" + acknowledge
+ "]: only \"auto\", \"manual\", and \"none\" supported.", ele);
}
return acknowledgeMode;
} else {
}
else {
return null;
}
}

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RetryInterceptorBuilder.java
View File

@@ -133,7 +133,7 @@ public abstract class RetryInterceptorBuilder<T extends MethodInterceptor> {
* @param maxInterval The max interval.
* @return this.
*/
public RetryInterceptorBuilder<T> backOffOptions(long initialInterval, double multiplier , long maxInterval) {
public RetryInterceptorBuilder<T> backOffOptions(long initialInterval, double multiplier, long maxInterval) {
Assert.isNull(this.retryOperations, "cannot set the back off policy when a custom retryOperations has been set");
Assert.isTrue(!this.backOffPolicySet, "cannot set the back off options when a back off policy has been set");
ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();
@@ -203,7 +203,7 @@ public abstract class RetryInterceptorBuilder<T extends MethodInterceptor> {
private RetryInterceptorBuilder() {
}
public static class StatefulRetryInterceptorBuilder extends RetryInterceptorBuilder<StatefulRetryOperationsInterceptor> {
public static final class StatefulRetryInterceptorBuilder extends RetryInterceptorBuilder<StatefulRetryOperationsInterceptor> {
private final StatefulRetryOperationsInterceptorFactoryBean factoryBean =
new StatefulRetryOperationsInterceptorFactoryBean();
@@ -289,7 +289,7 @@ public abstract class RetryInterceptorBuilder<T extends MethodInterceptor> {
}
public static class StatelessRetryInterceptorBuilder extends RetryInterceptorBuilder<RetryOperationsInterceptor> {
public static final class StatelessRetryInterceptorBuilder extends RetryInterceptorBuilder<RetryOperationsInterceptor> {
private final StatelessRetryOperationsInterceptorFactoryBean factoryBean =
new StatelessRetryOperationsInterceptorFactoryBean();

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/StatefulRetryOperationsInterceptorFactoryBean.java
View File

@@ -93,7 +93,8 @@ public class StatefulRetryOperationsInterceptorFactoryBean extends AbstractRetry
Message message = (Message) args[1];
if (messageRecoverer == null) {
logger.warn("Message dropped on recovery: " + message, cause);
} else {
}
else {
messageRecoverer.recover(message, cause);
}
// This is actually a normal outcome. It means the recovery was successful, but we don't want to consume

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/StatelessRetryOperationsInterceptorFactoryBean.java
View File

@@ -39,6 +39,7 @@ import org.springframework.retry.support.RetryTemplate;
* @see RetryOperations#execute(org.springframework.retry.RetryCallback, org.springframework.retry.RecoveryCallback)
*
* @author Dave Syer
* @author Gary Russell
*
*/
public class StatelessRetryOperationsInterceptorFactoryBean extends AbstractRetryOperationsInterceptorFactoryBean {
@@ -60,7 +61,8 @@ public class StatelessRetryOperationsInterceptorFactoryBean extends AbstractRetr
Message message = (Message) args[1];
if (messageRecoverer == null) {
logger.warn("Message dropped on recovery: " + message, cause);
} else {
}
else {
messageRecoverer.recover(message, cause);
}
return null;

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

@@ -300,7 +300,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
InetAddress localMachine = InetAddress.getLocalHost();
temp = localMachine.getHostName();
this.logger.debug("Using hostname [" + temp + "] for hostname.");
} catch (UnknownHostException e) {
}
catch (UnknownHostException e) {
this.logger.warn("Could not get host name, using 'localhost' as default value", e);
temp = "localhost";
}

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractRoutingConnectionFactory.java
View File

@@ -193,7 +193,7 @@ public abstract class AbstractRoutingConnectionFactory implements ConnectionFact
*/
protected void addTargetConnectionFactory(Object key, ConnectionFactory connectionFactory) {
this.targetConnectionFactories.put(key, connectionFactory);
for(ConnectionListener listener : this.connectionListeners) {
for (ConnectionListener listener : this.connectionListeners) {
connectionFactory.addConnectionListener(listener);
}
}

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

@@ -365,7 +365,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
@Override
public void shutdownCompleted(ShutdownSignalException cause) {
this.closeExceptionLogger.log(logger, "Channel shutdown" ,cause);
this.closeExceptionLogger.log(logger, "Channel shutdown", cause);
}
@Override
@@ -503,7 +503,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
}
else if (this.cacheMode == CacheMode.CONNECTION) {
if (!connection.isOpen()) {
synchronized(this.connectionMonitor) {
synchronized (this.connectionMonitor) {
this.allocatedConnectionNonTransactionalChannels.get(connection).clear();
this.allocatedConnectionTransactionalChannels.get(connection).clear();
connection.notifyCloseIfNecessary();
@@ -768,7 +768,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
public Properties getCacheProperties() {
Properties props = new Properties();
props.setProperty("cacheMode", this.cacheMode.name());
synchronized(this.connectionMonitor) {
synchronized (this.connectionMonitor) {
props.setProperty("channelCacheSize", Integer.toString(this.channelCacheSize));
if (this.cacheMode.equals(CacheMode.CONNECTION)) {
props.setProperty("connectionCacheSize", Integer.toString(this.connectionCacheSize));
@@ -827,7 +827,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
+ " " + super.toString() + "]";
}
private class CachedChannelInvocationHandler implements InvocationHandler {
private final class CachedChannelInvocationHandler implements InvocationHandler {
private final ChannelCachingConnectionProxy theConnection;
@@ -938,7 +938,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
* in the list, it means we're closing a cached channel (for which a permit
* has already been released).
*/
synchronized(this.channelList) {
synchronized (this.channelList) {
if (this.channelList.contains(proxy)) {
return;
}
@@ -1031,16 +1031,16 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
catch (Exception e) {}
catch (Exception e) { }
finally {
try {
if (channel.isOpen()) {
channel.close();
}
}
catch (IOException e) {}
catch (AlreadyClosedException e) {}
catch (TimeoutException e) {}
catch (IOException e) { }
catch (AlreadyClosedException e) { }
catch (TimeoutException e) { }
}
}
@@ -1062,7 +1062,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
}
private class ChannelCachingConnectionProxy implements Connection, ConnectionProxy {
private class ChannelCachingConnectionProxy implements Connection, ConnectionProxy { // NOSONAR - final (tests spy)
private volatile Connection target;

12
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java
View File

@@ -41,7 +41,11 @@ import com.rabbitmq.client.Channel;
* @author Gary Russell
* @author Artem Bilan
*/
public class ConnectionFactoryUtils {
public final class ConnectionFactoryUtils {
private ConnectionFactoryUtils() {
super();
}
/**
* Determine whether the given RabbitMQ Channel is transactional, that is, bound to the current thread by Spring's
@@ -126,7 +130,7 @@ public class ConnectionFactoryUtils {
if (resourceHolderToUse == null) {
resourceHolderToUse = new RabbitResourceHolder();
}
Connection connection = resourceFactory.getConnection(resourceHolderToUse);//NOSONAR
Connection connection = resourceFactory.getConnection(resourceHolderToUse); //NOSONAR
Channel channel = null;
try {
/*
@@ -152,7 +156,7 @@ public class ConnectionFactoryUtils {
}
catch (IOException ex) {
RabbitUtils.closeChannel(channel);//NOSONAR
RabbitUtils.closeChannel(channel); //NOSONAR
RabbitUtils.closeConnection(connection);
throw new AmqpIOException(ex);
}
@@ -250,7 +254,7 @@ public class ConnectionFactoryUtils {
* JtaTransactionManager transaction).
* @see org.springframework.transaction.jta.JtaTransactionManager
*/
private static class RabbitResourceSynchronization extends
private static final class RabbitResourceSynchronization extends
ResourceHolderSynchronization<RabbitResourceHolder, Object> {
private final boolean locallyTransacted;

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

@@ -32,12 +32,16 @@ import com.rabbitmq.client.Channel;
* @since 1.2
*
*/
public class ConsumerChannelRegistry {
public final class ConsumerChannelRegistry {
private static final Log logger = LogFactory.getLog(ConsumerChannelRegistry.class);
private static final ThreadLocal<ChannelHolder> consumerChannel = new ThreadLocal<ChannelHolder>();
private ConsumerChannelRegistry() {
super();
}
/**
* If a listener container is configured to use a RabbitTransactionManager, the
* consumer's channel is registered here so that it is used as the bound resource
@@ -99,7 +103,7 @@ public class ConsumerChannelRegistry {
return channel;
}
private static class ChannelHolder {
private static final class ChannelHolder {
private final Channel channel;

10
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactory.java
View File

@@ -104,7 +104,10 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
this.password = password;
this.useSSL = useSSL;
this.sslPropertiesLocation = sslPropertiesLocation;
this.keyStore = this.trustStore = this.keyStorePassPhrase = this.trustStorePassPhrase = null;
this.keyStore = null;
this.trustStore = null;
this.keyStorePassPhrase = null;
this.trustStorePassPhrase = null;
}
/**
@@ -169,7 +172,10 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
this.password = password;
this.useSSL = useSSL;
this.sslPropertiesLocation = sslPropertiesLocation;
this.keyStore = this.trustStore = this.keyStorePassPhrase = this.trustStorePassPhrase = null;
this.keyStore = null;
this.trustStore = null;
this.keyStorePassPhrase = null;
this.trustStorePassPhrase = null;
}
/**

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

@@ -153,7 +153,8 @@ public class RabbitResourceHolder extends ResourceHolderSupport {
}
channel.txCommit();
}
} catch (IOException e) {
}
catch (IOException e) {
throw new AmqpException("failed to commit RabbitMQ transaction", e);
}
}
@@ -174,7 +175,7 @@ public class RabbitResourceHolder extends ResourceHolderSupport {
logger.debug("Could not close synchronized Rabbit Channel after transaction", ex);
}
}
for (Connection con : this.connections) {//NOSONAR
for (Connection con : this.connections) { //NOSONAR
RabbitUtils.closeConnection(con);
}
this.connections.clear();
@@ -196,7 +197,8 @@ public class RabbitResourceHolder extends ResourceHolderSupport {
for (Long deliveryTag : this.deliveryTags.get(channel)) {
try {
channel.basicReject(deliveryTag, true);
} catch (IOException ex) {
}
catch (IOException ex) {
throw new AmqpIOException(ex);
}
}

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

@@ -53,7 +53,8 @@ public abstract class RabbitUtils {
if (connection != null) {
try {
connection.close();
} catch (Exception ex) {
}
catch (Exception ex) {
logger.debug("Ignoring Connection exception - assuming already closed: " + ex.getMessage(), ex);
}
}
@@ -91,7 +92,8 @@ public abstract class RabbitUtils {
Assert.notNull(channel, "Channel must not be null");
try {
channel.txCommit();
} catch (IOException ex) {
}
catch (IOException ex) {
throw new AmqpIOException(ex);
}
}
@@ -100,7 +102,8 @@ public abstract class RabbitUtils {
Assert.notNull(channel, "Channel must not be null");
try {
channel.txRollback();
} catch (IOException ex) {
}
catch (IOException ex) {
throw new AmqpIOException(ex);
}
}
@@ -124,7 +127,8 @@ public abstract class RabbitUtils {
* If not transactional then we are auto-acking (at least as of 1.0.0.M2) so there is nothing to recover.
* Messages are going to be lost in general.
*/
} catch (Exception ex) {
}
catch (Exception ex) {
throw RabbitExceptionTranslator.convertRabbitAccessException(ex);
}
}
@@ -137,7 +141,8 @@ public abstract class RabbitUtils {
public static void declareTransactional(Channel channel) {
try {
channel.txSelect();
} catch (IOException e) {
}
catch (IOException e) {
throw RabbitExceptionTranslator.convertRabbitAccessException(e);
}
}

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java
View File

@@ -52,7 +52,8 @@ public class SimpleConnection implements Connection {
channel.txSelect();
}
return channel;
} catch (IOException e) {
}
catch (IOException e) {
throw RabbitExceptionTranslator.convertRabbitAccessException(e);
}
}
@@ -62,7 +63,8 @@ public class SimpleConnection implements Connection {
try {
// let the physical close time out if necessary
this.delegate.close(this.closeTimeout);
} catch (IOException e) {
}
catch (IOException e) {
throw RabbitExceptionTranslator.convertRabbitAccessException(e);
}
}

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

@@ -178,7 +178,8 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
try {
channel.exchangeDelete(exchangeName);
} catch (IOException e) {
}
catch (IOException e) {
return false;
}
return true;
@@ -250,7 +251,8 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
public Boolean doInRabbit(Channel channel) throws Exception {
try {
channel.queueDelete(queueName);
} catch (IOException e) {
}
catch (IOException e) {
return false;
}
return true;
@@ -313,7 +315,8 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
channel.queueUnbind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
} else {
}
else {
channel.exchangeUnbind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
}
@@ -588,7 +591,8 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
catch (IOException e) {
logOrRethrowDeclarationException(queue, "queue", e);
}
} else if (this.logger.isDebugEnabled()) {
}
else if (this.logger.isDebugEnabled()) {
this.logger.debug("Queue with name that starts with 'amq.' cannot be declared.");
}
}
@@ -609,7 +613,8 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
}
} else {
}
else {
channel.exchangeBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
}

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

@@ -1169,7 +1169,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
protected Message doSendAndReceive(final String exchange, final String routingKey, final Message message,
CorrelationData correlationData) {
if (!this.evaluatedFastReplyTo) {
synchronized(this) {
synchronized (this) {
if (!this.evaluatedFastReplyTo) {
evaluateFastReplyTo();
}
@@ -1231,7 +1231,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
try {
channel.basicCancel(consumerTag);
}
catch (Exception e) {}
catch (Exception e) { }
}
return reply;
}

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

@@ -191,7 +191,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
Assert.notNull(queues, "'queues' cannot be null");
Assert.noNullElements(queues, "'queues' cannot contain null elements");
String[] queueNames = new String[queues.length];
for (int i = 0; i< queues.length; i++) {
for (int i = 0; i < queues.length; i++) {
queueNames[i] = queues[i].getName();
}
this.addQueueNames(queueNames);
@@ -218,7 +218,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
Assert.notNull(queues, "'queues' cannot be null");
Assert.noNullElements(queues, "'queues' cannot contain null elements");
String[] queueNames = new String[queues.length];
for (int i = 0; i< queues.length; i++) {
for (int i = 0; i < queues.length; i++) {
queueNames[i] = queues[i].getName();
}
return this.removeQueueNames(queueNames);
@@ -470,7 +470,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
this.lifecycleMonitor.notifyAll();
}
doInitialize();
} catch (Exception ex) {
}
catch (Exception ex) {
throw convertRabbitAccessException(ex);
}
}
@@ -488,9 +489,11 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
// Shut down the invokers.
try {
doShutdown();
} catch (Exception ex) {
}
catch (Exception ex) {
throw convertRabbitAccessException(ex);
} finally {
}
finally {
synchronized (this.lifecycleMonitor) {
this.running = false;
this.lifecycleMonitor.notifyAll();
@@ -545,7 +548,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
logger.debug("Starting Rabbit listener container.");
}
doStart();
} catch (Exception ex) {
}
catch (Exception ex) {
throw convertRabbitAccessException(ex);
}
}
@@ -572,9 +576,11 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
public void stop() {
try {
doStop();
} catch (Exception ex) {
}
catch (Exception ex) {
throw convertRabbitAccessException(ex);
} finally {
}
finally {
synchronized (this.lifecycleMonitor) {
this.running = false;
this.lifecycleMonitor.notifyAll();
@@ -844,13 +850,28 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
// Regular case: failed while active.
// Invoke ErrorHandler if available.
invokeErrorHandler(ex);
} else {
}
else {
// Rare case: listener thread failed after container shutdown.
// Log at debug level, to avoid spamming the shutdown log.
logger.debug("Listener exception after container shutdown", ex);
}
}
/**
* @param e The Exception.
* @param message The failed message.
* @return If 'e' is of type {@link ListenerExecutionFailedException} - return 'e' as it is, otherwise wrap it to
* {@link ListenerExecutionFailedException} and return.
*/
protected Exception wrapToListenerExecutionFailedExceptionIfNeeded(Exception e, Message message) {
if (!(e instanceof ListenerExecutionFailedException)) {
// Wrap exception to ListenerExecutionFailedException.
return new ListenerExecutionFailedException("Listener threw exception", e, message);
}
return e;
}
/**
* Exception that indicates that the initial setup of this container's shared Rabbit Connection failed. This is
* indicating to invokers that they need to establish the shared Connection themselves on first access.
@@ -867,17 +888,4 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
}
}
/**
* @param e The Exception.
* @param message The failed message.
* @return If 'e' is of type {@link ListenerExecutionFailedException} - return 'e' as it is, otherwise wrap it to
* {@link ListenerExecutionFailedException} and return.
*/
protected Exception wrapToListenerExecutionFailedExceptionIfNeeded(Exception e, Message message) {
if (!(e instanceof ListenerExecutionFailedException)) {
// Wrap exception to ListenerExecutionFailedException.
return new ListenerExecutionFailedException("Listener threw exception", e, message);
}
return e;
}
}

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java
View File

@@ -53,7 +53,7 @@ public class ActiveObjectCounter<T> {
Collection<T> objects = new HashSet<T>(this.locks.keySet());
for (T object : objects) {
CountDownLatch lock = this.locks.get(object);
if (lock==null) {
if (lock == null) {
continue;
}
t0 = System.currentTimeMillis();

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

@@ -404,7 +404,7 @@ public class BlockingQueueConsumer {
private void checkMissingQueues() {
long now = System.currentTimeMillis();
if (now - this.retryDeclarationInterval > this.lastRetryDeclaration) {
synchronized(this.missingQueues) {
synchronized (this.missingQueues) {
Iterator<String> iterator = this.missingQueues.iterator();
while (iterator.hasNext()) {
boolean available = true;
@@ -604,7 +604,108 @@ public class BlockingQueueConsumer {
this.consumer = null;
}
private class InternalConsumer extends DefaultConsumer {
/**
* Perform a rollback, handling rollback exceptions properly.
* @param ex the thrown application exception or error
* @throws Exception in case of a rollback error
*/
public void rollbackOnExceptionIfNecessary(Throwable ex) throws Exception {
boolean ackRequired = !this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual();
try {
if (this.transactional) {
if (logger.isDebugEnabled()) {
logger.debug("Initiating transaction rollback on application exception: " + ex);
}
RabbitUtils.rollbackIfNecessary(this.channel);
}
if (ackRequired) {
// We should always requeue if the container was stopping
boolean shouldRequeue = this.defaultRequeuRejected ||
ex instanceof MessageRejectedWhileStoppingException;
Throwable t = ex;
while (shouldRequeue && t != null) {
if (t instanceof AmqpRejectAndDontRequeueException) {
shouldRequeue = false;
}
t = t.getCause();
}
if (logger.isDebugEnabled()) {
logger.debug("Rejecting messages (requeue=" + shouldRequeue + ")");
}
for (Long deliveryTag : this.deliveryTags) {
// With newer RabbitMQ brokers could use basicNack here...
this.channel.basicReject(deliveryTag, shouldRequeue);
}
if (this.transactional) {
// Need to commit the reject (=nack)
RabbitUtils.commitIfNecessary(this.channel);
}
}
}
catch (Exception e) {
logger.error("Application exception overridden by rollback exception", ex);
throw e;
}
finally {
this.deliveryTags.clear();
}
}
/**
* Perform a commit or message acknowledgement, as appropriate.
* @param locallyTransacted Whether the channel is locally transacted.
* @throws IOException Any IOException.
* @return true if at least one delivery tag exists.
*/
public boolean commitIfNecessary(boolean locallyTransacted) throws IOException {
if (this.deliveryTags.isEmpty()) {
return false;
}
try {
boolean ackRequired = !this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual();
if (ackRequired) {
if (this.transactional && !locallyTransacted) {
// Not locally transacted but it is transacted so it
// could be synchronized with an external transaction
for (Long deliveryTag : this.deliveryTags) {
ConnectionFactoryUtils.registerDeliveryTag(this.connectionFactory, this.channel, deliveryTag);
}
}
else {
long deliveryTag = new ArrayList<Long>(this.deliveryTags).get(this.deliveryTags.size() - 1);
this.channel.basicAck(deliveryTag, true);
}
}
if (locallyTransacted) {
// For manual acks we still need to commit
RabbitUtils.commitIfNecessary(this.channel);
}
}
finally {
this.deliveryTags.clear();
}
return true;
}
@Override
public String toString() {
return "Consumer: tags=[" + (this.consumerTags.toString()) + "], channel=" + this.channel
+ ", acknowledgeMode=" + this.acknowledgeMode + " local queue size=" + this.queue.size();
}
private final class InternalConsumer extends DefaultConsumer {
private InternalConsumer(Channel channel) {
super(channel);
@@ -648,7 +749,7 @@ public class BlockingQueueConsumer {
if (logger.isDebugEnabled()) {
logger.debug("Received cancellation notice for tag " + consumerTag + "; " + BlockingQueueConsumer.this);
}
synchronized(BlockingQueueConsumer.this.consumerTags) {
synchronized (BlockingQueueConsumer.this.consumerTags) {
BlockingQueueConsumer.this.consumerTags.remove(consumerTag);
}
}
@@ -682,7 +783,7 @@ public class BlockingQueueConsumer {
private final byte[] body;
Delivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {//NOSONAR
Delivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { //NOSONAR
this.consumerTag = consumerTag;
this.envelope = envelope;
this.properties = properties;
@@ -707,7 +808,7 @@ public class BlockingQueueConsumer {
}
@SuppressWarnings("serial")
private static class DeclarationException extends AmqpException {
private static final class DeclarationException extends AmqpException {
private DeclarationException() {
super("Failed to declare queue(s):");
@@ -734,102 +835,4 @@ public class BlockingQueueConsumer {
}
@Override
public String toString() {
return "Consumer: tags=[" + (this.consumerTags.toString()) + "], channel=" + this.channel
+ ", acknowledgeMode=" + this.acknowledgeMode + " local queue size=" + this.queue.size();
}
/**
* Perform a rollback, handling rollback exceptions properly.
* @param ex the thrown application exception or error
* @throws Exception in case of a rollback error
*/
public void rollbackOnExceptionIfNecessary(Throwable ex) throws Exception {
boolean ackRequired = !this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual();
try {
if (this.transactional) {
if (logger.isDebugEnabled()) {
logger.debug("Initiating transaction rollback on application exception: " + ex);
}
RabbitUtils.rollbackIfNecessary(this.channel);
}
if (ackRequired) {
// We should always requeue if the container was stopping
boolean shouldRequeue = this.defaultRequeuRejected ||
ex instanceof MessageRejectedWhileStoppingException;
Throwable t = ex;
while (shouldRequeue && t != null) {
if (t instanceof AmqpRejectAndDontRequeueException) {
shouldRequeue = false;
}
t = t.getCause();
}
if (logger.isDebugEnabled()) {
logger.debug("Rejecting messages (requeue=" + shouldRequeue + ")");
}
for (Long deliveryTag : this.deliveryTags) {
// With newer RabbitMQ brokers could use basicNack here...
this.channel.basicReject(deliveryTag, shouldRequeue);
}
if (this.transactional) {
// Need to commit the reject (=nack)
RabbitUtils.commitIfNecessary(this.channel);
}
}
} catch (Exception e) {
logger.error("Application exception overridden by rollback exception", ex);
throw e;
} finally {
this.deliveryTags.clear();
}
}
/**
* Perform a commit or message acknowledgement, as appropriate.
* @param locallyTransacted Whether the channel is locally transacted.
* @throws IOException Any IOException.
* @return true if at least one delivery tag exists.
*/
public boolean commitIfNecessary(boolean locallyTransacted) throws IOException {
if (this.deliveryTags.isEmpty()) {
return false;
}
try {
boolean ackRequired = !this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual();
if (ackRequired) {
if (this.transactional && !locallyTransacted) {
// Not locally transacted but it is transacted so it
// could be synchronized with an external transaction
for (Long deliveryTag : this.deliveryTags) {
ConnectionFactoryUtils.registerDeliveryTag(this.connectionFactory, this.channel, deliveryTag);
}
} else {
long deliveryTag = new ArrayList<Long>(this.deliveryTags).get(this.deliveryTags.size() - 1);
this.channel.basicAck(deliveryTag, true);
}
}
if (locallyTransacted) {
// For manual acks we still need to commit
RabbitUtils.commitIfNecessary(this.channel);
}
}
finally {
this.deliveryTags.clear();
}
return true;
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2015 the original author or authors.
* Copyright 2014-2016 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.
@@ -194,7 +194,7 @@ public class RabbitListenerEndpointRegistrar implements BeanFactoryAware, Initia
}
private static class AmqpListenerEndpointDescriptor {
private static final class AmqpListenerEndpointDescriptor {
private final RabbitListenerEndpoint endpoint;

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistry.java
View File

@@ -282,7 +282,7 @@ public class RabbitListenerEndpointRegistry implements DisposableBean, SmartLife
}
private static class AggregatingCallback implements Runnable {
private static final class AggregatingCallback implements Runnable {
private final AtomicInteger count;

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

@@ -183,10 +183,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
private volatile ApplicationEventPublisher applicationEventPublisher;
public interface ContainerDelegate {
void invokeListener(Channel channel, Message message) throws Exception;
}
private final ContainerDelegate delegate = new ContainerDelegate() {
@Override
public void invokeListener(Channel channel, Message message) throws Exception {
@@ -276,7 +272,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
Assert.isTrue(concurrentConsumers <= this.maxConcurrentConsumers,
"'concurrentConsumers' cannot be more than 'maxConcurrentConsumers'");
}
synchronized(this.consumersMonitor) {
synchronized (this.consumersMonitor) {
if (logger.isDebugEnabled()) {
logger.debug("Changing consumers from " + this.concurrentConsumers + " to " + concurrentConsumers);
}
@@ -478,7 +474,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
public void setConsumerArguments(Map<String, Object> args) {
synchronized(this.consumersMonitor) {
synchronized (this.consumersMonitor) {
this.consumerArgs.clear();
this.consumerArgs.putAll(args);
}
@@ -881,7 +877,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
private boolean isActive(BlockingQueueConsumer consumer) {
Boolean consumerActive;
synchronized(this.consumersMonitor) {
synchronized (this.consumersMonitor) {
if (this.consumers != null) {
Boolean active = this.consumers.get(consumer);
consumerActive = active != null && active;
@@ -937,7 +933,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
catch (AmqpConnectException e) {
logger.info("Broker not available; cannot check queue declarations");
}
catch (AmqpIOException e){
catch (AmqpIOException e) {
if (RabbitUtils.isMismatchedQueueArgs(e)) {
throw new FatalListenerStartupException("Mismatched queues", e);
}
@@ -981,7 +977,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
private void considerAddingAConsumer() {
synchronized(this.consumersMonitor) {
synchronized (this.consumersMonitor) {
if (this.consumers != null
&& this.maxConcurrentConsumers != null && this.consumers.size() < this.maxConcurrentConsumers) {
long now = System.currentTimeMillis();
@@ -1148,7 +1144,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
catch (RuntimeException e) {
throw e;
}
catch (Throwable e) {//NOSONAR
catch (Throwable e) { //NOSONAR
// ok to catch Throwable here because we re-throw it below
throw new WrappedTransactionException(e);
}
@@ -1181,7 +1177,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
catch (ImmediateAcknowledgeAmqpException e) {
break;
}
catch (Throwable ex) {//NOSONAR
catch (Throwable ex) { //NOSONAR
consumer.rollbackOnExceptionIfNecessary(ex);
throw ex;
}
@@ -1196,6 +1192,44 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
return this.adviceChain;
}
@Override
protected void invokeListener(Channel channel, Message message) throws Exception {
this.proxy.invokeListener(channel, message);
}
/**
* 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}
* @throws Exception if the shared connection still can't be established
*/
protected void handleStartupFailure(BackOffExecution backOffExecution) throws Exception {
long recoveryInterval = backOffExecution.nextBackOff();
if (BackOffExecution.STOP == recoveryInterval) {
synchronized (this) {
if (isActive()) {
logger.warn("stopping container - restart recovery attempts exhausted");
stop();
}
}
return;
}
try {
if (logger.isDebugEnabled() && isActive()) {
logger.debug("Recovering consumer in " + recoveryInterval + " ms.");
}
long timeout = System.currentTimeMillis() + recoveryInterval;
while (isActive() && System.currentTimeMillis() < timeout) {
Thread.sleep(200);
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException("Unrecoverable interruption on consumer restart");
}
}
@Override
public String toString() {
return "SimpleMessageListenerContainer "
@@ -1205,7 +1239,13 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
+ ", queueNames=" + Arrays.toString(getQueueNames()) + "]";
}
private class AsyncMessageProcessingConsumer implements Runnable {
public interface ContainerDelegate {
void invokeListener(Channel channel, Message message) throws Exception;
}
private final class AsyncMessageProcessingConsumer implements Runnable {
private final BlockingQueueConsumer consumer;
@@ -1228,7 +1268,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
* @throws InterruptedException if the consumer startup is interrupted
*/
private FatalListenerStartupException getStartupException() throws TimeoutException, InterruptedException {
this.start.await(60000L, TimeUnit.MILLISECONDS);//NOSONAR - ignore return value
this.start.await(60000L, TimeUnit.MILLISECONDS); //NOSONAR - ignore return value
return this.startupException;
}
@@ -1263,7 +1303,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
catch (FatalListenerStartupException ex) {
throw ex;
}
catch (Throwable t) {//NOSONAR
catch (Throwable t) { //NOSONAR
this.start.countDown();
handleStartupFailure(this.consumer.getBackOffExecution());
throw t;
@@ -1380,12 +1420,12 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
this.logConsumerException(e);
}
}
catch (Error e) {//NOSONAR
catch (Error e) { //NOSONAR
// ok to catch Error - we're aborting so will stop
logger.error("Consumer thread error, thread abort.", e);
aborted = true;
}
catch (Throwable t) {//NOSONAR
catch (Throwable t) { //NOSONAR
// by now, it must be an exception
if (isActive()) {
this.logConsumerException(t);
@@ -1458,46 +1498,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
@Override
protected void invokeListener(Channel channel, Message message) throws Exception {
this.proxy.invokeListener(channel, message);
}
/**
* 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}
* @throws Exception if the shared connection still can't be established
*/
protected void handleStartupFailure(BackOffExecution backOffExecution) throws Exception {
long recoveryInterval = backOffExecution.nextBackOff();
if (BackOffExecution.STOP == recoveryInterval) {
synchronized (this) {
if (isActive()) {
logger.warn("stopping container - restart recovery attempts exhausted");
stop();
}
}
return;
}
try {
if (logger.isDebugEnabled() && isActive()) {
logger.debug("Recovering consumer in " + recoveryInterval + " ms.");
}
long timeout = System.currentTimeMillis() + recoveryInterval;
while (isActive() && System.currentTimeMillis() < timeout) {
Thread.sleep(200);
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException("Unrecoverable interruption on consumer restart");
}
}
@SuppressWarnings("serial")
private static class WrappedTransactionException extends RuntimeException {
private static final class WrappedTransactionException extends RuntimeException {
private WrappedTransactionException(Throwable cause) {
super(cause);

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

@@ -126,7 +126,7 @@ public class DelegatingInvocableHandler {
if (handler == null) {
throw new AmqpException("No method found for " + payloadClass);
}
this.cachedHandlers.putIfAbsent(payloadClass, handler);//NOSONAR
this.cachedHandlers.putIfAbsent(payloadClass, handler); //NOSONAR
setupReplyTo(handler);
}
return handler;
@@ -222,7 +222,7 @@ public class DelegatingInvocableHandler {
*/
public String getMethodNameFor(Object payload) {
InvocableHandlerMethod handlerForPayload = getHandlerForPayload(payload.getClass());
return handlerForPayload == null ? "no match" : handlerForPayload.getMethod().toGenericString();//NOSONAR
return handlerForPayload == null ? "no match" : handlerForPayload.getMethod().toGenericString(); //NOSONAR
}
}

21
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java
View File

@@ -182,7 +182,7 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
* If the inbound message has no type information and the configured message converter
* supports it, we attempt to infer the conversion type from the method signature.
*/
private class MessagingMessageConverterAdapter extends MessagingMessageConverter {
private final class MessagingMessageConverterAdapter extends MessagingMessageConverter {
private final Object bean;
@@ -223,11 +223,11 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
for (int i = 0; i < this.method.getParameterTypes().length; i++) {
MethodParameter methodParameter = new MethodParameter(this.method, i);
/*
* We're looking for a single non-annotated parameter, or one annotated with @Payload.
* We ignore parameters with type Message because they are not involved with conversion.
*/
if (eligibleParameter(methodParameter)
/*
* We're looking for a single non-annotated parameter, or one annotated with @Payload.
* We ignore parameters with type Message because they are not involved with conversion.
*/
if (isEligibleParameter(methodParameter)
&& (methodParameter.getParameterAnnotations().length == 0
|| methodParameter.hasParameterAnnotation(Payload.class))) {
if (genericParameterType == null) {
@@ -258,7 +258,7 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
* Don't consider parameter types that are available after conversion.
* Message, Message<?> and Channel.
*/
private boolean eligibleParameter(MethodParameter methodParameter) {
private boolean isEligibleParameter(MethodParameter methodParameter) {
Type parameterType = methodParameter.getGenericParameterType();
if (parameterType.equals(Channel.class)
|| parameterType.equals(org.springframework.amqp.core.Message.class)) {
@@ -267,12 +267,7 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
if (parameterType instanceof ParameterizedType) {
ParameterizedType parameterizedType = (ParameterizedType) parameterType;
if (parameterizedType.getRawType().equals(Message.class)) {
if(parameterizedType.getActualTypeArguments()[0] instanceof WildcardType) {
return false;
}
else {
return true;
}
return !(parameterizedType.getActualTypeArguments()[0] instanceof WildcardType);
}
}
return !parameterType.equals(Message.class); // could be Message without a generic type

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

@@ -528,7 +528,7 @@ public class AmqpAppender extends AppenderSkeleton {
@SuppressWarnings("rawtypes")
Map props = event.getProperties();
@SuppressWarnings("unchecked")
Set<Entry<?,?>> entrySet = props.entrySet();
Set<Entry<?, ?>> entrySet = props.entrySet();
for (Entry<?, ?> entry : entrySet) {
amqpProps.setHeader(entry.getKey().toString(), entry.getValue());
}
@@ -560,10 +560,10 @@ public class AmqpAppender extends AppenderSkeleton {
try {
message = new Message(msgBody.toString().getBytes(AmqpAppender.this.charset), amqpProps);
}
catch (UnsupportedEncodingException e) {/* fall back to default */}
catch (UnsupportedEncodingException e) { /* fall back to default */ }
}
if (message == null) {
message = new Message(msgBody.toString().getBytes(), amqpProps);//NOSONAR (default charset)
message = new Message(msgBody.toString().getBytes(), amqpProps); //NOSONAR (default charset)
}
message = postProcessMessageBeforeSend(message, event);
rabbitTemplate.send(AmqpAppender.this.exchangeName, routingKey, message);

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

@@ -240,7 +240,7 @@ public class AmqpAppender extends AbstractAppender {
@SuppressWarnings("rawtypes")
Map props = event.getProperties();
@SuppressWarnings("unchecked")
Set<Entry<?,?>> entrySet = props.entrySet();
Set<Entry<?, ?>> entrySet = props.entrySet();
for (Entry<?, ?> entry : entrySet) {
amqpProps.setHeader(entry.getKey().toString(), entry.getValue());
}
@@ -268,10 +268,10 @@ public class AmqpAppender extends AbstractAppender {
message = new Message(msgBody.toString().getBytes(AmqpAppender.this.manager.charset),
amqpProps);
}
catch (UnsupportedEncodingException e) {/* fall back to default */}
catch (UnsupportedEncodingException e) { /* fall back to default */ }
}
if (message == null) {
message = new Message(msgBody.toString().getBytes(), amqpProps);//NOSONAR (default charset)
message = new Message(msgBody.toString().getBytes(), amqpProps); //NOSONAR (default charset)
}
message = postProcessMessageBeforeSend(message, event);
rabbitTemplate.send(AmqpAppender.this.manager.exchangeName, routingKey, message);

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

@@ -528,7 +528,7 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
message = new Message(msgBody.getBytes(AmqpAppender.this.charset), amqpProps);
}
catch (UnsupportedEncodingException e) {
message = new Message(msgBody.getBytes(), amqpProps);//NOSONAR (default charset)
message = new Message(msgBody.getBytes(), amqpProps); //NOSONAR (default charset)
}
}

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

@@ -204,7 +204,7 @@ public class DefaultMessagePropertiesConverter implements MessagePropertiesConve
private Map<String, Object> convertHeadersIfNecessary(Map<String, Object> headers) {
if (CollectionUtils.isEmpty(headers)) {
return Collections.<String, Object> emptyMap();
return Collections.<String, Object>emptyMap();
}
Map<String, Object> writableHeaders = new HashMap<String, Object>();
for (Map.Entry<String, Object> entry : headers.entrySet()) {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2016 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,8 +18,8 @@ package org.springframework.amqp.rabbit.support;
import org.springframework.amqp.core.MessageProperties;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Envelope;
/**
* Strategy interface for converting between Spring AMQP {@link MessageProperties}

7
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java
View File

@@ -101,7 +101,7 @@ public class PublisherCallbackChannelImpl
private final ConcurrentMap<String, Listener> listeners = new ConcurrentHashMap<String, Listener>();
private final Map<Listener, SortedMap<Long, PendingConfirm>> pendingConfirms
= new ConcurrentHashMap<PublisherCallbackChannel.Listener, SortedMap<Long,PendingConfirm>>();
= new ConcurrentHashMap<PublisherCallbackChannel.Listener, SortedMap<Long, PendingConfirm>>();
private final SortedMap<Long, Listener> listenerForSeq = new ConcurrentSkipListMap<Long, Listener>();
@@ -111,7 +111,7 @@ public class PublisherCallbackChannelImpl
if (!conditionalMethodsChecked) {
// The following reflection is required to maintain compatibility with pre 3.6.x clients.
ReflectionUtils.doWithMethods(delegate.getClass(), new MethodCallback(){
ReflectionUtils.doWithMethods(delegate.getClass(), new MethodCallback() {
@Override
public void doWith(java.lang.reflect.Method method)
@@ -812,8 +812,7 @@ public class PublisherCallbackChannelImpl
String exchange,
String routingKey,
AMQP.BasicProperties properties,
byte[] body) throws IOException
{
byte[] body) throws IOException {
String uuidObject = properties.getHeaders().get(RETURN_CORRELATION_KEY).toString();
Listener listener = this.listeners.get(uuidObject);
if (listener == null || !listener.isReturnListener()) {

8
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -44,7 +44,11 @@ import com.rabbitmq.client.ShutdownSignalException;
* @since 1.2
*
*/
public class RabbitExceptionTranslator {
public final class RabbitExceptionTranslator {
private RabbitExceptionTranslator() {
super();
}
public static RuntimeException convertRabbitAccessException(Throwable ex) {
Assert.notNull(ex, "Exception must not be null");

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

@@ -156,7 +156,8 @@ public class RabbitTransactionManager extends AbstractPlatformTransactionManager
txObject.getResourceHolder().setTimeoutInSeconds(timeout);
}
TransactionSynchronizationManager.bindResource(getConnectionFactory(), txObject.getResourceHolder());
} catch (AmqpException ex) {
}
catch (AmqpException ex) {
if (resourceHolder != null) {
ConnectionFactoryUtils.releaseResources(resourceHolder);
}

1
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java
View File

@@ -43,7 +43,6 @@ import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;

128
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java
View File

@@ -97,18 +97,6 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
assertEquals(1, simpleFactory.getListenerContainers().size());
}
@Component
static class SampleBean {
@RabbitListener(queues = "myQueue")
public void defaultHandle(String msg) {
}
@RabbitListener(containerFactory = "simpleFactory", queues = "myQueue")
public void simpleHandle(String msg) {
}
}
/**
* Test for {@link FullBean} discovery. In this case, no default is set because
* all endpoints provide a default registry. This shows that the default factory
@@ -145,27 +133,6 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
}
}
@Component
static class FullBean {
@RabbitListener(id = "listener1", containerFactory = "simpleFactory", queues = {"queue1", "queue2"},
exclusive = true, priority = "34", admin = "rabbitAdmin")
public void fullHandle(String msg) {
}
}
@Component
static class FullConfigurableBean {
@RabbitListener(id = "${rabbit.listener.id}", containerFactory = "${rabbit.listener.containerFactory}",
queues = {"${rabbit.listener.queue}", "queue2"}, exclusive = true,
priority = "${rabbit.listener.priority}", admin = "${rabbit.listener.admin}")
public void fullHandle(String msg) {
}
}
/**
* Test for {@link CustomBean} and an manually endpoint registered
* with "myCustomEndpointId". The custom endpoint does not provide
@@ -195,14 +162,6 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
customRegistry.getListenerContainer("myCustomEndpointId"));
}
@Component
static class CustomBean {
@RabbitListener(id = "listenerId", containerFactory = "customFactory", queues = "myQueue")
public void customHandle(String msg) {
}
}
/**
* Test for {@link DefaultBean} that does not define the container
* factory to use as a default is registered with an explicit
@@ -224,13 +183,6 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
assertEquals(1, defaultFactory.getListenerContainers().size());
}
static class DefaultBean {
@RabbitListener(queues = "myQueue")
public void handleIt(String msg) {
}
}
/**
* Test for {@link ValidationBean} with a validator ({@link TestValidator}) specified
* in a custom {@link org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory}.
@@ -255,14 +207,6 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
listener.onMessage(amqpMessage, mock(Channel.class));
}
@Component
static class ValidationBean {
@RabbitListener(containerFactory = "defaultFactory", queues = "myQueue")
public void defaultHandle(@Validated String msg) {
}
}
/**
* Test for {@link RabbitListenersBean} that validates that the
* {@code @RabbitListener} annotations generate one specific container per annotation.
@@ -293,6 +237,70 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
assertEquals("class2", fourth.getQueueNames().iterator().next());
}
private void assertQueues(AbstractRabbitListenerEndpoint actual, String... expectedQueues) {
Collection<String> actualQueues = actual.getQueueNames();
for (String expectedQueue : expectedQueues) {
assertTrue("Queue '" + expectedQueue + "' not found", actualQueues.contains(expectedQueue));
}
assertEquals("Wrong number of queues", expectedQueues.length, actualQueues.size());
}
@Component
static class SampleBean {
@RabbitListener(queues = "myQueue")
public void defaultHandle(String msg) {
}
@RabbitListener(containerFactory = "simpleFactory", queues = "myQueue")
public void simpleHandle(String msg) {
}
}
@Component
static class FullBean {
@RabbitListener(id = "listener1", containerFactory = "simpleFactory", queues = {"queue1", "queue2"},
exclusive = true, priority = "34", admin = "rabbitAdmin")
public void fullHandle(String msg) {
}
}
@Component
static class FullConfigurableBean {
@RabbitListener(id = "${rabbit.listener.id}", containerFactory = "${rabbit.listener.containerFactory}",
queues = {"${rabbit.listener.queue}", "queue2"}, exclusive = true,
priority = "${rabbit.listener.priority}", admin = "${rabbit.listener.admin}")
public void fullHandle(String msg) {
}
}
@Component
static class CustomBean {
@RabbitListener(id = "listenerId", containerFactory = "customFactory", queues = "myQueue")
public void customHandle(String msg) {
}
}
static class DefaultBean {
@RabbitListener(queues = "myQueue")
public void handleIt(String msg) {
}
}
@Component
static class ValidationBean {
@RabbitListener(containerFactory = "defaultFactory", queues = "myQueue")
public void defaultHandle(@Validated String msg) {
}
}
@Component
static class RabbitListenersBean {
@@ -318,14 +326,6 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
}
private void assertQueues(AbstractRabbitListenerEndpoint actual, String... expectedQueues) {
Collection<String> actualQueues = actual.getQueueNames();
for (String expectedQueue : expectedQueues) {
assertTrue("Queue '" + expectedQueue + "' not found", actualQueues.contains(expectedQueue));
}
assertEquals("Wrong number of queues", expectedQueues.length, actualQueues.size());
}
static class TestValidator implements Validator {
@Override

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitCglibProxyTests.java
View File

@@ -121,7 +121,7 @@ public class EnableRabbitCglibProxyTests {
@Override
@Transactional
@RabbitListener(bindings = @QueueBinding(
value = @Queue(),
value = @Queue,
exchange = @Exchange(value = "auto.exch.test", autoDelete = "true"),
key = "auto.rk.test")
)

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIdleContainerTests.java
View File

@@ -141,7 +141,7 @@ public class EnableRabbitIdleContainerTests {
private boolean barEventReceived;
@RabbitListener(id="foo", queues="#{queue.name}")
@RabbitListener(id = "foo", queues = "#{queue.name}")
public String listenFoo(String foo) {
logger.info("foo: " + foo);
return foo.toUpperCase();
@@ -157,7 +157,7 @@ public class EnableRabbitIdleContainerTests {
this.latch.countDown();
}
@RabbitListener(id="bar", queues="#{queue.name}")
@RabbitListener(id = "bar", queues = "#{queue.name}")
public String listenBar(String bar) {
logger.info("bar: " + bar);
return bar.toUpperCase();

98
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
View File

@@ -175,25 +175,6 @@ public class EnableRabbitIntegrationTests {
@Autowired
private RabbitListenerEndpointRegistry registry;
/**
* Defer queue deletion until after the context has been stopped by the
* {@link DirtiesContext}.
*
*/
public static class DeleteQueuesExecutionListener extends AbstractTestExecutionListener {
@Override
public void afterTestClass(TestContext testContext) throws Exception {
brokerRunning.removeTestQueues();
}
@Override
public int getOrder() {
return Ordered.HIGHEST_PRECEDENCE;
}
}
@Test
public void autoDeclare() {
assertEquals("FOO", rabbitTemplate.convertSendAndReceive("auto.exch", "auto.rk", "foo"));
@@ -335,27 +316,21 @@ public class EnableRabbitIntegrationTests {
}
@Test
@DirtiesContext
public void simpleEndpointWithSendTo() throws InterruptedException {
rabbitTemplate.convertAndSend("test.sendTo", "bar");
int n = 0;
Object result = null;
while ((result = rabbitTemplate.receiveAndConvert("test.sendTo.reply")) == null && n++ < 100) {
Thread.sleep(100);
}
assertTrue(n < 100);
rabbitTemplate.setReceiveTimeout(10000);
Object result = rabbitTemplate.receiveAndConvert("test.sendTo.reply");
assertNotNull(result);
assertEquals("BAR", result);
}
@Test
@DirtiesContext
public void simpleEndpointWithSendToSpel() throws InterruptedException {
rabbitTemplate.convertAndSend("test.sendTo.spel", "bar");
int n = 0;
Object result = null;
while ((result = rabbitTemplate.receiveAndConvert("test.sendTo.reply.spel")) == null && n++ < 100) {
Thread.sleep(100);
}
assertTrue(n < 100);
rabbitTemplate.setReceiveTimeout(10000);
Object result = rabbitTemplate.receiveAndConvert("test.sendTo.reply.spel");
assertNotNull(result);
assertEquals("BARbar", result);
}
@@ -513,7 +488,7 @@ public class EnableRabbitIntegrationTests {
@Override
@RabbitListener(bindings = @QueueBinding(
value = @Queue(),
value = @Queue,
exchange = @Exchange(value = "auto.exch.tx", autoDelete = "true"),
key = "auto.rk.tx")
)
@@ -569,7 +544,7 @@ public class EnableRabbitIntegrationTests {
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "auto.declare.fanout", autoDelete = "true"),
exchange = @Exchange(value = "auto.exch.fanout", autoDelete = "true", type="fanout"))
exchange = @Exchange(value = "auto.exch.fanout", autoDelete = "true", type = "fanout"))
)
public String handleWithFanout(String foo) {
return foo.toUpperCase() + foo.toUpperCase();
@@ -577,7 +552,7 @@ public class EnableRabbitIntegrationTests {
@RabbitListener(bindings = {
@QueueBinding(
value = @Queue(),
value = @Queue,
exchange = @Exchange(value = "auto.exch", autoDelete = "true"),
key = "auto.anon.rk")}
)
@@ -586,7 +561,7 @@ public class EnableRabbitIntegrationTests {
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(autoDelete = "true", exclusive="true", durable="true"),
value = @Queue(autoDelete = "true", exclusive = "true", durable = "true"),
exchange = @Exchange(value = "auto.exch", autoDelete = "true"),
key = "auto.anon.atts.rk")
)
@@ -657,15 +632,15 @@ public class EnableRabbitIntegrationTests {
private final CountDownLatch latch = new CountDownLatch(1);
@RabbitListener(queues = "differentTypes", containerFactory="jsonListenerContainerFactory")
@RabbitListener(queues = "differentTypes", containerFactory = "jsonListenerContainerFactory")
public void handleDifferent(Foo2 foo) {
foos.add(foo);
latch.countDown();
}
@RabbitListener(id="notStarted", containerFactory = "rabbitAutoStartFalseListenerContainerFactory",
@RabbitListener(id = "notStarted", containerFactory = "rabbitAutoStartFalseListenerContainerFactory",
bindings = @QueueBinding(
value = @Queue(autoDelete = "true", exclusive="true", durable="true"),
value = @Queue(autoDelete = "true", exclusive = "true", durable = "true"),
exchange = @Exchange(value = "auto.start", autoDelete = "true"),
key = "auto.start")
)
@@ -710,11 +685,11 @@ public class EnableRabbitIntegrationTests {
public static class ProxiedListener {
@RabbitListener(queues="test.intercepted")
@RabbitListener(queues = "test.intercepted")
public void listen(String foo) {
}
@RabbitListener(queues="test.intercepted.withReply")
@RabbitListener(queues = "test.intercepted.withReply")
public String listenAndReply(String foo) {
return foo.toUpperCase();
}
@@ -1181,68 +1156,68 @@ public class EnableRabbitIntegrationTests {
public static class Foo2Service {
@RabbitListener(queues="test.converted")
@RabbitListener(queues = "test.converted")
public Foo2 foo2(Foo2 foo2) {
return foo2;
}
@RabbitListener(queues="test.converted.list")
@RabbitListener(queues = "test.converted.list")
public Foo2 foo2(List<Foo2> foo2s) {
Foo2 foo2 = foo2s.get(0);
foo2.setBar("BAZZZZ");
return foo2;
}
@RabbitListener(queues="test.converted.array")
@RabbitListener(queues = "test.converted.array")
public Foo2 foo2(Foo2[] foo2s) {
Foo2 foo2 = foo2s[0];
foo2.setBar("BAZZxx");
return foo2;
}
@RabbitListener(queues="test.converted.args1")
@RabbitListener(queues = "test.converted.args1")
public String foo2(Foo2 foo2, @Header("amqp_consumerQueue") String queue) {
return foo2 + queue;
}
@RabbitListener(queues="test.converted.args2")
@RabbitListener(queues = "test.converted.args2")
public String foo2a(@Payload Foo2 foo2, @Header("amqp_consumerQueue") String queue) {
return foo2 + queue;
}
@RabbitListener(queues="test.converted.message")
@RabbitListener(queues = "test.converted.message")
public String foo2Message(@Payload Foo2 foo2, Message message) {
return foo2.toString() + message.getMessageProperties().getTargetMethod().getName()
+ message.getMessageProperties().getTargetBean().getClass().getSimpleName();
}
@RabbitListener(queues="test.notconverted.message")
@RabbitListener(queues = "test.notconverted.message")
public String justMessage(Message message) {
return "foo" + message.getClass().getSimpleName();
}
@RabbitListener(queues="test.notconverted.channel")
@RabbitListener(queues = "test.notconverted.channel")
public String justChannel(Channel channel) {
return "barAndChannel";
}
@RabbitListener(queues="test.notconverted.messagechannel")
@RabbitListener(queues = "test.notconverted.messagechannel")
public String messageChannel(Foo2 foo2, Message message, Channel channel) {
return foo2 + message.getClass().getSimpleName() + "AndChannel";
}
@RabbitListener(queues="test.notconverted.messagingmessage")
@RabbitListener(queues = "test.notconverted.messagingmessage")
public String messagingMessage(org.springframework.messaging.Message<?> message) {
return message.getClass().getSimpleName() + message.getPayload().getClass().getSimpleName();
}
@RabbitListener(queues="test.converted.foomessage")
@RabbitListener(queues = "test.converted.foomessage")
public String messagingMessage(org.springframework.messaging.Message<Foo2> message,
@Header(value = "", required = false) String h) {
return message.getClass().getSimpleName() + message.getPayload().getClass().getSimpleName();
}
@RabbitListener(queues="test.notconverted.messagingmessagenotgeneric")
@RabbitListener(queues = "test.notconverted.messagingmessagenotgeneric")
public String messagingMessage(@SuppressWarnings("rawtypes") org.springframework.messaging.Message message,
@Header(value = "", required = false) Integer h) {
return message.getClass().getSimpleName() + message.getPayload().getClass().getSimpleName();
@@ -1250,4 +1225,23 @@ public class EnableRabbitIntegrationTests {
}
/**
* Defer queue deletion until after the context has been stopped by the
* {@link DirtiesContext}.
*
*/
public static class DeleteQueuesExecutionListener extends AbstractTestExecutionListener {
@Override
public void afterTestClass(TestContext testContext) throws Exception {
brokerRunning.removeTestQueues();
}
@Override
public int getOrder() {
return Ordered.HIGHEST_PRECEDENCE;
}
}
}

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/AdminParserTests.java
View File

@@ -81,7 +81,8 @@ public final class AdminParserTests {
RabbitAdmin admin;
if (StringUtils.hasText(adminBeanName)) {
admin = beanFactory.getBean(adminBeanName, RabbitAdmin.class);
} else {
}
else {
admin = beanFactory.getBean(RabbitAdmin.class);
}
assertEquals(expectedAutoStartup, admin.isAutoStartup());
@@ -109,7 +110,8 @@ public final class AdminParserTests {
if (!validContext) {
fail("Context " + resource + " failed to load");
}
} catch (BeanDefinitionParsingException e) {
}
catch (BeanDefinitionParsingException e) {
if (validContext) {
// Context expected to be valid - throw an exception up
throw e;

8
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ListenerContainerParserTests.java
View File

@@ -77,7 +77,7 @@ public class ListenerContainerParserTests {
assertEquals(beanFactory.getBean(TestBean.class), listenerAccessor.getPropertyValue("delegate"));
assertEquals("handle", listenerAccessor.getPropertyValue("defaultListenerMethod"));
Queue queue = beanFactory.getBean("bar", Queue.class);
assertEquals("[foo, "+queue.getName()+"]", Arrays.asList(container.getQueueNames()).toString());
assertEquals("[foo, " + queue.getName() + "]", Arrays.asList(container.getQueueNames()).toString());
assertEquals(5, ReflectionTestUtils.getField(container, "concurrentConsumers"));
assertEquals(6, ReflectionTestUtils.getField(container, "maxConcurrentConsumers"));
assertEquals(1234L, ReflectionTestUtils.getField(container, "startConsumerMinInterval"));
@@ -111,7 +111,7 @@ public class ListenerContainerParserTests {
public void testParseWithQueues() throws Exception {
SimpleMessageListenerContainer container = beanFactory.getBean("container2", SimpleMessageListenerContainer.class);
Queue queue = beanFactory.getBean("bar", Queue.class);
assertEquals("[foo, "+queue.getName()+"]", Arrays.asList(container.getQueueNames()).toString());
assertEquals("[foo, " + queue.getName() + "]", Arrays.asList(container.getQueueNames()).toString());
assertTrue(TestUtils.getPropertyValue(container, "missingQueuesFatal", Boolean.class));
assertFalse(TestUtils.getPropertyValue(container, "autoDeclare", Boolean.class));
}
@@ -187,12 +187,12 @@ public class ListenerContainerParserTests {
@Test
public void testIncompatibleTxAtts() {
try {
new ClassPathXmlApplicationContext(getClass().getSimpleName() + "-fail-context.xml", getClass()).close();;
new ClassPathXmlApplicationContext(getClass().getSimpleName() + "-fail-context.xml", getClass()).close();
fail("Parse exception exptected");
}
catch (BeanDefinitionParsingException e) {
assertTrue(e.getMessage().startsWith(
"Configuration problem: Listener Container - cannot set channel-transacted with acknowledge='NONE'"));
"Configuration problem: Listener Container - cannot set channel-transacted with acknowledge='NONE'"));
}
}

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ListenerContainerPlaceholderParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2010-2015 the original author or authors.
* Copyright 2010-2016 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.
@@ -55,7 +55,7 @@ public final class ListenerContainerPlaceholderParserTests {
@After
public void closeBeanFactory() throws Exception {
if (this.context!=null) {
if (this.context != null) {
CachingConnectionFactory cf = this.context.getBean(CachingConnectionFactory.class);
this.context.close();
assertTrue(TestUtils.getPropertyValue(cf, "deferredCloseExecutor", ThreadPoolExecutor.class)
@@ -73,7 +73,7 @@ public final class ListenerContainerPlaceholderParserTests {
assertEquals(this.context.getBean(TestBean.class), listenerAccessor.getPropertyValue("delegate"));
assertEquals("handle", listenerAccessor.getPropertyValue("defaultListenerMethod"));
Queue queue = this.context.getBean("bar", Queue.class);
assertEquals("[foo, "+queue.getName()+"]", Arrays.asList(container.getQueueNames()).toString());
assertEquals("[foo, " + queue.getName() + "]", Arrays.asList(container.getQueueNames()).toString());
}
}

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MismatchedQueueDeclarationTests.java
View File

@@ -68,7 +68,8 @@ public class MismatchedQueueDeclarationTests {
((DisposableBean) connectionFactory).destroy();
}
@Test @Ignore
@Test
@Ignore
public void testAdminFailsWithMismatchedQueue() throws Exception {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext();
context.setConfigLocation("org/springframework/amqp/rabbit/config/MismatchedQueueDeclarationTests-context.xml");

1
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueArgumentsParserTests.java
View File

@@ -22,6 +22,7 @@ import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserPlaceholderTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.config;
import org.junit.After;
import org.junit.Before;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.GenericXmlApplicationContext;
import org.springframework.core.io.ClassPathResource;
@@ -36,8 +37,8 @@ public class QueueParserPlaceholderTests extends QueueParserTests {
@After
public void closeBeanFactory() throws Exception {
if (beanFactory!=null) {
((ConfigurableApplicationContext)beanFactory).close();
if (beanFactory != null) {
((ConfigurableApplicationContext) beanFactory).close();
}
}

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserTests.java
View File

@@ -182,7 +182,7 @@ public class QueueParserTests {
assertFalse(queue.shouldDeclare());
}
@Test(expected=BeanDefinitionStoreException.class)
@Test(expected = BeanDefinitionStoreException.class)
public void testIllegalAnonymousQueue() throws Exception {
DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory();
XmlBeanDefinitionReader reader = new XmlBeanDefinitionReader(beanFactory);

13
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RabbitListenerContainerFactoryIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,13 +16,16 @@
package org.springframework.amqp.rabbit.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import com.rabbitmq.client.Channel;
import org.junit.Before;
import org.junit.Test;
@@ -41,11 +44,11 @@ import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.util.ReflectionUtils;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
import com.rabbitmq.client.Channel;
/**
* @author Stephane Nicoll
* @author Gary Russell
*/
public class RabbitListenerContainerFactoryIntegrationTests {

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandlerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 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.
@@ -54,7 +54,7 @@ public final class RabbitNamespaceHandlerTests {
public void setUp() throws Exception {
beanFactory = new DefaultListableBeanFactory();
XmlBeanDefinitionReader reader = new XmlBeanDefinitionReader(beanFactory);
reader.loadBeanDefinitions(new ClassPathResource(getClass().getSimpleName()+"-context.xml", getClass()));
reader.loadBeanDefinitions(new ClassPathResource(getClass().getSimpleName() + "-context.xml", getClass()));
}
@Test

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RetryInterceptorBuilderSupportTests.java
View File

@@ -143,7 +143,7 @@ public class RetryInterceptorBuilderSupportTests {
public void testWitCustomRetryPolicyTraverseCause() {
StatefulRetryOperationsInterceptor interceptor = RetryInterceptorBuilder.stateful()
.retryPolicy(new SimpleRetryPolicy(15, Collections
.<Class<? extends Throwable>, Boolean> singletonMap(Exception.class, true), true))
.<Class<? extends Throwable>, Boolean>singletonMap(Exception.class, true), true))
.build();
assertEquals(15, TestUtils.getPropertyValue(interceptor, "retryOperations.retryPolicy.maxAttempts"));
}

9
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerEndpointTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,6 +17,9 @@
package org.springframework.amqp.rabbit.config;
import static org.junit.Assert.assertSame;
import static org.mockito.Mockito.mock;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
@@ -26,11 +29,9 @@ import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import static org.junit.Assert.*;
import static org.mockito.Mockito.mock;
/**
* @author Stephane Nicoll
* @author Gary Russell
*/
public class SimpleRabbitListenerEndpointTests {

11
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java
View File

@@ -151,13 +151,13 @@ public class CachingConnectionFactoryIntegrationTests {
channels.add(connections.get(0).createChannel(false));
fail("Exception expected");
}
catch (AmqpTimeoutException e) {}
catch (AmqpTimeoutException e) { }
channels.add(connections.get(1).createChannel(false));
try {
channels.add(connections.get(1).createChannel(false));
fail("Exception expected");
}
catch (AmqpTimeoutException e) {}
catch (AmqpTimeoutException e) { }
channels.get(0).close();
channels.get(1).close();
channels.add(connections.get(0).createChannel(false));
@@ -346,7 +346,8 @@ public class CachingConnectionFactoryIntegrationTests {
}
});
fail("Expected AmqpIOException");
} catch (AmqpIOException e) {
}
catch (AmqpIOException e) {
// expected
}
template.convertAndSend(route, "message");
@@ -396,7 +397,7 @@ public class CachingConnectionFactoryIntegrationTests {
socket.close();
proxy.close();
}
catch (Exception ee) {}
catch (Exception ee) { }
}
}
}
@@ -413,7 +414,7 @@ public class CachingConnectionFactoryIntegrationTests {
socket.close();
proxy.close();
}
catch (Exception ee) {}
catch (Exception ee) { }
}
}
socket.close();

14
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
View File

@@ -255,7 +255,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
con.createChannel(false);
fail("Exception expected");
}
catch (AmqpTimeoutException e) {}
catch (AmqpTimeoutException e) { }
// should be ignored, and added last into channel cache.
channel1.close();
@@ -297,7 +297,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
ccf.createConnection();
fail("Exception expected");
}
catch (AmqpTimeoutException e) {}
catch (AmqpTimeoutException e) { }
// should be ignored, and added to cache
con1.close();
@@ -778,7 +778,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
called.incrementAndGet();
}
}));
((CachingConnectionFactory)connectionFactory).setChannelCacheSize(1);
((CachingConnectionFactory) connectionFactory).setChannelCacheSize(1);
Connection con = connectionFactory.createConnection();
Channel channel = con.createChannel(false);
@@ -830,7 +830,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
closed.set(connection);
}
});
((CachingConnectionFactory)connectionFactory).setChannelCacheSize(1);
((CachingConnectionFactory) connectionFactory).setChannelCacheSize(1);
Connection con = connectionFactory.createConnection();
Channel channel = con.createChannel(false);
@@ -908,7 +908,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
new AtomicReference<com.rabbitmq.client.Connection>();
final AtomicReference<com.rabbitmq.client.Connection> closedNotification =
new AtomicReference<com.rabbitmq.client.Connection>();
ccf.setConnectionListeners(Collections.singletonList(new ConnectionListener(){
ccf.setConnectionListeners(Collections.singletonList(new ConnectionListener() {
@Override
public void onCreate(Connection connection) {
@@ -1112,7 +1112,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
new AtomicReference<com.rabbitmq.client.Connection>();
final AtomicReference<com.rabbitmq.client.Connection> closedNotification =
new AtomicReference<com.rabbitmq.client.Connection>();
ccf.setConnectionListeners(Collections.singletonList(new ConnectionListener(){
ccf.setConnectionListeners(Collections.singletonList(new ConnectionListener() {
@Override
public void onCreate(Connection connection) {
@@ -1409,7 +1409,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
InOrder order = inOrder(mock);
order.verify(mock).setUri(uri);
order.verify(mock).newConnection((ExecutorService)null);
order.verify(mock).newConnection((ExecutorService) null);
verifyNoMoreInteractions(mock);
}

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryLifecycleTests.java
View File

@@ -86,7 +86,7 @@ public class ConnectionFactoryLifecycleTests {
private volatile boolean running;
public MyLifecycle (ConnectionFactory cf) {
public MyLifecycle(ConnectionFactory cf) {
this.admin = new RabbitAdmin(cf);
}

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SSLConnectionTests.java
View File

@@ -37,7 +37,8 @@ import com.rabbitmq.client.ConnectionFactory;
*/
public class SSLConnectionTests {
@Test @Ignore
@Test
@Ignore
public void test() throws Exception {
RabbitConnectionFactoryBean fb = new RabbitConnectionFactoryBean();
fb.setUseSSL(true);

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java
View File

@@ -236,7 +236,8 @@ public class SingleConnectionFactory extends AbstractConnectionFactory {
if (other.target != null) {
return false;
}
} else if (!target.equals(other.target)) {
}
else if (!target.equals(other.target)) {
return false;
}
return true;

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java
View File

@@ -463,10 +463,11 @@ public class BatchingRabbitTemplateTests {
}
private Message receive(BatchingRabbitTemplate template) throws InterruptedException {
Message message = null;
Message message = template.receive(ROUTE);
int n = 0;
while (n++ < 200 && (message = template.receive(ROUTE)) == null) {
while (n++ < 200 && message == null) {
Thread.sleep(50);
message = template.receive(ROUTE);
}
assertNotNull(message);
return message;

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java
View File

@@ -51,7 +51,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @since 1.3.6
*/
@ContextConfiguration(classes=FixedReplyQueueDeadLetterConfig.class)
@ContextConfiguration(classes = FixedReplyQueueDeadLetterConfig.class)
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class FixedReplyQueueDeadLetterTests {

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/Producer.java
View File

@@ -20,7 +20,11 @@ import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
public class Producer {
public final class Producer {
private Producer() {
super();
}
/**
* @param args

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/QueueUtils.java
View File

@@ -19,7 +19,11 @@ package org.springframework.amqp.rabbit.core;
import com.rabbitmq.client.AMQP.Queue;
import com.rabbitmq.client.Channel;
public class QueueUtils {
public final class QueueUtils {
private QueueUtils() {
super();
}
static void declareTestQueue(RabbitTemplate template, final String routingKey) {
// declare and bind queue

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminDeclarationTests.java
View File

@@ -310,7 +310,7 @@ public class RabbitAdminDeclarationTests {
queue.setAdminsThatShouldDeclare(null, admin1);
fail("Expected Exception");
}
catch (IllegalArgumentException e) {}
catch (IllegalArgumentException e) { }
}
@Configuration

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

@@ -96,7 +96,7 @@ public class RabbitAdminIntegrationTests {
if (context != null) {
context.close();
}
if (connectionFactory!=null) {
if (connectionFactory != null) {
connectionFactory.destroy();
}
}
@@ -124,7 +124,8 @@ public class RabbitAdminIntegrationTests {
new RabbitAdmin(connectionFactory1).declareQueue(queue);
try {
new RabbitAdmin(connectionFactory2).declareQueue(queue);
} finally {
}
finally {
// Need to release the connection so the exclusive queue is deleted
connectionFactory1.destroy();
connectionFactory2.destroy();
@@ -304,7 +305,8 @@ public class RabbitAdminIntegrationTests {
try {
rabbitAdmin.declareBinding(binding);
} catch (AmqpIOException ex) {
}
catch (AmqpIOException ex) {
Throwable cause = ex;
Throwable rootCause = null;
while (cause != null) {

16
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java
View File

@@ -101,7 +101,8 @@ public class RabbitBindingIntegrationTests {
result = getResult(consumer);
assertEquals("message", result);
} finally {
}
finally {
consumer.getChannel().basicCancel(tag);
}
@@ -141,7 +142,8 @@ public class RabbitBindingIntegrationTests {
result = getResult(consumer);
assertEquals("message", result);
} finally {
}
finally {
consumer.getChannel().basicCancel(tag);
}
@@ -216,7 +218,8 @@ public class RabbitBindingIntegrationTests {
template.convertAndSend("foo", "message");
String result = getResult(consumer);
assertEquals(null, result);
} finally {
}
finally {
consumer.stop();
}
@@ -237,7 +240,8 @@ public class RabbitBindingIntegrationTests {
template.convertAndSend("foo.end", "message");
String result = getResult(consumer);
assertEquals("message", result);
} finally {
}
finally {
consumer.stop();
}
@@ -270,7 +274,8 @@ public class RabbitBindingIntegrationTests {
template.convertAndSend("message");
String result = getResult(consumer);
assertEquals("message", result);
} finally {
}
finally {
consumer.stop();
}
@@ -309,4 +314,5 @@ public class RabbitBindingIntegrationTests {
}
return (String) new SimpleMessageConverter().fromMessage(response);
}
}

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitGatewaySupportTests.java
View File

@@ -65,5 +65,5 @@ public class RabbitGatewaySupportTests {
assertEquals("Correct RabbitTemplate", template, gateway.getRabbitTemplate());
assertEquals("initGateway called", test.size(), 1);
}
}

26
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitMessagingTemplateTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,6 +16,18 @@
package org.springframework.amqp.rabbit.core;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willThrow;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.io.Writer;
import java.util.HashMap;
import java.util.Map;
@@ -39,11 +51,9 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.converter.GenericMessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import static org.junit.Assert.*;
import static org.mockito.BDDMockito.*;
/**
* @author Stephane Nicoll
* @author Gary Russell
*/
public class RabbitMessagingTemplateTests {
@@ -406,8 +416,12 @@ public class RabbitMessagingTemplateTests {
}
private static org.springframework.amqp.core.Message anyAmqpMessage() {return any(org.springframework.amqp.core.Message.class);}
private static org.springframework.amqp.core.Message anyAmqpMessage() {
return any(org.springframework.amqp.core.Message.class);
}
private static MessageProperties anyMessageProperties() {return any(MessageProperties.class);}
private static MessageProperties anyMessageProperties() {
return any(MessageProperties.class);
}
}

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java
View File

@@ -102,7 +102,7 @@ public class RabbitTemplateHeaderTests {
Message replyMessage = new Message("!dlrow olleH".getBytes(), springProps);
template.onMessage(replyMessage);
return null;
}}
} }
).when(mockChannel).basicPublish(Mockito.any(String.class), Mockito.any(String.class), Mockito.anyBoolean(),
Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
Message reply = template.sendAndReceive(message);
@@ -153,7 +153,7 @@ public class RabbitTemplateHeaderTests {
Message replyMessage = new Message("!dlrow olleH".getBytes(), springProps);
template.onMessage(replyMessage);
return null;
}}
} }
).when(mockChannel).basicPublish(Mockito.any(String.class), Mockito.any(String.class), Mockito.anyBoolean(),
Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
Message reply = template.sendAndReceive(message);
@@ -209,7 +209,7 @@ public class RabbitTemplateHeaderTests {
}
template.onMessage(replyMessage);
return null;
}}
} }
).when(mockChannel).basicPublish(Mockito.any(String.class), Mockito.any(String.class), Mockito.anyBoolean(),
Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
Message reply = template.sendAndReceive(message);
@@ -260,7 +260,7 @@ public class RabbitTemplateHeaderTests {
Message replyMessage = new Message("!dlrow olleH".getBytes(), springProps);
template.onMessage(replyMessage);
return null;
}}
} }
).when(mockChannel).basicPublish(Mockito.any(String.class), Mockito.any(String.class), Mockito.anyBoolean(),
Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
Message reply = template.sendAndReceive(message);
@@ -318,7 +318,7 @@ public class RabbitTemplateHeaderTests {
}
template.onMessage(replyMessage);
return null;
}}
} }
).when(mockChannel).basicPublish(Mockito.any(String.class), Mockito.any(String.class), Mockito.anyBoolean(),
Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
Message reply = template.sendAndReceive(message);

12
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -469,11 +469,11 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
doReturn(new PublisherCallbackChannelImpl(mockChannel)).when(mockConnection).createChannel();
final AtomicInteger count = new AtomicInteger();
doAnswer(new Answer<Object>(){
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return count.incrementAndGet();
}}).when(mockChannel).getNextPublishSeqNo();
} }).when(mockChannel).getNextPublishSeqNo();
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
ccf.setPublisherConfirms(true);
@@ -514,7 +514,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
when(mockConnection.createChannel()).thenReturn(callbackChannel);
final AtomicInteger count = new AtomicInteger();
doAnswer(new Answer<Object>(){
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return count.incrementAndGet();
@@ -561,7 +561,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
when(mockConnection.createChannel()).thenReturn(callbackChannel);
final AtomicInteger count = new AtomicInteger();
doAnswer(new Answer<Object>(){
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return count.incrementAndGet();
@@ -887,7 +887,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
});
ExecutorService exec = Executors.newSingleThreadExecutor();
final AtomicInteger sent = new AtomicInteger();
doAnswer(new Answer<Boolean>(){
doAnswer(new Answer<Boolean>() {
@Override
public Boolean answer(InvocationOnMock invocation) throws Throwable {
@@ -903,7 +903,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
try {
template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc"));
}
catch (AmqpException e) {}
catch (AmqpException e) { }
}
sentAll.countDown();
}

17
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java
View File

@@ -101,19 +101,24 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests2 {
}
private void assertMessageCountEquals(long wanted) throws InterruptedException {
long messageCount;
long messageCount = determineMessageCount();
int n = 0;
while ((messageCount = this.templateWithConfirmsEnabled.execute(new ChannelCallback<Long>() {
while (messageCount < wanted && n++ < 100) {
Thread.sleep(100);
messageCount = determineMessageCount();
}
assertEquals(wanted, messageCount);
}
private Long determineMessageCount() {
return this.templateWithConfirmsEnabled.execute(new ChannelCallback<Long>() {
@Override
public Long doInRabbit(Channel channel) throws Exception {
return channel.messageCount(ROUTE);
}
})) < wanted && n++ < 100) {
Thread.sleep(100);
};
assertEquals(wanted, messageCount);
});
}
}

Some files were not shown because too many files have changed in this diff Show More