From 1405e3bbf01aba8c83b877120025126993dc2d4c Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 7 Aug 2012 17:24:40 -0400 Subject: [PATCH] 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 configuration. --- .../advice/ErrorMessageSendingRecoverer.java | 63 ++++++++++ ...ressionEvaluatingRequestHandlerAdvice.java | 12 ++ .../advice/RequestHandlerRetryAdvice.java | 1 - .../{ => advice}/RetryStateGenerator.java | 2 +- .../SpelExpressionRetryStateGenerator.java | 78 +++++++++++++ .../AdvisedMessageHandlerTests.java | 40 +++++-- ...essionRetryStateGeneratorTests-context.xml | 9 ++ ...pelExpressionRetryStateGeneratorTests.java | 109 ++++++++++++++++++ 8 files changed, 305 insertions(+), 9 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java rename spring-integration-core/src/main/java/org/springframework/integration/handler/{ => advice}/RetryStateGenerator.java (94%) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGenerator.java rename spring-integration-core/src/test/java/org/springframework/integration/handler/{ => advice}/AdvisedMessageHandlerTests.java (92%) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGeneratorTests-context.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGeneratorTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ErrorMessageSendingRecoverer.java new file mode 100644 index 0000000000..96621c5e58 --- /dev/null +++ b/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 { + + 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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdvice.java index 72a56825af..9bca05badb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdvice.java +++ b/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. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RequestHandlerRetryAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RequestHandlerRetryAdvice.java index 9a7eb59590..837f78d5bf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RequestHandlerRetryAdvice.java +++ b/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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/RetryStateGenerator.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RetryStateGenerator.java similarity index 94% rename from spring-integration-core/src/main/java/org/springframework/integration/handler/RetryStateGenerator.java rename to spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RetryStateGenerator.java index e8430d8d62..a25a172d04 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/RetryStateGenerator.java +++ b/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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGenerator.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGenerator.java new file mode 100644 index 0000000000..6d7f08dbd5 --- /dev/null +++ b/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 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 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); + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/AdvisedMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java similarity index 92% rename from spring-integration-core/src/test/java/org/springframework/integration/handler/AdvisedMessageHandlerTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java index 6e187893b0..23a1b50dfa 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/AdvisedMessageHandlerTests.java +++ b/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 adviceChain = new ArrayList(); 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() { 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()); - } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGeneratorTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGeneratorTests-context.xml new file mode 100644 index 0000000000..e2c0be33c5 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGeneratorTests-context.xml @@ -0,0 +1,9 @@ + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGeneratorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/SpelExpressionRetryStateGeneratorTests.java new file mode 100644 index 0000000000..bb1f33a374 --- /dev/null +++ b/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 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(false)); + RetryState state = generator.determineRetryState(message); + assertEquals("bar", state.getKey()); + assertFalse(((DefaultRetryState) state).isForceRefresh()); + assertFalse(state.rollbackFor(new RuntimeException())); + } +}