AMQP-489: Enhanced RepublishMessageRecoverer

JIRA: https://jira.spring.io/browse/AMQP-489

Support the addition of other headers to the republished message.

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<String, Object> 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<? extends String, ? extends Object> 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<? extends String, ? extends Object> additionalHeaders(Message message, Throwable cause) {
+		return null;
+	}
+
 	private String prefixedOriginalRoutingKey(Message message) {
 		return this.errorRoutingKeyPrefix + message.getMessageProperties().getReceivedRoutingKey();
 	}

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<? extends String, ? extends Object> 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