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

INT-3551: Idempotent Receiver: Add value-strategy

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3551

* Rename `MetadataKeyStrategy` -> `MetadataEntryStrategy`
* Add `valueStrategy` to the `MetadataStoreSelector`
* Add `value-strategy` and `value-expression` to the `<idempotent-receiver>`

INT-3551: Add `@IR` support on service methods

Rework `MetadataKeyStrategy` just to the `MessageProcessor`
Fix Docs

Minor Doc Polishing.
This commit is contained in:
Artem Bilan
2014-11-06 19:42:57 +02:00
committed by Gary Russell
parent 1018fcf361
commit 0cc9273a2e
14 changed files with 327 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/annotation/IdempotentReceiver.java
View File

@@ -23,7 +23,7 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* A {@code @Bean} that has a MessagingAnnotation (@code @ServiceActivator, @Router etc.)
* A {@code method} that has a MessagingAnnotation (@code @ServiceActivator, @Router etc.)
* that also has this annotation, has an
* {@link org.springframework.integration.handler.advice.IdempotentReceiverInterceptor} applied
* to the associated {@link org.springframework.messaging.MessageHandler#handleMessage} method.

26
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -27,6 +27,9 @@ import org.aopalliance.aop.Advice;
import org.springframework.aop.TargetSource;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.DefaultBeanFactoryPointcutAdvisor;
import org.springframework.aop.support.NameMatchMethodPointcut;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.annotation.Bean;
@@ -38,6 +41,7 @@ import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.core.env.Environment;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.annotation.IdempotentReceiver;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.IntegrationConfigUtils;
@@ -131,6 +135,28 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
handler = (MessageHandler) this.beanFactory.initializeBean(handler, handlerBeanName);
}
if (AnnotatedElementUtils.isAnnotated(method, IdempotentReceiver.class.getName())
&& !AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) {
String[] interceptors = AnnotationUtils.getAnnotation(method, IdempotentReceiver.class).value();
for (String interceptor : interceptors) {
DefaultBeanFactoryPointcutAdvisor advisor = new DefaultBeanFactoryPointcutAdvisor();
advisor.setAdviceBeanName(interceptor);
NameMatchMethodPointcut pointcut = new NameMatchMethodPointcut();
pointcut.setMappedName("handleMessage");
advisor.setPointcut(pointcut);
advisor.setBeanFactory(this.beanFactory);
if (handler instanceof Advised) {
((Advised) handler).addAdvisor(advisor);
}
else {
ProxyFactory proxyFactory = new ProxyFactory(bean);
proxyFactory.addAdvisor(advisor);
handler = (MessageHandler) proxyFactory.getProxy(this.beanFactory.getBeanClassLoader());
}
}
}
AbstractEndpoint endpoint = createEndpoint(handler, method, annotations);
if (endpoint != null) {
return endpoint;

69
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java
View File

@@ -22,11 +22,14 @@ import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.config.IdempotentReceiverAutoProxyCreatorInitializer;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor;
import org.springframework.integration.metadata.ExpressionMetadataKeyStrategy;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.integration.selector.MetadataStoreSelector;
import org.springframework.util.StringUtils;
@@ -46,28 +49,38 @@ public class IdempotentReceiverInterceptorParser extends AbstractBeanDefinitionP
boolean hasSelector = StringUtils.hasText(selector);
String store = element.getAttribute("metadata-store");
boolean hasStore = StringUtils.hasText(store);
String strategy = element.getAttribute("key-strategy");
boolean hasStrategy = StringUtils.hasText(strategy);
String expression = element.getAttribute("key-expression");
boolean hasExpression = StringUtils.hasText(expression);
String keyStrategy = element.getAttribute("key-strategy");
boolean hasKeyStrategy = StringUtils.hasText(keyStrategy);
String keyExpression = element.getAttribute("key-expression");
boolean hasKeyExpression = StringUtils.hasText(keyExpression);
String valueStrategy = element.getAttribute("value-strategy");
boolean hasValueStrategy = StringUtils.hasText(valueStrategy);
String valueExpression = element.getAttribute("value-expression");
boolean hasValueExpression = StringUtils.hasText(valueExpression);
String endpoints = element.getAttribute("endpoint");
if (!hasSelector & !(hasStrategy | hasExpression)) {
if (!hasSelector & !(hasKeyStrategy | hasKeyExpression)) {
parserContext.getReaderContext().error("One of the 'selector', 'key-strategy' or 'key-expression' " +
"attributes must be provided", source);
}
if (hasSelector & (hasStore | hasStrategy | hasExpression)) {
if (hasSelector & (hasStore | hasKeyStrategy | hasKeyExpression | hasValueStrategy | hasValueExpression)) {
parserContext.getReaderContext().error("The 'selector' attribute is mutually exclusive with " +
"'metadata-store', 'key-strategy' or 'key-expression'", source);
"'metadata-store', 'key-strategy', 'key-expression', 'value-strategy' " +
"or 'value-expression'", source);
}
if (hasStrategy & hasExpression) {
if (hasKeyStrategy & hasKeyExpression) {
parserContext.getReaderContext().error("The 'key-strategy' and 'key-expression' attributes " +
"are mutually exclusive", source);
}
if (hasValueStrategy & hasValueExpression) {
parserContext.getReaderContext().error("The 'value-strategy' and 'value-expression' attributes " +
"are mutually exclusive", source);
}
if (!StringUtils.hasText(endpoints)) {
parserContext.getReaderContext().error("The 'endpoint' attribute is required", source);
}
@@ -79,20 +92,44 @@ public class IdempotentReceiverInterceptorParser extends AbstractBeanDefinitionP
else {
BeanDefinitionBuilder selectorBuilder =
BeanDefinitionBuilder.genericBeanDefinition(MetadataStoreSelector.class);
BeanMetadataElement strategyBeanDefinition = null;
if (hasStrategy) {
strategyBeanDefinition = new RuntimeBeanReference(strategy);
BeanMetadataElement keyStrategyBeanDefinition = null;
if (hasKeyStrategy) {
keyStrategyBeanDefinition = new RuntimeBeanReference(keyStrategy);
}
else {
strategyBeanDefinition =
BeanDefinitionBuilder.genericBeanDefinition(ExpressionMetadataKeyStrategy.class)
.addConstructorArgValue(expression)
keyStrategyBeanDefinition =
BeanDefinitionBuilder.genericBeanDefinition(ExpressionEvaluatingMessageProcessor.class)
.addConstructorArgValue(
BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class).
addConstructorArgValue(keyExpression)
.getBeanDefinition()
)
.getBeanDefinition();
}
selectorBuilder.addConstructorArgValue(strategyBeanDefinition);
selectorBuilder.addConstructorArgValue(keyStrategyBeanDefinition);
BeanMetadataElement valueStrategyBeanDefinition = null;
if (hasValueStrategy) {
valueStrategyBeanDefinition = new RuntimeBeanReference(valueStrategy);
}
else if (hasValueExpression) {
valueStrategyBeanDefinition =
BeanDefinitionBuilder.genericBeanDefinition(ExpressionEvaluatingMessageProcessor.class)
.addConstructorArgValue(
BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class).
addConstructorArgValue(valueExpression)
.getBeanDefinition()
)
.getBeanDefinition();
}
selectorBuilder.addConstructorArgValue(valueStrategyBeanDefinition);
if (hasStore) {
selectorBuilder.addConstructorArgReference(store);
}
else {
selectorBuilder.addConstructorArgValue(new RootBeanDefinition(SimpleMetadataStore.class));
}
selectorBeanDefinition = selectorBuilder.getBeanDefinition();
}

11
spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java
View File

@@ -26,6 +26,7 @@ import org.springframework.util.Assert;
* with the Message itself as the root object within the evaluation context.
*
* @author Mark Fisher
* @author Artem Bilan
* @since 2.0
*/
public class ExpressionEvaluatingMessageProcessor<T> extends AbstractMessageProcessor<T> {
@@ -37,18 +38,15 @@ public class ExpressionEvaluatingMessageProcessor<T> extends AbstractMessageProc
/**
* Create an {@link ExpressionEvaluatingMessageProcessor} for the given expression.
*
* @param expression The expression.
*/
public ExpressionEvaluatingMessageProcessor(Expression expression) {
this(expression, null);
}
/**
* Create an {@link ExpressionEvaluatingMessageProcessor} for the given expression
* and expected type for its evaluation result.
*
* @param expression The expression.
* @param expectedType The expected type.
*/
@@ -63,11 +61,9 @@ public class ExpressionEvaluatingMessageProcessor<T> extends AbstractMessageProc
}
}
/**
* Processes the Message by evaluating the expression with that Message as the
* root object. The expression evaluation result Object will be returned.
*
* @param message The message.
* @return The result of processing the message.
*/
@@ -76,4 +72,9 @@ public class ExpressionEvaluatingMessageProcessor<T> extends AbstractMessageProc
return this.evaluateExpression(this.expression, message, this.expectedType);
}
@Override
public String toString() {
return "ExpressionEvaluatingMessageProcessor for: [" + this.expression.getExpressionString() + "]";
}
}

63
spring-integration-core/src/main/java/org/springframework/integration/metadata/ExpressionMetadataKeyStrategy.java
View File

@@ -1,63 +0,0 @@
/*
* Copyright 2014 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.metadata;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.messaging.Message;
/**
* The expression based {@link MetadataKeyStrategy} implementation.
* The provided {@link Message} is used as the evaluation context root object.
*
* @author Artem Bilan
* @since 4.1
*/
public class ExpressionMetadataKeyStrategy implements MetadataKeyStrategy, BeanFactoryAware {
private static final ExpressionParser PARSER = new SpelExpressionParser();
private final MessageProcessor<String> processor;
private final String expressionString;
public ExpressionMetadataKeyStrategy(String expressionString) {
this.processor = new ExpressionEvaluatingMessageProcessor<String>(PARSER.parseExpression(expressionString));
this.expressionString = expressionString;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
((BeanFactoryAware) this.processor).setBeanFactory(beanFactory);
}
@Override
public String getKey(Message<?> message) {
return this.processor.processMessage(message);
}
@Override
public String toString() {
return "ExpressionEvaluatingSelector for: [" + this.expressionString + "]";
}
}

32
spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataKeyStrategy.java
View File

@@ -1,32 +0,0 @@
/*
* Copyright 2014 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.metadata;
import org.springframework.messaging.Message;
/**
* The strategy to extract a {@code key} for the {@code MetadataStore}
* from the provided {@link Message}.
*
* @author Artem Bilan
* @since 4.1
*/
public interface MetadataKeyStrategy {
String getKey(Message<?> message);
}

37
spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java
View File

@@ -17,19 +17,19 @@
package org.springframework.integration.selector;
import org.springframework.integration.core.MessageSelector;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.metadata.ConcurrentMetadataStore;
import org.springframework.integration.metadata.MetadataKeyStrategy;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* The {@link MessageSelector} implementation using a {@link ConcurrentMetadataStore}
* and {@link MetadataKeyStrategy}.
* and {@link MessageProcessor<String>}.
* <p>
* The {@link #accept} method extracts {@code metadataKey} from the provided {@code message}
* using {@link MetadataKeyStrategy} and uses the {@code timestamp} header as the {@code value}
* (hex).
* using {@link MessageProcessor<String>} and uses the {@code timestamp} header as the {@code value}
* (hex) by default. The {@link #valueStrategy} can be provided to override the default behaviour.
* <p>
* The successful result of the {@link #accept} method is based on the
* {@link ConcurrentMetadataStore#putIfAbsent} return value. {@code true} is returned
@@ -52,23 +52,38 @@ public class MetadataStoreSelector implements MessageSelector {
private final ConcurrentMetadataStore metadataStore;
private final MetadataKeyStrategy keyStrategy;
private final MessageProcessor<String> keyStrategy;
public MetadataStoreSelector(MetadataKeyStrategy keyStrategy) {
this(keyStrategy, new SimpleMetadataStore());
private final MessageProcessor<String> valueStrategy;
public MetadataStoreSelector(MessageProcessor<String> keyStrategy) {
this(keyStrategy, (MessageProcessor<String>) null);
}
public MetadataStoreSelector(MetadataKeyStrategy keyStrategy, ConcurrentMetadataStore metadataStore) {
Assert.notNull(metadataStore);
public MetadataStoreSelector(MessageProcessor<String> keyStrategy, MessageProcessor<String> valueStrategy) {
this(keyStrategy, valueStrategy, new SimpleMetadataStore());
}
public MetadataStoreSelector(MessageProcessor<String> keyStrategy, ConcurrentMetadataStore metadataStore) {
this(keyStrategy, null, metadataStore);
}
public MetadataStoreSelector(MessageProcessor<String> keyStrategy, MessageProcessor<String> valueStrategy,
ConcurrentMetadataStore metadataStore) {
Assert.notNull(keyStrategy);
Assert.notNull(metadataStore);
this.metadataStore = metadataStore;
this.keyStrategy = keyStrategy;
this.valueStrategy = valueStrategy;
}
@Override
public boolean accept(Message<?> message) {
String key = this.keyStrategy.getKey(message);
String value = Long.toString(message.getHeaders().getTimestamp());
String key = this.keyStrategy.processMessage(message);
String value = (this.valueStrategy != null)
? this.valueStrategy.processMessage(message)
: Long.toString(message.getHeaders().getTimestamp());
return this.metadataStore.putIfAbsent(key, value) == null;
}

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

@@ -4364,11 +4364,11 @@ The list of component name patterns you want to track (e.g., tracked-components
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.metadata.MetadataKeyStrategy" />
<tool:expected-type type="org.springframework.integration.handler.MessageProcessor" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
The 'MetadataKeyStrategy' reference. Used by the underlying
A 'MessageProcessor' reference. Used by the underlying
'org.springframework.integration.selector.MetadataStoreSelector'.
Evaluates an 'idempotentKey' from the request Message.
Mutually exclusive with 'selector' and 'key-expression'.
@@ -4378,13 +4378,41 @@ The list of component name patterns you want to track (e.g., tracked-components
<xsd:attribute name="key-expression">
<xsd:annotation>
<xsd:documentation><![CDATA[
Expression to populate an 'org.springframework.integration.metadata.ExpressionMetadataKeyStrategy'.
Expression to populate an 'ExpressionEvaluatingMessageProcessor'.
Used by the underlying 'org.springframework.integration.selector.MetadataStoreSelector'.
Evaluates an 'idempotentKey' using the request Message as the evaluation context root object.
Mutually exclusive with 'selector' and 'key-strategy'.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-strategy" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.handler.MessageProcessor" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A 'MessageProcessor' reference. Used by the underlying
'org.springframework.integration.selector.MetadataStoreSelector'.
Evaluates a 'value' for the 'idempotentKey' from the request Message.
Mutually exclusive with 'selector' and 'value-expression'.
By default, the 'MetadataStoreSelector' uses the 'timestamp' message header as the Metadata 'value'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-expression">
<xsd:annotation>
<xsd:documentation><![CDATA[
Expression to populate an 'ExpressionEvaluatingMessageProcessor'.
Used by the underlying 'org.springframework.integration.selector.MetadataStoreSelector'.
Evaluates a 'value' for the 'idempotentKey' using the request Message as the evaluation context
root object.
Mutually exclusive with 'selector' and 'value-strategy'.
By default, the 'MetadataStoreSelector' uses the 'timestamp' message header as the Metadata 'value'.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="discard-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>

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

@@ -16,11 +16,19 @@
<idempotent-receiver id="selectorInterceptor" endpoint="foo" selector="selector"/>
<beans:bean id="keyStrategy" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.integration.metadata.MetadataKeyStrategy"/>
<beans:constructor-arg value="org.springframework.integration.handler.MessageProcessor"/>
</beans:bean>
<idempotent-receiver id="strategyInterceptor" endpoint="foo" key-strategy="keyStrategy"
discard-channel="nullChannel" throw-exception-on-rejection="true"/>
<beans:bean id="valueStrategy" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.integration.handler.MessageProcessor"/>
</beans:bean>
<idempotent-receiver id="strategyInterceptor"
endpoint="foo"
key-strategy="keyStrategy"
value-strategy="valueStrategy"
discard-channel="nullChannel"
throw-exception-on-rejection="true"/>
<beans:bean id="store" class="org.springframework.integration.metadata.SimpleMetadataStore"/>
@@ -31,7 +39,7 @@
<context:property-placeholder properties-ref="properties"/>
<idempotent-receiver id="expressionInterceptor" endpoint="foo, ${bar}"
metadata-store="store"
key-expression="headers.foo"/>
metadata-store="store"
key-expression="headers.foo"/>
</beans:beans>

68
spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests.java
View File

@@ -46,9 +46,9 @@ import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.InputStreamResource;
import org.springframework.integration.core.MessageSelector;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor;
import org.springframework.integration.metadata.ExpressionMetadataKeyStrategy;
import org.springframework.integration.metadata.MetadataKeyStrategy;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.integration.selector.MetadataStoreSelector;
import org.springframework.messaging.MessageChannel;
@@ -79,7 +79,10 @@ public class IdempotentReceiverParserTests {
private IdempotentReceiverInterceptor strategyInterceptor;
@Autowired
private MetadataKeyStrategy keyStrategy;
private MessageProcessor<String> keyStrategy;
@Autowired
private MessageProcessor<String> valueStrategy;
@Autowired
@Qualifier("nullChannel")
@@ -113,6 +116,7 @@ public class IdempotentReceiverParserTests {
Object messageSelector = getPropertyValue(this.strategyInterceptor, "messageSelector");
assertThat(messageSelector, instanceOf(MetadataStoreSelector.class));
assertSame(this.keyStrategy, getPropertyValue(messageSelector, "keyStrategy"));
assertSame(this.valueStrategy, getPropertyValue(messageSelector, "valueStrategy"));
@SuppressWarnings("unchecked")
Map<String, List<String>> idempotentEndpoints =
(Map<String, List<String>>) getPropertyValue(this.idempotentReceiverAutoProxyCreator,
@@ -129,7 +133,7 @@ public class IdempotentReceiverParserTests {
assertThat(messageSelector, instanceOf(MetadataStoreSelector.class));
assertSame(this.store, getPropertyValue(messageSelector, "metadataStore"));
Object keyStrategy = getPropertyValue(messageSelector, "keyStrategy");
assertThat(keyStrategy, instanceOf(ExpressionMetadataKeyStrategy.class));
assertThat(keyStrategy, instanceOf(ExpressionEvaluatingMessageProcessor.class));
assertThat(keyStrategy.toString(), containsString("headers.foo"));
@SuppressWarnings("unchecked")
Map<String, List<String>> idempotentEndpoints =
@@ -176,40 +180,66 @@ public class IdempotentReceiverParserTests {
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage(),
containsString("The 'selector' attribute is mutually exclusive with 'metadata-store', " +
"'key-strategy' or 'key-expression'"));
"'key-strategy', 'key-expression', 'value-strategy' or 'value-expression'"));
}
}
@Test
public void testSelectorAndStrategy() throws Exception {
public void testSelectorAndKeyStrategy() throws Exception {
try {
bootStrap("selector-and-strategy");
bootStrap("selector-and-key-strategy");
fail("BeanDefinitionParsingException expected");
}
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage(),
containsString("The 'selector' attribute is mutually exclusive with 'metadata-store', " +
"'key-strategy' or 'key-expression'"));
"'key-strategy', 'key-expression', 'value-strategy' or 'value-expression'"));
}
}
@Test
public void testSelectorAndExpression() throws Exception {
public void testSelectorAndKeyExpression() throws Exception {
try {
bootStrap("selector-and-expression");
bootStrap("selector-and-key-expression");
fail("BeanDefinitionParsingException expected");
}
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage(),
containsString("The 'selector' attribute is mutually exclusive with 'metadata-store', " +
"'key-strategy' or 'key-expression'"));
"'key-strategy', 'key-expression', 'value-strategy' or 'value-expression'"));
}
}
@Test
public void testStrategyAndExpression() throws Exception {
public void testSelectorAndValueStrategy() throws Exception {
try {
bootStrap("strategy-and-expression");
bootStrap("selector-and-value-strategy");
fail("BeanDefinitionParsingException expected");
}
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage(),
containsString("The 'selector' attribute is mutually exclusive with 'metadata-store', " +
"'key-strategy', 'key-expression', 'value-strategy' or 'value-expression'"));
}
}
@Test
public void testSelectorAndValueExpression() throws Exception {
try {
bootStrap("selector-and-value-expression");
fail("BeanDefinitionParsingException expected");
}
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage(),
containsString("The 'selector' attribute is mutually exclusive with 'metadata-store', " +
"'key-strategy', 'key-expression', 'value-strategy' or 'value-expression'"));
}
}
@Test
public void testKeyStrategyAndKeyExpression() throws Exception {
try {
bootStrap("key-strategy-and-key-expression");
fail("BeanDefinitionParsingException expected");
}
catch (BeanDefinitionParsingException e) {
@@ -218,6 +248,18 @@ public class IdempotentReceiverParserTests {
}
}
@Test
public void testValueStrategyAndValueExpression() throws Exception {
try {
bootStrap("value-strategy-and-value-expression");
fail("BeanDefinitionParsingException expected");
}
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage(),
containsString("The 'value-strategy' and 'value-expression' attributes are mutually exclusive"));
}
}
private ApplicationContext bootStrap(String configProperty) throws Exception {
PropertiesFactoryBean pfb = new PropertiesFactoryBean();
pfb.setLocation(new ClassPathResource(

15
spring-integration-core/src/test/java/org/springframework/integration/config/xml/idempotent-receiver-configs.properties
View File

@@ -13,8 +13,17 @@ without-endpoint=<int:idempotent-receiver endpoint="" selector="selector"/>
selector-and-store=<int:idempotent-receiver endpoint="foo" selector="selector" metadata-store="store"/>
selector-and-strategy=<int:idempotent-receiver endpoint="foo" selector="selector" key-strategy="strategy"/>
selector-and-key-strategy=<int:idempotent-receiver endpoint="foo" selector="selector" key-strategy="strategy"/>
selector-and-expression=<int:idempotent-receiver endpoint="foo" selector="selector" key-expression="expression"/>
selector-and-key-expression=<int:idempotent-receiver endpoint="foo" selector="selector" key-expression="expression"/>
strategy-and-expression=<int:idempotent-receiver endpoint="foo" key-strategy="strategy" key-expression="expression"/>
selector-and-value-strategy=<int:idempotent-receiver endpoint="foo" selector="selector" value-strategy="strategy"/>
selector-and-value-expression=<int:idempotent-receiver endpoint="foo" selector="selector" \
value-expression="expression"/>
key-strategy-and-key-expression=<int:idempotent-receiver endpoint="foo" key-strategy="strategy" \
key-expression="expression"/>
value-strategy-and-value-expression=<int:idempotent-receiver endpoint="foo" key-strategy="strategy" \
value-strategy="strategy" value-expression="expression"/>

9
spring-integration-core/src/test/java/org/springframework/integration/handler/advice/IdempotentReceiverTests.java
View File

@@ -33,16 +33,14 @@ import org.mockito.Mockito;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.metadata.ConcurrentMetadataStore;
import org.springframework.integration.metadata.ExpressionMetadataKeyStrategy;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.integration.selector.MetadataStoreSelector;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -86,7 +84,8 @@ public class IdempotentReceiverTests {
@Test
public void testIdempotentReceiverInterceptor() {
ConcurrentMetadataStore store = new SimpleMetadataStore();
ExpressionMetadataKeyStrategy idempotentKeyStrategy = new ExpressionMetadataKeyStrategy("payload");
ExpressionEvaluatingMessageProcessor<String> idempotentKeyStrategy =
new ExpressionEvaluatingMessageProcessor<>(new SpelExpressionParser().parseExpression("payload"));
BeanFactory beanFactory = Mockito.mock(BeanFactory.class);
idempotentKeyStrategy.setBeanFactory(beanFactory);
IdempotentReceiverInterceptor idempotentReceiverInterceptor =

66
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java
View File

@@ -23,6 +23,8 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
@@ -36,14 +38,15 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.annotation.IdempotentReceiver;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor;
import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport;
import org.springframework.integration.metadata.ConcurrentMetadataStore;
import org.springframework.integration.metadata.MetadataKeyStrategy;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.integration.selector.MetadataStoreSelector;
@@ -54,6 +57,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Component;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -82,15 +86,24 @@ public class IdempotentReceiverIntegrationTests {
@Autowired
private AtomicInteger adviceCalled;
@Autowired
private MessageChannel annotatedMethodChannel;
@Autowired
private FooService fooService;
@Test
public void testIdempotentReceiver() {
this.idempotentReceiverInterceptor.setThrowExceptionOnRejection(true);
TestUtils.getPropertyValue(this.store, "metadata", Map.class).clear();
Message<String> message = new GenericMessage<String>("foo");
this.input.send(message);
Message<?> receive = this.output.receive(10000);
assertNotNull(receive);
assertEquals(1, this.adviceCalled.get());
assertEquals(1, TestUtils.getPropertyValue(this.store, "metadata", Map.class).size());
assertNotNull(this.store.get("foo"));
String foo = this.store.get("foo");
assertEquals("FOO", foo);
try {
this.input.send(message);
@@ -108,6 +121,18 @@ public class IdempotentReceiverIntegrationTests {
assertEquals(1, TestUtils.getPropertyValue(store, "metadata", Map.class).size());
}
@Test
public void testIdempotentReceiverOnMethod() {
TestUtils.getPropertyValue(this.store, "metadata", Map.class).clear();
Message<String> message = new GenericMessage<String>("foo");
this.annotatedMethodChannel.send(message);
this.annotatedMethodChannel.send(message);
assertEquals(2, this.fooService.messages.size());
assertTrue(this.fooService.messages.get(1).getHeaders().get(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE,
Boolean.class));
}
@Configuration
@EnableIntegration
@EnableIntegrationMBeanExport(server = "mBeanServer")
@@ -125,17 +150,21 @@ public class IdempotentReceiverIntegrationTests {
@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
IdempotentReceiverInterceptor idempotentReceiverInterceptor =
new IdempotentReceiverInterceptor(new MetadataStoreSelector(new MetadataKeyStrategy() {
return new IdempotentReceiverInterceptor(new MetadataStoreSelector(new MessageProcessor<String>() {
@Override
public String getKey(Message<?> message) {
public String processMessage(Message<?> message) {
return message.getPayload().toString();
}
}, new MessageProcessor<String>() {
@Override
public String processMessage(Message<?> message) {
return message.getPayload().toString().toUpperCase();
}
}, store()));
idempotentReceiverInterceptor.setThrowExceptionOnRejection(true);
return idempotentReceiverInterceptor;
}
@Bean
@@ -182,6 +211,29 @@ public class IdempotentReceiverIntegrationTests {
};
}
@Bean
public MessageChannel annotatedMethodChannel() {
return new DirectChannel();
}
@Bean
public FooService fooService() {
return new FooService();
}
}
@Component
private static class FooService {
private List<Message<?>> messages = new ArrayList<Message<?>>();
@ServiceActivator(inputChannel = "annotatedMethodChannel")
@IdempotentReceiver("idempotentReceiverInterceptor")
public void handle(Message<?> message) {
this.messages.add(message);
}
}
}

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

@@ -616,12 +616,13 @@ public class MyAdvisedFilter {
<para>
To maintain <emphasis>state</emphasis> between messages and provide the ability to compare messages for the
idempotency, the <classname>MetadataStoreSelector</classname> is provided. It accepts a
<interfacename>MetadataKeyStrategy</interfacename> implementation (which creates a lookup key
<interfacename>MessageProcessor</interfacename> implementation (which creates a lookup key
based on the <classname>Message</classname>) and an optional
<interfacename>ConcurrentMetadataStore</interfacename> (<xref linkend="metadata-store"/>).
See the <classname>MetadataStoreSelector</classname> JavaDocs for more information. An
<classname>ExpressionMetadataKeyStrategy</classname> implementation is provided, allowing
simple SpEL expressions to be used to determine the key from the message.
See the <classname>MetadataStoreSelector</classname> JavaDocs for more information.
The <code>value</code> for <interfacename>ConcurrentMetadataStore</interfacename> also can be customized
using additional <interfacename>MessageProcessor</interfacename>. By default
<classname>MetadataStoreSelector</classname> uses <code>timestamp</code> message header.
</para>
<para>
For convenience, the <classname>MetadataStoreSelector</classname> options are configurable directly on
@@ -635,7 +636,9 @@ public class MyAdvisedFilter {
metadata-store="" ]]><co id="iri5" linkends="iri5-txt" /><![CDATA[
key-strategy="" ]]><co id="iri6" linkends="iri6-txt" /><![CDATA[
key-expression="" ]]><co id="iri7" linkends="iri7-txt" /><![CDATA[
throw-exception-on-rejection="" /> ]]><co id="iri8" linkends="iri8-txt" /></programlisting>
value-strategy="" ]]><co id="iri8" linkends="iri8-txt" /><![CDATA[
value-expression="" ]]><co id="iri9" linkends="iri9-txt" /><![CDATA[
throw-exception-on-rejection="" /> ]]><co id="iri10" linkends="iri10-txt" /></programlisting>
</section>
<calloutlist>
<callout arearefs="iri1" id="iri1-txt">
@@ -658,7 +661,9 @@ public class MyAdvisedFilter {
<para>
A <interfacename>MessageSelector</interfacename> bean reference.
Mutually exclusive with <code>metadata-store</code> and
<code>key-strategy (key-expression)</code>.
<code>key-strategy (key-expression)</code>. When <code>selector</code>
is not provided, one of <code>key-strategy</code> or <code>key-strategy-expression</code>
is required.
</para>
</callout>
@@ -681,23 +686,50 @@ public class MyAdvisedFilter {
<callout arearefs="iri6" id="iri6-txt">
<para>
A <interfacename>MetadataKeyStrategy</interfacename> reference. Used by the underlying
A <interfacename>MessageProcessor</interfacename> reference. Used by the underlying
<classname>MetadataStoreSelector</classname>.
Evaluates an <code>idempotentKey</code> from the request Message.
Mutually exclusive with <code>selector</code> and <code>key-expression</code>.
When a <code>selector</code>
is not provided, one of <code>key-strategy</code> or <code>key-strategy-expression</code>
is required.
</para>
</callout>
<callout arearefs="iri7" id="iri7-txt">
<para>
A SpEL expression to populate an <classname>ExpressionMetadataKeyStrategy</classname>.
A SpEL expression to populate an <classname>ExpressionEvaluatingMessageProcessor</classname>.
Used by the underlying <classname>MetadataStoreSelector</classname>.
Evaluates an <code>idempotentKey</code> using the request Message as the evaluation context root object.
Mutually exclusive with <code>selector</code> and <code>key-strategy</code>.
When a <code>selector</code>
is not provided, one of <code>key-strategy</code> or <code>key-strategy-expression</code>
is required.
</para>
</callout>
<callout arearefs="iri8" id="iri8-txt">
<para>
A <interfacename>MessageProcessor</interfacename> reference. Used by the underlying
<classname>MetadataStoreSelector</classname>.
Evaluates a <code>value</code> for the <code>idempotentKey</code> from the request Message.
Mutually exclusive with <code>selector</code> and <code>value-expression</code>.
By default, the 'MetadataStoreSelector' uses the 'timestamp' message header as the Metadata 'value'.
</para>
</callout>
<callout arearefs="iri8" id="iri8-txt">
<para>
A SpEL expression to populate an <classname>ExpressionEvaluatingMessageProcessor</classname>.
Used by the underlying <classname>MetadataStoreSelector</classname>.
Evaluates a <code>value</code> for the <code>idempotentKey</code> using the request Message
as the evaluation context root object.
Mutually exclusive with <code>selector</code> and <code>value-strategy</code>.
By default, the 'MetadataStoreSelector' uses the 'timestamp' message header as the Metadata 'value'.
</para>
</callout>
<callout arearefs="iri10" id="iri10-txt">
<para>
Throw an exception if the <classname>IdempotentReceiverInterceptor</classname> rejects the message
defaults to <code>false</code>.
@@ -707,7 +739,7 @@ public class MyAdvisedFilter {
</calloutlist>
<para>
For Java configuration, the method level <classname>IdempotentReceiver</classname> annotation is provided. It
is used to mark a <code>@Bean</code> that has a Messaging annotation (<code>@ServiceActivator</code>,
is used to mark a <code>method</code> that has a Messaging annotation (<code>@ServiceActivator</code>,
<code>@Router</code> etc.) to specify which <classname>IdempotentReceiverInterceptor</classname>s will be
applied to this endpoint:
</para>