INT-266 - adding namespace and annotation support for CorrelationStrategy

2009-02-23 16:18:08 +00:00
parent 03b99960e0
commit 0d960df263
17 changed files with 568 additions and 26 deletions
--- 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
@@ -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);
+    }
+    
+}
--- 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
@@ -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 {
+}
--- 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));
+                }
+            }
+        });
+    }
+
 }
--- 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 <em>aggregator</em> 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());
--- 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 @@
 					</xsd:attribute>
 					<xsd:attribute name="completion-strategy-method"
 						type="xsd:string" />
+                    <xsd:attribute name="correlation-strategy" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="java.lang.Object" />
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="correlation-strategy-method"
+						type="xsd:string" />
 					<xsd:attribute name="discard-channel" type="xsd:string">
 						<xsd:annotation>
 							<xsd:appinfo>
--- 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<Method> methodsWithName = new ArrayList<Method>();
 		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);
 			}
 		}
--- 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",
--- 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
@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
+
+    <beans:bean id="correlationStrategy" class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$FirstLetterCorrelationStrategy"/>
+
+    <beans:bean id="pojoCorrelationStrategy" class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$PojoCorrelationStrategy"/>
+
+    <beans:bean id="completionStrategy" class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$MessageCountCompletionStrategy">
+        <beans:constructor-arg value="3"/>
+    </beans:bean>
+
+    <beans:bean name="aggregator" class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$SimpleAggregator"/>
+
+    <channel id="inputChannel"/>
+
+    <channel id="outputChannel">
+        <queue capacity="10"/>
+    </channel>
+
+    <channel id="pojoInputChannel"/>
+
+    <channel id="pojoOutputChannel">
+        <queue capacity="10"/>
+    </channel>
+
+    <aggregator ref="aggregator"
+                completion-strategy="completionStrategy"
+                correlation-strategy="correlationStrategy"
+                input-channel="inputChannel"
+                output-channel="outputChannel"/>
+    
+    <aggregator ref="aggregator"
+                completion-strategy="completionStrategy"
+                correlation-strategy="pojoCorrelationStrategy" correlation-strategy-method="correlate"
+                input-channel="pojoInputChannel"
+                output-channel="pojoOutputChannel"/>
+
+
+</beans:beans>
--- 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
@@ -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<Message<?>> 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<Object> payloads) {
+            StringBuffer buffer = new StringBuffer();
+            for (Object payload: payloads) {
+                buffer.append(payload.toString());
+            }
+            return buffer.toString();
+        }
+
+    }
+
+}
--- 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
@@ -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
+        }
+
+    }
+
+}
--- 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
@@ -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");
+    }
+    
+}
--- 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 @@
 	<beans:bean id="completionStrategy"
 		class="org.springframework.integration.config.TestCompletionStrategy" />

+    <beans:bean id="correlationStrategy" class="org.springframework.integration.config.TestCorrelationStrategy"/>
+
 	<beans:bean id="pojoCompletionStrategy"
 		class="org.springframework.integration.config.MaxValueCompletionStrategy">
 		<beans:constructor-arg value="10" />
--- 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(
--- 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
@@ -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<String> payloads) {
+        StringBuffer buffer = new StringBuffer();
+        for (String s: payloads)  {
+            buffer.append(s);
+        }
+        return buffer.toString();
+    }
+
+    @CompletionStrategy
+    public boolean isComplete(List<String> payloads) {
+        return payloads.size() == 3;
+    }
+
+    @CorrelationStrategy
+    public String correlate(String payload) {
+        return payload.substring(0, 1);
+    }
+
+}
--- 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 @@
 	<context:component-scan base-package="org.springframework.integration.config" use-default-filters="false">
 		<context:include-filter type="regex"
 			expression="org\.springframework\.integration\.config\.annotation\.TestAnnotatedEndpoint.*"/>
-		<context:exclude-filter type="regex"
-			expression="org\.springframework\.integration\.config\.annotation\.TestAnnotatedEndpointWithCompletionStrategyOnly"/>
 	</context:component-scan>

 </beans:beans>
--- 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
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
+
+    <beans:bean id="correlationStrategy" class="org.springframework.integration.config.CorrelationStrategyInvalidConfigurationTests$VoidReturningCorrelationStrategy"/>
+
+    <beans:bean id="completionStrategy" class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$MessageCountCompletionStrategy">
+        <beans:constructor-arg value="3"/>
+    </beans:bean>
+
+    <beans:bean name="aggregator" class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$SimpleAggregator"/>
+
+    <channel id="inputChannel"/>
+
+    <channel id="outputChannel">
+        <queue capacity="10"/>
+    </channel>
+
+
+    <aggregator ref="aggregator"
+                completion-strategy="completionStrategy"
+                correlation-strategy="correlationStrategy" completion-strategy-method="invalidCorrelationMethod"
+                input-channel="inputChannel"
+                output-channel="outputChannel"/>
+    
+
+</beans:beans>
--- 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;
 		}
+
 	}

 }