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

AMQP-431: Add RabbitTemplate.recoveryCallback

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

AMQP-431: Address PR comments

Doc Polishing
This commit is contained in:
Artem Bilan
2014-10-09 14:03:14 +03:00
committed by Gary Russell
parent fdd5ac9b20
commit 421152d2de
8 changed files with 165 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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,

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

@@ -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<Object, SortedMap<Long, PendingConfirm>> pendingConfirms =
new ConcurrentHashMap<Object, SortedMap<Long, PendingConfirm>>();
private final Map<String, PendingReply> replyHolder = new ConcurrentHashMap<String, PendingReply>();
private final String uuid = UUID.randomUUID().toString();
private final StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
private final ReplyToAddressCallback<?> defaultReplyToAddressCallback = new ReplyToAddressCallback<Object>() {
@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<String, PendingReply> replyHolder = new ConcurrentHashMap<String, PendingReply>();
private volatile ConfirmCallback confirmCallback;
private volatile ReturnCallback returnCallback;
private final ConcurrentHashMap<Object, SortedMap<Long, PendingConfirm>> pendingConfirms =
new ConcurrentHashMap<Object, SortedMap<Long, PendingConfirm>>();
private volatile Expression mandatoryExpression = new ValueExpression<Boolean>(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<Object>() {
@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> T execute(final ChannelCallback<T> 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<T>) this.recoveryCallback);
}
catch (Exception e) {
if (e instanceof RuntimeException) {

12
spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.4.xsd
View File

@@ -1048,6 +1048,18 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="recovery-callback" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Reference to a RecoveryCallback - used in the 'retryTemplate.execute()'. Ignored if 'retry-template' isn't provided.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.retry.RecoveryCallback" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="send-connection-factory-selector-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[

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

@@ -34,6 +34,7 @@ import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.beans.factory.xml.XmlBeanDefinitionReader;
import org.springframework.core.io.ClassPathResource;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.support.RetryTemplate;
/**
@@ -90,6 +91,7 @@ public final class TemplateParserTests {
DirectFieldAccessor accessor = new DirectFieldAccessor(template);
assertEquals("foo", accessor.getPropertyValue("correlationKey"));
assertSame(beanFactory.getBean(RetryTemplate.class), accessor.getPropertyValue("retryTemplate"));
assertSame(beanFactory.getBean(RecoveryCallback.class), accessor.getPropertyValue("recoveryCallback"));
}
@Test

36
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java
View File

@@ -20,6 +20,7 @@ import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@@ -28,6 +29,7 @@ import static org.mockito.Mockito.when;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -48,6 +50,8 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.amqp.utils.SerializationUtils;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionException;
@@ -215,6 +219,38 @@ public class RabbitTemplateTests {
assertEquals(3, count.get());
}
@Test
public void testRecovery() throws Exception {
ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);
final AtomicInteger count = new AtomicInteger();
doAnswer(new Answer<Void>() {
@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<Object>() {
@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

49
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml
View File

@@ -1,44 +1,53 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<rabbit:template id="template" connection-factory="connectionFactory" />
<rabbit:template id="template" connection-factory="connectionFactory"/>
<rabbit:template id="kitchenSink" connection-factory="connectionFactory" channel-transacted="true" correlation-key="foo"
encoding="UTF-8" exchange="foo" queue="bar" routing-key="spam" message-converter="converter" reply-timeout="1000"
retry-template="retrier" />
<rabbit:template id="kitchenSink" connection-factory="connectionFactory" channel-transacted="true"
correlation-key="foo"
encoding="UTF-8" exchange="foo" queue="bar" routing-key="spam" message-converter="converter"
reply-timeout="1000"
retry-template="retrier"
recovery-callback="recoverer"/>
<rabbit:connection-factory id="connectionFactory" />
<rabbit:connection-factory id="connectionFactory"/>
<bean id="converter" class="org.springframework.amqp.support.converter.SerializerMessageConverter"/>
<rabbit:template id="withReplyQ" connection-factory="connectionFactory" reply-queue="replyQId" correlation-key="correlationId">
<rabbit:reply-listener />
<rabbit:template id="withReplyQ" connection-factory="connectionFactory" reply-queue="replyQId"
correlation-key="correlationId">
<rabbit:reply-listener/>
</rabbit:template>
<rabbit:queue id="replyQId" name="reply.queue" queue-arguments="args" />
<rabbit:queue id="replyQId" name="reply.queue" queue-arguments="args"/>
<rabbit:queue-arguments id="args">
<entry key="foo" value="bar" />
<entry key="foo" value="bar"/>
</rabbit:queue-arguments>
<rabbit:template id="withCallbacks" connection-factory="connectionFactory"
mandatory="true" return-callback="rcb" confirm-callback="ccb" />
mandatory="true" return-callback="rcb" confirm-callback="ccb"/>
<rabbit:template id="withMandatoryExpression" connection-factory="connectionFactory"
mandatory-expression="'true'"
send-connection-factory-selector-expression="'foo'"
receive-connection-factory-selector-expression="'foo'"/>
<beans:bean id="rcb" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.amqp.rabbit.core.RabbitTemplate$ReturnCallback" />
</beans:bean>
<bean id="rcb" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.amqp.rabbit.core.RabbitTemplate$ReturnCallback"/>
</bean>
<beans:bean id="ccb" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.amqp.rabbit.core.RabbitTemplate$ConfirmCallback" />
</beans:bean>
<bean id="ccb" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.amqp.rabbit.core.RabbitTemplate$ConfirmCallback"/>
</bean>
<bean id="retrier" class="org.springframework.retry.support.RetryTemplate"/>
<bean id="recoverer" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.retry.RecoveryCallback"/>
</bean>
<beans:bean id="retrier" class="org.springframework.retry.support.RetryTemplate" />
</beans>

36
src/reference/docbook/amqp.xml
View File

@@ -562,6 +562,42 @@ public AmqpTemplate rabbitTemplate();
template.setRetryTemplate(retryTemplate);
return template;
}]]></programlisting>
<para>
Starting with <emphasis>version 1.4</emphasis>, in addition to the <code>retryTemplate</code> property,
the <code>recoveryCallback</code> option is supported on the <classname>RabbitTemplate</classname>.
It is used as a second argument for the
<code>RetryTemplate.execute(RetryCallback&lt;T, E&gt; retryCallback,
RecoveryCallback&lt;T&gt;recoveryCallback)</code>.
</para>
<note>
The <interfacename>RecoveryCallback</interfacename> is somewhat limited in that the retry context
only contains the <code>lastThrowable</code> field. For more sophisticated use cases, you should
use an external <classname>RetryTemplate</classname> so that you can convey additional information
to the <interfacename>RecoveryCallback</interfacename> via the context's attributes:
</note>
<programlisting language="java"><![CDATA[retryTemplate.execute(
new RetryCallback<Object, Exception>() {
@Override
public Object doWithRetry(RetryContext context) throws Exception {
context.setAttribute("message", message);
return rabbitTemplate.convertAndSend(exchange, routingKey, message);
}
}, new RecoveryCallback<Object>() {
@Override
public Object recover(RetryContext context) throws Exception {
Object message = context.getAttribute("message");
Throwable t = context.getLastThrowable();
// Do something with message
return null;
}
});
}]]></programlisting>
<para>
In this case, you would <emphasis role="bold">not</emphasis> inject a <classname>RetryTemplate</classname>
into the <classname>RabbitTemplate</classname>.
</para>
</section>
<section id="template-confirms">
<title>Publisher Confirms and Returns</title>

8
src/reference/docbook/whats-new.xml
View File

@@ -112,6 +112,14 @@
See <xref linkend="routing-connection-factory"/>.
</para>
</section>
<section>
<title>RabbitTemplate: RecoveryCallback option</title>
<para>
The <code>recoveryCallback</code> property has been added to be used in the
<code>retryTemplate.execute()</code>.
See <xref linkend="template-retry"/>.
</para>
</section>
</section>
<section>