Merge pull request #693 from artembilan/INT-2858

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}.<p/> 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);
 		}
+
 	}
-}
+
+}

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<Object>(){
+			return retryTemplate.execute(new RetryCallback<Object>() {
 				public Object doWithRetry(RetryContext context) throws Exception {
 					try {
-						return callback.execute();
+						return callback.cloneAndExecute();
 					}
 					catch (MessagingException e) {
 						if (e.getFailedMessage() == null) {

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<Object>() {
 
 			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<Advice> adviceChain = new ArrayList<Advice>();
+
+		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<String>("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<Advice> adviceChain = new ArrayList<Advice>();
+
+		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<String>("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<Advice> adviceChain = new ArrayList<Advice>();
+
+		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<String>("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));
+
+	}
+
 
 }

src/reference/docbook/handler-advice.xml

@@ -66,9 +66,9 @@
 			standard Advices are provided:
 		</para>
 		<itemizedlist>
-			<listitem>MessageHandlerRetryAdvice</listitem>
-			<listitem>MessageHandlerCircuitBreakerAdvice</listitem>
-			<listitem>ExpressionEvaluatingMessageHandlerAdvice</listitem>
+			<listitem>RequestHandlerRetryAdvice</listitem>
+			<listitem>RequestHandlerCircuitBreakerAdvice</listitem>
+			<listitem>ExpressionEvaluatingRequestHandlerAdvice</listitem>
 		</itemizedlist>
 		<para>
 			These are each described in detail in the following sections.
@@ -148,7 +148,7 @@ DEBUG [task-scheduler-2]Retry failed last attempt: count=3]]></programlisting></
 				<para><emphasis>Simple Stateless Retry with Recovery</emphasis></para>
 				<para>
 					This example adds a <classname>RecoveryCallback</classname> to the
-					above example; it uses a <classname></classname>
+					above example; it uses a <classname>ErrorMessageSendingRecoverer</classname>
 					to send an <emphasis>ErrorMessage</emphasis> to a channel.
 				</para>
 				<para><programlisting><![CDATA[<int:service-activator input-channel="input" ref="failer" method="service">
@@ -448,5 +448,20 @@ Caused by: java.lang.RuntimeException: foo
 		return result;
 	}
 }]]></programlisting></para>
+		<note>
+			<para>
+				In addition to the <code>execute()</code> method, the <interfacename>ExecutionCallback</interfacename> provides
+				an additional method <code>cloneAndExecute()</code>. This method must be used in cases where the invocation
+				might be called multiple times within a single execution of <code>doInvoke()</code>, such as in the
+				<classname>RequestHandlerRetryAdvice</classname>. This is required because the Spring AOP
+				<classname>org.springframework.aop.framework.ReflectiveMethodInvocation</classname> object maintains state of which
+				advice in a chain was last invoked; this state must be reset for each call.
+			</para>
+			<para>
+				For more information, see the
+				<ulink url="http://static.springsource.org/spring-framework/docs/current/javadoc-api/org/springframework/aop/framework/ReflectiveMethodInvocation.html">
+				ReflectiveMethodInvocation</ulink> JavaDocs.
+			</para>
+		</note>
 	</section>
 </section>