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

INT-2426: Add Idempotent Receiver EIP

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

INT-2426: pushed `final` modifier fix for Java 6 compatibility

INT-2426: Rework logic to the `MetadataStore`

INT-2426: `IdempotentReceiver` -> `IdempotentReceiverInterceptor`

* Move `Idempotent Filtering` logic to the `IdempotentReceiverInterceptor`, which should be applied as a regular
AOP `Advice` to the `MessageHandler#handleMessage`
* Provide an xml component `<idempotent-receiver>`
* Introduce `IdempotentReceiverAutoProxyCreator` to get deal with `IdempotentReceiverInterceptor` and `MessageHandler`s.
The `Proxying` logic is based on the mapping between interceptor and `consumer endpoint` `ids`
* Introduce `MetadataStoreSelector` along side with `MetadataKeyStrategy` and `ExpressionMetadataKeyStrategy` implementation

INT-2426: Introduce `IdempotentReceiver` annotation

Add `IdempotentReceiverIntegrationTests` in the JMX module to be sure that all proxying works well.

INT-2426: Polishing according PR comments

* Rename `IdempotentReceiverAutoProxyCreatorInitializer`
* Add support for several `IRI` for the one `MH`
* Polishing JavaDocs

INT-2426: Add `What's New` note

Doc Polishing.

More Doc Polishing

Use Timestamp (hex) instead of Id for Value

Facilitate cleanup.
This commit is contained in:
Artem Bilan
2014-09-24 19:54:43 +03:00
committed by Gary Russell
parent 6e81950236
commit ff2b15ea9e
23 changed files with 1573 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java
View File

@@ -50,6 +50,8 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
public static final String ROUTING_SLIP = "routingSlip";
public static final String DUPLICATE_MESSAGE = "duplicateMessage";
public IntegrationMessageHeaderAccessor(Message<?> message) {
super(message);
}
@@ -107,6 +109,10 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
Assert.isTrue(Map.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName
+ "' header value must be an List.");
}
else if (IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE.equals(headerName)) {
Assert.isTrue(Boolean.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName
+ "' header value must be an Boolean.");
}
}
}

46
spring-integration-core/src/main/java/org/springframework/integration/annotation/IdempotentReceiver.java Normal file
View File

@@ -0,0 +1,46 @@
/*
* 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.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* A {@code @Bean} 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.
* The interceptor bean names are provided in the {@link #value()}.
*
* @author Artem Bilan
* @since 4.1
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface IdempotentReceiver {
/**
* @return the {@link org.springframework.integration.handler.advice.IdempotentReceiverInterceptor}
* bean references.
*/
String[] value();
}

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

@@ -165,13 +165,14 @@ public class ConsumerEndpointFactoryBean
}
if (!CollectionUtils.isEmpty(this.adviceChain)) {
/*
* ARPMHs advise the handleRequesMessage method internally and already have the advice chain injected.
* ARPMHs advise the handleRequestMessage 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);
Class<?> targetClass = AopUtils.getTargetClass(this.handler);
if (!(AbstractReplyProducingMessageHandler.class.isAssignableFrom(targetClass))) {
if (AopUtils.isAopProxy(this.handler)) {
for (Advice advice : this.adviceChain) {
NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(advice);
handlerAdvice.addMethodName("handleMessage");

124
spring-integration-core/src/main/java/org/springframework/integration/config/IdempotentReceiverAutoProxyCreator.java Normal file
View File

@@ -0,0 +1,124 @@
/*
* 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.config;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.springframework.aop.Advisor;
import org.springframework.aop.TargetSource;
import org.springframework.aop.framework.autoproxy.AbstractAutoProxyCreator;
import org.springframework.aop.support.DefaultBeanFactoryPointcutAdvisor;
import org.springframework.aop.support.NameMatchMethodPointcut;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.PatternMatchUtils;
/**
* The {@link AbstractAutoProxyCreator} implementation that applies {@code IdempotentReceiverInterceptor}s
* to {@link MessageHandler}s mapped by their {@code endpoint beanName}.
*
* @author Artem Bilan
* @since 4.1
*/
@SuppressWarnings("serial")
class IdempotentReceiverAutoProxyCreator extends AbstractAutoProxyCreator {
private List<Map<String, String>> idempotentEndpointsMapping;
private Map<String, List<String>> idempotentEndpoints;
public void setIdempotentEndpointsMapping(List<Map<String, String>> idempotentEndpointsMapping) {
Assert.notEmpty(idempotentEndpointsMapping);
this.idempotentEndpointsMapping = idempotentEndpointsMapping;
}
@Override
protected Object[] getAdvicesAndAdvisorsForBean(Class<?> beanClass, String beanName,
TargetSource customTargetSource) throws BeansException {
initIdempotentEndpointsIfNecessary();
if (MessageHandler.class.isAssignableFrom(beanClass)) {
List<Advisor> interceptors = new ArrayList<Advisor>();
for (Map.Entry<String, List<String>> entry : this.idempotentEndpoints.entrySet()) {
List<String> mappedNames = entry.getValue();
for (String mappedName : mappedNames) {
String pattern = mappedName + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX;
if (isMatch(pattern, beanName)) {
DefaultBeanFactoryPointcutAdvisor idempotentReceiverInterceptor
= new DefaultBeanFactoryPointcutAdvisor();
idempotentReceiverInterceptor.setAdviceBeanName(entry.getKey());
NameMatchMethodPointcut pointcut = new NameMatchMethodPointcut();
pointcut.setMappedName("handleMessage");
idempotentReceiverInterceptor.setPointcut(pointcut);
idempotentReceiverInterceptor.setBeanFactory(getBeanFactory());
interceptors.add(idempotentReceiverInterceptor);
}
}
}
if (!interceptors.isEmpty()) {
return interceptors.toArray();
}
}
return DO_NOT_PROXY;
}
private void initIdempotentEndpointsIfNecessary() {
if (this.idempotentEndpoints == null) {
synchronized (this) {
if (this.idempotentEndpoints == null) {
this.idempotentEndpoints = new LinkedHashMap<String, List<String>>();
for (Map<String, String> mapping : this.idempotentEndpointsMapping) {
Assert.isTrue(mapping.size() == 1, "The 'idempotentEndpointMapping' must be a SingletonMap");
String interceptor = mapping.keySet().iterator().next();
String endpoint = mapping.values().iterator().next();
Assert.hasText(interceptor, "The 'idempotentReceiverInterceptor' can't be empty String");
Assert.hasText(endpoint, "The 'idempotentReceiverEndpoint' can't be empty String");
List<String> endpoints = this.idempotentEndpoints.get(interceptor);
if (endpoints == null) {
endpoints = new ArrayList<String>();
this.idempotentEndpoints.put(interceptor, endpoints);
}
endpoints.add(endpoint);
}
}
}
}
}
private boolean isMatch(String mappedName, String beanName) {
boolean matched = PatternMatchUtils.simpleMatch(mappedName, beanName);
if (!matched) {
BeanFactory beanFactory = getBeanFactory();
if (beanFactory != null) {
String[] aliases = beanFactory.getAliases(beanName);
for (String alias : aliases) {
matched = PatternMatchUtils.simpleMatch(mappedName, alias);
if (matched) {
break;
}
}
}
}
return matched;
}
}

106
spring-integration-core/src/main/java/org/springframework/integration/config/IdempotentReceiverAutoProxyCreatorInitializer.java Normal file
View File

@@ -0,0 +1,106 @@
/*
* 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.config;
import java.util.List;
import java.util.Map;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.support.ManagedMap;
import org.springframework.core.type.MethodMetadata;
import org.springframework.integration.annotation.IdempotentReceiver;
import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* The {@link IntegrationConfigurationInitializer} that populates
* the {@link ConfigurableListableBeanFactory}
* with an {@link IdempotentReceiverAutoProxyCreator}
* when {@code IdempotentReceiverInterceptor} {@link BeanDefinition}s and their {@code mapping}
* to Consumer Endpoints are present.
*
* @author Artem Bilan
* @since 4.1
*/
public class IdempotentReceiverAutoProxyCreatorInitializer implements IntegrationConfigurationInitializer {
public static final String IDEMPOTENT_ENDPOINTS_MAPPING = "IDEMPOTENT_ENDPOINTS_MAPPING";
private static final String IDEMPOTENT_RECEIVER_AUTO_PROXY_CREATOR_BEAN_NAME =
IdempotentReceiverAutoProxyCreator.class.getName();
@Override
public void initialize(ConfigurableListableBeanFactory beanFactory) throws BeansException {
BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
List<Map<String, String>> idempotentEndpointsMapping = new ManagedList<Map<String, String>>();
for (String beanName : registry.getBeanDefinitionNames()) {
BeanDefinition beanDefinition = registry.getBeanDefinition(beanName);
if (IdempotentReceiverInterceptor.class.getName().equals(beanDefinition.getBeanClassName())) {
Object value = beanDefinition.removeAttribute(IDEMPOTENT_ENDPOINTS_MAPPING);
Assert.isInstanceOf(String.class, value,
"The 'mapping' of BeanDefinition 'IDEMPOTENT_ENDPOINTS_MAPPING' must be String.");
String mapping = (String) value;
String[] endpoints = StringUtils.tokenizeToStringArray(mapping, ",");
for (String endpoint : endpoints) {
Map<String, String> idempotentEndpoint = new ManagedMap<String, String>();
idempotentEndpoint.put(beanName, beanFactory.resolveEmbeddedValue(endpoint));
idempotentEndpointsMapping.add(idempotentEndpoint);
}
}
else if (beanDefinition instanceof AnnotatedBeanDefinition) {
if (beanDefinition.getSource() instanceof MethodMetadata) {
MethodMetadata beanMethod = (MethodMetadata) beanDefinition.getSource();
String annotationType = IdempotentReceiver.class.getName();
if (beanMethod.isAnnotated(annotationType)) {
Object value = beanMethod.getAnnotationAttributes(annotationType).get("value");
if (value != null) {
String[] interceptors = (String[]) value;
/*
MessageHandler beans, populated from @Bean methods, have a complex id,
including @Configuration bean name, method name and the Messaging annotation name.
The following pattern matches the bean name, regardless of the annotation name.
*/
String endpoint = beanDefinition.getFactoryBeanName() + "." + beanName + ".*";
for (String interceptor : interceptors) {
Map<String, String> idempotentEndpoint = new ManagedMap<String, String>();
idempotentEndpoint.put(interceptor, endpoint);
idempotentEndpointsMapping.add(idempotentEndpoint);
}
}
}
}
}
}
if (!idempotentEndpointsMapping.isEmpty()) {
BeanDefinition bd = BeanDefinitionBuilder.rootBeanDefinition(IdempotentReceiverAutoProxyCreator.class)
.addPropertyValue("idempotentEndpointsMapping", idempotentEndpointsMapping)
.getBeanDefinition();
registry.registerBeanDefinition(IDEMPOTENT_RECEIVER_AUTO_PROXY_CREATOR_BEAN_NAME, bd);
}
}
}

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

@@ -0,0 +1,120 @@
/*
* 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.config.xml;
import org.w3c.dom.Element;
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.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.IdempotentReceiverAutoProxyCreatorInitializer;
import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor;
import org.springframework.integration.metadata.ExpressionMetadataKeyStrategy;
import org.springframework.integration.selector.MetadataStoreSelector;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;idempotent-receiver/&gt; element.
*
* @author Artem Bilan
* @since 4.1
*/
public class IdempotentReceiverInterceptorParser extends AbstractBeanDefinitionParser {
@Override
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
Object source = parserContext.extractSource(element);
String selector = element.getAttribute("selector");
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 endpoints = element.getAttribute("endpoint");
if (!hasSelector & !(hasStrategy | hasExpression)) {
parserContext.getReaderContext().error("One of the 'selector', 'key-strategy' or 'key-expression' " +
"attributes must be provided", source);
}
if (hasSelector & (hasStore | hasStrategy | hasExpression)) {
parserContext.getReaderContext().error("The 'selector' attribute is mutually exclusive with " +
"'metadata-store', 'key-strategy' or 'key-expression'", source);
}
if (hasStrategy & hasExpression) {
parserContext.getReaderContext().error("The 'key-strategy' and 'key-expression' attributes " +
"are mutually exclusive", source);
}
if (!StringUtils.hasText(endpoints)) {
parserContext.getReaderContext().error("The 'endpoint' attribute is required", source);
}
BeanMetadataElement selectorBeanDefinition = null;
if (hasSelector) {
selectorBeanDefinition = new RuntimeBeanReference(selector);
}
else {
BeanDefinitionBuilder selectorBuilder =
BeanDefinitionBuilder.genericBeanDefinition(MetadataStoreSelector.class);
BeanMetadataElement strategyBeanDefinition = null;
if (hasStrategy) {
strategyBeanDefinition = new RuntimeBeanReference(strategy);
}
else {
strategyBeanDefinition =
BeanDefinitionBuilder.genericBeanDefinition(ExpressionMetadataKeyStrategy.class)
.addConstructorArgValue(expression)
.getBeanDefinition();
}
selectorBuilder.addConstructorArgValue(strategyBeanDefinition);
if (hasStore) {
selectorBuilder.addConstructorArgReference(store);
}
selectorBeanDefinition = selectorBuilder.getBeanDefinition();
}
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(IdempotentReceiverInterceptor.class)
.addConstructorArgValue(selectorBeanDefinition);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "throw-exception-on-rejection");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "discard-channel");
AbstractBeanDefinition interceptorBeanDefinition = builder.getBeanDefinition();
interceptorBeanDefinition.setAttribute(
IdempotentReceiverAutoProxyCreatorInitializer.IDEMPOTENT_ENDPOINTS_MAPPING,
endpoints);
return interceptorBeanDefinition;
}
@Override
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected boolean shouldFireEvents() {
return false;
}
}

1
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java
View File

@@ -82,6 +82,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan
registerBeanDefinitionParser("handler-retry-advice", retryParser);
registerBeanDefinitionParser("retry-advice", retryParser);
registerBeanDefinitionParser("scatter-gather", new ScatterGatherParser());
registerBeanDefinitionParser("idempotent-receiver", new IdempotentReceiverInterceptorParser());
}
}

8
spring-integration-core/src/main/java/org/springframework/integration/filter/AbstractMessageProcessingSelector.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-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.
@@ -29,10 +29,10 @@ import org.springframework.util.Assert;
/**
* A base class for {@link MessageSelector} implementations that delegate to
* a {@link MessageProcessor}.
*
*
* @author Mark Fisher
*/
abstract class AbstractMessageProcessingSelector implements MessageSelector, BeanFactoryAware {
public abstract class AbstractMessageProcessingSelector implements MessageSelector, BeanFactoryAware {
private final MessageProcessor<Boolean> messageProcessor;
@@ -43,7 +43,7 @@ abstract class AbstractMessageProcessingSelector implements MessageSelector, Bea
}
protected void setConversionService(ConversionService conversionService) {
public void setConversionService(ConversionService conversionService) {
if (this.messageProcessor instanceof AbstractMessageProcessor) {
((AbstractMessageProcessor<Boolean>) this.messageProcessor).setConversionService(conversionService);
}

174
spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java Normal file
View File

@@ -0,0 +1,174 @@
/*
* 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.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.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.core.MessageSelector;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
/**
* The {@link MethodInterceptor} implementation for the
* <a href="http://www.eaipatterns.com/IdempotentReceiver.html">Idempotent Receiver</a>
* E.I. Pattern.
* <p>
* This {@link MethodInterceptor} works like a {@code MessageFilter} if {@link #discardChannel}
* is provided or {@link #throwExceptionOnRejection} is set to {@code true}.
* However if those properties aren't provided, this interceptor will create an new {@link Message}
* with a {@link IntegrationMessageHeaderAccessor#DUPLICATE_MESSAGE} header when the
* {@code requestMessage} isn't accepted by {@link MessageSelector}.
* <p>
* The {@code idempotent filtering} logic depends on the provided {@link MessageSelector}.
* <p>
* This class is designed to be used only for the {@link MessageHandler#handleMessage},
* method.
*
* @author Artem Bilan
* @since 4.1
* @see org.springframework.integration.selector.MetadataStoreSelector
* @see org.springframework.integration.config.IdempotentReceiverAutoProxyCreatorInitializer
*/
public class IdempotentReceiverInterceptor implements MethodInterceptor, BeanFactoryAware {
protected final Log logger = LogFactory.getLog(this.getClass());
private final MessagingTemplate messagingTemplate = new MessagingTemplate();
private final MessageSelector messageSelector;
private volatile MessageChannel discardChannel;
private volatile boolean throwExceptionOnRejection;
private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
public IdempotentReceiverInterceptor(MessageSelector messageSelector) {
Assert.notNull(messageSelector);
this.messageSelector = messageSelector;
}
/**
* Specify the timeout value for sending to the discard channel.
* @param timeout the timeout in milliseconds
*/
public void setTimeout(long timeout) {
this.messagingTemplate.setSendTimeout(timeout);
}
/**
* Specify whether this interceptor should throw a
* {@link MessageRejectedException} when its selector does not accept a
* Message. The default value is <code>false</code> meaning that rejected
* Messages will be discarded or
* enriched with {@link IntegrationMessageHeaderAccessor#DUPLICATE_MESSAGE}
* header and returned as normal to the {@code invocation.proceed()}.
* Typically this value would <em>not</em> be <code>true</code> when
* a discard channel is provided, but if it is, it will cause the
* exception to be thrown <em>after</em>
* the Message is sent to the discard channel,
* @param throwExceptionOnRejection true if an exception should be thrown.
* @see #setDiscardChannel(MessageChannel)
*/
public void setThrowExceptionOnRejection(boolean throwExceptionOnRejection) {
this.throwExceptionOnRejection = throwExceptionOnRejection;
}
/**
* Specify a channel where rejected Messages should be sent. If the discard
* channel is null (the default), duplicate Messages will be enriched with
* {@link IntegrationMessageHeaderAccessor#DUPLICATE_MESSAGE} header
* and returned as normal to the {@code invocation.proceed()}. However,
* the 'throwExceptionOnRejection' flag determines whether rejected Messages
* trigger an exception. That value is evaluated regardless of the presence
* of a discard channel.
* <p>
* If there is needed just silently 'drop' rejected messages configure the
* {@link #discardChannel} to the {@code nullChannel}.
* @param discardChannel The discard channel.
* @see #setThrowExceptionOnRejection(boolean)
*/
public void setDiscardChannel(MessageChannel discardChannel) {
this.discardChannel = discardChannel;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
}
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
Method method = invocation.getMethod();
Object invocationThis = invocation.getThis();
Object[] arguments = invocation.getArguments();
boolean isMessageHandler = invocationThis != null && invocationThis instanceof MessageHandler;
boolean isMessageMethod = method.getName().equals("handleMessage")
&& (arguments.length == 1 && arguments[0] instanceof Message);
if (!isMessageHandler || !isMessageMethod) {
if (logger.isWarnEnabled()) {
String clazzName = invocationThis == null
? method.getDeclaringClass().getName()
: invocationThis.getClass().getName();
logger.warn("This advice " + this.getClass().getName() +
" can only be used for MessageHandlers; an attempt to advise method '"
+ method.getName() + "' in '" + clazzName + "' is ignored");
}
return invocation.proceed();
}
Message<?> message = (Message<?>) arguments[0];
boolean accept = this.messageSelector.accept(message);
if (!accept) {
boolean discarded = false;
if (this.discardChannel != null) {
this.messagingTemplate.send(this.discardChannel, message);
discarded = true;
}
if (this.throwExceptionOnRejection) {
throw new MessageRejectedException(message, "IdempotentReceiver '" + this
+ "' rejected duplicate Message: " + message);
}
if (!discarded) {
arguments[0] = this.messageBuilderFactory.fromMessage(message)
.setHeader(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE, true).build();
}
else {
return null;
}
}
return invocation.proceed();
}
}

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

@@ -0,0 +1,63 @@
/*
* 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 Normal file
View File

@@ -0,0 +1,32 @@
/*
* 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);
}

76
spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java Normal file
View File

@@ -0,0 +1,76 @@
/*
* 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.selector;
import org.springframework.integration.core.MessageSelector;
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}.
* <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).
* <p>
* The successful result of the {@link #accept} method is based on the
* {@link ConcurrentMetadataStore#putIfAbsent} return value. {@code true} is returned
* if {@code putIfAbsent} returns {@code null}.
* And, at the same time, it means that the value has been placed in the {@code MetadataStore}.
* Otherwise the messages isn't accepted because there is already a value in the
* {@code MetadataStore} associated with the {@code key}.
* <p>
* This {@link MessageSelector} is useful for an
* <a href="http://www.eaipatterns.com/IdempotentReceiver.html">Idempotent Receiver</a>
* implementation.
* <p>
* It can be used in a {@link org.springframework.integration.filter.MessageFilter}
* or {@link org.springframework.integration.handler.advice.IdempotentReceiverInterceptor}.
*
* @author Artem Bilan
* @since 4.1
*/
public class MetadataStoreSelector implements MessageSelector {
private final ConcurrentMetadataStore metadataStore;
private final MetadataKeyStrategy keyStrategy;
public MetadataStoreSelector(MetadataKeyStrategy keyStrategy) {
this(keyStrategy, new SimpleMetadataStore());
}
public MetadataStoreSelector(MetadataKeyStrategy keyStrategy, ConcurrentMetadataStore metadataStore) {
Assert.notNull(metadataStore);
Assert.notNull(keyStrategy);
this.metadataStore = metadataStore;
this.keyStrategy = keyStrategy;
}
@Override
public boolean accept(Message<?> message) {
String key = this.keyStrategy.getKey(message);
String value = Long.toString(message.getHeaders().getTimestamp());
return this.metadataStore.putIfAbsent(key, value) == null;
}
}

3
spring-integration-core/src/main/resources/META-INF/spring.factories
View File

@@ -1,3 +1,4 @@
org.springframework.integration.config.IntegrationConfigurationInitializer=\
org.springframework.integration.config.GlobalChannelInterceptorInitializer,\
org.springframework.integration.config.IntegrationConverterInitializer
org.springframework.integration.config.IntegrationConverterInitializer,\
org.springframework.integration.config.IdempotentReceiverAutoProxyCreatorInitializer

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

@@ -4302,6 +4302,104 @@ The list of component name patterns you want to track (e.g., tracked-components
</xsd:attribute>
</xsd:complexType>
<xsd:element name="idempotent-receiver">
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines an IdempotentReceiverInterceptor.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="endpoint" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>
Consumer Endpoint name(s) or patterns. To specify more than one name(pattern) use ',' 
(e.g. endpoint="xxx, xxx*, *xxx, *xxx*, xxx*yyy").
The endpoint 'id' is used to retrieve the target endpoint's
'MessageHandler' bean (using its '.handler' suffix),
for which the 'IdempotentReceiverInterceptor' will be applied.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="selector" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageSelector" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A 'MessageSelector' reference. Used to identify if the message should be accepted
by the target 'MessageHandler' endpoint.
Mutually exclusive with 'metadata-store' and 'key-strategy' or 'key-expression'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="metadata-store" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.metadata.ConcurrentMetadataStore" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
The 'MetadataStore' reference. Used by the underlying
'org.springframework.integration.selector.MetadataStoreSelector'.
Mutually exclusive with 'selector'.
Optional. By default 'MetadataStoreSelector' uses an internal 'SimpleMetadataStore'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-strategy" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.metadata.MetadataKeyStrategy" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
The 'MetadataKeyStrategy' reference. Used by the underlying
'org.springframework.integration.selector.MetadataStoreSelector'.
Evaluates an 'idempotentKey' from the request Message.
Mutually exclusive with 'selector' and 'key-expression'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-expression">
<xsd:annotation>
<xsd:documentation><![CDATA[
Expression to populate an 'org.springframework.integration.metadata.ExpressionMetadataKeyStrategy'.
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="discard-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Identifies the channel to send a message when 'IdempotentReceiverInterceptor' doesn't accept
the message. When omitted, duplicate messages are forwarded to the handler with a
'DUPLICATE_MESSAGE' header.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="throw-exception-on-rejection" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Throw an exception if the 'IdempotentReceiverInterceptor' rejects the message (default false).
It is applied regardless of whether or not a discard channel is provided.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="control-bus-type">
<xsd:annotation>
<xsd:documentation><![CDATA[

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

@@ -0,0 +1,37 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
<beans:bean id="selector" class="org.springframework.integration.selector.PayloadTypeSelector">
<beans:constructor-arg value="java.lang.String"/>
</beans:bean>
<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:bean>
<idempotent-receiver id="strategyInterceptor" endpoint="foo" key-strategy="keyStrategy"
discard-channel="nullChannel" throw-exception-on-rejection="true"/>
<beans:bean id="store" class="org.springframework.integration.metadata.SimpleMetadataStore"/>
<util:properties id="properties">
<beans:prop key="bar">bar*</beans:prop>
</util:properties>
<context:property-placeholder properties-ref="properties"/>
<idempotent-receiver id="expressionInterceptor" endpoint="foo, ${bar}"
metadata-store="store"
key-expression="headers.foo"/>
</beans:beans>

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

@@ -0,0 +1,237 @@
/*
* 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.config.xml;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.springframework.integration.test.util.TestUtils.getPropertyValue;
import java.io.ByteArrayInputStream;
import java.util.List;
import java.util.Map;
import java.util.Properties;
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.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.PropertiesFactoryBean;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.beans.factory.xml.XmlBeanDefinitionReader;
import org.springframework.context.ApplicationContext;
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.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;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Artem Bilan
* @since 4.1
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class IdempotentReceiverParserTests {
@Autowired
@Qualifier("org.springframework.integration.config.IdempotentReceiverAutoProxyCreator")
private BeanPostProcessor idempotentReceiverAutoProxyCreator;
@Autowired
private IdempotentReceiverInterceptor selectorInterceptor;
@Autowired
private MessageSelector selector;
@Autowired
private IdempotentReceiverInterceptor strategyInterceptor;
@Autowired
private MetadataKeyStrategy keyStrategy;
@Autowired
@Qualifier("nullChannel")
private MessageChannel nullChannel;
@Autowired
private IdempotentReceiverInterceptor expressionInterceptor;
@Autowired
private MetadataStore store;
@Test
public void testSelectorInterceptor() {
assertSame(this.selector, getPropertyValue(this.selectorInterceptor, "messageSelector"));
assertNull(getPropertyValue(this.selectorInterceptor, "discardChannel"));
assertFalse(getPropertyValue(this.selectorInterceptor, "throwExceptionOnRejection", Boolean.class));
@SuppressWarnings("unchecked")
Map<String, List<String>> idempotentEndpoints =
(Map<String, List<String>>) getPropertyValue(this.idempotentReceiverAutoProxyCreator,
"idempotentEndpoints", Map.class);
List<String> endpoints = idempotentEndpoints.get("selectorInterceptor");
assertNotNull(endpoints);
assertFalse(endpoints.isEmpty());
assertTrue(endpoints.contains("foo"));
}
@Test
public void testStrategyInterceptor() {
assertSame(this.nullChannel, getPropertyValue(this.strategyInterceptor, "discardChannel"));
assertTrue(getPropertyValue(this.strategyInterceptor, "throwExceptionOnRejection", Boolean.class));
Object messageSelector = getPropertyValue(this.strategyInterceptor, "messageSelector");
assertThat(messageSelector, instanceOf(MetadataStoreSelector.class));
assertSame(this.keyStrategy, getPropertyValue(messageSelector, "keyStrategy"));
@SuppressWarnings("unchecked")
Map<String, List<String>> idempotentEndpoints =
(Map<String, List<String>>) getPropertyValue(this.idempotentReceiverAutoProxyCreator,
"idempotentEndpoints", Map.class);
List<String> endpoints = idempotentEndpoints.get("strategyInterceptor");
assertNotNull(endpoints);
assertFalse(endpoints.isEmpty());
assertTrue(endpoints.contains("foo"));
}
@Test
public void testExpressionInterceptor() {
Object messageSelector = getPropertyValue(this.expressionInterceptor, "messageSelector");
assertThat(messageSelector, instanceOf(MetadataStoreSelector.class));
assertSame(this.store, getPropertyValue(messageSelector, "metadataStore"));
Object keyStrategy = getPropertyValue(messageSelector, "keyStrategy");
assertThat(keyStrategy, instanceOf(ExpressionMetadataKeyStrategy.class));
assertThat(keyStrategy.toString(), containsString("headers.foo"));
@SuppressWarnings("unchecked")
Map<String, List<String>> idempotentEndpoints =
(Map<String, List<String>>) getPropertyValue(this.idempotentReceiverAutoProxyCreator,
"idempotentEndpoints", Map.class);
List<String> endpoints = idempotentEndpoints.get("expressionInterceptor");
assertNotNull(endpoints);
assertFalse(endpoints.isEmpty());
assertTrue(endpoints.contains("foo"));
assertTrue(endpoints.contains("bar*"));
}
@Test
public void testEmpty() throws Exception {
try {
bootStrap("empty");
fail("BeanDefinitionParsingException expected");
}
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage(),
containsString("One of the 'selector', 'key-strategy' or 'key-expression' attributes " +
"must be provided"));
}
}
@Test
public void testWithoutEndpoint() throws Exception {
try {
bootStrap("without-endpoint");
fail("BeanDefinitionParsingException expected");
}
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage(),
containsString("he 'endpoint' attribute is required"));
}
}
@Test
public void testSelectorAndStore() throws Exception {
try {
bootStrap("selector-and-store");
fail("BeanDefinitionParsingException expected");
}
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage(),
containsString("The 'selector' attribute is mutually exclusive with 'metadata-store', " +
"'key-strategy' or 'key-expression'"));
}
}
@Test
public void testSelectorAndStrategy() throws Exception {
try {
bootStrap("selector-and-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'"));
}
}
@Test
public void testSelectorAndExpression() throws Exception {
try {
bootStrap("selector-and-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'"));
}
}
@Test
public void testStrategyAndExpression() throws Exception {
try {
bootStrap("strategy-and-expression");
fail("BeanDefinitionParsingException expected");
}
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage(),
containsString("The 'key-strategy' and 'key-expression' attributes are mutually exclusive"));
}
}
private ApplicationContext bootStrap(String configProperty) throws Exception {
PropertiesFactoryBean pfb = new PropertiesFactoryBean();
pfb.setLocation(new ClassPathResource(
"org/springframework/integration/config/xml/idempotent-receiver-configs.properties"));
pfb.afterPropertiesSet();
Properties prop = pfb.getObject();
ByteArrayInputStream stream = new ByteArrayInputStream((prop.getProperty("xmlheaders")
+ prop.getProperty(configProperty) + prop.getProperty("xmlfooter")).getBytes());
GenericApplicationContext ac = new GenericApplicationContext();
XmlBeanDefinitionReader reader = new XmlBeanDefinitionReader(ac);
reader.setValidationMode(XmlBeanDefinitionReader.VALIDATION_XSD);
reader.loadBeanDefinitions(new InputStreamResource(stream));
ac.refresh();
return ac;
}
}

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

@@ -0,0 +1,20 @@
xmlheaders=\
<?xml version="1.0" encoding="UTF-8"?> \
<beans xmlns="http://www.springframework.org/schema/beans" \
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" \
xmlns:int="http://www.springframework.org/schema/integration" \
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd \
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
xmlfooter= </beans>
empty=<int:idempotent-receiver endpoint=""/>
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-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"/>

39
spring-integration-core/src/test/java/org/springframework/integration/handler/advice/IdempotentReceiverTests-context.xml Normal file
View File

@@ -0,0 +1,39 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<beans:bean id="advice" class="org.springframework.integration.handler.advice.IdempotentReceiverTests$FooAdvice"/>
<beans:bean id="store" class="org.springframework.integration.metadata.SimpleMetadataStore"/>
<beans:bean id="store2" class="org.springframework.integration.metadata.SimpleMetadataStore"/>
<idempotent-receiver id="idempotentReceiverInterceptor"
endpoint="my*, output"
metadata-store="store"
key-expression="payload"
throw-exception-on-rejection="true"/>
<idempotent-receiver id="idempotentReceiverInterceptor2"
endpoint="myService2"
metadata-store="store2"
key-expression="payload.toUpperCase()"/>
<channel id="output">
<queue/>
</channel>
<service-activator id="myService" input-channel="input" output-channel="output" expression="payload">
<request-handler-advice-chain>
<ref bean="advice"/>
</request-handler-advice-chain>
</service-activator>
<service-activator id="myService2" input-channel="input2" output-channel="output" expression="payload"/>
</beans:beans>

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

@@ -0,0 +1,174 @@
/*
* 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.handler.advice;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.junit.runner.RunWith;
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.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.MessageRejectedException;
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;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Artem Bilan
* @since 4.1
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class IdempotentReceiverTests {
@Autowired
private MessageChannel input;
@Autowired
private MessageChannel input2;
@Autowired
private PollableChannel output;
@Autowired
private MetadataStore store;
@Autowired
private MetadataStore store2;
@Autowired
private IdempotentReceiverInterceptor idempotentReceiverInterceptor;
@Autowired
private FooAdvice fooAdvice;
@Test
public void testIdempotentReceiverInterceptor() {
ConcurrentMetadataStore store = new SimpleMetadataStore();
ExpressionMetadataKeyStrategy idempotentKeyStrategy = new ExpressionMetadataKeyStrategy("payload");
BeanFactory beanFactory = Mockito.mock(BeanFactory.class);
idempotentKeyStrategy.setBeanFactory(beanFactory);
IdempotentReceiverInterceptor idempotentReceiverInterceptor =
new IdempotentReceiverInterceptor(new MetadataStoreSelector(idempotentKeyStrategy, store));
idempotentReceiverInterceptor.setThrowExceptionOnRejection(true);
AtomicReference<Message<?>> handled = new AtomicReference<>();
MessageHandler idempotentReceiver = handled::set;
ProxyFactory proxyFactory = new ProxyFactory(idempotentReceiver);
proxyFactory.addAdvice(idempotentReceiverInterceptor);
idempotentReceiver = (MessageHandler) proxyFactory.getProxy();
idempotentReceiver.handleMessage(new GenericMessage<>("foo"));
assertEquals(1, TestUtils.getPropertyValue(store, "metadata", Map.class).size());
assertNotNull(store.get("foo"));
try {
idempotentReceiver.handleMessage(new GenericMessage<>("foo"));
fail("MessageRejectedException expected");
}
catch (Exception e) {
assertThat(e, instanceOf(MessageRejectedException.class));
}
idempotentReceiverInterceptor.setThrowExceptionOnRejection(false);
idempotentReceiver.handleMessage(new GenericMessage<>("foo"));
assertTrue(handled.get().getHeaders().get(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE,
Boolean.class));
assertEquals(1, TestUtils.getPropertyValue(store, "metadata", Map.class).size());
}
@Test
public void testIdempotentReceiver() {
Message<String> message = new GenericMessage<>("foo");
this.input.send(message);
Message<?> receive = this.output.receive(10000);
assertNotNull(receive);
assertEquals(1, this.fooAdvice.adviceCalled);
assertEquals(1, TestUtils.getPropertyValue(this.store, "metadata", Map.class).size());
assertNotNull(this.store.get("foo"));
try {
this.input.send(message);
fail("MessageRejectedException expected");
}
catch (Exception e) {
assertThat(e, instanceOf(MessageRejectedException.class));
}
this.idempotentReceiverInterceptor.setThrowExceptionOnRejection(false);
this.input.send(message);
receive = this.output.receive(10000);
assertNotNull(receive);
assertEquals(2, this.fooAdvice.adviceCalled);
assertTrue(receive.getHeaders().get(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE, Boolean.class));
assertEquals(1, TestUtils.getPropertyValue(this.store, "metadata", Map.class).size());
message = new GenericMessage<>("bar");
for (int i = 0; i < 2; i++) {
this.input2.send(message);
receive = this.output.receive(10000);
assertNotNull(receive);
}
assertTrue(receive.getHeaders().get(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE, Boolean.class));
assertEquals(2, TestUtils.getPropertyValue(this.store, "metadata", Map.class).size());
assertNotNull(this.store.get("bar"));
assertEquals(1, TestUtils.getPropertyValue(this.store2, "metadata", Map.class).size());
assertNotNull(this.store2.get("BAR"));
}
public static class FooAdvice extends AbstractRequestHandlerAdvice {
private int adviceCalled;
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
adviceCalled++;
return callback.execute();
}
}
}

187
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java Normal file
View File

@@ -0,0 +1,187 @@
/*
* 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.monitor;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.aopalliance.aop.Advice;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
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.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
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;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.transformer.Transformer;
import org.springframework.jmx.support.MBeanServerFactoryBean;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Artem Bilan
* @since 4.1
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class IdempotentReceiverIntegrationTests {
@Autowired
private MessageChannel input;
@Autowired
private PollableChannel output;
@Autowired
private MetadataStore store;
@Autowired
private IdempotentReceiverInterceptor idempotentReceiverInterceptor;
@Autowired
private AtomicInteger adviceCalled;
@Test
public void testIdempotentReceiver() {
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"));
try {
this.input.send(message);
fail("MessageRejectedException expected");
}
catch (Exception e) {
assertThat(e, instanceOf(MessageRejectedException.class));
}
this.idempotentReceiverInterceptor.setThrowExceptionOnRejection(false);
this.input.send(message);
receive = this.output.receive(10000);
assertNotNull(receive);
assertEquals(2, this.adviceCalled.get());
assertTrue(receive.getHeaders().get(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE, Boolean.class));
assertEquals(1, TestUtils.getPropertyValue(store, "metadata", Map.class).size());
}
@Configuration
@EnableIntegration
@EnableIntegrationMBeanExport(server = "mBeanServer")
public static class ContextConfiguration {
@Bean
public static MBeanServerFactoryBean mBeanServer() {
return new MBeanServerFactoryBean();
}
@Bean
public ConcurrentMetadataStore store() {
return new SimpleMetadataStore();
}
@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
IdempotentReceiverInterceptor idempotentReceiverInterceptor =
new IdempotentReceiverInterceptor(new MetadataStoreSelector(new MetadataKeyStrategy() {
@Override
public String getKey(Message<?> message) {
return message.getPayload().toString();
}
}, store()));
idempotentReceiverInterceptor.setThrowExceptionOnRejection(true);
return idempotentReceiverInterceptor;
}
@Bean
public MessageChannel input() {
return new DirectChannel();
}
@Bean
public PollableChannel output() {
return new QueueChannel();
}
@Bean
@org.springframework.integration.annotation.Transformer(inputChannel = "input",
outputChannel = "output", adviceChain = "fooAdvice")
@IdempotentReceiver("idempotentReceiverInterceptor")
public Transformer transformer() {
return new Transformer() {
@Override
public Message<?> transform(Message<?> message) {
return message;
}
};
}
@Bean
public AtomicInteger adviceCalled() {
return new AtomicInteger();
}
@Bean
public Advice fooAdvice(final AtomicInteger adviceCalled) {
return new AbstractRequestHandlerAdvice() {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message)
throws Exception {
adviceCalled.incrementAndGet();
return callback.execute();
}
};
}
}
}

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

@@ -582,4 +582,10 @@ public class MyAdvisedFilter {
the transaction, you would put the transaction advice first.
</para>
</section>
<section id="idempotent-receiver">
<title>Idempotent Receiver EIP Pattern</title>
<para>
TBD
</para>
</section>
</section>

7
src/reference/docbook/meta-data-store.xml
View File

@@ -47,8 +47,8 @@
<classname>RedisMetadataStore</classname> implement <interfacename>ConcurrentMetadataStore</interfacename>.
These provide for atomic updates and can be used across multiple component or application instances.
</para>
<section id="idempotent-receiver">
<title>Idempotent Receiver</title>
<section id="idempotent-receiver-pattern">
<title>Idempotent Receiver and Metadata Store</title>
<para>
The <emphasis>Metadata Store</emphasis> is useful for implementing the
EIP <ulink url="http://eaipatterns.com/IdempotentReceiver.html">Idempotent Receiver</ulink> pattern, when
@@ -70,6 +70,9 @@
The <code>value</code> of the idempotent entry may be some expiration date, after which that entry should
be removed from <emphasis>Metadata Store</emphasis> by some scheduled reaper.
</para>
<para>
Also see <xref linkend="idempotent-receiver"/>.
</para>
</section>
</section>

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

@@ -41,6 +41,17 @@
See <xref linkend="routing-slip"/> for more information.
</para>
</section>
<section id="4.1-idempotent-receiver">
<title>Idempotent Receiver Pattern</title>
<para>
The <emphasis>Idempotent Receiver</emphasis> EIP implementation is now provided
via the <code>&lt;idempotent-receiver&gt;</code> component in XML, or the
<classname>IdempotentReceiverInterceptor</classname> and
<interfacename>IdempotentReceiver</interfacename> annotation when using Java
Configuration.
See <xref linkend="idempotent-receiver"/> and their JavaDocs for more information.
</para>
</section>
<section id="4.1-BoonJsonObjectMapper">
<title>BoonJsonObjectMapper</title>
<para>