diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecoverer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecoverer.java index 9321fa32..5c1eb7d0 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecoverer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecoverer.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -42,7 +42,15 @@ import org.springframework.util.Assert; */ public class RepublishMessageRecoverer implements MessageRecoverer { - private static final Log logger = LogFactory.getLog(RepublishMessageRecoverer.class); + public static final String X_EXCEPTION_STACKTRACE = "x-exception-stacktrace"; + + public static final String X_EXCEPTION_MESSAGE = "x-exception-message"; + + public static final String X_ORIGINAL_EXCHANGE = "x-original-exchange"; + + public static final String X_ORIGINAL_ROUTING_KEY = "x-original-routingKey"; + + private final Log logger = LogFactory.getLog(getClass()); private final AmqpTemplate errorTemplate; @@ -91,10 +99,14 @@ public class RepublishMessageRecoverer implements MessageRecoverer { @Override public void recover(Message message, Throwable cause) { Map headers = message.getMessageProperties().getHeaders(); - headers.put("x-exception-stacktrace", getStackTraceAsString(cause)); - headers.put("x-exception-message", cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage()); - headers.put("x-original-exchange", message.getMessageProperties().getReceivedExchange()); - headers.put("x-original-routingKey", message.getMessageProperties().getReceivedRoutingKey()); + headers.put(X_EXCEPTION_STACKTRACE, getStackTraceAsString(cause)); + headers.put(X_EXCEPTION_MESSAGE, cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage()); + headers.put(X_ORIGINAL_EXCHANGE, message.getMessageProperties().getReceivedExchange()); + headers.put(X_ORIGINAL_ROUTING_KEY, message.getMessageProperties().getReceivedRoutingKey()); + Map additionalHeaders = additionalHeaders(message, cause); + if (additionalHeaders != null) { + headers.putAll(additionalHeaders); + } if (null != errorExchangeName) { String routingKey = errorRoutingKey != null ? errorRoutingKey : this.prefixedOriginalRoutingKey(message); @@ -112,6 +124,16 @@ public class RepublishMessageRecoverer implements MessageRecoverer { } } + /** + * Subclasses can override this method to add more headers to the republished message. + * @param message The failed message. + * @param cause The cause. + * @return A {@link Map} of additional headers to add. + */ + protected Map additionalHeaders(Message message, Throwable cause) { + return null; + } + private String prefixedOriginalRoutingKey(Message message) { return this.errorRoutingKeyPrefix + message.getMessageProperties().getReceivedRoutingKey(); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RetryInterceptorBuilderSupportTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RetryInterceptorBuilderSupportTests.java index f7a58dce..ec478a75 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RetryInterceptorBuilderSupportTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RetryInterceptorBuilderSupportTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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,12 +16,14 @@ package org.springframework.amqp.rabbit.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import java.util.Collections; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -194,15 +196,31 @@ public class RetryInterceptorBuilderSupportTests { AmqpTemplate amqpTemplate = mock(AmqpTemplate.class); RetryOperationsInterceptor interceptor = RetryInterceptorBuilder.stateless() - .recoverer(new RepublishMessageRecoverer(amqpTemplate)) + .recoverer(new RepublishMessageRecoverer(amqpTemplate) { + + @Override + protected Map additionalHeaders(Message message, Throwable cause) { + return Collections.singletonMap("fooHeader", "barValue"); + } + + }) .build(); final AtomicInteger count = new AtomicInteger(); Foo delegate = createDelegate(interceptor, count); - Message message = MessageBuilder.withBody("".getBytes()).setReceivedRoutingKey("foo").build(); + Message message = MessageBuilder + .withBody("".getBytes()) + .setReceivedExchange("exch") + .setReceivedRoutingKey("foo") + .build(); delegate.onMessage("", message); assertEquals(3, count.get()); verify(amqpTemplate).send("error.foo", message); + assertNotNull(message.getMessageProperties().getHeaders().get(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)); + assertNotNull(message.getMessageProperties().getHeaders().get(RepublishMessageRecoverer.X_EXCEPTION_MESSAGE)); + assertNotNull(message.getMessageProperties().getHeaders().get(RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE)); + assertNotNull(message.getMessageProperties().getHeaders().get(RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY)); + assertEquals("barValue", message.getMessageProperties().getHeaders().get("fooHeader")); } @Test