INT-2703 SpEL-Based Retry State Generator

Stateful retry requires some state to determine the
retry count for a resubmitted message.

State is provided by a retry state object generated by a
state generator. No standard state generators were
provided by Spring Integration.

Provide a SpEL-Based state generator to create a
RetryState object from a Message (for example, using
the jms message id header).

Add ErrorMessageSendingRecoverer - a RetryCallback
invoked when recovery attempts are exhausted.

Add string-based constructor to Spel Advice for
easier <bean/> configuration.

spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2002-2012 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.integration.handler.advice;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.integration.message.ErrorMessage;
+import org.springframework.retry.RecoveryCallback;
+import org.springframework.retry.RetryContext;
+import org.springframework.util.Assert;
+
+/**
+ * RecoveryCallback that sends the final throwable as an ErrorMessage after
+ * retry exhaustion.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class ErrorMessageSendingRecoverer implements RecoveryCallback<Object> {
+
+	private final static Log logger = LogFactory.getLog(ErrorMessageSendingRecoverer.class);
+
+	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
+
+	public ErrorMessageSendingRecoverer(MessageChannel channel) {
+		Assert.notNull(channel, "channel cannot be null");
+		this.messagingTemplate.setDefaultChannel(channel);
+	}
+
+	public void setSendTimeout(long sendTimeout) {
+		this.messagingTemplate.setSendTimeout(sendTimeout);
+	}
+
+	public Object recover(RetryContext context) throws Exception {
+		Throwable lastThrowable = context.getLastThrowable();
+		if (logger.isDebugEnabled()) {
+			String supplement = "";
+			if (lastThrowable instanceof MessagingException) {
+				supplement = ":failedMessage:" + ((MessagingException) lastThrowable).getFailedMessage();
+			}
+			logger.debug("Sending ErrorMessage " + supplement, lastThrowable);
+		}
+		messagingTemplate.send(new ErrorMessage(lastThrowable));
+		return null;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdvice.java

@@ -19,6 +19,7 @@ import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessageHeaders;
@@ -26,6 +27,7 @@ import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.util.StringUtils;
 
 /**
  * Used to advise {@link MessageHandler}s.
@@ -86,6 +88,16 @@ public class ExpressionEvaluatingRequestHandlerAdvice extends AbstractRequestHan
 
 	}
 
+	public ExpressionEvaluatingRequestHandlerAdvice(String onSuccessExpression, MessageChannel successChannel,
+			String onFailureExpression, MessageChannel failureChannel) {
+		this(!StringUtils.hasText(onSuccessExpression) ? null :
+				new SpelExpressionParser().parseExpression(onSuccessExpression),
+			 successChannel,
+			 !StringUtils.hasText(onFailureExpression) ? null :
+				new SpelExpressionParser().parseExpression(onFailureExpression),
+			 failureChannel);
+	}
+
 	/**
 	 * If true, any exception will be caught and null returned.
 	 * Default false.

spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RequestHandlerRetryAdvice.java

@@ -17,7 +17,6 @@ package org.springframework.integration.handler.advice;
 
 import org.springframework.integration.Message;
 import org.springframework.integration.MessagingException;
-import org.springframework.integration.handler.RetryStateGenerator;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.RetryCallback;
 import org.springframework.retry.RetryContext;

spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RetryStateGenerator.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.integration.handler;
+package org.springframework.integration.handler.advice;
 
 import org.springframework.integration.Message;
 import org.springframework.retry.RetryState;

spring-integration-core/src/main/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGenerator.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2002-2012 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.integration.handler.advice;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.classify.Classifier;
+import org.springframework.context.expression.BeanFactoryResolver;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+import org.springframework.integration.Message;
+import org.springframework.integration.util.ExpressionUtils;
+import org.springframework.retry.RetryState;
+import org.springframework.retry.support.DefaultRetryState;
+import org.springframework.util.Assert;
+
+/**
+ * Creates a DefaultRetryState from a {@link Message}.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class SpelExpressionRetryStateGenerator implements RetryStateGenerator, BeanFactoryAware {
+
+	private final StandardEvaluationContext evaluationContext;
+
+	private final Expression keyExpression;
+
+	private final Expression forceRefreshExpression;
+
+	private volatile Classifier<? super Throwable, Boolean> classifier;
+
+	public SpelExpressionRetryStateGenerator(String keyExpression) {
+		this(keyExpression, null);
+	}
+
+	public SpelExpressionRetryStateGenerator(String keyExpression, String forceRefreshExpression) {
+		Assert.notNull(keyExpression, "keyExpression must not be null");
+		this.keyExpression = new SpelExpressionParser().parseExpression(keyExpression);
+		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext();
+		if (forceRefreshExpression == null) {
+			this.forceRefreshExpression = null;
+		}
+		else {
+			this.forceRefreshExpression = new SpelExpressionParser().parseExpression(forceRefreshExpression);
+		}
+	}
+
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.evaluationContext.setBeanResolver(new BeanFactoryResolver(beanFactory));
+	}
+
+	public void setClassifier(Classifier<? super Throwable, Boolean> classifier) {
+		this.classifier = classifier;
+	}
+
+	public RetryState determineRetryState(Message<?> message) {
+		return new DefaultRetryState(this.keyExpression.getValue(this.evaluationContext, message),
+				this.forceRefreshExpression == null ? false :
+					this.forceRefreshExpression.getValue(this.evaluationContext, message, Boolean.class),
+				this.classifier);
+	}
+}

spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.integration.handler;
+package org.springframework.integration.handler.advice;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -34,9 +34,7 @@ import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.MessageHeaders;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.PollableChannel;
-import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
-import org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice;
-import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.RetryContext;
@@ -75,8 +73,8 @@ public class AdvisedMessageHandlerTests {
 		PollableChannel successChannel = new QueueChannel();
 		PollableChannel failureChannel = new QueueChannel();
 		ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice(
-				new SpelExpressionParser().parseExpression("'foo'"), successChannel,
-				new SpelExpressionParser().parseExpression("'bar'"), failureChannel);
+				"'foo'", successChannel,
+				"'bar'", failureChannel);
 
 		List<Advice> adviceChain = new ArrayList<Advice>();
 		adviceChain.add(advice);
@@ -448,6 +446,35 @@ public class AdvisedMessageHandlerTests {
 			}
 		});
 
+		defaultStatefulRetryRecoverAfterThirdTryGuts(counter, handler, replies, advice);
+
+	}
+
+	@Test
+	public void defaultStatefulRetryRecoverAfterThirdTrySpelState() {
+		final AtomicInteger counter = new AtomicInteger(3);
+		AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
+
+			@Override
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				if (counter.getAndDecrement() > 0) {
+					throw new RuntimeException("foo");
+				}
+				return "bar";
+			}
+		};
+		QueueChannel replies = new QueueChannel();
+		handler.setOutputChannel(replies);
+		RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
+
+		advice.setRetryStateGenerator(new SpelExpressionRetryStateGenerator("headers['id']"));
+
+		defaultStatefulRetryRecoverAfterThirdTryGuts(counter, handler, replies, advice);
+
+	}
+
+	private void defaultStatefulRetryRecoverAfterThirdTryGuts(final AtomicInteger counter,
+			AbstractReplyProducingMessageHandler handler, QueueChannel replies, RequestHandlerRetryAdvice advice) {
 		advice.setRecoveryCallback(new RecoveryCallback<Object>() {
 
 			public Object recover(RetryContext context) throws Exception {
@@ -472,6 +499,5 @@ public class AdvisedMessageHandlerTests {
 		Message<?> reply = replies.receive(1000);
 		assertNotNull(reply);
 		assertEquals("baz", reply.getPayload());
-
 	}
 }

spring-integration-core/src/test/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGeneratorTests-context.xml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<bean id="generator" class="org.springframework.integration.handler.advice.SpelExpressionRetryStateGenerator">
+		<constructor-arg value="headers['foo']"/>
+	</bean>
+</beans>

spring-integration-core/src/test/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGeneratorTests.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2002-2012 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.integration.handler.advice;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.classify.ClassifierSupport;
+import org.springframework.integration.Message;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.retry.RetryState;
+import org.springframework.retry.support.DefaultRetryState;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class SpelExpressionRetryStateGeneratorTests {
+
+	@Autowired
+	private RetryStateGenerator configGenerator;
+
+	private Message<String> message = MessageBuilder.withPayload("Hello, world!")
+				.setHeader("foo", "bar")
+				.setHeader("trueHeader", true)
+				.setHeader("falseHeader", false)
+				.build();
+
+	@Test
+	public void testBasic() {
+		SpelExpressionRetryStateGenerator generator =
+			new SpelExpressionRetryStateGenerator("headers['foo']");
+		RetryState state = generator.determineRetryState(message);
+		assertEquals("bar", state.getKey());
+		assertFalse(((DefaultRetryState) state).isForceRefresh());
+		assertTrue(state.rollbackFor(new RuntimeException()));
+	}
+
+	@Test
+	public void testBasicConfig() {
+		RetryState state = configGenerator.determineRetryState(message);
+		assertEquals("bar", state.getKey());
+		assertFalse(((DefaultRetryState) state).isForceRefresh());
+		assertTrue(state.rollbackFor(new RuntimeException()));
+	}
+
+	@Test
+	public void testForceRefreshTrue() {
+		SpelExpressionRetryStateGenerator generator =
+			new SpelExpressionRetryStateGenerator("headers['foo']", "headers['trueHeader']");
+		RetryState state = generator.determineRetryState(message);
+		assertEquals("bar", state.getKey());
+		assertTrue(((DefaultRetryState) state).isForceRefresh());
+		assertTrue(state.rollbackFor(new RuntimeException()));
+	}
+
+	@Test
+	public void testForceRefreshFalse() {
+		SpelExpressionRetryStateGenerator generator =
+			new SpelExpressionRetryStateGenerator("headers['foo']", "headers['falseHeader']");
+		RetryState state = generator.determineRetryState(message);
+		assertEquals("bar", state.getKey());
+		assertFalse(((DefaultRetryState) state).isForceRefresh());
+		assertTrue(state.rollbackFor(new RuntimeException()));
+	}
+
+	@Test
+	public void testForceRefreshElvis() {
+		SpelExpressionRetryStateGenerator generator =
+			new SpelExpressionRetryStateGenerator("headers['foo']", "headers['noHeader']?:true");
+		RetryState state = generator.determineRetryState(message);
+		assertEquals("bar", state.getKey());
+		assertTrue(((DefaultRetryState) state).isForceRefresh());
+		assertTrue(state.rollbackFor(new RuntimeException()));
+	}
+
+	@Test
+	public void testClassifier() {
+		SpelExpressionRetryStateGenerator generator =
+				new SpelExpressionRetryStateGenerator("headers['foo']");
+		generator.setClassifier(new ClassifierSupport<Throwable, Boolean>(false));
+		RetryState state = generator.determineRetryState(message);
+		assertEquals("bar", state.getKey());
+		assertFalse(((DefaultRetryState) state).isForceRefresh());
+		assertFalse(state.rollbackFor(new RuntimeException()));
+	}
+}