From 421152d2dea8b92344dd04cc8d71177e1ef7415f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 9 Oct 2014 14:03:14 +0300 Subject: [PATCH] AMQP-431: Add `RabbitTemplate.recoveryCallback` JIRA: https://jira.spring.io/browse/AMQP-431 AMQP-431: Address PR comments Doc Polishing --- .../amqp/rabbit/config/TemplateParser.java | 3 + .../amqp/rabbit/core/RabbitTemplate.java | 62 ++++++++++++------- .../amqp/rabbit/config/spring-rabbit-1.4.xsd | 12 ++++ .../rabbit/config/TemplateParserTests.java | 2 + .../amqp/rabbit/core/RabbitTemplateTests.java | 36 +++++++++++ .../config/TemplateParserTests-context.xml | 49 +++++++++------ src/reference/docbook/amqp.xml | 36 +++++++++++ src/reference/docbook/whats-new.xml | 8 +++ 8 files changed, 165 insertions(+), 43 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java index ec613f3f..b9083c7e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java @@ -63,6 +63,8 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { private static final String RETRY_TEMPLATE = "retry-template"; + private static final String RECOVERY_CALLBACK = "recovery-callback"; + @Override protected Class getBeanClass(Element element) { return RabbitTemplate.class; @@ -104,6 +106,7 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { NamespaceUtils.setReferenceIfAttributeDefined(builder, element, CONFIRM_CALLBACK_ATTRIBUTE); NamespaceUtils.setValueIfAttributeDefined(builder, element, CORRELATION_KEY); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, RETRY_TEMPLATE); + NamespaceUtils.setReferenceIfAttributeDefined(builder, element, RECOVERY_CALLBACK); BeanDefinition expressionDef = NamespaceUtils.createExpressionDefinitionFromValueOrExpression(MANDATORY_ATTRIBUTE, diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index b24b2f37..1bafac5d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -63,6 +63,7 @@ import org.springframework.context.expression.BeanFactoryResolver; import org.springframework.context.expression.MapAccessor; import org.springframework.expression.Expression; import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.retry.RecoveryCallback; import org.springframework.retry.RetryCallback; import org.springframework.retry.RetryContext; import org.springframework.retry.support.RetryTemplate; @@ -113,8 +114,8 @@ import com.rabbitmq.client.GetResponse; * @author Artem Bilan * @since 1.0 */ -public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, RabbitOperations, MessageListener, - PublisherCallbackChannel.Listener { +public class RabbitTemplate extends RabbitAccessor + implements BeanFactoryAware, RabbitOperations, MessageListener, PublisherCallbackChannel.Listener { /** Alias for amq.direct default exchange */ private static final String DEFAULT_EXCHANGE = ""; @@ -125,6 +126,24 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, private static final String DEFAULT_ENCODING = "UTF-8"; + private final ConcurrentHashMap> pendingConfirms = + new ConcurrentHashMap>(); + + private final Map replyHolder = new ConcurrentHashMap(); + + private final String uuid = UUID.randomUUID().toString(); + + private final StandardEvaluationContext evaluationContext = new StandardEvaluationContext(); + + private final ReplyToAddressCallback defaultReplyToAddressCallback = new ReplyToAddressCallback() { + + @Override + public Address getReplyToAddress(Message request, Object reply) { + return RabbitTemplate.this.getReplyToAddress(request); + } + + }; + private volatile String exchange = DEFAULT_EXCHANGE; private volatile String routingKey = DEFAULT_ROUTING_KEY; @@ -142,38 +161,22 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, private volatile Queue replyQueue; - private final Map replyHolder = new ConcurrentHashMap(); - private volatile ConfirmCallback confirmCallback; private volatile ReturnCallback returnCallback; - private final ConcurrentHashMap> pendingConfirms = - new ConcurrentHashMap>(); - private volatile Expression mandatoryExpression = new ValueExpression(false); - private final String uuid = UUID.randomUUID().toString(); - private volatile String correlationKey = null; private volatile RetryTemplate retryTemplate; + private volatile RecoveryCallback recoveryCallback; + private volatile Expression sendConnectionFactorySelectorExpression; private volatile Expression receiveConnectionFactorySelectorExpression; - private final StandardEvaluationContext evaluationContext = new StandardEvaluationContext(); - - private final ReplyToAddressCallback defaultReplyToAddressCallback = new ReplyToAddressCallback() { - - @Override - public Address getReplyToAddress(Message request, Object reply) { - return RabbitTemplate.this.getReplyToAddress(request); - } - - }; - /** * Convenient constructor for use with setter injection. Don't forget to set the connection factory. */ @@ -399,6 +402,18 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, this.retryTemplate = retryTemplate; } + /** + * Add a {@link RecoveryCallback} which is used for the {@code retryTemplate.execute}. + * If {@link #retryTemplate} isn't provided {@link #recoveryCallback} is ignored. + * {@link RecoveryCallback} should produce result compatible with + * {@link #execute(ChannelCallback, ConnectionFactory)} return type. + * @param recoveryCallback The retry recoveryCallback. + * @since 1.4 + */ + public void setRecoveryCallback(RecoveryCallback recoveryCallback) { + this.recoveryCallback = recoveryCallback; + } + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { this.evaluationContext.setBeanResolver(new BeanFactoryResolver(beanFactory)); @@ -406,8 +421,8 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, } /** - * Gets unconfirmed correlatiom data older than age and removes them. - * @param age in millseconds + * Gets unconfirmed correlation data older than age and removes them. + * @param age in milliseconds * @return the collection of correlation data for which confirms have * not been received. */ @@ -906,6 +921,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, return execute(action, null); } + @SuppressWarnings("unchecked") private T execute(final ChannelCallback action, final ConnectionFactory connectionFactory) { if (this.retryTemplate != null) { try { @@ -916,7 +932,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, return RabbitTemplate.this.doExecute(action, connectionFactory); } - }); + }, (RecoveryCallback) this.recoveryCallback); } catch (Exception e) { if (e instanceof RuntimeException) { diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.4.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.4.xsd index b5e4136e..2925c1de 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.4.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.4.xsd @@ -1048,6 +1048,18 @@ + + + + + + + + + + () { + + @Override + public Void answer(InvocationOnMock invocation) throws Throwable { + count.incrementAndGet(); + throw new AuthenticationFailureException("foo"); + } + }).when(mockConnectionFactory).newConnection((ExecutorService) null); + + RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + template.setRetryTemplate(new RetryTemplate()); + + final AtomicBoolean recoverInvoked = new AtomicBoolean(); + + template.setRecoveryCallback(new RecoveryCallback() { + + @Override + public Object recover(RetryContext context) throws Exception { + recoverInvoked.set(true); + return null; + } + + }); + template.convertAndSend("foo", "bar", "baz"); + assertEquals(3, count.get()); + assertTrue(recoverInvoked.get()); + } + public final static AtomicInteger LOOKUP_KEY_COUNT = new AtomicInteger(); @Test diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml index 5e02af86..dc6ed3f8 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml @@ -1,44 +1,53 @@ - - + - + - + - - + + - + - + + mandatory="true" return-callback="rcb" confirm-callback="ccb"/> - - - + + + - - - + + + + + + + + + - diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index 16fb062b..824849aa 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -562,6 +562,42 @@ public AmqpTemplate rabbitTemplate(); template.setRetryTemplate(retryTemplate); return template; }]]> + + Starting with version 1.4, in addition to the retryTemplate property, + the recoveryCallback option is supported on the RabbitTemplate. + It is used as a second argument for the + RetryTemplate.execute(RetryCallback<T, E> retryCallback, + RecoveryCallback<T>recoveryCallback). + + + The RecoveryCallback is somewhat limited in that the retry context + only contains the lastThrowable field. For more sophisticated use cases, you should + use an external RetryTemplate so that you can convey additional information + to the RecoveryCallback via the context's attributes: + + () { + + @Override + public Object doWithRetry(RetryContext context) throws Exception { + context.setAttribute("message", message); + return rabbitTemplate.convertAndSend(exchange, routingKey, message); + } + }, new RecoveryCallback() { + + @Override + public Object recover(RetryContext context) throws Exception { + Object message = context.getAttribute("message"); + Throwable t = context.getLastThrowable(); + // Do something with message + return null; + } + }); +}]]> + + In this case, you would not inject a RetryTemplate + into the RabbitTemplate. +
Publisher Confirms and Returns diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 2ce2cc6b..f64211db 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -112,6 +112,14 @@ See .
+
+ RabbitTemplate: RecoveryCallback option + + The recoveryCallback property has been added to be used in the + retryTemplate.execute(). + See . + +