diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/AbstractRequestHandlerAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/AbstractRequestHandlerAdvice.java index 8e5bbeafee..5095bb8dd3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/AbstractRequestHandlerAdvice.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/AbstractRequestHandlerAdvice.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -21,6 +21,7 @@ import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.aop.ProxyMethodInvocation; import org.springframework.integration.Message; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageHandler; @@ -33,11 +34,11 @@ import org.springframework.integration.handler.AbstractReplyProducingMessageHand * {@link MessageHandler#handleMessage(Message)} for other message handlers. * * @author Gary Russell + * @author Artem Bilan * @since 2.2 - * */ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupport - implements MethodInterceptor { + implements MethodInterceptor { protected final Log logger = LogFactory.getLog(this.getClass()); @@ -54,7 +55,7 @@ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupp else { Message message = (Message) arguments[0]; try { - return doInvoke(new ExecutionCallback(){ + return doInvoke(new ExecutionCallback() { public Object execute() throws Exception { try { @@ -64,6 +65,26 @@ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupp throw new ThrowableHolderException(e); } } + + public Object cloneAndExecute() throws Exception { + try { + /* + * If we don't copy the invocation carefully it won't keep a reference to the other + * interceptors in the chain. + */ + if (invocation instanceof ProxyMethodInvocation) { + return ((ProxyMethodInvocation) invocation).invocableClone().proceed(); + } + else { + throw new IllegalStateException( + "MethodInvocation of the wrong type detected - this should not happen with Spring AOP," + + " so please raise an issue if you see this exception"); + } + } + catch (Throwable e) { + throw new ThrowableHolderException(e); + } + } }, invocation.getThis(), message); } catch (Exception e) { @@ -80,17 +101,33 @@ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupp /** * Subclasses implement this method to apply behavior to the {@link MessageHandler}.

callback.execute() * invokes the handler method and returns its result, or null. + * * @param callback Subclasses invoke the execute() method on this interface to invoke the handler method. - * @param target The target handler. - * @param message The message that will be sent to the handler. + * @param target The target handler. + * @param message The message that will be sent to the handler. * @return the result after invoking the {@link MessageHandler}. * @throws Exception */ protected abstract Object doInvoke(ExecutionCallback callback, Object target, Message message) throws Exception; + /** + * Called by subclasses in doInvoke() to proceed() the invocation. + * + */ protected interface ExecutionCallback { + /** + * Call this for a normal invocation.proceed(). + */ Object execute() throws Exception; + + /** + * Call this when it is necessary to clone the invocation before + * calling proceed() - such as when the invocation might be called + * multiple times - for example in a retry advice. + */ + Object cloneAndExecute() throws Exception; + } @SuppressWarnings("serial") @@ -99,5 +136,7 @@ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupp public ThrowableHolderException(Throwable cause) { super(cause); } + } -} \ No newline at end of file + +} 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 a89660b799..bed7607600 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -32,12 +32,13 @@ import org.springframework.util.Assert; * exception is thrown but state is maintained to support * the retry policies. Stateful retry requires a * {@link RetryStateGenerator}. - * @author Gary Russell - * @since 2.2 * + * @author Gary Russell + * @author Artem Bilan + * @since 2.2 */ public class RequestHandlerRetryAdvice extends AbstractRequestHandlerAdvice - implements RetryListener { + implements RetryListener { private volatile RetryTemplate retryTemplate = new RetryTemplate(); @@ -80,10 +81,10 @@ public class RequestHandlerRetryAdvice extends AbstractRequestHandlerAdvice messageHolder.set(message); try { - return retryTemplate.execute(new RetryCallback(){ + return retryTemplate.execute(new RetryCallback() { public Object doWithRetry(RetryContext context) throws Exception { try { - return callback.execute(); + return callback.cloneAndExecute(); } catch (MessagingException e) { if (e.getFailedMessage() == null) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java index 5a48f708c3..dfaf4fd01b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -28,6 +28,8 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.aopalliance.aop.Advice; +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; import org.junit.Test; import org.springframework.integration.Message; import org.springframework.integration.MessageHandlingException; @@ -47,8 +49,8 @@ import org.springframework.retry.support.RetryTemplate; /** * @author Gary Russell + * @author Artem Bilan * @since 2.2 - * */ public class AdvisedMessageHandlerTests { @@ -359,7 +361,7 @@ public class AdvisedMessageHandlerTests { } @Test - public void defaultRetrySucceedonThirdTry() { + public void defaultRetrySucceedOnThirdTry() { final AtomicInteger counter = new AtomicInteger(2); AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { @@ -390,7 +392,7 @@ public class AdvisedMessageHandlerTests { } @Test - public void defaultStatefulRetrySucceedonThirdTry() { + public void defaultStatefulRetrySucceedOnThirdTry() { final AtomicInteger counter = new AtomicInteger(2); AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { @@ -484,7 +486,7 @@ public class AdvisedMessageHandlerTests { } private void defaultStatefulRetryRecoverAfterThirdTryGuts(final AtomicInteger counter, - AbstractReplyProducingMessageHandler handler, QueueChannel replies, RequestHandlerRetryAdvice advice) { + AbstractReplyProducingMessageHandler handler, QueueChannel replies, RequestHandlerRetryAdvice advice) { advice.setRecoveryCallback(new RecoveryCallback() { public Object recover(RetryContext context) throws Exception { @@ -576,5 +578,136 @@ public class AdvisedMessageHandlerTests { assertSame(message, ((MessagingException) error.getPayload()).getFailedMessage()); } + @Test + public void testINT2858RetryAdviceAsFirstInAdviceChain() { + final AtomicInteger counter = new AtomicInteger(3); + + AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + return "foo"; + } + }; + + List adviceChain = new ArrayList(); + + adviceChain.add(new RequestHandlerRetryAdvice()); + adviceChain.add(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + counter.getAndDecrement(); + throw new RuntimeException("intentional"); + } + }); + + handler.setAdviceChain(adviceChain); + handler.afterPropertiesSet(); + + try { + handler.handleMessage(new GenericMessage("test")); + } + catch (Exception e) { + Throwable cause = e.getCause(); + assertEquals("ThrowableHolderException", cause.getClass().getSimpleName()); + assertEquals("intentional", cause.getCause().getMessage()); + } + + assertTrue(counter.get() == 0); + } + + @Test + public void testINT2858RetryAdviceAsNestedInAdviceChain() { + final AtomicInteger counter = new AtomicInteger(0); + + AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + return "foo"; + } + }; + + QueueChannel replies = new QueueChannel(); + handler.setOutputChannel(replies); + + List adviceChain = new ArrayList(); + + ExpressionEvaluatingRequestHandlerAdvice expressionAdvice = new ExpressionEvaluatingRequestHandlerAdvice(); +// ThrowableHolderException / MessagingException / ThrowableHolderException / RuntimeException + expressionAdvice.setOnFailureExpression("#exception.cause.cause.cause.message"); + expressionAdvice.setReturnFailureExpressionResult(true); + final AtomicInteger outerCounter = new AtomicInteger(); + adviceChain.add(new AbstractRequestHandlerAdvice() { + + @Override + protected Object doInvoke(ExecutionCallback callback, Object target, Message message) throws Exception { + outerCounter.incrementAndGet(); + return callback.execute(); + } + }); + adviceChain.add(expressionAdvice); + adviceChain.add(new RequestHandlerRetryAdvice()); + adviceChain.add(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + throw new RuntimeException("intentional: " + counter.incrementAndGet()); + } + }); + + handler.setAdviceChain(adviceChain); + handler.afterPropertiesSet(); + + handler.handleMessage(new GenericMessage("test")); + Message receive = replies.receive(1000); + assertNotNull(receive); + assertEquals("intentional: 3", receive.getPayload()); + assertEquals(1, outerCounter.get()); + } + + + @Test + public void testINT2858ExpressionAdviceWithSendFailureOnEachRetry() { + final AtomicInteger counter = new AtomicInteger(0); + + AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { + @Override + protected Object handleRequestMessage(Message requestMessage) { + return "foo"; + } + }; + + QueueChannel errors = new QueueChannel(); + + List adviceChain = new ArrayList(); + + ExpressionEvaluatingRequestHandlerAdvice expressionAdvice = new ExpressionEvaluatingRequestHandlerAdvice(); + expressionAdvice.setOnFailureExpression("#exception.cause.message"); + expressionAdvice.setFailureChannel(errors); + + adviceChain.add(new RequestHandlerRetryAdvice()); + adviceChain.add(expressionAdvice); + adviceChain.add(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + throw new RuntimeException("intentional: " + counter.incrementAndGet()); + } + }); + + handler.setAdviceChain(adviceChain); + handler.afterPropertiesSet(); + + try { + handler.handleMessage(new GenericMessage("test")); + } + catch (Exception e) { + assertEquals("intentional: 3", e.getCause().getCause().getMessage()); + } + + for (int i = 1; i <= 3; i++) { + Message receive = errors.receive(1000); + assertNotNull(receive); + assertEquals("intentional: " + i, ((MessageHandlingExpressionEvaluatingAdviceException) receive.getPayload()).getEvaluationResult()); + } + + assertNull(errors.receive(1)); + + } + } diff --git a/src/reference/docbook/handler-advice.xml b/src/reference/docbook/handler-advice.xml index b561bcac0a..d2b5af55fb 100644 --- a/src/reference/docbook/handler-advice.xml +++ b/src/reference/docbook/handler-advice.xml @@ -66,9 +66,9 @@ standard Advices are provided: - MessageHandlerRetryAdvice - MessageHandlerCircuitBreakerAdvice - ExpressionEvaluatingMessageHandlerAdvice + RequestHandlerRetryAdvice + RequestHandlerCircuitBreakerAdvice + ExpressionEvaluatingRequestHandlerAdvice These are each described in detail in the following sections. @@ -148,7 +148,7 @@ DEBUG [task-scheduler-2]Retry failed last attempt: count=3]]>Simple Stateless Retry with Recovery This example adds a RecoveryCallback to the - above example; it uses a + above example; it uses a ErrorMessageSendingRecoverer to send an ErrorMessage to a channel. @@ -448,5 +448,20 @@ Caused by: java.lang.RuntimeException: foo return result; } }]]> + + + In addition to the execute() method, the ExecutionCallback provides + an additional method cloneAndExecute(). This method must be used in cases where the invocation + might be called multiple times within a single execution of doInvoke(), such as in the + RequestHandlerRetryAdvice. This is required because the Spring AOP + org.springframework.aop.framework.ReflectiveMethodInvocation object maintains state of which + advice in a chain was last invoked; this state must be reset for each call. + + + For more information, see the + + ReflectiveMethodInvocation JavaDocs. + +