From 4464d4d8a5397e90807294f45b33c3063adae6a0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 13 Mar 2018 15:08:19 -0400 Subject: [PATCH] AMQP-798: Master to 2.1.x; Fix Tangles JIRA: https://jira.spring.io/browse/AMQP-798 Also update amqp-client to 5.2.0 Don't use `MessageConverter` in `Message.toString()` (cycle) Fix cycle between template and admin (via containers) Fix outbound reference from support to connection Fix cycle between connection and listener via RabbitUtils Fix cycle between listener and core via ChannelAwareMessageListener and admin Fix cycle between listener.adapter and listener via RabbitListenerErrorHandler Fix cycle between amqp.support and amqp.core via Correlation Fix cycle between listener and transaction via ListenerFailedRuleBasedTransactionAttribute Fix Tests Polishing - PR Comments; rename schema Test polishing - travis failures Polishing according PR comments * Update Copyright for all affected classes * Optimize `DirectMessageListenerContainer.checkMissingQueues()` to cache `amqpAdmin` from the locally created * `RabbitUtils.DEFAULT_PORT` as not-used and deprecated in `2.0.x` * Remove commented test in the `LogAppenderUtils` * Deprecate `AbstractAdaptableMessageListener#setRabbitAdmin()` in favor of newly introduced `setAmqpAdmin()` --- build.gradle | 2 +- gradle.properties | 2 +- .../springframework/amqp/core/AmqpAdmin.java | 10 +- .../amqp/{support => core}/Correlation.java | 4 +- .../springframework/amqp/core/Message.java | 21 +- .../amqp/core/MessagePostProcessor.java | 1 - ...hiteListDeserializingMessageConverter.java | 19 +- .../amqp/utils/SerializationUtils.java | 65 ++++- .../amqp/rabbit/test/TestRabbitTemplate.java | 10 +- .../amqp/rabbit/AsyncRabbitTemplate.java | 7 +- .../rabbit/annotation/RabbitListener.java | 4 +- ...itListenerAnnotationBeanPostProcessor.java | 2 +- .../config/ListenerContainerFactoryBean.java | 38 +-- .../ConnectionFactoryConfigurationUtils.java | 61 +++++ .../amqp/rabbit/connection/RabbitUtils.java | 15 +- .../AbstractMessageListenerContainer.java | 72 ++++-- .../AbstractRabbitListenerEndpoint.java | 20 +- .../listener/BlockingQueueConsumer.java | 8 +- .../amqp/rabbit/listener/ContainerUtils.java | 62 +++++ .../DirectMessageListenerContainer.java | 33 ++- ...DirectReplyToMessageListenerContainer.java | 4 +- ...erFailedRuleBasedTransactionAttribute.java | 4 +- .../MethodRabbitListenerEndpoint.java | 3 +- .../AbstractAdaptableMessageListener.java | 3 +- .../adapter/MessageListenerAdapter.java | 6 +- .../MessagingMessageListenerAdapter.java | 4 +- .../api}/ChannelAwareMessageListener.java | 4 +- .../{ => api}/RabbitListenerErrorHandler.java | 4 +- .../rabbit/listener/api/package-info.java | 4 + .../amqp/rabbit/log4j2/AmqpAppender.java | 6 +- .../amqp/rabbit/logback/AmqpAppender.java | 7 +- .../amqp/rabbit/support/CorrelationData.java | 4 +- .../support/ListenerContainerAware.java | 2 +- .../amqp/rabbit/support/LogAppenderUtils.java | 50 ++-- .../main/resources/META-INF/spring.schemas | 4 +- ...g-rabbit-2.0.xsd => spring-rabbit-2.1.xsd} | 0 .../EnableRabbitIntegrationTests.java | 2 +- ...tenerContainerFactoryIntegrationTests.java | 4 +- ...atePublisherCallbacksIntegrationTests.java | 4 +- .../ContainerInitializationTests.java | 4 +- .../listener/ContainerShutDownTests.java | 46 ++-- ...sageListenerContainerIntegrationTests.java | 2 +- ...tReplyToMessageListenerContainerTests.java | 4 +- .../listener/ExternalTxManagerTests.java | 5 +- .../ListenFromAutoDeleteQueueTests.java | 14 +- .../listener/LocallyTransactedTests.java | 4 +- ...ContainerErrorHandlerIntegrationTests.java | 2 +- ...sageListenerManualAckIntegrationTests.java | 4 +- ...veryCachingConnectionIntegrationTests.java | 2 +- ...istenerRecoveryRepeatIntegrationTests.java | 4 +- ...MessageListenerTxSizeIntegrationTests.java | 4 +- ...ageListenerContainerIntegration2Tests.java | 4 +- ...sageListenerContainerIntegrationTests.java | 2 +- src/reference/asciidoc/amqp.adoc | 4 + src/reference/asciidoc/appendix.adoc | 234 ++++++++++++++++++ src/reference/asciidoc/whats-new.adoc | 170 +------------ 56 files changed, 714 insertions(+), 370 deletions(-) rename spring-amqp/src/main/java/org/springframework/amqp/{support => core}/Correlation.java (89%) create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryConfigurationUtils.java create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ContainerUtils.java rename spring-rabbit/src/main/java/org/springframework/amqp/rabbit/{transaction => listener}/ListenerFailedRuleBasedTransactionAttribute.java (92%) rename spring-rabbit/src/main/java/org/springframework/amqp/rabbit/{core => listener/api}/ChannelAwareMessageListener.java (92%) rename spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/{ => api}/RabbitListenerErrorHandler.java (93%) create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/package-info.java rename spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/{spring-rabbit-2.0.xsd => spring-rabbit-2.1.xsd} (100%) 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`.