diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java new file mode 100644 index 0000000000..00f976c9e5 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java @@ -0,0 +1,52 @@ +/* + * Copyright 2002-2009 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.aggregator; + +import java.lang.reflect.Method; + +import org.springframework.integration.core.Message; +import org.springframework.integration.handler.MessageMappingMethodInvoker; +import org.springframework.util.Assert; + +/** + * {@link CorrelationStrategy} implementation that works as an adapter to another bean. + * + * @author: Marius Bogoevici + */ +public class CorrelationStrategyAdapter implements CorrelationStrategy { + + private final MessageMappingMethodInvoker invoker; + + + public CorrelationStrategyAdapter(Object object, String methodName) { + this.invoker = new MessageMappingMethodInvoker(object, methodName, true); + } + + public CorrelationStrategyAdapter(Object object, Method method) { + Assert.notNull(object, "'object' must not be null"); + Assert.notNull(method, "'method' must not be null"); + Assert.isTrue(method.getParameterTypes().length == 1, "Method must accept exactly one parameter"); + Assert.isTrue(!Void.TYPE.equals(method.getReturnType()), "Method return type must not be void"); + this.invoker = new MessageMappingMethodInvoker(object, method); + } + + + public Object getCorrelationKey(Message message) { + return invoker.invokeMethod(message); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/CorrelationStrategy.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/CorrelationStrategy.java new file mode 100644 index 0000000000..c43a0113be --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/CorrelationStrategy.java @@ -0,0 +1,35 @@ +/* + * Copyright 2002-2009 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; + +/** + * Indicates that a given method is capable of determining the correlation key + * of a message sent as parameter. + * + * @author: Marius Bogoevici + */ +@Retention (RetentionPolicy.RUNTIME) +@Target (ElementType.METHOD) +@Documented +public @interface CorrelationStrategy { +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java index 81a86115b8..3a6a4148aa 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java @@ -24,8 +24,10 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.aggregator.AbstractMessageAggregator; import org.springframework.integration.aggregator.CompletionStrategyAdapter; import org.springframework.integration.aggregator.MethodInvokingAggregator; +import org.springframework.integration.aggregator.CorrelationStrategyAdapter; import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.CompletionStrategy; +import org.springframework.integration.annotation.CorrelationStrategy; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageHandler; import org.springframework.util.Assert; @@ -48,6 +50,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP protected MessageHandler createHandler(Object bean, Method method, Aggregator annotation) { MethodInvokingAggregator aggregator = new MethodInvokingAggregator(bean, method); this.configureCompletionStrategy(bean, aggregator); + this.configureCorrelationStrategy(bean, aggregator); String discardChannelName = annotation.discardChannel(); if (StringUtils.hasText(discardChannelName)) { MessageChannel discardChannel = this.channelResolver.resolveChannelName(discardChannelName); @@ -79,4 +82,15 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP }); } + private void configureCorrelationStrategy(final Object bean, final AbstractMessageAggregator aggregator) { + ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + Annotation annotation = AnnotationUtils.getAnnotation(method, CorrelationStrategy.class); + if (annotation != null) { + aggregator.setCorrelationStrategy(new CorrelationStrategyAdapter(bean, method)); + } + } + }); + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java index b643dfc9e4..7f19a66736 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java @@ -16,13 +16,13 @@ package org.springframework.integration.config.xml; -import org.w3c.dom.Element; - import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; + /** * Parser for the aggregator element of the integration namespace. * Registers the annotation-driven post-processors. @@ -36,6 +36,10 @@ public class AggregatorParser extends AbstractConsumerEndpointParser { private static final String COMPLETION_STRATEGY_METHOD_ATTRIBUTE = "completion-strategy-method"; + private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy"; + + private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method"; + private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel"; private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout"; @@ -50,6 +54,8 @@ public class AggregatorParser extends AbstractConsumerEndpointParser { private static final String COMPLETION_STRATEGY_PROPERTY = "completionStrategy"; + private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy"; + @Override protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { @@ -71,24 +77,31 @@ public class AggregatorParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE); - final String completionStrategyRef = element.getAttribute(COMPLETION_STRATEGY_REF_ATTRIBUTE); - final String completionStrategyMethod = element.getAttribute(COMPLETION_STRATEGY_METHOD_ATTRIBUTE); - if (StringUtils.hasText(completionStrategyRef)) { - if (StringUtils.hasText(completionStrategyMethod)) { - String adapterBeanName = this.createCompletionStrategyAdapter( - completionStrategyRef, completionStrategyMethod, parserContext); - builder.addPropertyReference(COMPLETION_STRATEGY_PROPERTY, adapterBeanName); - } - else { - builder.addPropertyReference(COMPLETION_STRATEGY_PROPERTY, completionStrategyRef); - } - } + this.injectPropertyWithBean(COMPLETION_STRATEGY_REF_ATTRIBUTE, COMPLETION_STRATEGY_METHOD_ATTRIBUTE, COMPLETION_STRATEGY_PROPERTY, + "CompletionStrategyAdapter", element, builder, parserContext); + this.injectPropertyWithBean(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY, + "CorrelationStrategyAdapter", element, builder, parserContext); return builder; } - private String createCompletionStrategyAdapter(String ref, String method, ParserContext parserContext) { + private void injectPropertyWithBean(String beanRefAttribute, String methodRefAttribute, String beanProperty, String adapterClass, + Element element, BeanDefinitionBuilder builder, ParserContext parserContext) { + final String beanRef = element.getAttribute(beanRefAttribute); + final String beanMethod = element.getAttribute(methodRefAttribute); + if (StringUtils.hasText(beanRef)) { + if (StringUtils.hasText(beanMethod)) { + String adapterBeanName = this.createAdapter(beanRef, beanMethod, adapterClass, parserContext); + builder.addPropertyReference(beanProperty, adapterBeanName); + } + else { + builder.addPropertyReference(beanProperty, beanRef); + } + } + } + + private String createAdapter(String ref, String method, String unqualifiedClassName, ParserContext parserContext) { BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CompletionStrategyAdapter"); + IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator." + unqualifiedClassName); builder.addConstructorArgReference(ref); builder.addConstructorArgValue(method); return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry()); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd index 8e8e9260b9..7e7afa2722 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd @@ -823,6 +823,17 @@ + + + + + + + + + + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageMappingMethodInvoker.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageMappingMethodInvoker.java index a28d749a87..aeb8d49d3e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageMappingMethodInvoker.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageMappingMethodInvoker.java @@ -26,9 +26,6 @@ import java.util.List; import java.util.Map; import java.util.Set; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageHandlingException; @@ -40,6 +37,9 @@ import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + /** * A base or helper class for any Messaging component that acts as an adapter * by invoking a "plain" (not Message-aware) method on a given target object. @@ -47,6 +47,7 @@ import org.springframework.util.StringUtils; * 'methodName', or an Annotation type must be provided. * * @author Mark Fisher + * @author Marius Bogoevici */ public class MessageMappingMethodInvoker { @@ -78,13 +79,17 @@ public class MessageMappingMethodInvoker { this.methodResolver = this.createResolverForAnnotation(annotationType); } - public MessageMappingMethodInvoker(Object object, String methodName) { + public MessageMappingMethodInvoker(Object object, String methodName) { + this(object, methodName, false); + } + + public MessageMappingMethodInvoker(Object object, String methodName, boolean requiresReturnValue) { Assert.notNull(object, "object must not be null"); Assert.notNull(methodName, "methodName must not be null"); this.object = object; - this.methodResolver = this.createResolverForMethodName(methodName); + this.methodResolver = this.createResolverForMethodName(methodName, requiresReturnValue); } - + public Object invokeMethod(Message message) { Assert.notNull(message, "message must not be null"); @@ -164,11 +169,12 @@ public class MessageMappingMethodInvoker { return mapper.fromMessage(message); } - private HandlerMethodResolver createResolverForMethodName(String methodName) { + private HandlerMethodResolver createResolverForMethodName(String methodName, boolean requiresReturnValue) { List methodsWithName = new ArrayList(); Method[] defaultCandidateMethods = HandlerMethodUtils.getCandidateHandlerMethods(this.object); for (Method method : defaultCandidateMethods) { - if (method.getName().equals(methodName)) { + if (method.getName().equals(methodName) + && (!requiresReturnValue || !Void.TYPE.equals(method.getReturnType()))) { methodsWithName.add(method); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index 54bfa46e59..b0e1bff6bf 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -30,6 +30,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.aggregator.CompletionStrategy; import org.springframework.integration.aggregator.CompletionStrategyAdapter; +import org.springframework.integration.aggregator.CorrelationStrategy; import org.springframework.integration.aggregator.MethodInvokingAggregator; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; @@ -76,6 +77,7 @@ public class AggregatorParserTests { EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("completelyDefinedAggregator"); CompletionStrategy completionStrategy = (CompletionStrategy) context.getBean("completionStrategy"); + CorrelationStrategy correlationStrategy = (CorrelationStrategy) context.getBean("correlationStrategy"); MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel"); MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel"); Object consumer = new DirectFieldAccessor(endpoint).getPropertyValue("handler"); @@ -87,6 +89,8 @@ public class AggregatorParserTests { Assert.assertEquals( "The AggregatorEndpoint is not injected with the appropriate CompletionStrategy instance", completionStrategy, accessor.getPropertyValue("completionStrategy")); + Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate CorrelationStrategy instance", + correlationStrategy, accessor.getPropertyValue("correlationStrategy")); Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate output channel", outputChannel, accessor.getPropertyValue("outputChannel")); Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate discard channel", diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests-context.xml new file mode 100644 index 0000000000..05c7c0300b --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests-context.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java new file mode 100644 index 0000000000..7ae6a9c946 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java @@ -0,0 +1,150 @@ +/* + * Copyright 2002-2009 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 org.junit.Assert; +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.integration.aggregator.CompletionStrategy; +import org.springframework.integration.aggregator.CorrelationStrategy; +import org.springframework.integration.annotation.Aggregator; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author: Marius Bogoevici + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class AggregatorWithCorrelationStrategyTests { + + @Autowired + @Qualifier("inputChannel") + MessageChannel inputChannel; + + @Autowired + @Qualifier("outputChannel") + PollableChannel outputChannel; + + @Autowired + @Qualifier("pojoInputChannel") + MessageChannel pojoInputChannel; + + @Autowired + @Qualifier("pojoOutputChannel") + PollableChannel pojoOutputChannel; + + + @Test + public void testCorrelationAndCompletion() { + inputChannel.send(MessageBuilder.withPayload("A1").build()); + inputChannel.send(MessageBuilder.withPayload("B2").build()); + inputChannel.send(MessageBuilder.withPayload("C3").build()); + inputChannel.send(MessageBuilder.withPayload("A4").build()); + inputChannel.send(MessageBuilder.withPayload("B5").build()); + inputChannel.send(MessageBuilder.withPayload("C6").build()); + inputChannel.send(MessageBuilder.withPayload("A7").build()); + inputChannel.send(MessageBuilder.withPayload("B8").build()); + inputChannel.send(MessageBuilder.withPayload("C9").build()); + receiveAndCompare(outputChannel, "A1A4A7"); + receiveAndCompare(outputChannel, "B2B5B8"); + receiveAndCompare(outputChannel, "C3C6C9"); + } + + @Test + public void testCorrelationAndCompletionWithPojo() { + // the test verifies how a pojo strategy is applied + // Strings are correlated by their first letter, integers are correlated by the last digit + pojoInputChannel.send(MessageBuilder.withPayload("X1").build()); + pojoInputChannel.send(MessageBuilder.withPayload("Y2").build()); + pojoInputChannel.send(MessageBuilder.withPayload(93).build()); + pojoInputChannel.send(MessageBuilder.withPayload("X4").build()); + pojoInputChannel.send(MessageBuilder.withPayload("Y5").build()); + pojoInputChannel.send(MessageBuilder.withPayload(113).build()); + pojoInputChannel.send(MessageBuilder.withPayload("X7").build()); + pojoInputChannel.send(MessageBuilder.withPayload("Y8").build()); + pojoInputChannel.send(MessageBuilder.withPayload(213).build()); + receiveAndCompare(pojoOutputChannel, "X1X4X7"); + receiveAndCompare(pojoOutputChannel, "Y2Y5Y8"); + receiveAndCompare(pojoOutputChannel, "93113213"); + } + + private void receiveAndCompare(PollableChannel outputChannel, String expectedValue) { + Message firstResult = outputChannel.receive(500); + Assert.assertNotNull(firstResult); + Assert.assertEquals(expectedValue, firstResult.getPayload()); + } + + + public static class MessageCountCompletionStrategy implements CompletionStrategy { + + private final int expectedSize; + + + public MessageCountCompletionStrategy(int expectedSize) { + this.expectedSize = expectedSize; + } + + public boolean isComplete(List> messages) { + return messages.size() == expectedSize; + } + + } + + public static class FirstLetterCorrelationStrategy implements CorrelationStrategy { + + public Object getCorrelationKey(Message message) { + return message.getPayload().toString().subSequence(0,1); + } + + } + + public static class PojoCorrelationStrategy { + + public String correlate(String message) { + return message.substring(0,1); + } + + public String correlate(Integer mesage) { + return Integer.toString(mesage % 10); + } + + } + + public static class SimpleAggregator { + + @Aggregator + protected String concatenate(List payloads) { + StringBuffer buffer = new StringBuffer(); + for (Object payload: payloads) { + buffer.append(payload.toString()); + } + return buffer.toString(); + } + + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/CorrelationStrategyInvalidConfigurationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/CorrelationStrategyInvalidConfigurationTests.java new file mode 100644 index 0000000000..b990711537 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/CorrelationStrategyInvalidConfigurationTests.java @@ -0,0 +1,42 @@ +/* + * Copyright 2002-2009 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 org.junit.Test; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.context.support.ClassPathXmlApplicationContext; + +/** + * @author: Marius Bogoevici + */ +public class CorrelationStrategyInvalidConfigurationTests { + + @Test(expected = BeanCreationException.class) + public void testCorrelationStrategyWithVoidReturningMethods() throws Exception { + new ClassPathXmlApplicationContext("correlationStrategyWithVoidMethods.xml", CorrelationStrategyInvalidConfigurationTests.class); + } + + public static class VoidReturningCorrelationStrategy { + + public void invalidCorrelationMethod(String string) { + //do nothing + } + + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestCorrelationStrategy.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestCorrelationStrategy.java new file mode 100644 index 0000000000..59c14ba8f4 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestCorrelationStrategy.java @@ -0,0 +1,31 @@ +/* + * Copyright 2002-2009 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 org.springframework.integration.aggregator.CorrelationStrategy; +import org.springframework.integration.core.Message; + +/** + * @author: Marius Bogoevici + */ +public class TestCorrelationStrategy implements CorrelationStrategy { + + public Object getCorrelationKey(Message message) { + throw new UnsupportedOperationException("for configuration test only"); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml index 449283371c..12e75f7229 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml @@ -24,6 +24,7 @@ discard-channel="discardChannel" ref="aggregatorBean" completion-strategy="completionStrategy" + correlation-strategy="correlationStrategy" send-timeout="86420000" send-partial-result-on-timeout="true" reaper-interval="135" @@ -55,6 +56,8 @@ + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index 90a76892b1..dbed9f979a 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -34,9 +34,13 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.aggregator.AbstractMessageAggregator; import org.springframework.integration.aggregator.CompletionStrategyAdapter; import org.springframework.integration.aggregator.SequenceSizeCompletionStrategy; +import org.springframework.integration.aggregator.CorrelationStrategyAdapter; import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.annotation.CorrelationStrategy; +import org.springframework.integration.handler.HandlerMethodResolver; +import org.springframework.integration.handler.StaticHandlerMethodResolver; /** * @author Marius Bogoevici @@ -100,6 +104,27 @@ public class AggregatorAnnotationTests { assertEquals("completionChecker", completionCheckerMethod.getName()); } + @Test + public void testAnnotationWithCustomCorrelationStrategy() throws Exception { + ApplicationContext context = new ClassPathXmlApplicationContext( + new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); + final String endpointName = "endpointWithCorrelationStrategy"; + AbstractMessageAggregator aggregator = this.getAggregator(context, endpointName); + Object correlationStrategy = getPropertyValue(aggregator, "correlationStrategy"); + Assert.assertTrue(correlationStrategy instanceof CorrelationStrategyAdapter); + CorrelationStrategyAdapter completionStrategyAdapter = (CorrelationStrategyAdapter) correlationStrategy; + DirectFieldAccessor invokerAccessor = new DirectFieldAccessor( + new DirectFieldAccessor(completionStrategyAdapter).getPropertyValue("invoker")); + Object targetObject = invokerAccessor.getPropertyValue("object"); + assertSame(context.getBean(endpointName), targetObject); + HandlerMethodResolver completionCheckerMethodResolver = (HandlerMethodResolver) invokerAccessor.getPropertyValue("methodResolver"); + assertTrue(completionCheckerMethodResolver instanceof StaticHandlerMethodResolver); + DirectFieldAccessor resolverAccessor = new DirectFieldAccessor(completionCheckerMethodResolver); + Method completionCheckerMethod = (Method) resolverAccessor.getPropertyValue("method"); + assertEquals("correlate", completionCheckerMethod.getName()); + } + + private AbstractMessageAggregator getAggregator(ApplicationContext context, final String endpointName) { EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean( diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCorrelationStrategy.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCorrelationStrategy.java new file mode 100644 index 0000000000..43ebafbb7b --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCorrelationStrategy.java @@ -0,0 +1,51 @@ +/* + * Copyright 2002-2008 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.annotation; + +import java.util.List; + +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Aggregator; +import org.springframework.integration.annotation.CompletionStrategy; +import org.springframework.integration.annotation.CorrelationStrategy; + +/** + * @author: Marius Bogoevici + */ +@MessageEndpoint("endpointWithCorrelationStrategy") +public class TestAnnotatedEndpointWithCorrelationStrategy { + + @Aggregator(inputChannel = "inputChannel") + public String aggregatingMethod(List payloads) { + StringBuffer buffer = new StringBuffer(); + for (String s: payloads) { + buffer.append(s); + } + return buffer.toString(); + } + + @CompletionStrategy + public boolean isComplete(List payloads) { + return payloads.size() == 3; + } + + @CorrelationStrategy + public String correlate(String payload) { + return payload.substring(0, 1); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml index 4d9312fdbc..fa003e12d4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml @@ -20,8 +20,6 @@ - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/correlationStrategyWithVoidMethods.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/correlationStrategyWithVoidMethods.xml new file mode 100644 index 0000000000..3947188915 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/correlationStrategyWithVoidMethods.xml @@ -0,0 +1,31 @@ + + + + + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/MessageMappingMethodInvokerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/MessageMappingMethodInvokerTests.java index c13c7c9dc6..8785eb31ca 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/MessageMappingMethodInvokerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/MessageMappingMethodInvokerTests.java @@ -17,6 +17,8 @@ package org.springframework.integration.handler; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.fail; import java.lang.reflect.Method; import java.util.Map; @@ -32,6 +34,7 @@ import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher + * @author Marius Bogoevici */ public class MessageMappingMethodInvokerTests { @@ -93,6 +96,25 @@ public class MessageMappingMethodInvokerTests { assertEquals("testing-123", result); } + @Test + public void testVoidMethodsIncludedbyDefault() { + MessageMappingMethodInvoker invoker = new MessageMappingMethodInvoker(new TestBean(), "testVoidReturningMethods"); + assertNull(invoker.invokeMethod(MessageBuilder.withPayload("Something").build())); + assertEquals(12, invoker.invokeMethod(MessageBuilder.withPayload(12).build())); + } + + @Test + public void testVoidMethodsExcludedByFlag() { + MessageMappingMethodInvoker invoker = new MessageMappingMethodInvoker(new TestBean(), "testVoidReturningMethods", true); + assertEquals(12, invoker.invokeMethod(MessageBuilder.withPayload(12).build())); + try { + assertNull(invoker.invokeMethod(MessageBuilder.withPayload("Something").build())); + fail(); + } catch(IllegalArgumentException ex){ + + } + } + @Test public void messageOnlyWithAnnotatedMethod() throws Exception { AnnotatedTestService service = new AnnotatedTestService(); @@ -182,6 +204,15 @@ public class MessageMappingMethodInvokerTests { public String acceptPayloadAndHeaderAndReturnObject(String s, @Header("number") Integer n) { return s + "-" + n; } + + public void testVoidReturningMethods(String s) { + // do nothing + } + + public int testVoidReturningMethods(int i) { + return i; + } + } private static class AnnotatedTestService { @@ -222,6 +253,7 @@ public class MessageMappingMethodInvokerTests { public Integer integerMethod(Integer i) { return i; } + } }