From ff2b15ea9e92487c558d7b66b778524b0c10daa6 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 24 Sep 2014 19:54:43 +0300 Subject: [PATCH] INT-2426: Add `Idempotent Receiver` EIP 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 `` * 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. --- .../IntegrationMessageHeaderAccessor.java | 6 + .../annotation/IdempotentReceiver.java | 46 ++++ .../config/ConsumerEndpointFactoryBean.java | 9 +- .../IdempotentReceiverAutoProxyCreator.java | 124 +++++++++ ...ntReceiverAutoProxyCreatorInitializer.java | 106 ++++++++ .../IdempotentReceiverInterceptorParser.java | 120 +++++++++ .../xml/IntegrationNamespaceHandler.java | 1 + .../AbstractMessageProcessingSelector.java | 8 +- .../advice/IdempotentReceiverInterceptor.java | 174 +++++++++++++ .../ExpressionMetadataKeyStrategy.java | 63 +++++ .../metadata/MetadataKeyStrategy.java | 32 +++ .../selector/MetadataStoreSelector.java | 76 ++++++ .../main/resources/META-INF/spring.factories | 3 +- .../config/xml/spring-integration-4.1.xsd | 98 ++++++++ .../IdempotentReceiverParserTests-context.xml | 37 +++ .../xml/IdempotentReceiverParserTests.java | 237 ++++++++++++++++++ .../idempotent-receiver-configs.properties | 20 ++ .../IdempotentReceiverTests-context.xml | 39 +++ .../advice/IdempotentReceiverTests.java | 174 +++++++++++++ .../IdempotentReceiverIntegrationTests.java | 187 ++++++++++++++ src/reference/docbook/handler-advice.xml | 6 + src/reference/docbook/meta-data-store.xml | 7 +- src/reference/docbook/whats-new.xml | 11 + 23 files changed, 1573 insertions(+), 11 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/annotation/IdempotentReceiver.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/config/IdempotentReceiverAutoProxyCreator.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/config/IdempotentReceiverAutoProxyCreatorInitializer.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/metadata/ExpressionMetadataKeyStrategy.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataKeyStrategy.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests-context.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/xml/idempotent-receiver-configs.properties create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/handler/advice/IdempotentReceiverTests-context.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/handler/advice/IdempotentReceiverTests.java create mode 100644 spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java index 41430437b7..42a9848f49 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java @@ -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."); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/IdempotentReceiver.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/IdempotentReceiver.java new file mode 100644 index 0000000000..9cf5ed22f4 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/IdempotentReceiver.java @@ -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(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 225103fe9f..c298f97a17 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -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"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IdempotentReceiverAutoProxyCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IdempotentReceiverAutoProxyCreator.java new file mode 100644 index 0000000000..94cbd6d6d1 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IdempotentReceiverAutoProxyCreator.java @@ -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> idempotentEndpointsMapping; + + private Map> idempotentEndpoints; + + public void setIdempotentEndpointsMapping(List> 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 interceptors = new ArrayList(); + for (Map.Entry> entry : this.idempotentEndpoints.entrySet()) { + List 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>(); + for (Map 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 endpoints = this.idempotentEndpoints.get(interceptor); + if (endpoints == null) { + endpoints = new ArrayList(); + 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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IdempotentReceiverAutoProxyCreatorInitializer.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IdempotentReceiverAutoProxyCreatorInitializer.java new file mode 100644 index 0000000000..3d501d8357 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IdempotentReceiverAutoProxyCreatorInitializer.java @@ -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> idempotentEndpointsMapping = new ManagedList>(); + + 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 idempotentEndpoint = new ManagedMap(); + 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 idempotentEndpoint = new ManagedMap(); + 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); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java new file mode 100644 index 0000000000..d1a7d0fdec --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java @@ -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 <idempotent-receiver/> 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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java index 59f9b4d1fd..adc0a34a4c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java @@ -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()); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/filter/AbstractMessageProcessingSelector.java b/spring-integration-core/src/main/java/org/springframework/integration/filter/AbstractMessageProcessingSelector.java index a755938bfe..71c568a87c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/filter/AbstractMessageProcessingSelector.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/filter/AbstractMessageProcessingSelector.java @@ -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 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) this.messageProcessor).setConversionService(conversionService); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java new file mode 100644 index 0000000000..cb4eb8f4c3 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java @@ -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 + * Idempotent Receiver + * E.I. Pattern. + *

+ * 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}. + *

+ * The {@code idempotent filtering} logic depends on the provided {@link MessageSelector}. + *

+ * 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 false 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 not be true when + * a discard channel is provided, but if it is, it will cause the + * exception to be thrown after + * 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. + *

+ * 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(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/metadata/ExpressionMetadataKeyStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/ExpressionMetadataKeyStrategy.java new file mode 100644 index 0000000000..cab93683ad --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/ExpressionMetadataKeyStrategy.java @@ -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 processor; + + private final String expressionString; + + public ExpressionMetadataKeyStrategy(String expressionString) { + this.processor = new ExpressionEvaluatingMessageProcessor(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 + "]"; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataKeyStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataKeyStrategy.java new file mode 100644 index 0000000000..c1b5a0a94a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataKeyStrategy.java @@ -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); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java b/spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java new file mode 100644 index 0000000000..a9b37b88fc --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java @@ -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}. + *

+ * 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). + *

+ * 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}. + *

+ * This {@link MessageSelector} is useful for an + * Idempotent Receiver + * implementation. + *

+ * 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; + } + +} diff --git a/spring-integration-core/src/main/resources/META-INF/spring.factories b/spring-integration-core/src/main/resources/META-INF/spring.factories index 549179dfbe..4ac88568a7 100644 --- a/spring-integration-core/src/main/resources/META-INF/spring.factories +++ b/spring-integration-core/src/main/resources/META-INF/spring.factories @@ -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 diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd index a591a88fc4..a7f00daa58 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd @@ -4302,6 +4302,104 @@ The list of component name patterns you want to track (e.g., tracked-components + + + + + + + + + + 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. + + + + + + + + + + + + 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'. + + + + + + + + + + + + The 'MetadataStore' reference. Used by the underlying + 'org.springframework.integration.selector.MetadataStoreSelector'. + Mutually exclusive with 'selector'. + Optional. By default 'MetadataStoreSelector' uses an internal 'SimpleMetadataStore'. + + + + + + + + + + + + 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'. + + + + + + + + + + + + + + + + + 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. + + + + + + + Throw an exception if the 'IdempotentReceiverInterceptor' rejects the message (default false). + It is applied regardless of whether or not a discard channel is provided. + + + + + + + + + + + + + + + + + + + + + + + + bar* + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests.java new file mode 100644 index 0000000000..9bc3560dfc --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests.java @@ -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> idempotentEndpoints = + (Map>) getPropertyValue(this.idempotentReceiverAutoProxyCreator, + "idempotentEndpoints", Map.class); + List 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> idempotentEndpoints = + (Map>) getPropertyValue(this.idempotentReceiverAutoProxyCreator, + "idempotentEndpoints", Map.class); + List 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> idempotentEndpoints = + (Map>) getPropertyValue(this.idempotentReceiverAutoProxyCreator, + "idempotentEndpoints", Map.class); + List 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; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/idempotent-receiver-configs.properties b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/idempotent-receiver-configs.properties new file mode 100644 index 0000000000..0b05605ba6 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/idempotent-receiver-configs.properties @@ -0,0 +1,20 @@ +xmlheaders=\ + \ + +xmlfooter= + +empty= + +without-endpoint= + +selector-and-store= + +selector-and-strategy= + +selector-and-expression= + +strategy-and-expression= diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/IdempotentReceiverTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/IdempotentReceiverTests-context.xml new file mode 100644 index 0000000000..42a0268339 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/IdempotentReceiverTests-context.xml @@ -0,0 +1,39 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/IdempotentReceiverTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/IdempotentReceiverTests.java new file mode 100644 index 0000000000..00f9baa3e5 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/IdempotentReceiverTests.java @@ -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> 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 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(); + } + + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java new file mode 100644 index 0000000000..5a2b298b25 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java @@ -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 message = new GenericMessage("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(); + } + + }; + } + + } + +} diff --git a/src/reference/docbook/handler-advice.xml b/src/reference/docbook/handler-advice.xml index d3993c61dd..232feb7f73 100644 --- a/src/reference/docbook/handler-advice.xml +++ b/src/reference/docbook/handler-advice.xml @@ -582,4 +582,10 @@ public class MyAdvisedFilter { the transaction, you would put the transaction advice first. +

+ Idempotent Receiver EIP Pattern + + TBD + +
diff --git a/src/reference/docbook/meta-data-store.xml b/src/reference/docbook/meta-data-store.xml index 66cd1a2afa..44a9de0a37 100644 --- a/src/reference/docbook/meta-data-store.xml +++ b/src/reference/docbook/meta-data-store.xml @@ -47,8 +47,8 @@ RedisMetadataStore implement ConcurrentMetadataStore. These provide for atomic updates and can be used across multiple component or application instances. -
- Idempotent Receiver +
+ Idempotent Receiver and Metadata Store The Metadata Store is useful for implementing the EIP Idempotent Receiver pattern, when @@ -70,6 +70,9 @@ The value of the idempotent entry may be some expiration date, after which that entry should be removed from Metadata Store by some scheduled reaper. + + Also see . +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 4ecf70eade..b1bec277b2 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -41,6 +41,17 @@ See for more information. +
+ Idempotent Receiver Pattern + + The Idempotent Receiver EIP implementation is now provided + via the <idempotent-receiver> component in XML, or the + IdempotentReceiverInterceptor and + IdempotentReceiver annotation when using Java + Configuration. + See and their JavaDocs for more information. + +
BoonJsonObjectMapper