INT-2214, INT-343, INT-2250 MessageHandler Advice

Add general capability to advise just the handleRequestMessage
part of an AbstractReplyProducingMessageHandler.

This is to advise just the immediate operation, and not the
entire downstream flow.

Uses include:

* outbound gateway post processing
* adding retry behavior using spring-retry
* adding circuit breaker functionality

Initial commit for review.

Also need to advise simple message handlers (such as file
etc) to allow them to post-process file operations
with payload.delete(), payload.renameTo(...) etc.

INT-2250 Add Circuit Breaker Advice

INT-343 Add Retry Advice

Stateless and Stateful retry using spring-retry. Stateless
means the RetryTemplate performs the retries internally.
Stateful means the exception is thrown (e.g. to JMS container)
and the retry state is maintained by spring-retry.

INT-2215, INT-343, INT-2250 Refactoring

Factor out common abstract Advice class.

INT-2214 Catch Evaluation Expression Exceptions

If an onSuccess expression evaluation fails, add an
option so the user can decide whether such an exception is
caught, or propagated to the caller.

INT-2214 etc PR Review Polishing

INT-2214 etc Namespace Core, File, FTP

Add <request-handler-advice-chain/> to outbound endpoints.

INT-2214 etc. More Namespace Support

amqp, event, gemfire, groovy, http, ip, jdbc, jms, jmx, jpa, mail, rmi, sftp, twitter, ws, xmpp

INT-2214 etc Polishing

PR Review

INT-2214 etc Polishing

Don't catch Throwable.

Move Advice classes to handler.advice package.

spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java

@@ -91,6 +91,8 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
 
 	public static final String DISPOSITION_RESULT = "dispositionResult";
 
+	public static final String POSTPROCESS_RESULT = "postProcessResult";
+
 
 	private final Map<String, Object> headers;
 
@@ -164,12 +166,12 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
 	}
 
 	@Override
-	public boolean equals(Object obj) {
-		if (this == obj) {
+	public boolean equals(Object object) {
+		if (this == object) {
 			return true;
 		}
-		if (obj != null && obj instanceof MessageHeaders) {
-			MessageHeaders other = (MessageHeaders) obj;
+		if (object != null && object instanceof MessageHeaders) {
+			MessageHeaders other = (MessageHeaders) object;
 			return this.headers.equals(other.headers);
 		}
 		return false;

spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java

@@ -1,11 +1,11 @@
 /*
- * Copyright 2002-2011 the original author or authors.
- * 
+ * 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.
@@ -13,6 +13,9 @@
 
 package org.springframework.integration.config;
 
+import java.util.List;
+
+import org.aopalliance.aop.Advice;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
@@ -23,11 +26,14 @@ import org.springframework.integration.MessageChannel;
 import org.springframework.integration.context.Orderable;
 import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 
 /**
  * @author Dave Syer
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public abstract class AbstractSimpleMessageHandlerFactoryBean<H extends MessageHandler> implements FactoryBean<MessageHandler>, BeanFactoryAware {
 
@@ -43,6 +49,8 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean<H extends MessageH
 
 	private final Object initializationMonitor = new Object();
 
+	private volatile List<Advice> adviceChain;
+
 
 	public AbstractSimpleMessageHandlerFactoryBean() {
 		super();
@@ -64,6 +72,10 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean<H extends MessageH
 		return this.beanFactory;
 	}
 
+	public void setAdviceChain(List<Advice> adviceChain) {
+		this.adviceChain = adviceChain;
+	}
+
 	public H getObject() throws Exception {
 		if (this.handler == null) {
 			this.handler = this.createHandlerInternal();
@@ -71,9 +83,6 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean<H extends MessageH
 			if (this.handler instanceof MessageProducer && this.outputChannel != null) {
 				((MessageProducer) this.handler).setOutputChannel(this.outputChannel);
 			}
-			if (this.handler instanceof BeanFactoryAware) {
-				((BeanFactoryAware) this.handler).setBeanFactory(beanFactory);
-			}
 			if (this.handler instanceof Orderable && this.order != null) {
 				((Orderable) this.handler).setOrder(this.order.intValue());
 			}
@@ -91,6 +100,10 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean<H extends MessageH
 			if (handler instanceof BeanFactoryAware) {
 				((BeanFactoryAware) handler).setBeanFactory(getBeanFactory());
 			}
+			if (!CollectionUtils.isEmpty(this.adviceChain) &&
+					this.handler instanceof AbstractReplyProducingMessageHandler) {
+				((AbstractReplyProducingMessageHandler) this.handler).setAdviceChain(this.adviceChain);
+			}
 			this.initialized = true;
 		}
 		if (handler instanceof InitializingBean) {

spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * 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.
@@ -15,10 +15,15 @@
  */
 package org.springframework.integration.config;
 
+import java.util.List;
+
+import org.aopalliance.aop.Advice;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.aop.framework.Advised;
+import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.AopUtils;
+import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
@@ -36,8 +41,10 @@ import org.springframework.integration.core.SubscribableChannel;
 import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.endpoint.PollingConsumer;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -56,13 +63,13 @@ public class ConsumerEndpointFactoryBean
 	private volatile String inputChannelName;
 
 	private volatile PollerMetadata pollerMetadata;
-	
+
 	private volatile boolean autoStartup = true;
 
 	private volatile MessageChannel inputChannel;
 
 	private volatile ConfigurableBeanFactory beanFactory;
-	
+
 	private volatile ClassLoader beanClassLoader;
 
 	private volatile AbstractEndpoint endpoint;
@@ -75,6 +82,8 @@ public class ConsumerEndpointFactoryBean
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
+	private volatile List<Advice> adviceChain;
+
 	public void setHandler(MessageHandler handler) {
 		Assert.notNull(handler, "handler must not be null");
 		synchronized (this.handlerMonitor) {
@@ -94,7 +103,7 @@ public class ConsumerEndpointFactoryBean
 	public void setPollerMetadata(PollerMetadata pollerMetadata) {
 		this.pollerMetadata = pollerMetadata;
 	}
-	
+
 	public void setBeanClassLoader(ClassLoader classLoader) {
 		this.beanClassLoader = classLoader;
 	}
@@ -112,6 +121,11 @@ public class ConsumerEndpointFactoryBean
 		this.beanFactory = (ConfigurableBeanFactory) beanFactory;
 	}
 
+	public void setAdviceChain(List<Advice> adviceChain) {
+		Assert.notNull(adviceChain, "adviceChain must not be null");
+		this.adviceChain = adviceChain;
+	}
+
 	public void afterPropertiesSet() throws Exception {
 		try {
 			if (!this.beanName.startsWith("org.springframework")) {
@@ -132,6 +146,32 @@ public class ConsumerEndpointFactoryBean
 						+ this.handler + " for " + this.beanName + " :" + e.getMessage());
 			}
 		}
+		if (!CollectionUtils.isEmpty(this.adviceChain)) {
+			/*
+			 *  ARPMHs advise the handleRequesMessage method internally and already have the advice chain injected.
+			 *  So we only advise handlers that are not reply-producing. If the handler is already advised,
+			 *  add the configured advices to its chain, otherwise create a proxy.
+			 */
+			if (!(this.handler instanceof AbstractReplyProducingMessageHandler)) {
+				if (AopUtils.isAopProxy(this.handler) && this.handler instanceof Advised) {
+					Class<?> targetClass = AopUtils.getTargetClass(this.handler);
+					for (Advice advice : this.adviceChain) {
+						NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(advice);
+						handlerAdvice.addMethodName("handleMessage");
+						if (AopUtils.canApply(handlerAdvice.getPointcut(), targetClass)) {
+							((Advised) this.handler).addAdvice(advice);
+						}
+					}
+				}
+				else {
+					ProxyFactory proxyFactory = new ProxyFactory(this.handler);
+					for (Advice advice : this.adviceChain) {
+						proxyFactory.addAdvice(advice);
+					}
+					this.handler = (MessageHandler) proxyFactory.getProxy(this.beanClassLoader);
+				}
+			}
+		}
 		this.initializeEndpoint();
 	}
 
@@ -184,9 +224,9 @@ public class ConsumerEndpointFactoryBean
 				pollingConsumer.setTrigger(this.pollerMetadata.getTrigger());
 				pollingConsumer.setAdviceChain(this.pollerMetadata.getAdviceChain());
 				pollingConsumer.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
-				
+
 				pollingConsumer.setErrorHandler(this.pollerMetadata.getErrorHandler());
-				
+
 				pollingConsumer.setReceiveTimeout(this.pollerMetadata.getReceiveTimeout());
 				pollingConsumer.setBeanClassLoader(beanClassLoader);
 				pollingConsumer.setBeanFactory(beanFactory);

spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java

@@ -19,8 +19,7 @@ package org.springframework.integration.config.xml;
 import java.util.Collection;
 import java.util.List;
 
-import org.w3c.dom.Element;
-
+import org.springframework.beans.factory.BeanDefinitionStoreException;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.ConstructorArgumentValues;
 import org.springframework.beans.factory.config.ConstructorArgumentValues.ValueHolder;
@@ -32,10 +31,10 @@ import org.springframework.beans.factory.support.ManagedSet;
 import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
-import org.springframework.beans.factory.BeanDefinitionStoreException;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
+import org.w3c.dom.Element;
 
 /**
  * Base class parser for elements that create Message Endpoints.
@@ -92,6 +91,11 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
 			return handlerBeanDefinition;
 		}
 
+		Element adviceChainElement = DomUtils.getChildElementByTagName(element,
+				IntegrationNamespaceUtils.REQUEST_HANDLER_ADVICE_CHAIN);
+		IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(adviceChainElement, null,
+				handlerBuilder, parserContext);
+
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ConsumerEndpointFactoryBean.class);
 
 		String handlerBeanName = BeanDefinitionReaderUtils.generateBeanName(handlerBeanDefinition, parserContext.getRegistry());
@@ -99,6 +103,7 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
 		parserContext.registerBeanComponent(new BeanComponentDefinition(handlerBeanDefinition, handlerBeanName, handlerAlias));
 
 		builder.addPropertyReference("handler", handlerBeanName);
+
 		String inputChannelName = element.getAttribute(inputChannelAttributeName);
 
 		if (!parserContext.getRegistry().containsBeanDefinition(inputChannelName)){
@@ -136,4 +141,5 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
 		parserContext.registerBeanComponent(new BeanComponentDefinition(beanDefinition, beanName));
 		return null;
 	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractOutboundChannelAdapterParser.java

@@ -16,16 +16,18 @@
 
 package org.springframework.integration.config.xml;
 
-import org.w3c.dom.Element;
-
+import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.parsing.BeanComponentDefinition;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
+import org.w3c.dom.Element;
 
 /**
  * Base class for outbound Channel Adapter parsers.
@@ -34,7 +36,7 @@ import org.springframework.util.xml.DomUtils;
  * an {@link org.springframework.integration.endpoint.AbstractEndpoint} depending on the channel type.
  * If this component is defined as nested element (e.g., inside of the chain) it will produce
  * a {@link org.springframework.integration.core.MessageHandler}.
- * 
+ *
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
@@ -48,7 +50,8 @@ public abstract class AbstractOutboundChannelAdapterParser extends AbstractChann
 		}
 		BeanDefinitionBuilder builder =  BeanDefinitionBuilder.genericBeanDefinition(ConsumerEndpointFactoryBean.class);
 		Element pollerElement = DomUtils.getChildElementByTagName(element, "poller");
-		builder.addPropertyReference("handler", this.parseAndRegisterConsumer(element, parserContext));
+		BeanComponentDefinition handlerBeanComponentDefinition = this.doParseAndRegisterConsumer(element, parserContext);
+		builder.addPropertyReference("handler", handlerBeanComponentDefinition.getBeanName());
 		if (pollerElement != null) {
 			if (!StringUtils.hasText(channelName)) {
 				parserContext.getReaderContext().error(
@@ -58,6 +61,40 @@ public abstract class AbstractOutboundChannelAdapterParser extends AbstractChann
 		}
 		builder.addPropertyValue("inputChannelName", channelName);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
+
+		Element adviceChainElement = DomUtils.getChildElementByTagName(element,
+				IntegrationNamespaceUtils.REQUEST_HANDLER_ADVICE_CHAIN);
+		@SuppressWarnings("rawtypes")
+		ManagedList adviceChain = IntegrationNamespaceUtils.configureAdviceChain(adviceChainElement, null,
+				builder, parserContext);
+		if (adviceChain != null) {
+			BeanDefinition handlerBeanDefinition = handlerBeanComponentDefinition.getBeanDefinition();
+			/*
+			 * For ARPMH, the advice chain is injected so just the handleRequestMessage method is advised.
+			 * Sometime ARPMHs do double duty as a gateway and a channel adapter. The parser subclass
+			 * can indicate this by overriding isUsingReplyProducer(), or we can try to determine it from
+			 * the bean class.
+			 */
+			boolean isReplyProducer = this.isUsingReplyProducer();
+			if (!isReplyProducer) {
+				Class<?> beanClass = null;
+				if (handlerBeanDefinition instanceof AbstractBeanDefinition) {
+					AbstractBeanDefinition abstractBeanDefinition = (AbstractBeanDefinition) handlerBeanDefinition;
+					if (abstractBeanDefinition.hasBeanClass()) {
+						beanClass = abstractBeanDefinition.getBeanClass();
+					}
+				}
+				isReplyProducer = beanClass != null && AbstractReplyProducingMessageHandler.class.isAssignableFrom(beanClass);
+			}
+
+			if (isReplyProducer) {
+				handlerBeanDefinition.getPropertyValues().add("adviceChain", adviceChain);
+			}
+			else {
+				builder.addPropertyValue("adviceChain", adviceChain);
+			}
+		}
+
 		return builder.getBeanDefinition();
 	}
 
@@ -65,8 +102,19 @@ public abstract class AbstractOutboundChannelAdapterParser extends AbstractChann
 	 * Override this method to control the registration process and return the bean name.
 	 * If parsing a bean definition whose name can be auto-generated, consider using
 	 * {@link #parseConsumer(Element, ParserContext)} instead.
+	 * @deprecated Use {@link #doParseAndRegisterConsumer(Element, ParserContext)}
 	 */
+	@Deprecated
 	protected String parseAndRegisterConsumer(Element element, ParserContext parserContext) {
+		return doParseAndRegisterConsumer(element, parserContext).getBeanName();
+	}
+
+	/**
+	 * Override this method to control the registration process and return the bean name.
+	 * If parsing a bean definition whose name can be auto-generated, consider using
+	 * {@link #parseConsumer(Element, ParserContext)} instead.
+	 */
+	protected BeanComponentDefinition doParseAndRegisterConsumer(Element element, ParserContext parserContext) {
 		AbstractBeanDefinition definition = this.parseConsumer(element, parserContext);
 		if (definition == null) {
 			parserContext.getReaderContext().error("Consumer parsing must return an AbstractBeanDefinition.", element);
@@ -77,8 +125,9 @@ public abstract class AbstractOutboundChannelAdapterParser extends AbstractChann
 		}
 		String beanName = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry());
 		String[] handlerAlias = IntegrationNamespaceUtils.generateAlias(element);
-		parserContext.registerBeanComponent(new BeanComponentDefinition(definition, beanName, handlerAlias));
-		return beanName;
+		BeanComponentDefinition beanComponentDefinition = new BeanComponentDefinition(definition, beanName, handlerAlias);
+		parserContext.registerBeanComponent(beanComponentDefinition);
+		return beanComponentDefinition;
 	}
 
 	/**
@@ -87,4 +136,13 @@ public abstract class AbstractOutboundChannelAdapterParser extends AbstractChann
 	 */
 	protected abstract AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext);
 
+	/**
+	 * Override this to signal that this channel adapter is actually using a AbstractReplyProducingMessageHandler
+	 * while it is not possible for this parser to determine that because, say, a FactoryBean is being used.
+	 * @return false, unless overridden.
+	 */
+	protected boolean isUsingReplyProducer() {
+		return false;
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java

@@ -19,9 +19,11 @@ import java.util.List;
 
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.BeanDefinitionHolder;
+import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.config.TypedStringValue;
 import org.springframework.beans.factory.parsing.BeanComponentDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.core.Conventions;
@@ -33,6 +35,8 @@ import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
 import org.w3c.dom.Element;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
 
 /**
  * Shared utility methods for integration namespace parsers.
@@ -54,6 +58,7 @@ public abstract class IntegrationNamespaceUtils {
 	static final String ORDER = "order";
 	static final String EXPRESSION_ATTRIBUTE = "expression";
 	public static final String HANDLER_ALIAS_SUFFIX = ".handler";
+	public static final String REQUEST_HANDLER_ADVICE_CHAIN = "request-handler-advice-chain";
 
 	/**
 	 * Property name on ChannelInitializer used to configure the default max subscribers for
@@ -308,4 +313,56 @@ public abstract class IntegrationNamespaceUtils {
 		}
 		return handlerAlias;
 	}
+
+	@SuppressWarnings({ "rawtypes" })
+	public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement, Element txElement,
+			BeanDefinitionBuilder parentBuilder, ParserContext parserContext) {
+		ManagedList adviceChain = configureAdviceChain(adviceChainElement, txElement, parentBuilder, parserContext);
+		if (adviceChain != null) {
+			parentBuilder.addPropertyValue("adviceChain", adviceChain);
+		}
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	public static ManagedList configureAdviceChain(Element adviceChainElement, Element txElement,
+			BeanDefinitionBuilder parentBuilder, ParserContext parserContext) {
+		ManagedList adviceChain = null;
+		// Schema validation ensures txElement and adviceChainElement are mutually exclusive
+		if (txElement != null) {
+			adviceChain = new ManagedList();
+			adviceChain.add(IntegrationNamespaceUtils.configureTransactionAttributes(txElement));
+		}
+		if (adviceChainElement != null) {
+			adviceChain = new ManagedList();
+			NodeList childNodes = adviceChainElement.getChildNodes();
+			for (int i = 0; i < childNodes.getLength(); i++) {
+				Node child = childNodes.item(i);
+				if (child.getNodeType() == Node.ELEMENT_NODE) {
+					Element childElement = (Element) child;
+					String localName = child.getLocalName();
+					if ("bean".equals(localName)) {
+						BeanDefinitionHolder holder = parserContext.getDelegate().parseBeanDefinitionElement(
+								childElement, parentBuilder.getBeanDefinition());
+						parserContext.registerBeanComponent(new BeanComponentDefinition(holder));
+						adviceChain.add(new RuntimeBeanReference(holder.getBeanName()));
+					}
+					else if ("ref".equals(localName)) {
+						String ref = childElement.getAttribute("bean");
+						adviceChain.add(new RuntimeBeanReference(ref));
+					}
+					else {
+						BeanDefinition customBeanDefinition = parserContext.getDelegate().parseCustomElement(
+								childElement, parentBuilder.getBeanDefinition());
+						if (customBeanDefinition == null) {
+							parserContext.getReaderContext().error(
+									"failed to parse custom element '" + localName + "'", childElement);
+						}
+						adviceChain.add(customBeanDefinition);
+					}
+				}
+			}
+		}
+		return adviceChain;
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/xml/PollerParser.java

@@ -20,14 +20,9 @@ import java.util.ArrayList;
 import java.util.List;
 
 import org.springframework.beans.factory.BeanDefinitionStoreException;
-import org.springframework.beans.factory.config.BeanDefinition;
-import org.springframework.beans.factory.config.BeanDefinitionHolder;
-import org.springframework.beans.factory.config.RuntimeBeanReference;
-import org.springframework.beans.factory.parsing.BeanComponentDefinition;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
-import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
@@ -40,8 +35,6 @@ import org.springframework.scheduling.support.PeriodicTrigger;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
 import org.w3c.dom.Element;
-import org.w3c.dom.Node;
-import org.w3c.dom.NodeList;
 
 /**
  * Parser for the <poller> element.
@@ -94,7 +87,8 @@ public class PollerParser extends AbstractBeanDefinitionParser {
 
 		Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
 		Element adviceChainElement = DomUtils.getChildElementByTagName(element, "advice-chain");
-		configureAdviceChain(adviceChainElement, txElement, metadataBuilder, parserContext);
+		IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(adviceChainElement, txElement,
+				metadataBuilder, parserContext);
 
 		Element pseudoTxElement = DomUtils.getChildElementByTagName(element, "psuedo-transactional");
 		if (pseudoTxElement != null && txElement != null) {
@@ -174,48 +168,6 @@ public class PollerParser extends AbstractBeanDefinitionParser {
 		targetBuilder.addPropertyReference("trigger", triggerBeanNames.get(0));
 	}
 
-	/**
-	 * Parses the 'advice-chain' element's sub-elements.
-	 */
-	@SuppressWarnings({ "unchecked", "rawtypes" })
-	private void configureAdviceChain(Element adviceChainElement, Element txElement, BeanDefinitionBuilder targetBuilder, ParserContext parserContext) {
-		ManagedList adviceChain = new ManagedList();
-		// Schema validation ensures txElement and adviceChainElement are mutually exclusive
-		if (txElement != null) {
-			adviceChain.add(IntegrationNamespaceUtils.configureTransactionAttributes(txElement));
-		}
-		if (adviceChainElement != null) {
-			NodeList childNodes = adviceChainElement.getChildNodes();
-			for (int i = 0; i < childNodes.getLength(); i++) {
-				Node child = childNodes.item(i);
-				if (child.getNodeType() == Node.ELEMENT_NODE) {
-					Element childElement = (Element) child;
-					String localName = child.getLocalName();
-					if ("bean".equals(localName)) {
-						BeanDefinitionHolder holder = parserContext.getDelegate().parseBeanDefinitionElement(
-								childElement, targetBuilder.getBeanDefinition());
-						parserContext.registerBeanComponent(new BeanComponentDefinition(holder));
-						adviceChain.add(new RuntimeBeanReference(holder.getBeanName()));
-					}
-					else if ("ref".equals(localName)) {
-						String ref = childElement.getAttribute("bean");
-						adviceChain.add(new RuntimeBeanReference(ref));
-					}
-					else {
-						BeanDefinition customBeanDefinition = parserContext.getDelegate().parseCustomElement(
-								childElement, targetBuilder.getBeanDefinition());
-						if (customBeanDefinition == null) {
-							parserContext.getReaderContext().error(
-									"failed to parse custom element '" + localName + "'", childElement);
-						}
-						adviceChain.add(customBeanDefinition);
-					}
-				}
-			}
-		}
-		targetBuilder.addPropertyValue("adviceChain", adviceChain);
-	}
-
 	private void configureTransactionSync(Element element, BeanDefinitionBuilder metadataBuilder,
 			ParserContext parserContext) {
 		if (element != null) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * 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.
@@ -16,6 +16,11 @@
 
 package org.springframework.integration.handler;
 
+import java.util.List;
+
+import org.aopalliance.aop.Advice;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessageDeliveryException;
@@ -26,6 +31,8 @@ import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.support.channel.ChannelResolutionException;
 import org.springframework.integration.support.channel.ChannelResolver;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.CollectionUtils;
 
 /**
  * Base class for MessageHandlers that are capable of producing replies.
@@ -33,8 +40,10 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @author Iwein Fuld
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
-public abstract class AbstractReplyProducingMessageHandler extends AbstractMessageHandler implements MessageProducer {
+public abstract class AbstractReplyProducingMessageHandler extends AbstractMessageHandler
+		implements MessageProducer, BeanClassLoaderAware {
 
 	private MessageChannel outputChannel;
 
@@ -42,6 +51,13 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 
 	private final MessagingTemplate messagingTemplate;
 
+	private volatile RequestHandler advisedRequestHandler;
+
+	private volatile List<Advice> adviceChain;
+
+	private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
+
+
 
 	public AbstractReplyProducingMessageHandler() {
 		this.messagingTemplate = new MessagingTemplate();
@@ -82,11 +98,30 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 		return this.messagingTemplate;
 	}
 
+
+	public void setAdviceChain(List<Advice> adviceChain) {
+		Assert.notNull(adviceChain, "adviceChain cannot be null");
+		this.adviceChain = adviceChain;
+	}
+
+
+	public void setBeanClassLoader(ClassLoader beanClassLoader) {
+		this.beanClassLoader = beanClassLoader;
+	}
+
+
 	@Override
 	protected void onInit() {
 		if (this.getBeanFactory() != null) {
 			this.messagingTemplate.setBeanFactory(getBeanFactory());
 		}
+		if (!CollectionUtils.isEmpty(this.adviceChain)) {
+			ProxyFactory proxyFactory = new ProxyFactory(new AdvisedRequestHandler());
+			for (Advice advice : this.adviceChain) {
+				proxyFactory.addAdvice(advice);
+			}
+			this.advisedRequestHandler = (RequestHandler) proxyFactory.getProxy(this.beanClassLoader);
+		}
 	}
 
 	/**
@@ -94,7 +129,13 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 	 */
 	@Override
 	protected final void handleMessageInternal(Message<?> message) {
-		Object result = this.handleRequestMessage(message);
+		Object result;
+		if (this.advisedRequestHandler == null) {
+			result = this.handleRequestMessage(message);
+		}
+		else {
+			result = this.advisedRequestHandler.handleRequestMessage(message);
+		}
 		if (result != null) {
 			MessageHeaders requestHeaders = message.getHeaders();
 			this.handleResult(result, requestHeaders);
@@ -149,7 +190,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 	 * 'outputChannel' is <code>null</code>. In that case, the header value must not also be
 	 * <code>null</code>, and it must be an instance of either String or {@link MessageChannel}.
 	 * @param replyMessage the reply Message to send
-	 * @param replyChannelHeaderValue the 'replyChannel' header value from the original request 
+	 * @param replyChannelHeaderValue the 'replyChannel' header value from the original request
 	 */
 	private final void sendReplyMessage(Message<?> replyMessage, final Object replyChannelHeaderValue) {
 		if (logger.isDebugEnabled()) {
@@ -207,4 +248,26 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 	 */
 	protected abstract Object handleRequestMessage(Message<?> requestMessage);
 
+
+	private interface RequestHandler {
+
+		Object handleRequestMessage(Message<?> requestMessage);
+
+		String toString();
+	}
+
+	private class AdvisedRequestHandler implements RequestHandler {
+
+		public Object handleRequestMessage(Message<?> requestMessage) {
+			return AbstractReplyProducingMessageHandler.this.handleRequestMessage(requestMessage);
+		}
+
+		@Override
+		public String toString() {
+			return AbstractReplyProducingMessageHandler.this.toString();
+		}
+
+
+	}
+
 }

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

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+import org.springframework.integration.Message;
+import org.springframework.retry.RetryState;
+
+/**
+ * Strategy interface for generating a {@link RetryState} instance
+ * based on a message.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public interface RetryStateGenerator {
+
+	RetryState determineRetryState(Message<?> message);
+}

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

@@ -0,0 +1,87 @@
+/*
+ * 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 java.lang.reflect.Method;
+
+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.integration.Message;
+import org.springframework.integration.core.MessageHandler;
+
+/**
+ * Base class for {@link MessageHandler} advice classes.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public abstract class AbstractRequestHandlerAdvice implements MethodInterceptor {
+
+	protected final Log logger = LogFactory.getLog(this.getClass());
+
+	public final Object invoke(final MethodInvocation invocation) throws Throwable {
+
+		Method method = invocation.getMethod();
+		Object[] arguments = invocation.getArguments();
+		boolean isMessageMethod = (method.getName().equals("handleRequestMessage") || method.getName().equals("handleMessage"))
+				&& (arguments.length == 1 && arguments[0] instanceof Message);
+
+		if (!isMessageMethod) {
+			return invocation.proceed();
+		}
+		else {
+			Message<?> message = (Message<?>) arguments[0];
+			try {
+				return doInvoke(new ExecutionCallback(){
+
+					public Object execute() throws Exception {
+						try {
+							return invocation.proceed();
+						}
+						catch (Throwable e) {
+							throw new ThrowableHolderException(e);
+						}
+					}
+				}, invocation.getThis(), message);
+			}
+			catch (Exception e) {
+				if (e instanceof ThrowableHolderException) {
+					throw e.getCause();
+				}
+				else {
+					throw e;
+				}
+			}
+		}
+	}
+
+	protected abstract Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception;
+
+	protected interface ExecutionCallback {
+
+		Object execute() throws Exception;
+	}
+
+	@SuppressWarnings("serial")
+	private class ThrowableHolderException extends RuntimeException {
+
+		public ThrowableHolderException(Throwable cause) {
+			super(cause);
+		}
+	}
+}

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

@@ -0,0 +1,166 @@
+/*
+ * 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.expression.Expression;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.MessageHeaders;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
+import org.springframework.integration.support.MessageBuilder;
+
+/**
+ * Used to advise {@link MessageHandler}s.
+ * Two expressions 'onSuccessExpression' and 'onFailureExpression' are evaluated when
+ * appropriate. If the evaluation returns a result, a message is sent to the onSuccessChannel
+ * or onFailureChannel as appropriate; the message is the input message with a header
+ * {@link MessageHeaders#POSTPROCESS_RESULT} containing the evaluation result.
+ * The failure expression is NOT evaluated if the success expression throws an exception.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class ExpressionEvaluatingRequestHandlerAdvice extends AbstractRequestHandlerAdvice
+		implements BeanFactoryAware {
+
+	private final ExpressionEvaluatingMessageProcessor<Object> onSuccessMessageProcessor;
+
+	private final MessageChannel successChannel;
+
+	private final ExpressionEvaluatingMessageProcessor<Object> onFailureMessageProcessor;
+
+	private final MessageChannel failureChannel;
+
+	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
+
+	private volatile boolean trapException = false;
+
+	private volatile boolean returnFailureExpressionResult = false;
+
+	private volatile BeanFactory beanFactory;
+
+	private volatile boolean propagateOnSuccessEvaluationFailures;
+
+	/**
+	 * @param onSuccessExpression
+	 * @param successChannel
+	 * @param onFailureExpression
+	 * @param failureChannel
+	 */
+	public ExpressionEvaluatingRequestHandlerAdvice(Expression onSuccessExpression, MessageChannel successChannel,
+			Expression onFailureExpression, MessageChannel failureChannel) {
+		if (onSuccessExpression != null) {
+			this.onSuccessMessageProcessor = new ExpressionEvaluatingMessageProcessor<Object>(onSuccessExpression);
+			this.onSuccessMessageProcessor.setBeanFactory(this.beanFactory);
+		}
+		else {
+			this.onSuccessMessageProcessor = null;
+		}
+		this.successChannel = successChannel;
+		if (onFailureExpression != null) {
+			this.onFailureMessageProcessor = new ExpressionEvaluatingMessageProcessor<Object>(onFailureExpression);
+			onFailureMessageProcessor.setBeanFactory(this.beanFactory);
+		}
+		else {
+			this.onFailureMessageProcessor = null;
+		}
+		this.failureChannel = failureChannel;
+
+	}
+
+	/**
+	 * If true, any exception will be caught and null returned.
+	 * Default false.
+	 * @param trapException
+	 */
+	public void setTrapException(boolean trapException) {
+		this.trapException = trapException;
+	}
+
+	/**
+	 * If true, the result of evaluating the onFailureExpression will
+	 * be returned as the result of AbstractReplyProducingMessageHandler.handleRequestMessage(Message).
+	 * @param returnFailureExpressionResult
+	 */
+	public void setReturnFailureExpressionResult(boolean returnFailureExpressionResult) {
+		this.returnFailureExpressionResult = returnFailureExpressionResult;
+	}
+
+	/**
+	 * If true and an onSuccess expression evaluation fails with an exception, the exception will be thrown to the
+	 * caller. If false, the exception is caught. Default false. Ignored for onFailure expression evaluation - the
+	 * original exception will be propagated (unless trapException is true).
+	 * @param propagateOnSuccessEvaluationFailures
+	 */
+	public void setPropagateEvaluationFailures(boolean propagateOnSuccessEvaluationFailures) {
+		this.propagateOnSuccessEvaluationFailures = propagateOnSuccessEvaluationFailures;
+	}
+
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+	}
+
+	@Override
+	protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
+		try {
+			Object result = callback.execute();
+			if (onSuccessMessageProcessor != null) {
+				evaluateExpression(message, this.onSuccessMessageProcessor, this.successChannel, this.propagateOnSuccessEvaluationFailures);
+			}
+			return result;
+		}
+		catch (Exception e) {
+			Object evalResult = evaluateExpression(message, this.onFailureMessageProcessor, this.failureChannel, false);
+			if (this.returnFailureExpressionResult) {
+				return evalResult;
+			}
+			if (!this.trapException) {
+				throw e;
+			}
+			return null;
+		}
+	}
+
+	private Object evaluateExpression(Message<?> message,
+			ExpressionEvaluatingMessageProcessor<Object> expressionEvaluatingMessageProcessor,
+			MessageChannel resultChannel, boolean propagateEvaluationFailure) throws Exception {
+		Object evalResult;
+		boolean evaluationFailed = false;
+		try {
+			evalResult = expressionEvaluatingMessageProcessor.processMessage(message);
+		}
+		catch (Exception e) {
+			evalResult = e;
+			evaluationFailed = true;
+		}
+		if (evalResult != null && resultChannel != null) {
+			message = MessageBuilder.fromMessage(message)
+					.setHeader(MessageHeaders.POSTPROCESS_RESULT, evalResult)
+					.build();
+			this.messagingTemplate.send(resultChannel, message);
+		}
+		if (evaluationFailed && propagateEvaluationFailure) {
+			throw (Exception) evalResult;
+		}
+		return evalResult;
+	}
+
+}

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

@@ -0,0 +1,95 @@
+/*
+ * 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 java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.MessagingException;
+
+/**
+ * A circuit breaker that stops calling a failing service after threshold
+ * failures, until halfOpenAfter milliseconds has elapsed. A successful
+ * call resets the failure counter.
+ *
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class RequestHandlerCircuitBreakerAdvice extends AbstractRequestHandlerAdvice {
+
+	private volatile int threshold = 5;
+
+	private volatile long halfOpenAfter = 1000;
+
+	private final ConcurrentMap<Object, AdvisedMetadata> metadataMap = new ConcurrentHashMap<Object, AdvisedMetadata>();
+
+	public void setThreshold(int threshold) {
+		this.threshold = threshold;
+	}
+
+	public void setHalfOpenAfter(long halfOpenAfter) {
+		this.halfOpenAfter = halfOpenAfter;
+	}
+
+	@Override
+	protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
+		AdvisedMetadata metadata = this.metadataMap.get(target);
+		if (metadata == null) {
+			this.metadataMap.putIfAbsent(target, new AdvisedMetadata());
+			metadata = this.metadataMap.get(target);
+		}
+		if (metadata.getFailures().get() >= this.threshold &&
+				System.currentTimeMillis() - metadata.getLastFailure() < this.halfOpenAfter) {
+			throw new MessagingException("Circuit Breaker is Open for " + target);
+		}
+		try {
+			Object result = callback.execute();
+			if (logger.isDebugEnabled() && metadata.getFailures().get() > 0) {
+				logger.debug("Closing Circuit Breaker for " + target);
+			}
+			metadata.getFailures().set(0);
+			return result;
+		}
+		catch (Exception e) {
+			metadata.getFailures().incrementAndGet();
+			metadata.setLastFailure(System.currentTimeMillis());
+			throw e;
+		}
+	}
+
+	private class AdvisedMetadata {
+
+		private final AtomicInteger failures = new AtomicInteger();
+
+		private volatile long lastFailure;
+
+		private long getLastFailure() {
+			return lastFailure;
+		}
+
+		private void setLastFailure(long lastFailure) {
+			this.lastFailure = lastFailure;
+		}
+
+		private AtomicInteger getFailures() {
+			return failures;
+		}
+	}
+
+}

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

@@ -0,0 +1,87 @@
+/*
+ * 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.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;
+import org.springframework.retry.RetryState;
+import org.springframework.retry.support.RetryTemplate;
+
+/**
+ * Uses spring-retry to perform stateless or stateful retry.
+ * Stateless retry means the retries are performed internally
+ * by the {@link RetryTemplate}; stateful retry means the
+ * exception is thrown but state is maintained to support
+ * the retry policies. Stateful retry requires a
+ * {@link RetryStateGenerator}.
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class RequestHandlerRetryAdvice extends AbstractRequestHandlerAdvice {
+
+	private volatile RetryTemplate retryTemplate = new RetryTemplate();
+
+	private volatile RecoveryCallback<Object> recoveryCallback;
+
+	// Stateless unless a state generator is provided
+	private volatile RetryStateGenerator retryStateGenerator =
+			new RetryStateGenerator() {
+				public RetryState determineRetryState(Message<?> message) {
+					return null;
+				}
+			};
+
+	public void setRetryTemplate(RetryTemplate retryTemplate) {
+		this.retryTemplate = retryTemplate;
+	}
+
+	public void setRecoveryCallback(RecoveryCallback<Object> recoveryCallback) {
+		this.recoveryCallback = recoveryCallback;
+	}
+
+	public void setRetryStateGenerator(RetryStateGenerator retryStateGenerator) {
+		this.retryStateGenerator = retryStateGenerator;
+	}
+
+	@Override
+	protected Object doInvoke(final ExecutionCallback callback, Object target, final Message<?> message) throws Exception {
+		RetryState retryState = null;
+		retryState = this.retryStateGenerator.determineRetryState(message);
+
+		return retryTemplate.execute(new RetryCallback<Object>(){
+			public Object doWithRetry(RetryContext context) throws Exception {
+				try {
+					return callback.execute();
+				}
+				catch (MessagingException e) {
+					if (e.getFailedMessage() == null) {
+						e.setFailedMessage(message);
+					}
+					throw e;
+				}
+				catch (Exception e) {
+					throw new MessagingException(message, "Failed to invoke handler", e);
+				}
+			}
+		}, this.recoveryCallback, retryState);
+	}
+
+}

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

@@ -0,0 +1,6 @@
+/**
+ * Provides classes that are used to advise
+ * {@link org.springframework.integration.core.MessageHandler}s with
+ * cross-cutting concerns.
+ */
+package org.springframework.integration.handler.advice;

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd

@@ -1124,6 +1124,7 @@
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:element>
+			<xsd:element name="request-handler-advice-chain" type="adviceChainType" minOccurs="0" maxOccurs="1" />
 		</xsd:sequence>
 		<xsd:attribute name="request-channel" type="xsd:string" use="optional">
 			<xsd:annotation>
@@ -1481,28 +1482,7 @@
 		<xsd:sequence>
 			<xsd:choice>
 				<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1" />
-				<xsd:element name="advice-chain" minOccurs="0" maxOccurs="1">
-					<xsd:complexType>
-						<xsd:sequence>
-							<xsd:choice minOccurs="0" maxOccurs="unbounded">
-								<xsd:element name="ref" minOccurs="0" maxOccurs="unbounded">
-									<xsd:complexType>
-										<xsd:attribute name="bean" type="xsd:string" use="required">
-											<xsd:annotation>
-												<xsd:appinfo>
-													<tool:annotation kind="ref">
-														<tool:expected-type type="java.lang.Object" />
-													</tool:annotation>
-												</xsd:appinfo>
-											</xsd:annotation>
-										</xsd:attribute>
-									</xsd:complexType>
-								</xsd:element>
-								<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="unbounded" />
-							</xsd:choice>
-						</xsd:sequence>
-					</xsd:complexType>
-				</xsd:element>
+				<xsd:element name="advice-chain" type="adviceChainType" minOccurs="0" maxOccurs="1" />
 			</xsd:choice>
 			<xsd:choice minOccurs="0" maxOccurs="1">
 				<xsd:element name="psuedo-transactional" type="pseudoTransactionalType" minOccurs="0" maxOccurs="1">
@@ -3422,12 +3402,34 @@ is provided, the return value is expected to match a channel name exactly.
 		<xsd:attributeGroup ref="transactionSyncAttributeGroup" />
 	</xsd:complexType>
 
+	<xsd:complexType name="adviceChainType">
+		<xsd:sequence>
+			<xsd:choice minOccurs="0" maxOccurs="unbounded">
+				<xsd:element name="ref" minOccurs="0" maxOccurs="unbounded">
+					<xsd:complexType>
+						<xsd:attribute name="bean" type="xsd:string" use="required">
+							<xsd:annotation>
+								<xsd:appinfo>
+									<tool:annotation kind="ref">
+										<tool:expected-type type="java.lang.Object" />
+									</tool:annotation>
+								</xsd:appinfo>
+							</xsd:annotation>
+						</xsd:attribute>
+					</xsd:complexType>
+				</xsd:element>
+				<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="unbounded" />
+			</xsd:choice>
+		</xsd:sequence>
+	</xsd:complexType>
+
 	<xsd:complexType name="expressionOrInnerEndpointDefinitionAware">
 		<xsd:complexContent>
 			<xsd:extension base="handlerEndpointType">
-				<xsd:choice minOccurs="0" maxOccurs="2">
+				<xsd:choice minOccurs="0" maxOccurs="3">
 					<xsd:element name="poller" type="basePollerType" minOccurs="0" maxOccurs="1" />
 					<xsd:element name="expression" type="innerExpressionType" minOccurs="0" maxOccurs="1" />
+					<xsd:element name="request-handler-advice-chain" type="adviceChainType" minOccurs="0" maxOccurs="1" />
 					<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="1" />
 				</xsd:choice>
 				<xsd:attribute name="expression" type="xsd:string">

spring-integration-core/src/test/java/org/springframework/integration/config/FilterParserTests-context.xml

@@ -23,7 +23,11 @@
 		<queue capacity="1"/>
 	</channel>
 
-	<filter ref="selectorBean" method="hasText" input-channel="adapterInput" output-channel="adapterOutput"/>
+	<filter ref="selectorBean" method="hasText" input-channel="adapterInput" output-channel="adapterOutput">
+		<request-handler-advice-chain>
+			<beans:bean class="org.springframework.integration.config.FilterParserTests$FooFilter" />
+		</request-handler-advice-chain>
+	</filter>
 
 	<beans:bean id="selectorBean"
 		class="org.springframework.integration.config.FilterParserTests$TestSelectorBean"/>

spring-integration-core/src/test/java/org/springframework/integration/config/FilterParserTests.java

@@ -22,7 +22,6 @@ import static org.junit.Assert.assertNull;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.integration.Message;
@@ -30,6 +29,7 @@ import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessageRejectedException;
 import org.springframework.integration.core.MessageSelector;
 import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -69,13 +69,16 @@ public class FilterParserTests {
 	@Autowired @Qualifier("discardAndExceptionOutput")
 	PollableChannel discardAndExceptionOutput;
 
+	private static volatile int adviceCalled;
 
 	@Test
 	public void filterWithSelectorAdapterAccepts() {
+		adviceCalled = 0;
 		adapterInput.send(new GenericMessage<String>("test"));
 		Message<?> reply = adapterOutput.receive(0);
 		assertNotNull(reply);
 		assertEquals("test", reply.getPayload());
+		assertEquals(1, adviceCalled);
 	}
 
 	@Test
@@ -156,4 +159,13 @@ public class FilterParserTests {
 		}
 	}
 
+	public static class FooFilter extends AbstractRequestHandlerAdvice {
+
+		@Override
+		protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
+			adviceCalled++;
+			return callback.execute();
+		}
+
+	}
 }

spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml

@@ -22,6 +22,10 @@
 		<property name="name" expression="payload.sourceName"/>
 		<property name="age" value="42"/>
 		<property name="gender" expression="@testBean"/>
+		
+		<request-handler-advice-chain>
+			<beans:bean class="org.springframework.integration.config.xml.EnricherParserTests$FooAdvice" />
+		</request-handler-advice-chain>
 	</enricher>
 
 	<beans:bean id="testBean" class="java.lang.String">

spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java

@@ -35,6 +35,7 @@ import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.core.SubscribableChannel;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.integration.transformer.ContentEnricher;
@@ -44,7 +45,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 /**
  * @author Mark Fisher
  * @author Gunnar Hillert
- * 
+ * @author Gary Russell
+ *
  * @since 2.1
  */
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -54,6 +56,7 @@ public class EnricherParserTests {
 	@Autowired
 	private ApplicationContext context;
 
+	private static volatile int adviceCalled;
 
 	@Test
 	@SuppressWarnings("unchecked")
@@ -84,13 +87,14 @@ public class EnricherParserTests {
 				throw new IllegalStateException("expected 'name', 'age', and 'gender' only, not: " + e.getKey().getExpressionString());
 			}
 		}
+
 	}
 
 	@Test
 	public void configurationCheckTimeoutParameters() {
-		
+
 		Object endpoint = context.getBean("enricher");
-		
+
 		Long requestTimeout = TestUtils.getPropertyValue(endpoint, "handler.requestTimeout", Long.class);
 		Long replyTimeout   = TestUtils.getPropertyValue(endpoint, "handler.replyTimeout", Long.class);
 
@@ -98,18 +102,18 @@ public class EnricherParserTests {
 		assertEquals(Long.valueOf(9876L), replyTimeout);
 
 	}
-	
+
 	@Test
 	public void configurationCheckRequiresReply() {
-		
+
 		Object endpoint = context.getBean("enricher");
-		
+
 		boolean requiresReply = TestUtils.getPropertyValue(endpoint, "handler.requiresReply", Boolean.class);
 
 		assertTrue("Was expecting requiresReply to be 'false'", requiresReply);
 
 	}
-	
+
 	@Test
 	public void integrationTest() {
 		SubscribableChannel requests = context.getBean("requests", SubscribableChannel.class);
@@ -128,6 +132,7 @@ public class EnricherParserTests {
 		assertEquals(42, enriched.getAge());
 		assertEquals("male", enriched.getGender());
 		assertNotSame(original, enriched);
+		assertEquals(1, adviceCalled);
 	}
 
 	private static class Source {
@@ -176,6 +181,7 @@ public class EnricherParserTests {
 			this.gender = gender;
 		}
 
+		@Override
 		public Object clone() {
 			Target copy = new Target();
 			copy.setName(this.name);
@@ -184,4 +190,13 @@ public class EnricherParserTests {
 		}
 	}
 
+	public static class FooAdvice extends AbstractRequestHandlerAdvice {
+
+		@Override
+		protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
+			adviceCalled++;
+			return callback.execute();
+		}
+
+	}
 }

spring-integration-core/src/test/java/org/springframework/integration/config/xml/ServiceActivatorParserTests-context.xml

@@ -21,4 +21,9 @@
 
 	<beans:bean id="testBean" class="org.springframework.integration.config.xml.ServiceActivatorParserTests$TestBean"/>
 
+	<service-activator id="withAdvice" input-channel="advisedInput" expression="'foo'">
+		<request-handler-advice-chain>
+			<beans:bean class="org.springframework.integration.config.xml.ServiceActivatorParserTests$BarAdvice" />
+		</request-handler-advice-chain>
+	</service-activator>
 </beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/config/xml/ServiceActivatorParserTests.java

@@ -20,12 +20,13 @@ import static org.junit.Assert.assertEquals;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.handler.ServiceActivatingHandler;
+import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
@@ -56,6 +57,9 @@ public class ServiceActivatorParserTests {
 	@Autowired
 	private MessageChannel multipleArgsFromPayloadInput;
 
+	@Autowired
+	private MessageChannel advisedInput;
+
 	@SuppressWarnings("unused") // testing auto wiring only
 	@Autowired
 	@Qualifier("org.springframework.integration.config.ServiceActivatorFactoryBean#0")
@@ -102,6 +106,11 @@ public class ServiceActivatorParserTests {
 		assertEquals("JohnDoe", result);
 	}
 
+	@Test
+	public void advised() {
+		Object result = this.sendAndReceive(advisedInput, "hello");
+		assertEquals("bar", result);
+	}
 
 	private Object sendAndReceive(MessageChannel channel, Object payload) {
 		MessagingTemplate template = new MessagingTemplate(channel);
@@ -152,4 +161,13 @@ public class ServiceActivatorParserTests {
 		}
 	}
 
+	public static class BarAdvice extends AbstractRequestHandlerAdvice {
+
+		@Override
+		protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
+			callback.execute();
+			return "bar";
+		}
+
+	}
 }

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

@@ -0,0 +1,477 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.aopalliance.aop.Advice;
+import org.junit.Test;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.Message;
+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.message.GenericMessage;
+import org.springframework.retry.RecoveryCallback;
+import org.springframework.retry.RetryContext;
+import org.springframework.retry.RetryState;
+import org.springframework.retry.support.DefaultRetryState;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class AdvisedMessageHandlerTests {
+
+	@Test
+	public void successFailureAdvice() {
+		final AtomicBoolean doFail = new AtomicBoolean();
+		AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
+			@Override
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				if (doFail.get()) {
+					throw new RuntimeException("qux");
+				}
+				return "baz";
+			}
+		};
+		QueueChannel replies = new QueueChannel();
+		handler.setOutputChannel(replies);
+		Message<String> message = new GenericMessage<String>("Hello, world!");
+
+		// no advice
+		handler.handleMessage(message);
+		Message<?> reply = replies.receive(1000);
+		assertNotNull(reply);
+		assertEquals("baz", reply.getPayload());
+
+		PollableChannel successChannel = new QueueChannel();
+		PollableChannel failureChannel = new QueueChannel();
+		ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice(
+				new SpelExpressionParser().parseExpression("'foo'"), successChannel,
+				new SpelExpressionParser().parseExpression("'bar'"), failureChannel);
+
+		List<Advice> adviceChain = new ArrayList<Advice>();
+		adviceChain.add(advice);
+		handler.setAdviceChain(adviceChain);
+		handler.afterPropertiesSet();
+
+		// advice with success
+		handler.handleMessage(message);
+		reply = replies.receive(1000);
+		assertNotNull(reply);
+		assertEquals("baz", reply.getPayload());
+
+		Message<?> success = successChannel.receive(1000);
+		assertNotNull(success);
+		assertEquals("Hello, world!", success.getPayload());
+		assertEquals("foo", success.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT));
+
+		// advice with failure, not trapped
+		doFail.set(true);
+		try {
+			handler.handleMessage(message);
+			fail("Expected exception");
+		}
+		catch (Exception e) {
+			assertEquals("qux", e.getCause().getMessage());
+		}
+
+		Message<?> failure = failureChannel.receive(1000);
+		assertNotNull(failure);
+		assertEquals("Hello, world!", failure.getPayload());
+		assertEquals("bar", failure.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT));
+
+		// advice with failure, trapped
+		advice.setTrapException(true);
+		handler.handleMessage(message);
+		failure = failureChannel.receive(1000);
+		assertNotNull(failure);
+		assertEquals("Hello, world!", failure.getPayload());
+		assertEquals("bar", failure.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT));
+		assertNull(replies.receive(1));
+
+		// advice with failure, eval is result
+		advice.setReturnFailureExpressionResult(true);
+		handler.handleMessage(message);
+		failure = failureChannel.receive(1000);
+		assertNotNull(failure);
+		assertEquals("Hello, world!", failure.getPayload());
+		assertEquals("bar", failure.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT));
+
+		reply = replies.receive(1000);
+		assertNotNull(reply);
+		assertEquals("bar", reply.getPayload());
+
+	}
+
+	@Test
+	public void propagateOnSuccessExpressionFailures() {
+		final AtomicBoolean doFail = new AtomicBoolean();
+		AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
+			@Override
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				if (doFail.get()) {
+					throw new RuntimeException("qux");
+				}
+				return "baz";
+			}
+		};
+		QueueChannel replies = new QueueChannel();
+		handler.setOutputChannel(replies);
+		Message<String> message = new GenericMessage<String>("Hello, world!");
+
+		PollableChannel successChannel = new QueueChannel();
+		PollableChannel failureChannel = new QueueChannel();
+		ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice(
+				new SpelExpressionParser().parseExpression("1/0"), successChannel,
+				new SpelExpressionParser().parseExpression("1/0"), failureChannel);
+
+		List<Advice> adviceChain = new ArrayList<Advice>();
+		adviceChain.add(advice);
+		handler.setAdviceChain(adviceChain);
+		handler.afterPropertiesSet();
+
+		// failing advice with success
+		handler.handleMessage(message);
+		Message<?> reply = replies.receive(1000);
+		assertNotNull(reply);
+		assertEquals("baz", reply.getPayload());
+
+		Message<?> success = successChannel.receive(1000);
+		assertNotNull(success);
+		assertEquals("Hello, world!", success.getPayload());
+		assertEquals(MessageHandlingException.class, success.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT).getClass());
+		assertEquals("Expression evaluation failed: 1/0", ((Exception) success.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT)).getMessage());
+
+		// propagate failing advice with success
+		advice.setPropagateEvaluationFailures(true);
+		try {
+			handler.handleMessage(message);
+			fail("Expected Exception");
+		}
+		catch (MessageHandlingException e) {
+			assertEquals("Expression evaluation failed: 1/0", e.getMessage());
+		}
+		reply = replies.receive(1);
+		assertNull(reply);
+
+		success = successChannel.receive(1000);
+		assertNotNull(success);
+		assertEquals("Hello, world!", success.getPayload());
+		assertEquals(MessageHandlingException.class, success.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT).getClass());
+		assertEquals("Expression evaluation failed: 1/0", ((Exception) success.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT)).getMessage());
+
+	}
+
+	@Test
+	public void propagateOnFailureExpressionFailures() {
+		final AtomicBoolean doFail = new AtomicBoolean(true);
+		AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
+			@Override
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				if (doFail.get()) {
+					throw new RuntimeException("qux");
+				}
+				return "baz";
+			}
+		};
+		QueueChannel replies = new QueueChannel();
+		handler.setOutputChannel(replies);
+		Message<String> message = new GenericMessage<String>("Hello, world!");
+
+		PollableChannel successChannel = new QueueChannel();
+		PollableChannel failureChannel = new QueueChannel();
+		ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice(
+				new SpelExpressionParser().parseExpression("1/0"), successChannel,
+				new SpelExpressionParser().parseExpression("1/0"), failureChannel);
+
+		List<Advice> adviceChain = new ArrayList<Advice>();
+		adviceChain.add(advice);
+		handler.setAdviceChain(adviceChain);
+		handler.afterPropertiesSet();
+
+		// failing advice with failure
+		try {
+			handler.handleMessage(message);
+			fail("Expected exception");
+		}
+		catch (Exception e) {
+			assertEquals("qux", e.getCause().getMessage());
+		}
+		Message<?> reply = replies.receive(1);
+		assertNull(reply);
+
+		Message<?> failure = failureChannel.receive(1000);
+		assertNotNull(failure);
+		assertEquals("Hello, world!", failure.getPayload());
+		assertEquals(MessageHandlingException.class, failure.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT).getClass());
+		assertEquals("Expression evaluation failed: 1/0", ((Exception) failure.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT)).getMessage());
+
+		// propagate failing advice with failure; expect original exception
+		advice.setPropagateEvaluationFailures(true);
+		try {
+			handler.handleMessage(message);
+			fail("Expected Exception");
+		}
+		catch (MessageHandlingException e) {
+			assertEquals("qux", e.getCause().getMessage());
+		}
+		reply = replies.receive(1);
+		assertNull(reply);
+
+		failure = failureChannel.receive(1000);
+		assertNotNull(failure);
+		assertEquals("Hello, world!", failure.getPayload());
+		assertEquals(MessageHandlingException.class, failure.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT).getClass());
+		assertEquals("Expression evaluation failed: 1/0", ((Exception) failure.getHeaders().get(MessageHeaders.POSTPROCESS_RESULT)).getMessage());
+
+	}
+
+	@Test
+	public void circuitBreakerTests() throws Exception {
+		final AtomicBoolean doFail = new AtomicBoolean();
+		AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
+
+			@Override
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				if (doFail.get()) {
+					throw new RuntimeException("foo");
+				}
+				return "bar";
+			}
+
+		};
+		handler.setBeanName("baz");
+		handler.setOutputChannel(new QueueChannel());
+		RequestHandlerCircuitBreakerAdvice advice = new RequestHandlerCircuitBreakerAdvice();
+		/*
+		 * Circuit breaker opens after 2 failures; allows a new attempt after 100ms and
+		 * immediately opens again if that attempt fails. After a successful attempt,
+		 * we reset the failure counter.
+		 */
+		advice.setThreshold(2);
+		advice.setHalfOpenAfter(100);
+
+		List<Advice> adviceChain = new ArrayList<Advice>();
+		adviceChain.add(advice);
+		handler.setAdviceChain(adviceChain);
+		handler.afterPropertiesSet();
+
+		doFail.set(true);
+		Message<String> message = new GenericMessage<String>("Hello, world!");
+		try {
+			handler.handleMessage(message);
+			fail("Expected failure");
+		}
+		catch (Exception e) {
+			assertEquals("foo", e.getCause().getMessage());
+		}
+		try {
+			handler.handleMessage(message);
+			fail("Expected failure");
+		}
+		catch (Exception e) {
+			assertEquals("foo", e.getCause().getMessage());
+		}
+		try {
+			handler.handleMessage(message);
+			fail("Expected failure");
+		}
+		catch (Exception e) {
+			assertEquals("Circuit Breaker is Open for baz", e.getMessage());
+		}
+		Thread.sleep(100);
+		try {
+			handler.handleMessage(message);
+			fail("Expected failure");
+		}
+		catch (Exception e) {
+			assertEquals("foo", e.getCause().getMessage());
+		}
+		try {
+			handler.handleMessage(message);
+			fail("Expected failure");
+		}
+		catch (Exception e) {
+			assertEquals("Circuit Breaker is Open for baz", e.getMessage());
+		}
+		Thread.sleep(100);
+		doFail.set(false);
+		handler.handleMessage(message);
+		doFail.set(true);
+		try {
+			handler.handleMessage(message);
+			fail("Expected failure");
+		}
+		catch (Exception e) {
+			assertEquals("foo", e.getCause().getMessage());
+		}
+		try {
+			handler.handleMessage(message);
+			fail("Expected failure");
+		}
+		catch (Exception e) {
+			assertEquals("foo", e.getCause().getMessage());
+		}
+		try {
+			handler.handleMessage(message);
+			fail("Expected failure");
+		}
+		catch (Exception e) {
+			assertEquals("Circuit Breaker is Open for baz", e.getMessage());
+		}
+	}
+
+	@Test
+	public void defaultRetrySucceedonThirdTry() {
+		final AtomicInteger counter = new AtomicInteger(2);
+		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();
+
+		List<Advice> adviceChain = new ArrayList<Advice>();
+		adviceChain.add(advice);
+		handler.setAdviceChain(adviceChain);
+		handler.afterPropertiesSet();
+
+		Message<String> message = new GenericMessage<String>("Hello, world!");
+		handler.handleMessage(message);
+		assertTrue(counter.get() == -1);
+		Message<?> reply = replies.receive(1000);
+		assertNotNull(reply);
+		assertEquals("bar", reply.getPayload());
+
+	}
+
+	@Test
+	public void defaultStatefulRetrySucceedonThirdTry() {
+		final AtomicInteger counter = new AtomicInteger(2);
+		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 RetryStateGenerator() {
+			public RetryState determineRetryState(Message<?> message) {
+				return new DefaultRetryState(message.getHeaders().getId());
+			}
+		});
+
+		List<Advice> adviceChain = new ArrayList<Advice>();
+		adviceChain.add(advice);
+		handler.setAdviceChain(adviceChain);
+		handler.afterPropertiesSet();
+
+		Message<String> message = new GenericMessage<String>("Hello, world!");
+		for (int i = 0; i < 3; i++) {
+			try {
+				handler.handleMessage(message);
+			}
+			catch (Exception e) {
+				assertTrue(i < 2);
+			}
+		}
+		assertTrue(counter.get() == -1);
+		Message<?> reply = replies.receive(1000);
+		assertNotNull(reply);
+		assertEquals("bar", reply.getPayload());
+
+	}
+
+	@Test
+	public void defaultStatefulRetryRecoverAfterThirdTry() {
+		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 RetryStateGenerator() {
+			public RetryState determineRetryState(Message<?> message) {
+				return new DefaultRetryState(message.getHeaders().getId());
+			}
+		});
+
+		advice.setRecoveryCallback(new RecoveryCallback<Object>() {
+
+			public Object recover(RetryContext context) throws Exception {
+				return "baz";
+			}
+		});
+
+		List<Advice> adviceChain = new ArrayList<Advice>();
+		adviceChain.add(advice);
+		handler.setAdviceChain(adviceChain);
+		handler.afterPropertiesSet();
+
+		Message<String> message = new GenericMessage<String>("Hello, world!");
+		for (int i = 0; i < 4; i++) {
+			try {
+				handler.handleMessage(message);
+			}
+			catch (Exception e) {
+			}
+		}
+		assertTrue(counter.get() == 0);
+		Message<?> reply = replies.receive(1000);
+		assertNotNull(reply);
+		assertEquals("baz", reply.getPayload());
+
+	}
+}

spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * 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.
@@ -19,19 +19,22 @@ package org.springframework.integration.transformer;
 import static org.junit.Assert.assertEquals;
 
 import org.junit.Test;
-
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
 import org.springframework.integration.message.GenericMessage;
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public class TransformerContextTests {
 
+	private static volatile int adviceCalled;
+
 	@Test
 	public void methodInvokingTransformer() {
 		ApplicationContext context = new ClassPathXmlApplicationContext(
@@ -41,6 +44,16 @@ public class TransformerContextTests {
 		input.send(new GenericMessage<String>("foo"));
 		Message<?> reply = output.receive(0);
 		assertEquals("FOO", reply.getPayload());
+		assertEquals(1, adviceCalled);
 	}
 
+	public static class FooAdvice extends AbstractRequestHandlerAdvice {
+
+		@Override
+		protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
+			adviceCalled++;
+			return callback.execute();
+		}
+
+	}
 }

spring-integration-core/src/test/java/org/springframework/integration/transformer/transformerContextTests.xml

@@ -13,7 +13,11 @@
 		<queue capacity="50"/>
 	</channel>
 
-	<transformer input-channel="input" ref="testBean" method="upperCase" output-channel="output"/>
+	<transformer input-channel="input" ref="testBean" method="upperCase" output-channel="output">
+		<request-handler-advice-chain>
+			<beans:bean class="org.springframework.integration.transformer.TransformerContextTests$FooAdvice" />
+		</request-handler-advice-chain>
+	</transformer>
 
 	<beans:bean id="testBean" class="org.springframework.integration.transformer.TestBean"/>