Merge pull request #348 from olegz/INT-2396

2012-03-27 15:30:29 -04:00
parent 6bb460d825 fc31317aa4
commit 6975da83da
4 changed files with 160 additions and 88 deletions
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java
@@ -18,6 +18,9 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler;
 import org.springframework.util.StringUtils;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.w3c.dom.Element;

 /**
@@ -29,6 +32,8 @@ import org.w3c.dom.Element;
 */
 public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractConsumerEndpointParser {

+	private final Log logger = LogFactory.getLog(this.getClass());
+
 	private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy";

 	private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method";
@@ -58,23 +63,36 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo
 	protected void injectPropertyWithAdapter(String beanRefAttribute, String methodRefAttribute,
 			String expressionAttribute, String beanProperty, String adapterClass, Element element,
 			BeanDefinitionBuilder builder, BeanMetadataElement processor, ParserContext parserContext) {
+
 		final String beanRef = element.getAttribute(beanRefAttribute);
 		final String beanMethod = element.getAttribute(methodRefAttribute);
 		final String expression = element.getAttribute(expressionAttribute);
+
+		final boolean hasBeanRef = StringUtils.hasText(beanRef);
+		final boolean hasExpression = StringUtils.hasText(expression);
+
+		if (hasBeanRef && hasExpression) {
+			this.logger.warn("Exactly one of the '" + beanRefAttribute + "' or '" + expressionAttribute +
+					"' attribute is allowed. The '" + expressionAttribute +
+					"' is ignored when both are provided." +
+					"NOTE: This is a warning message only, to avoid a breaking change in a point release and should " +
+					"be treated as an error. In a future release this condition will result in the actual exception");
+		}
+
 		BeanMetadataElement adapter = null;
-		if (StringUtils.hasText(beanRef)) {
+		if (hasBeanRef) {
 			adapter = this.createAdapter(new RuntimeBeanReference(beanRef), beanMethod, adapterClass, parserContext);
 		}
-		else if (processor != null) {
-			adapter = this.createAdapter(processor, beanMethod, adapterClass, parserContext);
-		}
-		else if (StringUtils.hasText(expression)) {
+		else if (hasExpression) {
 			BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
 					.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.ExpressionEvaluating"
 							+ adapterClass);
 			adapterBuilder.addConstructorArgValue(expression);
 			adapter = adapterBuilder.getBeanDefinition();
 		}
+		else if (processor != null) {
+			adapter = this.createAdapter(processor, beanMethod, adapterClass, parserContext);
+		}
 		else {
 			adapter = this.createAdapter(null, beanMethod, adapterClass, parserContext);
 		}
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
@@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicReference;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
+
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.context.ApplicationContext;
@@ -35,6 +36,9 @@ import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.MessageRejectedException;
 import org.springframework.integration.aggregator.AggregatingMessageHandler;
 import org.springframework.integration.aggregator.CorrelationStrategy;
+import org.springframework.integration.aggregator.ExpressionEvaluatingCorrelationStrategy;
+import org.springframework.integration.aggregator.ExpressionEvaluatingReleaseStrategy;
+import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
 import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
 import org.springframework.integration.aggregator.ReleaseStrategy;
 import org.springframework.integration.core.MessageHandler;
@@ -47,6 +51,7 @@ import org.springframework.integration.test.util.TestUtils;
 import static org.hamcrest.CoreMatchers.is;

 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;

@@ -217,10 +222,29 @@ public class AggregatorParserTests {
 		context = new ClassPathXmlApplicationContext("invalidReleaseStrategyMethod.xml", this.getClass());
 	}

+	@Test
+	public void testAggregationWithExpressionsAndPojoAggregator() {
+		EventDrivenConsumer aggregatorConsumer = (EventDrivenConsumer) context.getBean("aggregatorWithExpressionsAndPojoAggregator");
+		AggregatingMessageHandler aggregatingMessageHandler = (AggregatingMessageHandler) TestUtils.getPropertyValue(aggregatorConsumer, "handler");
+		MethodInvokingMessageGroupProcessor messageGroupProcessor = (MethodInvokingMessageGroupProcessor) TestUtils.getPropertyValue(aggregatingMessageHandler, "outputProcessor");
+		Object messageGroupProcessorTargetObject = TestUtils.getPropertyValue(messageGroupProcessor, "processor.delegate.targetObject");
+		assertSame(context.getBean("aggregatorBean"), messageGroupProcessorTargetObject);
+		ReleaseStrategy releaseStrategy = (ReleaseStrategy) TestUtils.getPropertyValue(aggregatingMessageHandler, "releaseStrategy");
+		CorrelationStrategy correlationStrategy = (CorrelationStrategy) TestUtils.getPropertyValue(aggregatingMessageHandler, "correlationStrategy");
+
+		assertTrue(ExpressionEvaluatingReleaseStrategy.class.equals(releaseStrategy.getClass()));
+		assertTrue(ExpressionEvaluatingCorrelationStrategy.class.equals(correlationStrategy.getClass()));
+	}
+
+	// should be re-enabled for INT-2497
+//	@Test(expected=BeanDefinitionParsingException.class)
+//	public void testAggregatorFailureIfMutuallyExclusivityPresent() {
+//		this.context = new ClassPathXmlApplicationContext("aggregatorParserFailTests.xml", this.getClass());
+//	}
+
 	private static <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
 			MessageChannel outputChannel) {
 		return MessageBuilder.withPayload(payload).setCorrelationId(correlationId).setSequenceSize(sequenceSize)
 				.setSequenceNumber(sequenceNumber).setReplyChannel(outputChannel).build();
 	}
-
 }
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserFailTests.xml
+++ b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserFailTests.xml
@@ -0,0 +1,24 @@
+<?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.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<channel id="aggregatorWithExpressionsAndPojoAggregatorInput"/>
+	<aggregator id="aggregatorWithExpressionsAndPojoAggregator"
+		input-channel="aggregatorWithExpressionsAndPojoAggregatorInput"
+		ref="aggregatorBean"
+		release-strategy="releaseStrategy"
+		release-strategy-expression="size() == 2"
+        correlation-strategy-expression="headers['foo']"/>
+
+	<beans:bean id="aggregatorBean"
+		class="org.springframework.integration.config.TestAggregatorBean" />
+
+	<beans:bean id="releaseStrategy"
+		class="org.springframework.integration.config.TestReleaseStrategy" />
+
+</beans:beans>
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml
+++ b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml
@@ -1,82 +1,88 @@
-<?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.xsd
-			http://www.springframework.org/schema/integration
-			http://www.springframework.org/schema/integration/spring-integration.xsd">
-
-	<channel id="outputChannel">
-		<queue capacity="5"/>
-	</channel>
-	<channel id="discardChannel">
-		<queue capacity="5"/>
-	</channel>
-
-	<channel id="aggregatorWithReferenceInput"/>
-	<aggregator id="aggregatorWithReference" ref="aggregatorBean"
-              input-channel="aggregatorWithReferenceInput" output-channel="outputChannel"/>
-
-	<channel id="completelyDefinedAggregatorInput"/>
-	<aggregator id="completelyDefinedAggregator"
-		input-channel="completelyDefinedAggregatorInput"
-		output-channel="outputChannel"
-		discard-channel="discardChannel"
-		ref="aggregatorBean" 
-		release-strategy="releaseStrategy"
-        correlation-strategy="correlationStrategy"
-		send-timeout="86420000"
-		send-partial-result-on-expiry="true"/>
-
-	<channel id="aggregatorWithExpressionsInput"/>
-	<channel id="aggregatorWithExpressionsOutput"/>
-	<aggregator id="aggregatorWithExpressions"
-		input-channel="aggregatorWithExpressionsInput"
-		output-channel="aggregatorWithExpressionsOutput"
-		expression="?[payload.startsWith('1')].![payload]"
-		release-strategy-expression="#root.size()>2"
-        correlation-strategy-expression="headers['foo']"/>
-
-	<channel id="aggregatorWithReferenceAndMethodInput"/>
-	<aggregator id="aggregatorWithReferenceAndMethod"
-		ref="adderBean"
-		method="add"
-		input-channel="aggregatorWithReferenceAndMethodInput"
-		output-channel="outputChannel"/>
-
-	<channel id="aggregatorWithPojoReleaseStrategyInput"/>
-	<aggregator id="aggregatorWithPojoReleaseStrategy"
-		input-channel="aggregatorWithPojoReleaseStrategyInput"
-		output-channel="outputChannel"
-		ref="adderBean"
-		method="add"
-		release-strategy="pojoReleaseStrategy"
-		release-strategy-method="checkCompletenessAsList"/>
-		
-	<channel id="aggregatorWithPojoReleaseStrategyInputAsCollection"/>
-	<aggregator id="aggregatorWithPojoReleaseStrategyAsCollection"
-		input-channel="aggregatorWithPojoReleaseStrategyInputAsCollection"
-		output-channel="outputChannel"
-		ref="adderBean"
-		method="add"
-		release-strategy="pojoReleaseStrategy"
-		release-strategy-method="checkCompletenessAsCollection"/>
-
-	<beans:bean id="aggregatorBean"
-		class="org.springframework.integration.config.TestAggregatorBean" />
-
-	<beans:bean id="adderBean"
-		class="org.springframework.integration.config.Adder" />
-
-	<beans:bean id="releaseStrategy"
-		class="org.springframework.integration.config.TestReleaseStrategy" />
-
-    <beans:bean id="correlationStrategy" class="org.springframework.integration.config.TestCorrelationStrategy"/>
-
-	<beans:bean id="pojoReleaseStrategy"
-		class="org.springframework.integration.config.MaxValueReleaseStrategy">
-		<beans:constructor-arg value="10" />
-	</beans:bean>
-
-</beans:beans>
+<?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.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<channel id="outputChannel">
+		<queue capacity="5"/>
+	</channel>
+	<channel id="discardChannel">
+		<queue capacity="5"/>
+	</channel>
+
+	<channel id="aggregatorWithReferenceInput"/>
+	<aggregator id="aggregatorWithReference" ref="aggregatorBean"
+              input-channel="aggregatorWithReferenceInput" output-channel="outputChannel"/>
+
+	<channel id="completelyDefinedAggregatorInput"/>
+	<aggregator id="completelyDefinedAggregator"
+		input-channel="completelyDefinedAggregatorInput"
+		output-channel="outputChannel"
+		discard-channel="discardChannel"
+		ref="aggregatorBean"
+		release-strategy="releaseStrategy"
+        correlation-strategy="correlationStrategy"
+		send-timeout="86420000"
+		send-partial-result-on-expiry="true"/>
+
+	<channel id="aggregatorWithExpressionsInput"/>
+	<channel id="aggregatorWithExpressionsOutput"/>
+	<aggregator id="aggregatorWithExpressions"
+		input-channel="aggregatorWithExpressionsInput"
+		output-channel="aggregatorWithExpressionsOutput"
+		expression="?[payload.startsWith('1')].![payload]"
+		release-strategy-expression="#root.size()>2"
+        correlation-strategy-expression="headers['foo']"/>
+
+	<channel id="aggregatorWithReferenceAndMethodInput"/>
+	<aggregator id="aggregatorWithReferenceAndMethod"
+		ref="adderBean"
+		method="add"
+		input-channel="aggregatorWithReferenceAndMethodInput"
+		output-channel="outputChannel"/>
+
+	<channel id="aggregatorWithPojoReleaseStrategyInput"/>
+	<aggregator id="aggregatorWithPojoReleaseStrategy"
+		input-channel="aggregatorWithPojoReleaseStrategyInput"
+		output-channel="outputChannel"
+		ref="adderBean"
+		method="add"
+		release-strategy="pojoReleaseStrategy"
+		release-strategy-method="checkCompletenessAsList"/>
+
+	<channel id="aggregatorWithPojoReleaseStrategyInputAsCollection"/>
+	<aggregator id="aggregatorWithPojoReleaseStrategyAsCollection"
+		input-channel="aggregatorWithPojoReleaseStrategyInputAsCollection"
+		output-channel="outputChannel"
+		ref="adderBean"
+		method="add"
+		release-strategy="pojoReleaseStrategy"
+		release-strategy-method="checkCompletenessAsCollection"/>
+
+	<channel id="aggregatorWithExpressionsAndPojoAggregatorInput"/>
+	<aggregator id="aggregatorWithExpressionsAndPojoAggregator"
+		input-channel="aggregatorWithExpressionsAndPojoAggregatorInput"
+		ref="aggregatorBean"
+		release-strategy-expression="size() == 2"
+        correlation-strategy-expression="headers['foo']"/>
+
+	<beans:bean id="aggregatorBean"
+		class="org.springframework.integration.config.TestAggregatorBean" />
+
+	 <beans:bean id="adderBean"
+		class="org.springframework.integration.config.Adder" />
+
+	<beans:bean id="releaseStrategy"
+		class="org.springframework.integration.config.TestReleaseStrategy" />
+
+    <beans:bean id="correlationStrategy" class="org.springframework.integration.config.TestCorrelationStrategy"/>
+
+	<beans:bean id="pojoReleaseStrategy"
+		class="org.springframework.integration.config.MaxValueReleaseStrategy">
+		<beans:constructor-arg value="10" />
+	</beans:bean>
+</beans:beans>