diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java index c9b27d8c2d..7051c6f5d1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 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. @@ -27,8 +27,9 @@ import org.springframework.util.StringUtils; /** * Factory bean for creating a Message Filter. - * + * * @author Mark Fisher + * @author Gary Russell * @since 2.0 */ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean { @@ -39,6 +40,7 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean private volatile Long sendTimeout; + private volatile Boolean discardWithinAdvice; public void setDiscardChannel(MessageChannel discardChannel) { this.discardChannel = discardChannel; @@ -52,6 +54,10 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean this.sendTimeout = sendTimeout; } + public void setDiscardWithinAdvice(boolean discardWithinAdvice) { + this.discardWithinAdvice = discardWithinAdvice; + } + @Override MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) { MessageSelector selector = null; @@ -83,6 +89,9 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean if (this.sendTimeout != null) { filter.setSendTimeout(this.sendTimeout.longValue()); } + if (this.discardWithinAdvice != null) { + filter.setDiscardWithinAdvice(this.discardWithinAdvice); + } return filter; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/FilterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/FilterParser.java index b8d33f68fb..eb668d4326 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/FilterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/FilterParser.java @@ -19,6 +19,7 @@ package org.springframework.integration.config.xml; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.FilterFactoryBean; +import org.springframework.util.xml.DomUtils; import org.w3c.dom.Element; /** @@ -43,6 +44,11 @@ public class FilterParser extends AbstractDelegatingConsumerEndpointParser { void postProcess(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "discard-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "throw-exception-on-rejection"); + Element adviceChainElement = DomUtils.getChildElementByTagName(element, + IntegrationNamespaceUtils.REQUEST_HANDLER_ADVICE_CHAIN); + if (adviceChainElement != null) { + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, adviceChainElement, "discard-within-advice"); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java index 022cd410aa..a40e1fe12a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java @@ -21,7 +21,7 @@ import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.MessageRejectedException; import org.springframework.integration.core.MessageSelector; -import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.handler.AbstractReplyProducingPostProcessingMessageHandler; import org.springframework.util.Assert; /** @@ -35,8 +35,9 @@ import org.springframework.util.Assert; * * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ -public class MessageFilter extends AbstractReplyProducingMessageHandler { +public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHandler { private final MessageSelector selector; @@ -82,6 +83,15 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler { this.discardChannel = discardChannel; } + /** + * Set to 'true' if you wish the discard processing to occur within any + * request handler advice applied to this filter. Also applies to + * throwing an exception on rejection. Default: true. + */ + public void setDiscardWithinAdvice(boolean discardWithinAdvice) { + this.setPostProcessWithinAdvice(discardWithinAdvice); + } + @Override public String getComponentType() { return "filter"; @@ -99,17 +109,26 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler { } @Override - protected Object handleRequestMessage(Message message) { + protected Object doHandleRequestMessage(Message message) { if (this.selector.accept(message)) { return message; } - if (this.discardChannel != null) { - this.getMessagingTemplate().send(this.discardChannel, message); + else { + return null; } - if (this.throwExceptionOnRejection) { - throw new MessageRejectedException(message, "MessageFilter '" + this.getComponentName() + "' rejected Message"); + } + + @Override + public Object postProcess(Message message, Object result) { + if (result == null) { + if (this.discardChannel != null) { + this.getMessagingTemplate().send(this.discardChannel, message); + } + if (this.throwExceptionOnRejection) { + throw new MessageRejectedException(message, "MessageFilter '" + this.getComponentName() + "' rejected Message"); + } } - return null; + return result; } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 57df4f7f69..a16f66d383 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -104,6 +104,9 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa this.adviceChain = adviceChain; } + protected boolean hasAdviceChain() { + return this.adviceChain != null && this.adviceChain.size() > 0; + } public void setBeanClassLoader(ClassLoader beanClassLoader) { this.beanClassLoader = beanClassLoader; @@ -134,7 +137,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa result = this.handleRequestMessage(message); } else { - result = this.advisedRequestHandler.handleRequestMessage(message); + result = doInvokeAdvisedRequestHandler(message); } if (result != null) { MessageHeaders requestHeaders = message.getHeaders(); @@ -149,6 +152,10 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa } } + protected Object doInvokeAdvisedRequestHandler(Message message) { + return this.advisedRequestHandler.handleRequestMessage(message); + } + private void handleResult(Object result, MessageHeaders requestHeaders) { if (result instanceof Iterable && this.shouldSplitReply((Iterable) result)) { for (Object o : (Iterable) result) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingPostProcessingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingPostProcessingMessageHandler.java new file mode 100644 index 0000000000..a6a4a9f871 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingPostProcessingMessageHandler.java @@ -0,0 +1,61 @@ +/* + * 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. + * 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; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public abstract class AbstractReplyProducingPostProcessingMessageHandler + extends AbstractReplyProducingMessageHandler + implements PostProcessingMessageHandler { + + private volatile boolean postProcessWithinAdvice = true; + + /** + * Specify whether the post processing should occur within + * the scope of any configured advice classes. If false, the + * post processing will occur after the advice chain returns. Default true. + * This is only applicable if there is in fact an advice chain present. + */ + public void setPostProcessWithinAdvice(boolean postProcessWithinAdvice) { + this.postProcessWithinAdvice = postProcessWithinAdvice; + } + + @Override + protected final Object handleRequestMessage(Message requestMessage) { + Object result = this.doHandleRequestMessage(requestMessage); + if (this.postProcessWithinAdvice || !this.hasAdviceChain()) { + this.postProcess(requestMessage, result); + } + return result; + } + + @Override + protected final Object doInvokeAdvisedRequestHandler(Message message) { + Object result = super.doInvokeAdvisedRequestHandler(message); + if (!this.postProcessWithinAdvice) { + this.postProcess(message, result); + } + return result; + } + + protected abstract Object doHandleRequestMessage(Message requestMessage); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/PostProcessingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/PostProcessingMessageHandler.java new file mode 100644 index 0000000000..e4b28463a8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/PostProcessingMessageHandler.java @@ -0,0 +1,41 @@ +/* + * 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. + * 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; + + +/** + * Implementations of this interface are subclasses of + * {@link AbstractMessageHandler} that perform post processing after the + * {@link AbstractMessageHandler#handleMessageInternal(org.springframework.integration.Message)} + * call. + * + * @author Gary Russell + * @since 3.0 + * + */ +public interface PostProcessingMessageHandler { + + /** + * Take some further action on the result and/or message. + * @param result The result from {@link AbstractMessageHandler#handleMessageInternal(Message)}. + * @param message The message. + * @return The post-processed result. + */ + public Object postProcess(Message message, Object result); + +} diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd index 228a481c0b..320a234a26 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd @@ -2417,7 +2417,26 @@ - + + + + + + + + + + + + + + + + @@ -3584,6 +3603,25 @@ is provided, the return value is expected to match a channel name exactly. + + + + + + + + + + + + A SpEL expression to be evaluated against the input Message as its root object. + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/FilterParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/FilterParserTests-context.xml index 90b06b9931..7a5fd884d9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/FilterParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/FilterParserTests-context.xml @@ -23,8 +23,8 @@ - - + + @@ -32,7 +32,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/FilterParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/FilterParserTests.java index 6ac12485c8..40035831dc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/FilterParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/FilterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 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. @@ -17,8 +17,10 @@ package org.springframework.integration.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import org.junit.Test; import org.junit.runner.RunWith; @@ -29,14 +31,17 @@ 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.filter.MessageFilter; import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.StringUtils; /** * @author Mark Fisher + * @author Gary Russell */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration @@ -69,8 +74,20 @@ public class FilterParserTests { @Autowired @Qualifier("discardAndExceptionOutput") PollableChannel discardAndExceptionOutput; + @Autowired @Qualifier("advised.handler") + MessageFilter advised; + + @Autowired @Qualifier("notAdvised.handler") + MessageFilter notAdvised; + private static volatile int adviceCalled; + @Test + public void adviseDiscard() { + assertFalse(TestUtils.getPropertyValue(this.advised, "postProcessWithinAdvice", Boolean.class)); + assertTrue(TestUtils.getPropertyValue(this.notAdvised, "postProcessWithinAdvice", Boolean.class)); + } + @Test public void filterWithSelectorAdapterAccepts() { adviceCalled = 0; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java index 856553ee58..086db65776 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java @@ -54,8 +54,10 @@ import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessagingException; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.core.MessageSelector; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.filter.MessageFilter; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice.MessageHandlingExpressionEvaluatingAdviceException; import org.springframework.integration.message.AdviceMessage; @@ -852,6 +854,77 @@ public class AdvisedMessageHandlerTests { "an attempt to advise method 'call' in 'java.util.concurrent.Callable' is ignored")); } + public void filterDiscardNoAdvice() { + MessageFilter filter = new MessageFilter(new MessageSelector() { + @Override + public boolean accept(Message message) { + return false; + } + }); + QueueChannel discardChannel = new QueueChannel(); + filter.setDiscardChannel(discardChannel); + filter.handleMessage(new GenericMessage("foo")); + assertNotNull(discardChannel.receive(0)); + } + + @Test + public void filterDiscardWithinAdvice() { + MessageFilter filter = new MessageFilter(new MessageSelector() { + @Override + public boolean accept(Message message) { + return false; + } + }); + final QueueChannel discardChannel = new QueueChannel(); + filter.setDiscardChannel(discardChannel); + List adviceChain = new ArrayList(); + final AtomicReference> discardedWithinAdvice = new AtomicReference>(); + adviceChain.add(new AbstractRequestHandlerAdvice() { + @Override + protected Object doInvoke(ExecutionCallback callback, Object target, Message message) throws Exception { + Object result = callback.execute(); + discardedWithinAdvice.set(discardChannel.receive(0)); + return result; + } + }); + filter.setAdviceChain(adviceChain); + filter.afterPropertiesSet(); + filter.handleMessage(new GenericMessage("foo")); + assertNotNull(discardedWithinAdvice.get()); + assertNull(discardChannel.receive(0)); + } + + @Test + public void filterDiscardOutsideAdvice() { + MessageFilter filter = new MessageFilter(new MessageSelector() { + @Override + public boolean accept(Message message) { + return false; + } + }); + final QueueChannel discardChannel = new QueueChannel(); + filter.setDiscardChannel(discardChannel); + List adviceChain = new ArrayList(); + final AtomicReference> discardedWithinAdvice = new AtomicReference>(); + final AtomicBoolean adviceCalled = new AtomicBoolean(); + adviceChain.add(new AbstractRequestHandlerAdvice() { + @Override + protected Object doInvoke(ExecutionCallback callback, Object target, Message message) throws Exception { + Object result = callback.execute(); + discardedWithinAdvice.set(discardChannel.receive(0)); + adviceCalled.set(true); + return result; + } + }); + filter.setAdviceChain(adviceChain); + filter.setDiscardWithinAdvice(false); + filter.afterPropertiesSet(); + filter.handleMessage(new GenericMessage("foo")); + assertTrue(adviceCalled.get()); + assertNull(discardedWithinAdvice.get()); + assertNotNull(discardChannel.receive(0)); + } + private interface Bar { Object handleRequestMessage(Message message) throws Throwable; } diff --git a/src/reference/docbook/handler-advice.xml b/src/reference/docbook/handler-advice.xml index fdeccb417e..e6f3100af0 100644 --- a/src/reference/docbook/handler-advice.xml +++ b/src/reference/docbook/handler-advice.xml @@ -458,4 +458,20 @@ protected abstract Object doInvoke(ExecutionCallback callback, Object target, Me +
+ Advising Filters + + There is an additional consideration when advising Filters. By default, any discard + actions (when the filter returns false) are performed within the scope of the + advice chain. This could include all the flow downstream of the discard channel. + So, for example if an element downstream of the discard-channel throws an exception, + and there is a retry advice, the process will be retried. This is also the case if + throwExceptionOnRejection is set to true (the exception is thrown within the + scope of the advice). + + + Setting discard-within-advice to "false" modifies this behavior and the discard + (or exception) occurs after the advice chain is called. + +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index e9d7b0dc65..b6c8d80064 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -82,6 +82,16 @@ for at least this number of milliseconds. For more information see . +
+ Advising Filters + + Previously, when a <filter/> had a <request-handler-advice-chain/>, the discard + action was all performed within the scope of the advice chain (including any downstream flow + on the discard-channel). The filter element now has an attribute + discard-within-advice (default true), to allow the discard action to + be performed after the advice chain completes. See . + +
ObjectToStringTransformer Improvements