Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2938 Filter - Provide Option for Discard

Browse Source 
Provide an option to determine whether or not the discard
(and exception throwing if configured) occurs within the
advice chain (default), or after it completes.

INT-2938 Filter Discard Advice Option - Schema

Schema, parser, tests for option on filter element.

INT-2938 Polishing - PR Comments + Doc

INT-2938 Add to "What's New"

INT-2938 Polishing - Javadoc Typo

while merging: simplified to use boolean var only
This commit is contained in:
Gary Russell
2013-02-21 10:45:33 -05:00
committed by Mark Fisher
parent 2bf6b9ba14
commit d1d367a97f
12 changed files with 313 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java
View File

@@ -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;
}

6
spring-integration-core/src/main/java/org/springframework/integration/config/xml/FilterParser.java
View File

@@ -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");
}
}
}

35
spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java
View File

@@ -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

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

@@ -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) {

61
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingPostProcessingMessageHandler.java Normal file
View File

@@ -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);
}

41
spring-integration-core/src/main/java/org/springframework/integration/handler/PostProcessingMessageHandler.java Normal file
View File

@@ -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);
}

40
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd
View File

@@ -2417,7 +2417,26 @@
<xsd:complexType name="filter-type">
<xsd:complexContent>
<xsd:extension base="expressionOrInnerEndpointDefinitionAware">
<xsd:extension base="expressionOrInnerEndpointDefinitionAwareNoAdviceChain">
<xsd:sequence>
<xsd:element name="request-handler-advice-chain" minOccurs="0" maxOccurs="1">
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="adviceChainType">
<xsd:attribute name="discard-within-advice" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
When true (default) any discard action (and exception thrown) will occur
within the scope of the advice class(es) in the chain. Otherwise, these actions
will occur after the advice chain returns.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="discard-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
@@ -3584,6 +3603,25 @@ is provided, the return value is expected to match a channel name exactly.
</xsd:complexContent>
</xsd:complexType>
<xsd:complexType name="expressionOrInnerEndpointDefinitionAwareNoAdviceChain">
<xsd:complexContent>
<xsd:extension base="handlerEndpointType">
<xsd:choice minOccurs="0" maxOccurs="2">
<xsd:element name="poller" type="basePollerType" minOccurs="0" maxOccurs="1" />
<xsd:element name="expression" type="innerExpressionType" minOccurs="0" maxOccurs="1" />
<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="1" />
</xsd:choice>
<xsd:attribute name="expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
A SpEL expression to be evaluated against the input Message as its root object.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
<xsd:complexType name="innerExpressionType">
<xsd:attribute name="key" type="xsd:string" use="required">
<xsd:annotation>

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

@@ -23,8 +23,8 @@
<queue capacity="1"/>
</channel>
<filter ref="selectorBean" method="hasText" input-channel="adapterInput" output-channel="adapterOutput">
<request-handler-advice-chain>
<filter id="advised" ref="selectorBean" method="hasText" input-channel="adapterInput" output-channel="adapterOutput">
<request-handler-advice-chain discard-within-advice="false">
<beans:bean class="org.springframework.integration.config.FilterParserTests$FooFilter" />
</request-handler-advice-chain>
</filter>
@@ -32,7 +32,7 @@
<beans:bean id="selectorBean"
class="org.springframework.integration.config.FilterParserTests$TestSelectorBean"/>
<filter ref="selectorImpl" input-channel="implementationInput" output-channel="implementationOutput"/>
<filter id="notAdvised" ref="selectorImpl" input-channel="implementationInput" output-channel="implementationOutput"/>
<filter ref="selectorImpl" input-channel="exceptionInput" output-channel="implementationOutput" throw-exception-on-rejection="true"/>

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

@@ -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;

73
spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java
View File

@@ -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<String>("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<Advice> adviceChain = new ArrayList<Advice>();
final AtomicReference<Message<?>> discardedWithinAdvice = new AtomicReference<Message<?>>();
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<String>("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<Advice> adviceChain = new ArrayList<Advice>();
final AtomicReference<Message<?>> discardedWithinAdvice = new AtomicReference<Message<?>>();
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<String>("foo"));
assertTrue(adviceCalled.get());
assertNull(discardedWithinAdvice.get());
assertNotNull(discardChannel.receive(0));
}
private interface Bar {
Object handleRequestMessage(Message<?> message) throws Throwable;
}

16
src/reference/docbook/handler-advice.xml
View File

@@ -458,4 +458,20 @@ protected abstract Object doInvoke(ExecutionCallback callback, Object target, Me
</para>
</note>
</section>
<section id="advising-filters">
<title>Advising Filters</title>
<para>
There is an additional consideration when advising <classname>Filter</classname>s. By default, any discard
actions (when the filter returns false) are performed <emphasis>within</emphasis> the scope of the
advice chain. This could include all the flow downstream of the <emphasis>discard channel</emphasis>.
So, for example if an element downstream of the <emphasis>discard-channel</emphasis> throws an exception,
and there is a retry advice, the process will be retried. This is also the case if
<emphasis>throwExceptionOnRejection</emphasis> is set to true (the exception is thrown within the
scope of the advice).
</para>
<para>
Setting <emphasis>discard-within-advice</emphasis> to "false" modifies this behavior and the discard
(or exception) occurs after the advice chain is called.
</para>
</section>
</section>

10
src/reference/docbook/whats-new.xml
View File

@@ -82,6 +82,16 @@
for at least this number of milliseconds. For more information see <xref linkend="aggregator-config"/>.
</para>
</section>
<section id="3.0-advising-filters">
<title>Advising Filters</title>
<para>
Previously, when a &lt;filter/&gt; had a &lt;request-handler-advice-chain/&gt;, the discard
action was all performed within the scope of the advice chain (including any downstream flow
on the <code>discard-channel</code>). The filter element now has an attribute
<code>discard-within-advice</code> (default <code>true</code>), to allow the discard action to
be performed after the advice chain completes. See <xref linkend="advising-filters"/>.
</para>
</section>
<section id="3.0-o-t-s-t">
<title>ObjectToStringTransformer Improvements</title>
<para>