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 index 2443f491ac..fa358cc0e2 100644 --- 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 @@ -62,8 +62,7 @@ class IdempotentReceiverAutoProxyCreator extends AbstractAutoProxyCreator { 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)) { + if (isMatch(mappedName, beanName)) { DefaultBeanFactoryPointcutAdvisor idempotentReceiverInterceptor = new DefaultBeanFactoryPointcutAdvisor(); idempotentReceiverInterceptor.setAdviceBeanName(entry.getKey()); 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 index 3d501d8357..2922ee2409 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -28,8 +28,10 @@ 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.core.type.StandardMethodMetadata; import org.springframework.integration.annotation.IdempotentReceiver; import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor; +import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -66,7 +68,8 @@ public class IdempotentReceiverAutoProxyCreatorInitializer implements Integratio String[] endpoints = StringUtils.tokenizeToStringArray(mapping, ","); for (String endpoint : endpoints) { Map idempotentEndpoint = new ManagedMap(); - idempotentEndpoint.put(beanName, beanFactory.resolveEmbeddedValue(endpoint)); + idempotentEndpoint.put(beanName, + beanFactory.resolveEmbeddedValue(endpoint) + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX); idempotentEndpointsMapping.add(idempotentEndpoint); } } @@ -78,12 +81,17 @@ public class IdempotentReceiverAutoProxyCreatorInitializer implements Integratio 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 + ".*"; + String endpoint = beanName; + if (!MessageHandler.class.isAssignableFrom( + ((StandardMethodMetadata) beanMethod).getIntrospectedMethod().getReturnType())) { + /* + 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. + */ + endpoint = beanDefinition.getFactoryBeanName() + "." + beanName + + ".*" + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX; + } for (String interceptor : interceptors) { Map idempotentEndpoint = new ManagedMap(); idempotentEndpoint.put(interceptor, endpoint); 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 index 9592b706b4..c4aa8cccb6 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -106,7 +106,7 @@ public class IdempotentReceiverParserTests { List endpoints = idempotentEndpoints.get("selectorInterceptor"); assertNotNull(endpoints); assertFalse(endpoints.isEmpty()); - assertTrue(endpoints.contains("foo")); + assertTrue(endpoints.contains("foo.handler")); } @Test @@ -124,7 +124,7 @@ public class IdempotentReceiverParserTests { List endpoints = idempotentEndpoints.get("strategyInterceptor"); assertNotNull(endpoints); assertFalse(endpoints.isEmpty()); - assertTrue(endpoints.contains("foo")); + assertTrue(endpoints.contains("foo.handler")); } @Test @@ -142,8 +142,8 @@ public class IdempotentReceiverParserTests { List endpoints = idempotentEndpoints.get("expressionInterceptor"); assertNotNull(endpoints); assertFalse(endpoints.isEmpty()); - assertTrue(endpoints.contains("foo")); - assertTrue(endpoints.contains("bar*")); + assertTrue(endpoints.contains("foo.handler")); + assertTrue(endpoints.contains("bar*.handler")); } @Test 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 index 8c8ce38311..4a9e448ff9 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -16,8 +16,10 @@ package org.springframework.integration.monitor; +import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; @@ -43,6 +45,7 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.handler.MessageProcessor; +import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor; import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport; @@ -50,11 +53,15 @@ import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.metadata.MetadataStore; import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.integration.selector.MetadataStoreSelector; +import org.springframework.integration.support.MessageBuilder; 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.MessageHandler; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.stereotype.Component; @@ -92,6 +99,12 @@ public class IdempotentReceiverIntegrationTests { @Autowired private FooService fooService; + @Autowired + private MessageChannel annotatedBeanMessageHandlerChannel; + + @Autowired + private MessageChannel annotatedBeanMessageHandlerChannel2; + @Test public void testIdempotentReceiver() { this.idempotentReceiverInterceptor.setThrowExceptionOnRejection(true); @@ -133,6 +146,33 @@ public class IdempotentReceiverIntegrationTests { Boolean.class)); } + @Test + public void testIdempotentReceiverOnBeanMessageHandler() { + PollableChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("bar").setReplyChannel(replyChannel).build(); + this.annotatedBeanMessageHandlerChannel.send(message); + + Message receive = replyChannel.receive(10000); + assertNotNull(receive); + assertFalse(receive.getHeaders().containsKey(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE)); + + this.annotatedBeanMessageHandlerChannel.send(message); + receive = replyChannel.receive(10000); + assertNotNull(receive); + assertTrue(receive.getHeaders().containsKey(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE)); + assertTrue(receive.getHeaders().get(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE, Boolean.class)); + + this.annotatedBeanMessageHandlerChannel2.send(new GenericMessage("baz")); + try { + this.annotatedBeanMessageHandlerChannel2.send(new GenericMessage("baz")); + fail("MessageHandlingException expected"); + } + catch (Exception e) { + assertThat(e.getMessage(), containsString("duplicate message has been received")); + } + } + + @Configuration @EnableIntegration @EnableIntegrationMBeanExport(server = "mBeanServer") @@ -221,6 +261,36 @@ public class IdempotentReceiverIntegrationTests { return new FooService(); } + @Bean + @ServiceActivator(inputChannel = "annotatedBeanMessageHandlerChannel") + @IdempotentReceiver("idempotentReceiverInterceptor") + public MessageHandler messageHandler() { + return new ServiceActivatingHandler(new MessageProcessor() { + + @Override + public Object processMessage(Message message) { + return message; + } + + }); + } + + @Bean + @ServiceActivator(inputChannel = "annotatedBeanMessageHandlerChannel2") + @IdempotentReceiver("idempotentReceiverInterceptor") + public MessageHandler messageHandler2() { + return new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + if (message.getHeaders().containsKey(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE)) { + throw new MessageHandlingException(message, "duplicate message has been received"); + } + } + + }; + } + } @Component