diff --git a/build.gradle b/build.gradle index a93bb307..e0ede24b 100644 --- a/build.gradle +++ b/build.gradle @@ -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' diff --git a/gradle.properties b/gradle.properties index 046f8fb8..3117720d 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1,2 +1,2 @@ -version=2.0.3.BUILD-SNAPSHOT +version=2.1.0.BUILD-SNAPSHOT org.gradle.daemon=true diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java index c1c73bdd..ac84c65a 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -105,4 +105,12 @@ public interface AmqpAdmin { */ Properties getQueueProperties(String queueName); + /** + * Initialize the admin. + * @since 2.1 + */ + default void initialize() { + // no op + } + } diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/Correlation.java b/spring-amqp/src/main/java/org/springframework/amqp/core/Correlation.java similarity index 89% rename from spring-amqp/src/main/java/org/springframework/amqp/support/Correlation.java rename to spring-amqp/src/main/java/org/springframework/amqp/core/Correlation.java index 571c60af..2fa12bb7 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/Correlation.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/Correlation.java @@ -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. diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/Message.java b/spring-amqp/src/main/java/org/springframework/amqp/core/Message.java index 07bfc816..d0efacf1 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/Message.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/Message.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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 whiteListPatterns = new LinkedHashSet( + 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) diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/MessagePostProcessor.java b/spring-amqp/src/main/java/org/springframework/amqp/core/MessagePostProcessor.java index ed04f75f..c2d2d95b 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/MessagePostProcessor.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/MessagePostProcessor.java @@ -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 diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/WhiteListDeserializingMessageConverter.java b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/WhiteListDeserializingMessageConverter.java index 84095eb4..136ce914 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/WhiteListDeserializingMessageConverter.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/WhiteListDeserializingMessageConverter.java @@ -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); } } diff --git a/spring-amqp/src/main/java/org/springframework/amqp/utils/SerializationUtils.java b/spring-amqp/src/main/java/org/springframework/amqp/utils/SerializationUtils.java index 3455ef6c..d031cdba 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/utils/SerializationUtils.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/utils/SerializationUtils.java @@ -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 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 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); + } + } diff --git a/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java b/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java index 3bad4810..1d68bf52 100644 --- a/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java +++ b/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/TestRabbitTemplate.java @@ -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(); } + } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java index d4d911c8..4103d6c9 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java @@ -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 returnType, boolean enableConfirms) { + this.userPostProcessor = userPostProcessor; this.returnType = returnType; this.enableConfirms = enableConfirms; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java index 8eda2048..32285995 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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 diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java index 78bb8529..cc911004 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java @@ -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; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java index 101beea6..453cd4d4 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java @@ -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 0) { + Map clientProps = + connectionFactory.getRabbitConnectionFactory() + .getClientProperties(); + + for (String prop : props) { + String[] aProp = prop.split(":"); + if (aProp.length == 2) { + clientProps.put(aProp[0].trim(), aProp[1].trim()); + } + } + } + } + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java index 343ec6b1..f2505ea5 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java @@ -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 physicalCloseRequired = new ThreadLocal(); + private static final ThreadLocal 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) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index be4003b9..81274856 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -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 admins = this.getApplicationContext().getBeansOfType(RabbitAdmin.class); + if (this.amqpAdmin == null && this.getApplicationContext() != null) { + Map 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 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)); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java index b88b574b..1481976b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java @@ -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 queues = new ArrayList(); + private final Collection queues = new ArrayList<>(); - private final Collection queueNames = new ArrayList(); + private final Collection 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); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 7306a07e..e382f46b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -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); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ContainerUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ContainerUtils.java new file mode 100644 index 00000000..7d15ddc1 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ContainerUtils.java @@ -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; + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java index a418c464..5795486a 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java @@ -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 ctor = (Constructor) 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); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java index 8424ee6f..2d75ded3 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java @@ -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; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/ListenerFailedRuleBasedTransactionAttribute.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerFailedRuleBasedTransactionAttribute.java similarity index 92% rename from spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/ListenerFailedRuleBasedTransactionAttribute.java rename to spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerFailedRuleBasedTransactionAttribute.java index dd8ed03c..1ba32b16 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/ListenerFailedRuleBasedTransactionAttribute.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ListenerFailedRuleBasedTransactionAttribute.java @@ -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; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java index ef8d218c..8d3798c8 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java @@ -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; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java index aab6fea7..080a8d0c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java @@ -27,7 +27,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 */ diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java index 0ff05c1c..d1217509 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java @@ -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 { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java index 8a784379..f9cc55a0 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,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; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/ChannelAwareMessageListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/ChannelAwareMessageListener.java similarity index 92% rename from spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/ChannelAwareMessageListener.java rename to spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/ChannelAwareMessageListener.java index 9f6625ba..5f8742f3 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/ChannelAwareMessageListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/ChannelAwareMessageListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerErrorHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java similarity index 93% rename from spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerErrorHandler.java rename to spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java index 7e674a75..2195bef4 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerErrorHandler.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java @@ -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; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/package-info.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/package-info.java new file mode 100644 index 00000000..c25b7c53 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides Additional APIs for listeners. + */ +package org.springframework.amqp.rabbit.listener.api; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java index 61321631..80f937fa 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java @@ -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(); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java index 4acde65c..b1f66de7 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java @@ -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 { 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(); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java index de9b1711..f27118da 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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. diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ListenerContainerAware.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ListenerContainerAware.java index 02f563b6..e648eb7b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ListenerContainerAware.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ListenerContainerAware.java @@ -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 diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/LogAppenderUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/LogAppenderUtils.java index 2b3cc35c..dc013db3 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/LogAppenderUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/LogAppenderUtils.java @@ -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 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); } } diff --git a/spring-rabbit/src/main/resources/META-INF/spring.schemas b/spring-rabbit/src/main/resources/META-INF/spring.schemas index d6f04382..a4c46500 100644 --- a/spring-rabbit/src/main/resources/META-INF/spring.schemas +++ b/spring-rabbit/src/main/resources/META-INF/spring.schemas @@ -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 diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.0.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.1.xsd similarity index 100% rename from spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.0.xsd rename to spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.1.xsd diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index b1b1cabb..4b99d553 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RabbitListenerContainerFactoryIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RabbitListenerContainerFactoryIntegrationTests.java index 62be2b76..8b2f564f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RabbitListenerContainerFactoryIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RabbitListenerContainerFactoryIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index 529e7900..cebcca44 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java index 1c42fcc0..82803e3f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java @@ -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")); } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerShutDownTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerShutDownTests.java index 52381e89..3b7c0668 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerShutDownTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerShutDownTests.java @@ -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(); + } } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java index 8fd64cf6..7d644be3 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainerTests.java index 365cbe95..89955c97 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainerTests.java @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ExternalTxManagerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ExternalTxManagerTests.java index 32e94596..4dd23d03 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ExternalTxManagerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ExternalTxManagerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ListenFromAutoDeleteQueueTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ListenFromAutoDeleteQueueTests.java index 21142960..701b87f1 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ListenFromAutoDeleteQueueTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ListenFromAutoDeleteQueueTests.java @@ -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 diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/LocallyTransactedTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/LocallyTransactedTests.java index a8928448..df0e5bbd 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/LocallyTransactedTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/LocallyTransactedTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java index c0849de8..a40a0d4b 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java index cdd760a4..c101310f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java index e5856d5b..5a1fd3e9 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryRepeatIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryRepeatIntegrationTests.java index a8174785..e525721e 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryRepeatIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryRepeatIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java index d81537b7..7017dfda 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java index 745bd895..b819d804 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java @@ -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++) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java index b48b6857..372db32d 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java @@ -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; diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index cf736bc3..aab8d06d 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -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. diff --git a/src/reference/asciidoc/appendix.adoc b/src/reference/asciidoc/appendix.adoc index 9a37f701..0edab871 100644 --- a/src/reference/asciidoc/appendix.adoc +++ b/src/reference/asciidoc/appendix.adoc @@ -7,6 +7,240 @@ See <>. [[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 <> 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 <>. + +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 <> and <> 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 <> and <> 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 <> 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 <> for more information. + +The `AsyncRabbitTemplate` now supports Direct reply-to; see <> for more information. + +The `RabbitTemplate` and `AsyncRabbitTemplate` now have `receiveAndConvert` and `convertSendAndReceiveAsType` methods that take a `ParameterizedTypeReference` 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 <>, <>, <>, and <> for more information. + +You can now use a `RabbitTemplate` to perform multiple operations on a dedicated channel. +See <> for more information. + +===== Listener Adapter + +A convenient `FunctionalInterface` is available for using lambdas with the `MessageListenerAdapter`. +See <> 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 <> 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 <> 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 <> and <> for more information. + +A new `ConnectionNameStrategy` is now provided to populate the application-specific identification of the target RabbitMQ connection from the `AbstractConnectionFactory`. +See <> for more information. + +===== Retry Changes + +The `MissingMessageIdAdvice` is no longer provided; it's functionality is now built-in; see <> for more information. + +===== Anonymous Queue Naming + +By default, `AnonymousQueues` are now named with the default `Base64UrlNamingStrategy` instead of a simple `UUID` string. +See <> for more information. + +===== @RabbitListener Changes + +You can now provide simple queue declarations (only bound to the default exchange) in `@RabbitListener` annotations. +See <> 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 <> 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 <> 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 <> 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 <> 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 <> 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 <> 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 <> 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 <>. + +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 <> 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 <> 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 <> 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 <> 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 <> 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 <> for more information. + +==== Earlier Releases + +See <> for changes in previous versions. + ==== Changes in 1.6 Since 1.5 ===== Testing Support diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index c34b564b..125b1a34 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -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 <> 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 <>. 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 <> and <> 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 <> and <> 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 <> 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 <> 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 <> for more information. - -The `AsyncRabbitTemplate` now supports Direct reply-to; see <> for more information. - -The `RabbitTemplate` and `AsyncRabbitTemplate` now have `receiveAndConvert` and `convertSendAndReceiveAsType` methods that take a `ParameterizedTypeReference` 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 <>, <>, <>, and <> for more information. - -You can now use a `RabbitTemplate` to perform multiple operations on a dedicated channel. -See <> for more information. - -===== Listener Adapter - -A convenient `FunctionalInterface` is available for using lambdas with the `MessageListenerAdapter`. -See <> 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 <> 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 <> 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 <> and <> for more information. - -A new `ConnectionNameStrategy` is now provided to populate the application-specific identification of the target RabbitMQ connection from the `AbstractConnectionFactory`. -See <> for more information. - -===== Retry Changes - -The `MissingMessageIdAdvice` is no longer provided; it's functionality is now built-in; see <> for more information. - -===== Anonymous Queue Naming - -By default, `AnonymousQueues` are now named with the default `Base64UrlNamingStrategy` instead of a simple `UUID` string. -See <> for more information. - -===== @RabbitListener Changes - -You can now provide simple queue declarations (only bound to the default exchange) in `@RabbitListener` annotations. -See <> 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 <> 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 <> 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 <> 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 <> 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 <> 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 <> 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 <> for more information. - -==== Earlier Releases - -See <> 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`.