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

AMQP-798: Master to 2.1.x; Fix Tangles

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-798

Also update amqp-client to 5.2.0

Don't use `MessageConverter` in `Message.toString()` (cycle)

Fix cycle between template and admin (via containers)

Fix outbound reference from support to connection

Fix cycle between connection and listener via RabbitUtils

Fix cycle between listener and core via ChannelAwareMessageListener and admin

Fix cycle between listener.adapter and listener via RabbitListenerErrorHandler
Fix cycle between amqp.support and amqp.core via Correlation

Fix cycle between listener and transaction via ListenerFailedRuleBasedTransactionAttribute

Fix Tests

Polishing - PR Comments; rename schema

Test polishing - travis failures

Polishing according PR comments

* Update Copyright for all affected classes
* Optimize `DirectMessageListenerContainer.checkMissingQueues()` to
cache `amqpAdmin` from the locally created
* `RabbitUtils.DEFAULT_PORT` as not-used and deprecated in `2.0.x`
* Remove commented test in the `LogAppenderUtils`
* Deprecate `AbstractAdaptableMessageListener#setRabbitAdmin()` in favor
of newly introduced `setAmqpAdmin()`
This commit is contained in:
Gary Russell
2018-03-13 15:08:19 -04:00
committed by Artem Bilan
parent 1d38c14935
commit 4464d4d8a5
56 changed files with 714 additions and 370 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -94,7 +94,7 @@ subprojects { subproject ->
log4jVersion = '2.8.2'
logbackVersion = '1.2.3'
mockitoVersion = '2.11.0'
rabbitmqVersion = project.hasProperty('rabbitmqVersion') ? project.rabbitmqVersion : '5.1.2'
rabbitmqVersion = project.hasProperty('rabbitmqVersion') ? project.rabbitmqVersion : '5.2.0'
rabbitmqHttpClientVersion = '2.0.1.RELEASE'
springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.0.4.RELEASE'

2
gradle.properties
View File

@@ -1,2 +1,2 @@
version=2.0.3.BUILD-SNAPSHOT
version=2.1.0.BUILD-SNAPSHOT
org.gradle.daemon=true

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -105,4 +105,12 @@ public interface AmqpAdmin {
*/
Properties getQueueProperties(String queueName);
/**
* Initialize the admin.
* @since 2.1
*/
default void initialize() {
// no op
}
}

4
spring-amqp/src/main/java/org/springframework/amqp/support/Correlation.java → spring-amqp/src/main/java/org/springframework/amqp/core/Correlation.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.amqp.support;
package org.springframework.amqp.core;
/**
* A marker interface for data used to correlate information about sent messages.

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,11 +16,15 @@
package org.springframework.amqp.core;
import java.io.ByteArrayInputStream;
import java.io.Serializable;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.LinkedHashSet;
import java.util.Set;
import org.springframework.amqp.support.converter.SerializerMessageConverter;
import org.springframework.amqp.utils.SerializationUtils;
import org.springframework.util.Assert;
/**
* The 0-8 and 0-9-1 AMQP specifications do not define an Message class or interface. Instead, when performing an
@@ -42,11 +46,8 @@ public class Message implements Serializable {
private static final String ENCODING = Charset.defaultCharset().name();
private static final SerializerMessageConverter SERIALIZER_MESSAGE_CONVERTER = new SerializerMessageConverter();
static {
SERIALIZER_MESSAGE_CONVERTER.setWhiteListPatterns(Arrays.asList("java.util.*", "java.lang.*"));
}
private static final Set<String> whiteListPatterns = new LinkedHashSet<String>(
Arrays.asList("java.util.*", "java.lang.*"));
private final MessageProperties messageProperties;
@@ -70,7 +71,8 @@ public class Message implements Serializable {
* @since 1.5.7
*/
public static void addWhiteListPatterns(String... patterns) {
SERIALIZER_MESSAGE_CONVERTER.addWhiteListPatterns(patterns);
Assert.notNull(patterns, "'patterns' cannot be null");
whiteListPatterns.addAll(Arrays.asList(patterns));
}
public byte[] getBody() {
@@ -100,7 +102,8 @@ public class Message implements Serializable {
try {
String contentType = (this.messageProperties != null) ? this.messageProperties.getContentType() : null;
if (MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT.equals(contentType)) {
return SERIALIZER_MESSAGE_CONVERTER.fromMessage(this).toString();
return SerializationUtils.deserialize(new ByteArrayInputStream(this.body), whiteListPatterns,
getClass().getClassLoader()).toString();
}
if (MessageProperties.CONTENT_TYPE_TEXT_PLAIN.equals(contentType)
|| MessageProperties.CONTENT_TYPE_JSON.equals(contentType)

1
spring-amqp/src/main/java/org/springframework/amqp/core/MessagePostProcessor.java
View File

@@ -17,7 +17,6 @@
package org.springframework.amqp.core;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.support.Correlation;
/**
* Used in several places in the framework, such as

19
spring-amqp/src/main/java/org/springframework/amqp/support/converter/WhiteListDeserializingMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,7 +22,7 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import org.springframework.util.PatternMatchUtils;
import org.springframework.amqp.utils.SerializationUtils;
/**
* MessageConverters that potentially use Java deserialization.
@@ -59,20 +59,7 @@ public abstract class WhiteListDeserializingMessageConverter extends AbstractMes
}
protected void checkWhiteList(Class<?> clazz) throws IOException {
if (this.whiteListPatterns.isEmpty()) {
return;
}
if (clazz.isArray() || clazz.isPrimitive() || clazz.equals(String.class)
|| Number.class.isAssignableFrom(clazz)) {
return;
}
String className = clazz.getName();
for (String pattern : this.whiteListPatterns) {
if (PatternMatchUtils.simpleMatch(pattern, className)) {
return;
}
}
throw new SecurityException("Attempt to deserialize unauthorized " + clazz);
SerializationUtils.checkWhiteList(clazz, this.whiteListPatterns);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2017 the original author or authors.
* Copyright 2006-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -19,8 +19,16 @@ package org.springframework.amqp.utils;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.ObjectStreamClass;
import java.util.Set;
import org.springframework.core.ConfigurableObjectInputStream;
import org.springframework.core.NestedIOException;
import org.springframework.util.ObjectUtils;
import org.springframework.util.PatternMatchUtils;
/**
* Static utility to help with serialization.
@@ -91,4 +99,59 @@ public final class SerializationUtils {
}
}
/**
* Deserialize the stream.
* @param inputStream the stream.
* @param whiteListPatterns allowed classes.
* @param classLoader the class loader.
* @return the result.
* @throws IOException IO Exception.
* @since 2.1
*/
public static Object deserialize(InputStream inputStream, Set<String> whiteListPatterns, ClassLoader classLoader)
throws IOException {
try (
ObjectInputStream objectInputStream = new ConfigurableObjectInputStream(inputStream, classLoader) {
@Override
protected Class<?> resolveClass(ObjectStreamClass classDesc)
throws IOException, ClassNotFoundException {
Class<?> clazz = super.resolveClass(classDesc);
checkWhiteList(clazz, whiteListPatterns);
return clazz;
}
}) {
return objectInputStream.readObject();
}
catch (ClassNotFoundException ex) {
throw new NestedIOException("Failed to deserialize object type", ex);
}
}
/**
* Verify that the class is in the white list.
* @param clazz the class.
* @param whiteListPatterns the patterns.
* @since 2.1
*/
public static void checkWhiteList(Class<?> clazz, Set<String> whiteListPatterns) {
if (ObjectUtils.isEmpty(whiteListPatterns)) {
return;
}
if (clazz.isArray() || clazz.isPrimitive() || clazz.equals(String.class)
|| Number.class.isAssignableFrom(clazz)) {
return;
}
String className = clazz.getName();
for (String pattern : whiteListPatterns) {
if (PatternMatchUtils.simpleMatch(pattern, className)) {
return;
}
}
throw new SecurityException("Attempt to deserialize unauthorized " + clazz);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,7 +22,6 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.mock;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
@@ -34,11 +33,11 @@ import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
import org.springframework.beans.BeansException;
@@ -57,6 +56,7 @@ import com.rabbitmq.client.Envelope;
* It does not currently support publisher confirms/returns.
*
* @author Gary Russell
*
* @since 2.0
*
*/
@@ -111,7 +111,8 @@ public class TestRabbitTemplate extends RabbitTemplate implements ApplicationCon
@Override
protected void sendToRabbit(Channel channel, String exchange, String routingKey, boolean mandatory,
Message message) throws IOException {
Message message) {
Listeners listeners = this.listeners.get(routingKey);
if (listeners == null) {
throw new IllegalArgumentException("No listener for " + routingKey);
@@ -193,6 +194,7 @@ public class TestRabbitTemplate extends RabbitTemplate implements ApplicationCon
}
return this.iterator.next();
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -31,11 +31,11 @@ import org.springframework.amqp.core.Address;
import org.springframework.amqp.core.AmqpMessageReturnedException;
import org.springframework.amqp.core.AmqpReplyTimeoutException;
import org.springframework.amqp.core.AsyncAmqpTemplate;
import org.springframework.amqp.core.Correlation;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
@@ -43,9 +43,9 @@ import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannel;
import org.springframework.amqp.support.Correlation;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SmartMessageConverter;
import org.springframework.beans.factory.BeanNameAware;
@@ -833,6 +833,7 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa
AsyncCorrelationData(MessagePostProcessor userPostProcessor, ParameterizedTypeReference<C> returnType,
boolean enableConfirms) {
this.userPostProcessor = userPostProcessor;
this.returnType = returnType;
this.enableConfirms = enableConfirms;

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2017 the original author or authors.
* Copyright 2014-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -208,7 +208,7 @@ public @interface RabbitListener {
String returnExceptions() default "";
/**
* Set an {@link org.springframework.amqp.rabbit.listener.RabbitListenerErrorHandler}
* Set an {@link org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler}
* to invoke if the listener method throws an exception.
* @return the error handler.
* @since 2.0

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
View File

@@ -46,7 +46,7 @@ import org.springframework.amqp.rabbit.listener.MultiMethodRabbitListenerEndpoin
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.rabbit.listener.RabbitListenerErrorHandler;
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;

38
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -486,7 +486,7 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
container.setMessagePropertiesConverter(this.messagePropertiesConverter);
}
if (this.rabbitAdmin != null) {
container.setRabbitAdmin(this.rabbitAdmin);
container.setAmqpAdmin(this.rabbitAdmin);
}
if (this.missingQueuesFatal != null) {
container.setMissingQueuesFatal(this.missingQueuesFatal);
@@ -562,23 +562,6 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
}
}
/**
* The container type.
*/
public enum Type {
/**
* {@link SimpleMessageListenerContainer}.
*/
simple,
/**
* {@link DirectMessageListenerContainer}.
*/
direct
}
@Override
public void start() {
if (this.container != null) {
@@ -615,4 +598,21 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
}
}
/**
* The container type.
*/
public enum Type {
/**
* {@link SimpleMessageListenerContainer}.
*/
simple,
/**
* {@link DirectMessageListenerContainer}.
*/
direct
}
}

61
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryConfigurationUtils.java Normal file
View File

@@ -0,0 +1,61 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit.connection;
import java.util.Map;
/**
* Utility methods for configuring connection factories.
*
* @author Gary Russell
*
* @since 2.1
*
*/
public final class ConnectionFactoryConfigurationUtils {
private ConnectionFactoryConfigurationUtils() {
super();
}
/**
* Parse the properties {@code key:value[,key:value]...} and add them to the
* underlying connection factory client properties.
* @param connectionFactory the connection factory.
* @param clientConnectionProperties the properties.
*/
public static void updateClientConnectionProperties(AbstractConnectionFactory connectionFactory,
String clientConnectionProperties) {
if (clientConnectionProperties != null) {
String[] props = clientConnectionProperties.split(",");
if (props.length > 0) {
Map<String, Object> clientProps =
connectionFactory.getRabbitConnectionFactory()
.getClientProperties();
for (String prop : props) {
String[] aProp = prop.split(":");
if (aProp.length == 2) {
clientProps.put(aProp[0].trim(), aProp[1].trim());
}
}
}
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -39,14 +39,13 @@ import com.rabbitmq.client.impl.recovery.AutorecoveringChannel;
* @author Mark Fisher
* @author Mark Pollack
* @author Gary Russell
* @author Artem Bilan
*/
public abstract class RabbitUtils {
public static final int DEFAULT_PORT = AMQP.PROTOCOL.PORT;
private static final Log logger = LogFactory.getLog(RabbitUtils.class);
private static final ThreadLocal<Boolean> physicalCloseRequired = new ThreadLocal<Boolean>();
private static final ThreadLocal<Boolean> physicalCloseRequired = new ThreadLocal<>();
/**
* Close the given RabbitMQ Connection and ignore any thrown exception. This is useful for typical
@@ -159,7 +158,6 @@ public abstract class RabbitUtils {
/**
* Declare to that broker that a channel is going to be used transactionally, and convert exceptions that arise.
*
* @param channel the channel to use
*/
public static void declareTransactional(Channel channel) {
@@ -324,10 +322,15 @@ public abstract class RabbitUtils {
* @param logger the logger to use for debug.
* @return true to requeue.
* @since 2.0
* @deprecated in favor of {@code ContainerUtils#shouldRequeue(boolean, Throwable, Log)}.
*/
@Deprecated
public static boolean shouldRequeue(boolean defaultRequeueRejected, Throwable throwable, Log logger) {
logger.warn("Use ContainerUtils.shouldRequeue()");
// compare class by name to avoid tangle
boolean shouldRequeue = defaultRequeueRejected ||
throwable instanceof MessageRejectedWhileStoppingException;
throwable.getClass().getName().equals(
"org.springframework.amqp.rabbit.listener.MessageRejectedWhileStoppingException");
Throwable t = throwable;
while (shouldRequeue && t != null) {
if (t instanceof AmqpRejectAndDontRequeueException) {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -37,6 +37,7 @@ import org.springframework.amqp.AmqpIOException;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessagePostProcessor;
@@ -49,8 +50,7 @@ import org.springframework.amqp.rabbit.connection.RabbitAccessor;
import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.amqp.rabbit.connection.RoutingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerExecutionException;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
@@ -93,6 +93,7 @@ import com.rabbitmq.client.ShutdownSignalException;
* @author Alex Panchenko
* @author Johno Crawford
* @author Arnaud Cogoluègnes
* @author Artem Bilan
*/
public abstract class AbstractMessageListenerContainer extends RabbitAccessor
implements MessageListenerContainer, ApplicationContextAware, BeanNameAware, DisposableBean,
@@ -137,7 +138,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
private MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
private RabbitAdmin rabbitAdmin;
private AmqpAdmin amqpAdmin;
private boolean missingQueuesFatal = true;
@@ -841,20 +842,45 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
return this.messagePropertiesConverter;
}
protected RabbitAdmin getRabbitAdmin() {
return this.rabbitAdmin;
/**
* Return the admin.
* @return the admin.
* @deprecated in favor of {@link #getAmqpAdmin()}
*/
@Deprecated
protected AmqpAdmin getRabbitAdmin() {
return getAmqpAdmin();
}
protected AmqpAdmin getAmqpAdmin() {
return this.amqpAdmin;
}
/**
* Set the {@link RabbitAdmin}, used to declare any auto-delete queues, bindings
* Set the {@link AmqpAdmin}, used to declare any auto-delete queues, bindings
* etc when the container is started. Only needed if those queues use conditional
* declaration (have a 'declared-by' attribute). If not specified, an internal
* admin will be used which will attempt to declare all elements not having a
* 'declared-by' attribute.
* @param rabbitAdmin The admin.
* @param amqpAdmin the AmqpAdmin to use
* @since 2.1
*/
public final void setRabbitAdmin(RabbitAdmin rabbitAdmin) {
this.rabbitAdmin = rabbitAdmin;
public void setAmqpAdmin(AmqpAdmin amqpAdmin) {
this.amqpAdmin = amqpAdmin;
}
/**
* Set the {@link AmqpAdmin}, used to declare any auto-delete queues, bindings
* etc when the container is started. Only needed if those queues use conditional
* declaration (have a 'declared-by' attribute). If not specified, an internal
* admin will be used which will attempt to declare all elements not having a
* 'declared-by' attribute.
* @param amqpAdmin The admin.
* @deprecated in favor of {@link #setAmqpAdmin(AmqpAdmin)}
*/
@Deprecated
public final void setRabbitAdmin(AmqpAdmin amqpAdmin) {
setAmqpAdmin(amqpAdmin);
}
/**
@@ -1530,22 +1556,22 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
}
protected void configureAdminIfNeeded() {
if (this.rabbitAdmin == null && this.getApplicationContext() != null) {
Map<String, RabbitAdmin> admins = this.getApplicationContext().getBeansOfType(RabbitAdmin.class);
if (this.amqpAdmin == null && this.getApplicationContext() != null) {
Map<String, AmqpAdmin> admins = this.getApplicationContext().getBeansOfType(AmqpAdmin.class);
if (admins.size() == 1) {
this.rabbitAdmin = admins.values().iterator().next();
this.amqpAdmin = admins.values().iterator().next();
}
else {
if (isAutoDeclare() || isMismatchedQueuesFatal()) {
if (logger.isDebugEnabled()) {
logger.debug("For 'autoDeclare' and 'mismatchedQueuesFatal' to work, there must be exactly one "
+ "RabbitAdmin in the context or you must inject one into this container; found: "
+ "AmqpAdmin in the context or you must inject one into this container; found: "
+ admins.size() + " for container " + this.toString());
}
}
if (isMismatchedQueuesFatal()) {
throw new IllegalStateException("When 'mismatchedQueuesFatal' is 'true', there must be exactly "
+ "one RabbitAdmin in the context or you must inject one into this container; found: "
+ "one AmqpAdmin in the context or you must inject one into this container; found: "
+ admins.size() + " for container " + this.toString());
}
}
@@ -1553,9 +1579,9 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
}
protected void checkMismatchedQueues() {
if (this.mismatchedQueuesFatal && this.rabbitAdmin != null) {
if (this.mismatchedQueuesFatal && this.amqpAdmin != null) {
try {
this.rabbitAdmin.initialize();
this.amqpAdmin.initialize();
}
catch (AmqpConnectException e) {
logger.info("Broker not available; cannot check queue declarations");
@@ -1572,7 +1598,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
}
/**
* Use {@link RabbitAdmin#initialize()} to redeclare everything if necessary.
* Use {@link AmqpAdmin#initialize()} to redeclare everything if necessary.
* Since auto deletion of a queue can cause upstream elements
* (bindings, exchanges) to be deleted too, everything needs to be redeclared if
* a queue is missing.
@@ -1589,8 +1615,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
* fail with a fatal error if mismatches occur.
*/
protected synchronized void redeclareElementsIfNecessary() {
RabbitAdmin rabbitAdmin = getRabbitAdmin();
if (rabbitAdmin == null || !isAutoDeclare()) {
AmqpAdmin amqpAdmin = getAmqpAdmin();
if (amqpAdmin == null || !isAutoDeclare()) {
return;
}
try {
@@ -1601,11 +1627,11 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
for (Entry<String, Queue> entry : queueBeans.entrySet()) {
Queue queue = entry.getValue();
if (isMismatchedQueuesFatal() || (queueNames.contains(queue.getName()) &&
rabbitAdmin.getQueueProperties(queue.getName()) == null)) {
amqpAdmin.getQueueProperties(queue.getName()) == null)) {
if (logger.isDebugEnabled()) {
logger.debug("Redeclaring context exchanges, queues, bindings.");
}
rabbitAdmin.initialize();
amqpAdmin.initialize();
return;
}
}
@@ -1654,7 +1680,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
protected void prepareHolderForRollback(RabbitResourceHolder resourceHolder, RuntimeException exception) {
if (resourceHolder != null) {
resourceHolder.setRequeueOnRollback(isAlwaysRequeueWithTxManagerRollback() ||
RabbitUtils.shouldRequeue(isDefaultRequeueRejected(), exception, logger));
ContainerUtils.shouldRequeue(isDefaultRequeueRejected(), exception, logger));
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2017 the original author or authors.
* Copyright 2014-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,6 +22,7 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerEndpoint;
@@ -41,7 +42,10 @@ import org.springframework.util.Assert;
*
* @author Stephane Nicoll
* @author Gary Russell
* @author Artem Bilan
*
* @since 1.4
*
* @see MethodRabbitListenerEndpoint
* @see SimpleRabbitListenerEndpoint
*/
@@ -49,9 +53,9 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
private String id;
private final Collection<Queue> queues = new ArrayList<Queue>();
private final Collection<Queue> queues = new ArrayList<>();
private final Collection<String> queueNames = new ArrayList<String>();
private final Collection<String> queueNames = new ArrayList<>();
private boolean exclusive;
@@ -59,7 +63,7 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
private String concurrency;
private RabbitAdmin admin;
private AmqpAdmin admin;
private BeanFactory beanFactory;
@@ -204,15 +208,15 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
* Set the {@link RabbitAdmin} instance to use.
* @param admin the {@link RabbitAdmin} instance.
*/
public void setAdmin(RabbitAdmin admin) {
public void setAdmin(AmqpAdmin admin) {
this.admin = admin;
}
/**
* @return the {@link RabbitAdmin} instance to use or {@code null} if
* @return the {@link AmqpAdmin} instance to use or {@code null} if
* none is configured.
*/
public RabbitAdmin getAdmin() {
public AmqpAdmin getAdmin() {
return this.admin;
}
@@ -271,7 +275,7 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
}
if (getAdmin() != null) {
container.setRabbitAdmin(getAdmin());
container.setAmqpAdmin(getAdmin());
}
setupMessageListener(listenerContainer);
}

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

@@ -763,7 +763,7 @@ public class BlockingQueueConsumer implements RecoveryListener {
OptionalLong deliveryTag = this.deliveryTags.stream().mapToLong(l -> l).max();
if (deliveryTag.isPresent()) {
this.channel.basicNack(deliveryTag.getAsLong(), true,
RabbitUtils.shouldRequeue(this.defaultRequeueRejected, ex, logger));
ContainerUtils.shouldRequeue(this.defaultRequeueRejected, ex, logger));
}
if (this.transactional) {
// Need to commit the reject (=nack)
@@ -967,6 +967,7 @@ public class BlockingQueueConsumer implements RecoveryListener {
}
@Override
public void handleConsumeOk(String consumerTag) {
this.consumerTag = consumerTag;
this.delegate.handleConsumeOk(consumerTag);
@@ -975,24 +976,29 @@ public class BlockingQueueConsumer implements RecoveryListener {
}
}
@Override
public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
this.delegate.handleShutdownSignal(consumerTag, sig);
}
@Override
public void handleCancel(String consumerTag) throws IOException {
this.delegate.handleCancel(consumerTag);
}
@Override
public void handleCancelOk(String consumerTag) {
this.delegate.handleCancelOk(consumerTag);
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
this.delegate.handleDelivery(consumerTag, envelope, properties, body);
}
@Override
public void handleRecoverOk(String consumerTag) {
this.delegate.handleRecoverOk(consumerTag);
}

62
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ContainerUtils.java Normal file
View File

@@ -0,0 +1,62 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit.listener;
import org.apache.commons.logging.Log;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
/**
* Utility methods for listener containers.
*
* @author Gary Russell
*
* @since 2.1
*
*/
public final class ContainerUtils {
private ContainerUtils() {
super();
}
/**
* Determine whether a message should be requeued; returns true if the throwable is a
* {@link MessageRejectedWhileStoppingException} or defaultRequeueRejected is true and
* there is not an {@link AmqpRejectAndDontRequeueException} in the cause chain.
* @param defaultRequeueRejected the default requeue rejected.
* @param throwable the throwable.
* @param logger the logger to use for debug.
* @return true to requeue.
*/
public static boolean shouldRequeue(boolean defaultRequeueRejected, Throwable throwable, Log logger) {
boolean shouldRequeue = defaultRequeueRejected ||
throwable instanceof MessageRejectedWhileStoppingException;
Throwable t = throwable;
while (shouldRequeue && t != null) {
if (t instanceof AmqpRejectAndDontRequeueException) {
shouldRequeue = false;
}
t = t.getCause();
}
if (logger.isDebugEnabled()) {
logger.debug("Rejecting messages (requeue=" + shouldRequeue + ")");
}
return shouldRequeue;
}
}

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

@@ -17,6 +17,7 @@
package org.springframework.amqp.rabbit.listener;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -41,6 +42,7 @@ import org.springframework.amqp.AmqpConnectException;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.AmqpIOException;
import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.Connection;
@@ -50,7 +52,6 @@ import org.springframework.amqp.rabbit.connection.ConsumerChannelRegistry;
import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.amqp.rabbit.connection.SimpleResourceHolder;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -59,6 +60,7 @@ import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
@@ -510,18 +512,33 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
private void checkMissingQueues(String[] queueNames) {
if (isMissingQueuesFatal()) {
RabbitAdmin checkAdmin = getRabbitAdmin();
AmqpAdmin checkAdmin = getAmqpAdmin();
if (checkAdmin == null) {
/*
* Checking queue existence doesn't require an admin in the context or injected into
* the container. If there's no such admin, just create a local one here.
* Use reflection to avoid class tangles.
*/
checkAdmin = new RabbitAdmin(getConnectionFactory());
try {
Class<?> clazz = ClassUtils.forName("org.springframework.amqp.rabbit.core.RabbitAdmin",
getClass().getClassLoader());
@SuppressWarnings("unchecked")
Constructor<AmqpAdmin> ctor = (Constructor<AmqpAdmin>) clazz
.getConstructor(ConnectionFactory.class);
checkAdmin = ctor.newInstance(getConnectionFactory());
setAmqpAdmin(checkAdmin);
}
catch (Exception e) {
this.logger.error("Failed to create a RabbitAdmin", e);
}
}
for (String queue : queueNames) {
Properties queueProperties = checkAdmin.getQueueProperties(queue);
if (queueProperties == null && isMissingQueuesFatal()) {
throw new IllegalStateException("At least one of the configured queues is missing");
if (checkAdmin != null) {
for (String queue : queueNames) {
Properties queueProperties = checkAdmin.getQueueProperties(queue);
if (queueProperties == null && isMissingQueuesFatal()) {
throw new IllegalStateException("At least one of the configured queues is missing");
}
}
}
}
@@ -962,7 +979,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
}
}
getChannel().basicNack(deliveryTag, true,
RabbitUtils.shouldRequeue(isDefaultRequeueRejected(), e, this.logger));
ContainerUtils.shouldRequeue(isDefaultRequeueRejected(), e, this.logger));
}
catch (IOException e1) {
this.logger.error("Failed to nack message", e1);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,7 +23,7 @@ import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Address;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.util.Assert;
import com.rabbitmq.client.Channel;

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/ListenerFailedRuleBasedTransactionAttribute.java → spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerFailedRuleBasedTransactionAttribute.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.amqp.rabbit.transaction;
package org.springframework.amqp.rabbit.listener;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2016 the original author or authors.
* Copyright 2014-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@ import java.util.Arrays;
import org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter;
import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.messaging.handler.annotation.SendTo;

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

@@ -27,7 +27,7 @@ import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
@@ -56,6 +56,7 @@ import com.rabbitmq.client.Channel;
* @author Artem Bilan
*
* @since 1.4
*
* @see MessageListener
* @see ChannelAwareMessageListener
*/

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -27,7 +27,7 @@ import org.springframework.amqp.AmqpIllegalStateException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
@@ -123,7 +123,7 @@ import com.rabbitmq.client.Channel;
* @see #setResponseRoutingKey(String)
* @see #setMessageConverter
* @see org.springframework.amqp.support.converter.SimpleMessageConverter
* @see org.springframework.amqp.rabbit.core.ChannelAwareMessageListener
* @see org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener
* @see org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer#setMessageListener
*/
public class MessageListenerAdapter extends AbstractAdaptableMessageListener {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,7 +22,7 @@ import java.lang.reflect.Type;
import java.lang.reflect.WildcardType;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.RabbitListenerErrorHandler;
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.support.AmqpHeaderMapper;
import org.springframework.amqp.support.converter.MessageConversionException;

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/ChannelAwareMessageListener.java → spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/ChannelAwareMessageListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.amqp.rabbit.core;
package org.springframework.amqp.rabbit.listener.api;
import org.springframework.amqp.core.Message;

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerErrorHandler.java → spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.amqp.rabbit.listener;
package org.springframework.amqp.rabbit.listener.api;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides Additional APIs for listeners.
*/
package org.springframework.amqp.rabbit.listener.api;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -62,11 +62,11 @@ import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactoryConfigurationUtils;
import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean;
import org.springframework.amqp.rabbit.core.DeclareExchangeConnectionListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.LogAppenderUtils;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.retry.RetryPolicy;
@@ -622,7 +622,7 @@ public class AmqpAppender extends AbstractAppender {
this.connectionFactory.setAddresses(this.addresses);
}
if (this.clientConnectionProperties != null) {
LogAppenderUtils.updateClientConnectionProperties(this.connectionFactory,
ConnectionFactoryConfigurationUtils.updateClientConnectionProperties(this.connectionFactory,
this.clientConnectionProperties);
}
setUpExchangeDeclaration();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2017 the original author or authors.
* Copyright 2014-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -44,11 +44,11 @@ import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactoryConfigurationUtils;
import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean;
import org.springframework.amqp.rabbit.core.DeclareExchangeConnectionListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.LogAppenderUtils;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
@@ -612,7 +612,8 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
if (this.addresses != null) {
this.connectionFactory.setAddresses(this.addresses);
}
LogAppenderUtils.updateClientConnectionProperties(this.connectionFactory, this.clientConnectionProperties);
ConnectionFactoryConfigurationUtils.updateClientConnectionProperties(this.connectionFactory,
this.clientConnectionProperties);
updateConnectionClientProperties(this.connectionFactory.getRabbitConnectionFactory().getClientProperties());
setUpExchangeDeclaration();
this.senderPool = Executors.newCachedThreadPool();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,7 +16,7 @@
package org.springframework.amqp.rabbit.support;
import org.springframework.amqp.support.Correlation;
import org.springframework.amqp.core.Correlation;
/**
* Base class for correlating publisher confirms to sent messages.

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

@@ -19,7 +19,7 @@ package org.springframework.amqp.rabbit.support;
import java.util.Collection;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
/**
* {@link MessageListener}s and {@link ChannelAwareMessageListener}s that also implement this

50
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/LogAppenderUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,9 +16,9 @@
package org.springframework.amqp.rabbit.support;
import java.util.Map;
import java.lang.reflect.Method;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.util.ClassUtils;
/**
* Utility methods for log appenders.
@@ -26,11 +26,33 @@ import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
* @author Gary Russell
* @since 1.5.6
*
* @deprecated in favor of {@code ConnectionFactoryConfigurationUtils}.
*
*/
@Deprecated
public final class LogAppenderUtils {
private static final Method util;
static {
Method method;
try {
Class<?> utils = ClassUtils.forName(
"org.springframework.amqp.rabbit.connection.ConnectionFactoryConfigurationUtils",
LogAppenderUtils.class.getClassLoader());
Class<?> abstractCF = ClassUtils.forName(
"org.springframework.amqp.rabbit.connection.AbstractConnectionFactory",
LogAppenderUtils.class.getClassLoader());
method = utils.getDeclaredMethod("updateClientConnectionProperties", abstractCF, String.class);
}
catch (Exception e) {
method = null;
}
util = method;
}
private LogAppenderUtils() {
// empty
super();
}
/**
@@ -39,20 +61,14 @@ public final class LogAppenderUtils {
* @param connectionFactory the connection factory.
* @param clientConnectionProperties the properties.
*/
public static void updateClientConnectionProperties(AbstractConnectionFactory connectionFactory,
public static void updateClientConnectionProperties(Object connectionFactory,
String clientConnectionProperties) {
if (clientConnectionProperties != null) {
String[] props = clientConnectionProperties.split(",");
if (props.length > 0) {
Map<String, Object> clientProps = connectionFactory.getRabbitConnectionFactory()
.getClientProperties();
for (String prop : props) {
String[] aProp = prop.split(":");
if (aProp.length == 2) {
clientProps.put(aProp[0].trim(), aProp[1].trim());
}
}
}
try {
util.invoke(null, connectionFactory, clientConnectionProperties);
}
catch (Exception e) {
throw new RuntimeException("Failed to set properties", e);
}
}

4
spring-rabbit/src/main/resources/META-INF/spring.schemas
View File

@@ -1,2 +1,2 @@
http\://www.springframework.org/schema/rabbit/spring-rabbit-2.0.xsd=org/springframework/amqp/rabbit/config/spring-rabbit-2.0.xsd
http\://www.springframework.org/schema/rabbit/spring-rabbit.xsd=org/springframework/amqp/rabbit/config/spring-rabbit-2.0.xsd
http\://www.springframework.org/schema/rabbit/spring-rabbit-2.1.xsd=org/springframework/amqp/rabbit/config/spring-rabbit-2.1.xsd
http\://www.springframework.org/schema/rabbit/spring-rabbit.xsd=org/springframework/amqp/rabbit/config/spring-rabbit-2.1.xsd

0
spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.0.xsd → spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.1.xsd
View File

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

@@ -78,8 +78,8 @@ import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.rabbit.listener.RabbitListenerErrorHandler;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.rabbit.test.MessageTestUtils;
import org.springframework.amqp.support.AmqpHeaders;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -32,10 +32,10 @@ import org.junit.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.test.MessageTestUtils;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -60,6 +60,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Correlation;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
@@ -74,7 +75,6 @@ import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.rabbit.support.PendingConfirm;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannel.Listener;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl;
import org.springframework.amqp.support.Correlation;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.DirectFieldAccessor;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -75,7 +75,7 @@ public class ContainerInitializationTests {
catch (ApplicationContextException e) {
assertThat(e.getCause().getCause(), instanceOf(IllegalStateException.class));
assertThat(e.getMessage(), containsString("When 'mismatchedQueuesFatal' is 'true', there must be "
+ "exactly one RabbitAdmin in the context or you must inject one into this container; found: 0"));
+ "exactly one AmqpAdmin in the context or you must inject one into this container; found: 0"));
}
}

46
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerShutDownTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,7 +23,6 @@ import static org.junit.Assert.assertTrue;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.AfterClass;
import org.junit.ClassRule;
@@ -68,16 +67,14 @@ public class ContainerShutDownTests {
container.setShutdownTimeout(500);
container.setQueueNames("test.shutdown");
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean testEnded = new AtomicBoolean();
final CountDownLatch testEnded = new CountDownLatch(1);
container.setMessageListener(m -> {
while (!testEnded.get()) {
try {
latch.countDown();
Thread.sleep(100);
}
catch (InterruptedException e) {
// Thread.currentThread().interrupt(); // eat it
}
try {
latch.countDown();
testEnded.await(30, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
final CountDownLatch startLatch = new CountDownLatch(1);
@@ -86,19 +83,24 @@ public class ContainerShutDownTests {
startLatch.countDown();
}
});
container.start();
assertTrue(startLatch.await(10, TimeUnit.SECONDS));
RabbitTemplate template = new RabbitTemplate(cf);
template.convertAndSend("test.shutdown", "foo");
assertTrue(latch.await(10, TimeUnit.SECONDS));
Connection connection = cf.createConnection();
Map<?, ?> channels = TestUtils.getPropertyValue(connection, "target.delegate._channelManager._channelMap", Map.class);
assertThat(channels.size(), equalTo(2));
container.stop();
assertThat(channels.size(), equalTo(1));
Map<?, ?> channels = TestUtils.getPropertyValue(connection, "target.delegate._channelManager._channelMap",
Map.class);
container.start();
try {
assertTrue(startLatch.await(30, TimeUnit.SECONDS));
RabbitTemplate template = new RabbitTemplate(cf);
template.convertAndSend("test.shutdown", "foo");
assertTrue(latch.await(30, TimeUnit.SECONDS));
assertThat(channels.size(), equalTo(2));
}
finally {
container.stop();
assertThat(channels.size(), equalTo(1));
cf.destroy();
testEnded.set(true);
cf.destroy();
testEnded.countDown();
}
}
}

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java
View File

@@ -54,13 +54,13 @@ import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.support.ArgumentBuilder;
import org.springframework.amqp.rabbit.test.LogLevelAdjuster;
import org.springframework.amqp.support.ConsumerTagStrategy;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -32,9 +32,9 @@ import org.junit.Test;
import org.springframework.amqp.core.Address;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.DirectFieldAccessor;

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ExternalTxManagerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -49,9 +49,8 @@ import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.transaction.ListenerFailedRuleBasedTransactionAttribute;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.transaction.TransactionDefinition;

14
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ListenFromAutoDeleteQueueTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2015 the original author or authors.
* Copyright 2014-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -89,8 +89,8 @@ public class ListenFromAutoDeleteQueueTests {
template.convertAndSend("testContainerWithAutoDeleteQueues", "anon", "foo");
assertNotNull(queue.poll(10, TimeUnit.SECONDS));
this.listenerContainer1.stop();
RabbitAdmin admin = spy(TestUtils.getPropertyValue(this.listenerContainer1, "rabbitAdmin", RabbitAdmin.class));
new DirectFieldAccessor(this.listenerContainer1).setPropertyValue("rabbitAdmin", admin);
RabbitAdmin admin = spy(TestUtils.getPropertyValue(this.listenerContainer1, "amqpAdmin", RabbitAdmin.class));
new DirectFieldAccessor(this.listenerContainer1).setPropertyValue("amqpAdmin", admin);
this.listenerContainer1.start();
template.convertAndSend("testContainerWithAutoDeleteQueues", "anon", "foo");
assertNotNull(queue.poll(10, TimeUnit.SECONDS));
@@ -117,8 +117,8 @@ public class ListenFromAutoDeleteQueueTests {
SimpleMessageListenerContainer listenerContainer = context.getBean("container3",
SimpleMessageListenerContainer.class);
listenerContainer.stop();
RabbitAdmin admin = spy(TestUtils.getPropertyValue(listenerContainer, "rabbitAdmin", RabbitAdmin.class));
new DirectFieldAccessor(listenerContainer).setPropertyValue("rabbitAdmin", admin);
RabbitAdmin admin = spy(TestUtils.getPropertyValue(listenerContainer, "amqpAdmin", RabbitAdmin.class));
new DirectFieldAccessor(listenerContainer).setPropertyValue("amqpAdmin", admin);
int n = 0;
while (admin.getQueueProperties(this.expiringQueue.getName()) != null && n < 100) {
Thread.sleep(100);
@@ -139,8 +139,8 @@ public class ListenFromAutoDeleteQueueTests {
SimpleMessageListenerContainer listenerContainer = context.getBean("container4",
SimpleMessageListenerContainer.class);
listenerContainer.stop();
RabbitAdmin admin = spy(TestUtils.getPropertyValue(listenerContainer, "rabbitAdmin", RabbitAdmin.class));
new DirectFieldAccessor(listenerContainer).setPropertyValue("rabbitAdmin", admin);
RabbitAdmin admin = spy(TestUtils.getPropertyValue(listenerContainer, "amqpAdmin", RabbitAdmin.class));
new DirectFieldAccessor(listenerContainer).setPropertyValue("amqpAdmin", admin);
listenerContainer = spy(listenerContainer);
//Prevent a long 'passiveDeclare' process

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -47,8 +47,8 @@ import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.DirectFieldAccessor;
import com.rabbitmq.client.AMQP.BasicProperties;

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java
View File

@@ -54,12 +54,12 @@ import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.utils.test.TestUtils;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -34,11 +34,11 @@ import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.test.LogLevelAdjuster;
import org.springframework.beans.factory.DisposableBean;

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java
View File

@@ -45,13 +45,13 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionProxy;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LongRunningIntegrationTest;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.test.LogLevelAdjuster;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.factory.DisposableBean;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -37,11 +37,11 @@ import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerExecutionException;
import org.springframework.amqp.rabbit.test.LogLevelAdjuster;
import org.springframework.amqp.rabbit.test.RepeatProcessor;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -34,11 +34,11 @@ import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.test.LogLevelAdjuster;
import org.springframework.beans.factory.DisposableBean;

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

@@ -67,7 +67,6 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
@@ -75,6 +74,7 @@ import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LongRunningIntegrationTest;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.support.ConsumerCancelledException;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl;
import org.springframework.amqp.utils.test.TestUtils;
@@ -305,7 +305,7 @@ public class SimpleMessageListenerContainerIntegration2Tests {
container.setApplicationContext(context);
RabbitAdmin admin = new RabbitAdmin(this.template.getConnectionFactory());
admin.setApplicationContext(context);
container.setRabbitAdmin(admin);
container.setAmqpAdmin(admin);
container.afterPropertiesSet();
container.start();
for (int i = 0; i < 10; i++) {

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java
View File

@@ -43,12 +43,12 @@ import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LongRunningIntegrationTest;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.test.LogLevelAdjuster;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.transaction.TransactionDefinition;

4
src/reference/asciidoc/amqp.adoc
View File

@@ -1408,6 +1408,8 @@ public interface ChannelAwareMessageListener {
}
----
IMPORTANT: In _version 2.1_, this interface moved from package `o.s.amqp.rabbit.core` to `o.s.amqp.rabbit.listener.api`.
[[message-listener-adapter]]
====== MessageListenerAdapter
@@ -2364,6 +2366,8 @@ The stack trace of the server exception will be synthesized by merging the serve
IMPORTANT: This mechanism will generally only work with the default `SimpleMessageConverter`, which uses Java serialization; exceptions are generally not "Jackson-friendly" so can't be serialized to JSON.
If you are using JSON, consider using an `errorHandler` to return some other Jackson-friendly `Error` object when an exception is thrown.
IMPORTANT: In _version 2.1_, this interface moved from package `o.s.amqp.rabbit.listener` to `o.s.amqp.rabbit.listener.api`.
====== Container Management
Containers created for annotations are not registered with the application context.

234
src/reference/asciidoc/appendix.adoc
View File

@@ -7,6 +7,240 @@ See <<whats-new>>.
[[previous-whats-new]]
=== Previous Releases
==== Changes in 2.0 Since 1.7
===== CachingConnectionFactory
Starting with _version 2.0.2_, the `RabbitTemplate` can be configured to use a different connection to that used by listener containers.
This is to avoid deadlocked consumers when producers are blocked for any reason.
See <<separate-connection>> for more information.
===== AMQP Client library
Spring AMQP now uses the new 5.0.x version of the `amqp-client` library provided by the RabbitMQ team.
This client has auto recovery configured by default; see <<auto-recovery>>.
NOTE: As of version 4.0, the client enables automatic recovery by default; while compatible with this feature, Spring AMQP has its own recovery mechanisms and the client recovery feature generally isn't needed.
It is recommended to disable `amqp-client` automatic recovery, to avoid getting `AutoRecoverConnectionNotCurrentlyOpenException` s when the broker is available, but the connection has not yet recovered.
Starting with _version 1.7.1_, Spring AMQP disables it unless you explicitly create your own RabbitMQ connection factory and provide it to the `CachingConnectionFactory`.
RabbitMQ `ConnectionFactory` instances created by the `RabbitConnectionFactoryBean` will also have the option disabled by default.
===== General Changes
The `ExchangeBuilder` now builds durable exchanges by default.
The `@Exchange` annotation used within a `@QeueueBinding` also declares durable exchanges by default.
The `@Queue` annotation used within a `@RabbitListener` by default declares durable queues if named and non-durable if anonymous.
See <<builder-api>> and <<async-annotation-driven>> for more information.
===== Deleted classes
`UniquelyNameQueue` is no longer provided. It is unusual to create a durable non auto-delete queue with a unique name.
This class has been deleted; if you require its functionality, use `new Queue(UUID.randomUUID().toString())`.
===== New Listener Container
The `DirectMessageListenerContainer` has been added alongside the existing `SimpleMessageListenerContainer`.
See <<choose-container>> and <<containerAttributes>> for information about choosing which container to use as well as how to configure them.
===== Log4j Appender
This appender is no longer available due to the end-of-life of log4j.
See <<logging>> for information about the available log appenders.
===== RabbitTemplate Changes
IMPORTANT: Previously, a non-transactional `RabbitTemplate` participated in an existing transaction if it ran on a transactional listener container thread.
This was a serious bug; however, users might have relied on this behavior.
Starting with _version 1.6.2_, you must set the `channelTransacted` boolean on the template for it to participate in the container transaction.
The `RabbitTemplate` now uses a `DirectReplyToMessageListenerContainer` (by default) instead of creating a new consumer for each request.
See <<direct-reply-to>> for more information.
The `AsyncRabbitTemplate` now supports Direct reply-to; see <<async-template>> for more information.
The `RabbitTemplate` and `AsyncRabbitTemplate` now have `receiveAndConvert` and `convertSendAndReceiveAsType` methods that take a `ParameterizedTypeReference<T>` argument, allowing the caller to specify the type to convert the result to.
This is particularly useful for complex types or when type information is not conveyed in message headers.
Requires a `SmartMessageConverter` such as the `Jackson2JsonMessageConverter`.
See <<receiving-messages>>, <<request-reply>>, <<async-template>>, and <<json-complex>> for more information.
You can now use a `RabbitTemplate` to perform multiple operations on a dedicated channel.
See <<scoped-operations>> for more information.
===== Listener Adapter
A convenient `FunctionalInterface` is available for using lambdas with the `MessageListenerAdapter`.
See <<message-listener-adapter>> for more information.
===== Listener Container Changes
====== Prefetch default value
The prefetch default value used to be 1, which could lead to under-utilization of efficient consumers.
The default prefetch value is now 250, which should keep consumers busy in most common scenarios and
thus improve throughput.
IMPORTANT: There are nevertheless scenarios where the prefetch value should
be low: for example, with large messages, especially if the processing is slow (messages could add up
to a large amount of memory in the client process), and if strict message ordering is necessary
(the prefetch value should be set back to 1 in this case).
Also, with low-volume messaging and multiple consumers (including concurrency within a single listener container instance), you may wish to reduce the prefetch to get a more even distribution of messages across consumers.
For more background about prefetch, see this post about https://www.rabbitmq.com/blog/2014/04/14/finding-bottlenecks-with-rabbitmq-3-3/[consumer utilization in RabbitMQ]
and this post about https://www.rabbitmq.com/blog/2012/05/11/some-queuing-theory-throughput-latency-and-bandwidth/[queuing theory].
====== Message Count
Previously, `MessageProperties.getMessageCount()` returned `0` for messages emitted by the container.
This property only applies when using `basicGet` (e.g. from `RabbitTemplate.receive()` methods) and is now initialized to `null` for container messages.
====== Transaction Rollback behavior
Message requeue on transaction rollback is now consistent, regardless of whether or not a transaction manager is configured.
See <<transaction-rollback>> for more information.
====== Shutdown Behavior
If the container threads do not respond to a shutdown within `shutdownTimeout`, the channel(s) will be forced closed, by default.
See <<containerAttributes>> for more information.
====== After Receive Message Post Processors
If a `MessagePostProcessor` in the `afterReceiveMessagePostProcessors` property returns `null`, the message is discarded (and acknowledged if appropriate).
===== Connection Factory Changes
The connection and channel listener interfaces now provide a mechanism to obtain information about exceptions.
See <<connection-channel-listeners>> and <<publishing-is-async>> for more information.
A new `ConnectionNameStrategy` is now provided to populate the application-specific identification of the target RabbitMQ connection from the `AbstractConnectionFactory`.
See <<connections>> for more information.
===== Retry Changes
The `MissingMessageIdAdvice` is no longer provided; it's functionality is now built-in; see <<retry>> for more information.
===== Anonymous Queue Naming
By default, `AnonymousQueues` are now named with the default `Base64UrlNamingStrategy` instead of a simple `UUID` string.
See <<anonymous-queue>> for more information.
===== @RabbitListener Changes
You can now provide simple queue declarations (only bound to the default exchange) in `@RabbitListener` annotations.
See <<async-annotation-driven>> for more information.
You can now configure `@RabbitListener` annotations so that any exceptions thrown will be returned to the sender.
You can also configure a `RabbitListenerErrorHandler` to handle exceptions.
See <<annotation-error-handling>> for more information.
You can now bind a queue with multiple routing keys when using the `@QueueBinding` annotation.
Also `@QueueBinding.exchange()` now supports custom exchange types and declares durable exchanges by default.
You can now set the `concurrency` of the listener container at the annotation level rather than having to configure a different container factory for different concurrency settings.
You can now set the `autoStartup` property of the listener container at the annotation level, overriding the default setting in the container factory.
You can now set after receive and before send (reply) `MessagePostProcessor` s in the `RabbitListener` container factories.
See <<async-annotation-driven>> for more information.
Starting with _version 2.0.3_, one of the `@RabbitHandler` s on a class-level `@RabbitListener` can be designated as the default.
See <<annotation-method-selection>> for more information.
===== Container Conditional Rollback
When using an external transaction manager (e.g. JDBC), rule-based rollback is now supported when providing the container with a transaction attribute.
It is also now more flexible when using a transaction advice.
See <<conditional-rollback>> for more information.
===== Remove Jackson 1.x support
Deprecated in previous versions, Jackson `1.x` converters and related components have now been deleted; use similar components based on Jackson 2.x.
See <<json-message-converter>> for more information.
===== JSON Message Converter
When the `__TypeId__` is set to `Hashtable` for an inbound JSON message, the default conversion type is now `LinkedHashMap`; previously it was `Hashtable`.
To revert to a `Hashtable` use `setDefaultMapType` on the `DefaultClassMapper`.
===== XML Parsers
When parsing `Queue` and `Exchange` XML components, the parsers no longer register the `name` attribute value as a bean alias if an `id` attribute is present.
See <<note-id-name>> for more information.
===== Blocked Connection
The `com.rabbitmq.client.BlockedListener` can now be injected into the `org.springframework.amqp.rabbit.connection.Connection` object.
Also the `ConnectionBlockedEvent` and `ConnectionUnblockedEvent` events are emitted by the `ConnectionFactory`, when the connection is blocked or unblocked by the Broker.
See <<connections>> for more information.
==== Changes in 1.7 Since 1.6
===== AMQP Client library
Spring AMQP now uses the new 4.0.x version of the `amqp-client` library provided by the RabbitMQ team.
This client has auto recovery configured by default; see <<auto-recovery>>.
NOTE: The 4.0.x client enables automatic recovery by default; while compatible with this feature, Spring AMQP has its own recovery mechanisms and the client recovery feature generally isn't needed.
It is recommended to disable `amqp-client` automatic recovery, to avoid getting `AutoRecoverConnectionNotCurrentlyOpenException` s when the broker is available, but the connection has not yet recovered.
Starting with _version 1.7.1_, Spring AMQP disables it unless you explicitly create your own RabbitMQ connection factory and provide it to the `CachingConnectionFactory`.
RabbitMQ `ConnectionFactory` instances created by the `RabbitConnectionFactoryBean` will also have the option disabled by default.
===== Log4j2 upgrade
The minimum Log4j2 version (for the `AmqpAppender`) is now `2.7`.
The framework is no longer compatible with previous versions.
See <<logging>> for more information.
===== Logback Appender
This appender no longer captures caller data (method, line number) by default; it can be re-enabled by setting the `includeCallerData` configuration option.
See <<logging>> for information about the available log appenders.
===== Spring Retry upgrade
The minimum Spring Retry version is now `1.2`.
The framework is no longer compatible with previous versions.
====== Shutdown Behavior
You can now set `forceCloseChannel` to `true` so that, if the container threads do not respond to a shutdown within `shutdownTimeout`, the channel(s) will be forced closed,
causing any unacked messages to be requeued.
See <<containerAttributes>> for more information.
===== FasterXML Jackson upgrade
The minimum Jackson version is now `2.8`.
The framework is no longer compatible with previous versions.
===== JUnit @Rules
Rules that have up until now been used internally by the framework have now been made available in a separate jar `spring-rabbit-junit`.
See <<junit-rules>> for more information.
===== Container Conditional Rollback
When using an external transaction manager (e.g. JDBC), rule-based rollback is now supported when providing the container with a transaction attribute.
It is also now more flexible when using a transaction advice.
===== Connection Naming Strategy
A new `ConnectionNameStrategy` is now provided to populate the application-specific identification of the target RabbitMQ connection from the `AbstractConnectionFactory`.
See <<connections>> for more information.
===== Listener Container Changes
====== Transaction Rollback behavior
Message requeue on transaction rollback can now be configured to be consistent, regardless of whether or not a transaction manager is configured.
See <<transaction-rollback>> for more information.
==== Earlier Releases
See <<previous-whats-new>> for changes in previous versions.
==== Changes in 1.6 Since 1.5
===== Testing Support

170
src/reference/asciidoc/whats-new.adoc
View File

@@ -1,17 +1,11 @@
[[whats-new]]
=== What's New
==== Changes in 2.0 Since 1.7
===== CachingConnectionFactory
Starting with _version 2.0.2_, the `RabbitTemplate` can be configured to use a different connection to that used by listener containers.
This is to avoid deadlocked consumers when producers are blocked for any reason.
See <<separate-connection>> for more information.
==== Changes in 2.1 Since 2.0
===== AMQP Client library
Spring AMQP now uses the new 5.0.x version of the `amqp-client` library provided by the RabbitMQ team.
Spring AMQP now uses the new 5.2.x version of the `amqp-client` library provided by the RabbitMQ team.
This client has auto recovery configured by default; see <<auto-recovery>>.
NOTE: As of version 4.0, the client enables automatic recovery by default; while compatible with this feature, Spring AMQP has its own recovery mechanisms and the client recovery feature generally isn't needed.
@@ -19,162 +13,8 @@ It is recommended to disable `amqp-client` automatic recovery, to avoid getting
Starting with _version 1.7.1_, Spring AMQP disables it unless you explicitly create your own RabbitMQ connection factory and provide it to the `CachingConnectionFactory`.
RabbitMQ `ConnectionFactory` instances created by the `RabbitConnectionFactoryBean` will also have the option disabled by default.
===== General Changes
The `ExchangeBuilder` now builds durable exchanges by default.
The `@Exchange` annotation used within a `@QeueueBinding` also declares durable exchanges by default.
The `@Queue` annotation used within a `@RabbitListener` by default declares durable queues if named and non-durable if anonymous.
See <<builder-api>> and <<async-annotation-driven>> for more information.
===== Package Changes
===== Deleted classes
`UniquelyNameQueue` is no longer provided. It is unusual to create a durable non auto-delete queue with a unique name.
This class has been deleted; if you require its functionality, use `new Queue(UUID.randomUUID().toString())`.
===== New Listener Container
The `DirectMessageListenerContainer` has been added alongside the existing `SimpleMessageListenerContainer`.
See <<choose-container>> and <<containerAttributes>> for information about choosing which container to use as well as how to configure them.
===== Log4j Appender
This appender is no longer available due to the end-of-life of log4j.
See <<logging>> for information about the available log appenders.
===== Logback Appender
This appender no longer captures caller data (method, line number) by default; it can be re-enabled by setting the `includeCallerData` configuration option.
See <<logging>> for information about the available log appenders.
===== RabbitTemplate Changes
IMPORTANT: Previously, a non-transactional `RabbitTemplate` participated in an existing transaction if it ran on a transactional listener container thread.
This was a serious bug; however, users might have relied on this behavior.
Starting with _version 1.6.2_, you must set the `channelTransacted` boolean on the template for it to participate in the container transaction.
The `RabbitTemplate` now uses a `DirectReplyToMessageListenerContainer` (by default) instead of creating a new consumer for each request.
See <<direct-reply-to>> for more information.
The `AsyncRabbitTemplate` now supports Direct reply-to; see <<async-template>> for more information.
The `RabbitTemplate` and `AsyncRabbitTemplate` now have `receiveAndConvert` and `convertSendAndReceiveAsType` methods that take a `ParameterizedTypeReference<T>` argument, allowing the caller to specify the type to convert the result to.
This is particularly useful for complex types or when type information is not conveyed in message headers.
Requires a `SmartMessageConverter` such as the `Jackson2JsonMessageConverter`.
See <<receiving-messages>>, <<request-reply>>, <<async-template>>, and <<json-complex>> for more information.
You can now use a `RabbitTemplate` to perform multiple operations on a dedicated channel.
See <<scoped-operations>> for more information.
===== Listener Adapter
A convenient `FunctionalInterface` is available for using lambdas with the `MessageListenerAdapter`.
See <<message-listener-adapter>> for more information.
===== Listener Container Changes
====== Prefetch default value
The prefetch default value used to be 1, which could lead to under-utilization of efficient consumers.
The default prefetch value is now 250, which should keep consumers busy in most common scenarios and
thus improve throughput.
IMPORTANT: There are nevertheless scenarios where the prefetch value should
be low: for example, with large messages, especially if the processing is slow (messages could add up
to a large amount of memory in the client process), and if strict message ordering is necessary
(the prefetch value should be set back to 1 in this case).
Also, with low-volume messaging and multiple consumers (including concurrency within a single listener container instance), you may wish to reduce the prefetch to get a more even distribution of messages across consumers.
For more background about prefetch, see this post about https://www.rabbitmq.com/blog/2014/04/14/finding-bottlenecks-with-rabbitmq-3-3/[consumer utilization in RabbitMQ]
and this post about https://www.rabbitmq.com/blog/2012/05/11/some-queuing-theory-throughput-latency-and-bandwidth/[queuing theory].
====== Message Count
Previously, `MessageProperties.getMessageCount()` returned `0` for messages emitted by the container.
This property only applies when using `basicGet` (e.g. from `RabbitTemplate.receive()` methods) and is now initialized to `null` for container messages.
====== Transaction Rollback behavior
Message requeue on transaction rollback is now consistent, regardless of whether or not a transaction manager is configured.
See <<transaction-rollback>> for more information.
====== Shutdown Behavior
If the container threads do not respond to a shutdown within `shutdownTimeout`, the channel(s) will be forced closed, by default.
See <<containerAttributes>> for more information.
====== After Receive Message Post Processors
If a `MessagePostProcessor` in the `afterReceiveMessagePostProcessors` property returns `null`, the message is discarded (and acknowledged if appropriate).
===== Connection Factory Changes
The connection and channel listener interfaces now provide a mechanism to obtain information about exceptions.
See <<connection-channel-listeners>> and <<publishing-is-async>> for more information.
A new `ConnectionNameStrategy` is now provided to populate the application-specific identification of the target RabbitMQ connection from the `AbstractConnectionFactory`.
See <<connections>> for more information.
===== Retry Changes
The `MissingMessageIdAdvice` is no longer provided; it's functionality is now built-in; see <<retry>> for more information.
===== Anonymous Queue Naming
By default, `AnonymousQueues` are now named with the default `Base64UrlNamingStrategy` instead of a simple `UUID` string.
See <<anonymous-queue>> for more information.
===== @RabbitListener Changes
You can now provide simple queue declarations (only bound to the default exchange) in `@RabbitListener` annotations.
See <<async-annotation-driven>> for more information.
You can now configure `@RabbitListener` annotations so that any exceptions thrown will be returned to the sender.
You can also configure a `RabbitListenerErrorHandler` to handle exceptions.
See <<annotation-error-handling>> for more information.
You can now bind a queue with multiple routing keys when using the `@QueueBinding` annotation.
Also `@QueueBinding.exchange()` now supports custom exchange types and declares durable exchanges by default.
You can now set the `concurrency` of the listener container at the annotation level rather than having to configure a different container factory for different concurrency settings.
You can now set the `autoStartup` property of the listener container at the annotation level, overriding the default setting in the container factory.
You can now set after receive and before send (reply) `MessagePostProcessor` s in the `RabbitListener` container factories.
See <<async-annotation-driven>> for more information.
Starting with _version 2.0.3_, one of the `@RabbitHandler` s on a class-level `@RabbitListener` can be designated as the default.
See <<annotation-method-selection>> for more information.
===== Container Conditional Rollback
When using an external transaction manager (e.g. JDBC), rule-based rollback is now supported when providing the container with a transaction attribute.
It is also now more flexible when using a transaction advice.
See <<conditional-rollback>> for more information.
===== Remove Jackson 1.x support
Deprecated in previous versions, Jackson `1.x` converters and related components have now been deleted; use similar components based on Jackson 2.x.
See <<json-message-converter>> for more information.
===== JSON Message Converter
When the `__TypeId__` is set to `Hashtable` for an inbound JSON message, the default conversion type is now `LinkedHashMap`; previously it was `Hashtable`.
To revert to a `Hashtable` use `setDefaultMapType` on the `DefaultClassMapper`.
===== XML Parsers
When parsing `Queue` and `Exchange` XML components, the parsers no longer register the `name` attribute value as a bean alias if an `id` attribute is present.
See <<note-id-name>> for more information.
===== Blocked Connection
The `com.rabbitmq.client.BlockedListener` can now be injected into the `org.springframework.amqp.rabbit.connection.Connection` object.
Also the `ConnectionBlockedEvent` and `ConnectionUnblockedEvent` events are emitted by the `ConnectionFactory`, when the connection is blocked or unblocked by the Broker.
See <<connections>> for more information.
==== Earlier Releases
See <<previous-whats-new>> for changes in previous versions.
Certain classes have moved to different packages; most are internal classes and won't affect user applications.
Two exceptions are `ChannelAwareMessageListener` and `RabbitListenerErrorHandler`; these interfaces are now in `org.springframework.amqp.rabbit.listener.api`.