INT-3317: Remove comparator from resequencer

JIRA: https://jira.spring.io/browse/INT-3317

In addition add `AbstractCorrelatingMessageHandler.discardChannelName` for convenience with JavaConfig

INT-3317: Add `AbstractCorrMH.outputChannelName`

INT-3317: Fix further `BF` population

Also add `MessageFilter.discardChannelName`

Move `XPathExpressionEvaluatingHeaderValueMessageProcessor` to the `xml.transformer.support` package

Additional polishing

Polishing

PR comments.

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -22,7 +22,9 @@ import java.util.concurrent.locks.Lock;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.core.MessageProducer;
@@ -40,8 +42,10 @@ import org.springframework.integration.util.UUIDConverter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Abstract Message handler that holds a buffer of correlated messages in a
@@ -63,6 +67,7 @@ import org.springframework.util.CollectionUtils;
  * @author Dave Syer
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.0
  */
 public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer {
@@ -81,9 +86,13 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 
 	private MessageChannel outputChannel;
 
+	private String outputChannelName;
+
 	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
 
-	private volatile MessageChannel discardChannel = new NullChannel();
+	private volatile MessageChannel discardChannel;
+
+	private volatile String discardChannelName;
 
 	private boolean sendPartialResultOnExpiry = false;
 
@@ -95,6 +104,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 
 	private volatile long minimumTimeoutForEmptyGroups;
 
+	private volatile boolean releasePartialSequences;
+
 	public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
 									 CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
 		Assert.notNull(processor);
@@ -151,13 +162,60 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 		this.outputChannel = outputChannel;
 	}
 
+	public void setOutputChannelName(String outputChannelName) {
+		this.outputChannelName = outputChannelName;
+	}
+
 	@Override
 	protected void onInit() throws Exception {
 		super.onInit();
 		BeanFactory beanFactory = this.getBeanFactory();
 		if (beanFactory != null) {
 			this.messagingTemplate.setBeanFactory(beanFactory);
+			if (StringUtils.hasText(this.discardChannelName)) {
+				Assert.isNull(this.discardChannel, "'outputChannelName' and 'discardChannel' are mutually exclusive.");
+				try {
+					this.discardChannel = beanFactory.getBean(this.discardChannelName, MessageChannel.class);
+				}
+				catch (BeansException e) {
+					throw new DestinationResolutionException("Failed to look up MessageChannel with name '"
+							+ this.discardChannelName + "' in the BeanFactory.");
+				}
+			}
+
+			if (StringUtils.hasText(this.outputChannelName)) {
+				Assert.isNull(this.outputChannel, "'outputChannelName' and 'outputChannel' are mutually exclusive.");
+				try {
+					this.outputChannel = this.getBeanFactory().getBean(this.outputChannelName, MessageChannel.class);
+				}
+				catch (BeansException e) {
+					throw new DestinationResolutionException("Failed to look up MessageChannel with name '"
+							+ this.outputChannelName + "' in the BeanFactory.");
+				}
+			}
+
+			if (this.outputProcessor instanceof BeanFactoryAware) {
+				((BeanFactoryAware) this.outputProcessor).setBeanFactory(beanFactory);
+			}
+			if (this.correlationStrategy instanceof BeanFactoryAware) {
+				((BeanFactoryAware) this.correlationStrategy).setBeanFactory(beanFactory);
+			}
+			if (this.releaseStrategy instanceof BeanFactoryAware) {
+				((BeanFactoryAware) this.releaseStrategy).setBeanFactory(beanFactory);
+			}
 		}
+
+		if (this.discardChannel == null) {
+			this.discardChannel = new NullChannel();
+		}
+
+		if (this.releasePartialSequences) {
+			Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy,
+					"Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName()
+							+ "] cannot release partial sequences. Use the default SequenceSizeReleaseStrategy instead.");
+			((SequenceSizeReleaseStrategy)this.releaseStrategy).setReleasePartialSequences(releasePartialSequences);
+		}
+
 		/*
 		 * Disallow any further changes to the lock registry
 		 * (checked in the setter).
@@ -166,9 +224,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 	}
 
 	public void setDiscardChannel(MessageChannel discardChannel) {
+		Assert.notNull(discardChannel, "'discardChannel' cannot be null");
 		this.discardChannel = discardChannel;
 	}
 
+	public void setDiscardChannelName(String discardChannelName) {
+		this.discardChannelName = discardChannelName;
+	}
+
 	public void setSendTimeout(long sendTimeout) {
 		this.messagingTemplate.setSendTimeout(sendTimeout);
 	}
@@ -192,11 +255,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 		this.minimumTimeoutForEmptyGroups = minimumTimeoutForEmptyGroups;
 	}
 
-	public void setReleasePartialSequences(boolean releasePartialSequences){
-		Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy,
-				"Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName()
-						+ "] cannot release partial sequences. Use the default SequenceSizeReleaseStrategy instead.");
-		((SequenceSizeReleaseStrategy)this.releaseStrategy).setReleasePartialSequences(releasePartialSequences);
+	public void setReleasePartialSequences(boolean releasePartialSequences) {
+		this.releasePartialSequences = releasePartialSequences;
 	}
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2014 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,13 +19,12 @@ package org.springframework.integration.aggregator;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.expression.Expression;
 import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.SpelParserConfiguration;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
-import org.springframework.messaging.Message;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
+import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
 /**
@@ -33,13 +32,12 @@ import org.springframework.util.Assert;
  *
  * @author Dave Syer
  * @author Oleg Zhurakousky
+ * @author Artem Bilan
  */
-public class ExpressionEvaluatingCorrelationStrategy implements CorrelationStrategy, BeanFactoryAware, InitializingBean{
+public class ExpressionEvaluatingCorrelationStrategy implements CorrelationStrategy, BeanFactoryAware {
 
 	private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
 
-	private volatile BeanFactory beanFactory;
-
 	private final ExpressionEvaluatingMessageProcessor<Object> processor;
 
 
@@ -58,12 +56,8 @@ public class ExpressionEvaluatingCorrelationStrategy implements CorrelationStrat
 	}
 
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
-		this.beanFactory = beanFactory;
-	}
-
-	public void afterPropertiesSet() throws Exception {
-		if (this.beanFactory != null){
-			this.processor.setBeanFactory(this.beanFactory);
+		if (beanFactory != null){
+			this.processor.setBeanFactory(beanFactory);
 		}
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingCorrelationStrategy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2014 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,8 +18,12 @@ package org.springframework.integration.aggregator;
 
 import java.lang.reflect.Method;
 
-import org.springframework.messaging.Message;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.integration.handler.MessageProcessor;
 import org.springframework.integration.handler.MethodInvokingMessageProcessor;
+import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
 /**
@@ -27,10 +31,11 @@ import org.springframework.util.Assert;
  *
  * @author Marius Bogoevici
  * @author Dave Syer
+ * @author Artem Bilan
  */
-public class MethodInvokingCorrelationStrategy implements CorrelationStrategy {
+public class MethodInvokingCorrelationStrategy implements CorrelationStrategy, BeanFactoryAware {
 
-	private final MethodInvokingMessageProcessor<Object> processor;
+	private final MessageProcessor<?> processor;
 
 	public MethodInvokingCorrelationStrategy(Object object, String methodName) {
 		this.processor = new MethodInvokingMessageProcessor<Object>(object, methodName, true);
@@ -43,8 +48,15 @@ public class MethodInvokingCorrelationStrategy implements CorrelationStrategy {
 		this.processor = new MethodInvokingMessageProcessor<Object>(object, method);
 	}
 
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		if (beanFactory != null && this.processor instanceof BeanFactoryAware) {
+			((BeanFactoryAware) this.processor).setBeanFactory(beanFactory);
+		}
+	}
+
+	@Override
 	public Object getCorrelationKey(Message<?> message) {
 		return processor.processMessage(message);
 	}
-
 }

spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2014 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,16 +19,17 @@ package org.springframework.integration.aggregator;
 import java.lang.reflect.Method;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.integration.store.MessageGroup;
 
 /**
- * A {@link ReleaseStrategy} that invokes a method on a plain old Java object. 
- * 
+ * A {@link ReleaseStrategy} that invokes a method on a plain old Java object.
+ *
  * @author Marius Bogoevici
  * @author Dave Syer
  */
-public class MethodInvokingReleaseStrategy implements ReleaseStrategy {
+public class MethodInvokingReleaseStrategy implements ReleaseStrategy, BeanFactoryAware {
 
 	private final MethodInvokingMessageListProcessor<Boolean> adapter;
 
@@ -46,10 +47,12 @@ public class MethodInvokingReleaseStrategy implements ReleaseStrategy {
 		this.adapter.setConversionService(conversionService);
 	}
 
+	@Override
 	public void setBeanFactory(BeanFactory beanFactory) {
 		this.adapter.setBeanFactory(beanFactory);
 	}
 
+	@Override
 	public boolean canRelease(MessageGroup messages) {
 		return this.adapter.process(messages.getMessages(), null);
 	}

spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java

@@ -16,11 +16,9 @@ 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.integration.aggregator.ResequencingMessageGroupProcessor;
 import org.springframework.integration.aggregator.ResequencingMessageHandler;
-import org.springframework.integration.store.SimpleMessageStore;
 
 /**
  * Parser for the <resequencer> element.
@@ -32,9 +30,6 @@ import org.springframework.integration.store.SimpleMessageStore;
  */
 public class ResequencerParser extends AbstractCorrelatingMessageHandlerParser {
 
-
-	private static final String COMPARATOR_REF_ATTRIBUTE = "comparator";
-
 	private static final String RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE = "release-partial-sequences";
 
 	@Override
@@ -43,17 +38,7 @@ public class ResequencerParser extends AbstractCorrelatingMessageHandlerParser {
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ResequencingMessageHandler.class);
 		BeanDefinitionBuilder processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(ResequencingMessageGroupProcessor.class);
 
-		// Comparator
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(processorBuilder, element, COMPARATOR_REF_ATTRIBUTE);
-
-		String processorRef = BeanDefinitionReaderUtils.registerWithGeneratedName(processorBuilder.getBeanDefinition(),
-				parserContext.getRegistry());
-
-		// Message group processor
-		builder.addConstructorArgReference(processorRef);
-
-		// Message store
-		builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(SimpleMessageStore.class).getBeanDefinition());
+		builder.addConstructorArgValue(processorBuilder.getBeanDefinition());
 
 		this.doParse(builder, element, processorBuilder.getBeanDefinition(), parserContext);
 

spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java

@@ -16,13 +16,16 @@
 
 package org.springframework.integration.filter;
 
+import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.integration.MessageRejectedException;
 import org.springframework.integration.core.MessageSelector;
 import org.springframework.integration.handler.AbstractReplyProducingPostProcessingMessageHandler;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * Message Handler that delegates to a {@link MessageSelector}. If and only if
@@ -36,6 +39,7 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHandler {
 
@@ -45,6 +49,7 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa
 
 	private volatile MessageChannel discardChannel;
 
+	private volatile String discardChannelName;
 
 	/**
 	 * Create a MessageFilter that will delegate to the given
@@ -90,6 +95,10 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa
 		this.discardChannel = discardChannel;
 	}
 
+	public void setDiscardChannelName(String discardChannelName) {
+		this.discardChannelName = discardChannelName;
+	}
+
 	/**
 	 * Set to 'true' if you wish the discard processing to occur within any
 	 * request handler advice applied to this filter. Also applies to
@@ -108,6 +117,16 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa
 
 	@Override
 	protected void doInit() {
+		if (StringUtils.hasText(this.discardChannelName)) {
+			Assert.isNull(this.discardChannel, "'outputChannelName' and 'discardChannel' are mutually exclusive.");
+			try {
+				this.discardChannel = this.getBeanFactory().getBean(this.discardChannelName, MessageChannel.class);
+			}
+			catch (BeansException e) {
+				throw new DestinationResolutionException("Failed to look up MessageChannel with name '"
+						+ this.discardChannelName + "' in the BeanFactory.");
+			}
+		}
 		if (this.selector instanceof AbstractMessageProcessingSelector) {
 			((AbstractMessageProcessingSelector) this.selector).setConversionService(this.getConversionService());
 		}

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd

@@ -3520,20 +3520,6 @@
 	<xsd:complexType name="resequencer-type">
 		<xsd:complexContent>
 			<xsd:extension base="correlating-message-handler-type">
-				<xsd:attribute name="comparator" type="xsd:string">
-					<xsd:annotation>
-						<xsd:documentation>
-							Comparator for messages used to sort the sequence when released. Defaults to comparing
-							the
-							sequence number header.
-						</xsd:documentation>
-						<xsd:appinfo>
-							<tool:annotation kind="ref">
-								<tool:expected-type type="java.util.Comparator" />
-							</tool:annotation>
-						</xsd:appinfo>
-					</xsd:annotation>
-				</xsd:attribute>
 				<xsd:attribute name="release-partial-sequences" type="xsd:string">
 					<xsd:annotation>
 						<xsd:documentation>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java

@@ -47,6 +47,7 @@ public class AggregatorTests {
 	@Before
 	public void configureAggregator() {
 		this.aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), store);
+		this.aggregator.afterPropertiesSet();
 	}
 
 

spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java

@@ -16,20 +16,21 @@
 
 package org.springframework.integration.aggregator;
 
-import org.junit.Before;
-import org.junit.Test;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.integration.store.MessageGroupStore;
-import org.springframework.integration.store.SimpleMessageStore;
-import org.springframework.integration.support.MessageBuilder;
-
 import static org.mockito.Matchers.isA;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.store.SimpleMessageStore;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
 public class CorrelatingMessageHandlerIntegrationTests {
 
 	private MessageGroupStore store = new SimpleMessageStore(100);
@@ -45,6 +46,7 @@ public class CorrelatingMessageHandlerIntegrationTests {
 		when(outputChannel.send(isA(Message.class))).thenReturn(true);
 		defaultHandler.setOutputChannel(outputChannel);
 		defaultHandler.setSendTimeout(-1);
+		defaultHandler.afterPropertiesSet();
 	}
 
 

spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategyTests.java

@@ -61,7 +61,6 @@ public class ExpressionEvaluatingCorrelationStrategyTests {
 		Expression expression = parser.parseExpression("payload.substring(0,1)");
 		strategy = new ExpressionEvaluatingCorrelationStrategy(expression);
 		strategy.setBeanFactory(mock(BeanFactory.class));
-		strategy.afterPropertiesSet();
 		Object correlationKey = strategy.getCorrelationKey(new GenericMessage<String>("bla"));
 		assertThat(correlationKey, is(instanceOf(String.class)));
 		assertThat((String) correlationKey, is("b"));

spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java

@@ -54,6 +54,7 @@ public class ResequencerTests {
 	@Before
 	public void configureResequencer() {
 		this.resequencer = new ResequencingMessageHandler(processor, store, null, null);
+		this.resequencer.afterPropertiesSet();
 	}
 
 	@Test
@@ -81,6 +82,7 @@ public class ResequencerTests {
 		SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy();
 		releaseStrategy.setReleasePartialSequences(true);
 		this.resequencer = new ResequencingMessageHandler(processor, store, null, releaseStrategy);
+		this.resequencer.afterPropertiesSet();
 
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage("123", "ABC", 3, 1, replyChannel);
@@ -104,6 +106,8 @@ public class ResequencerTests {
 				return "A";
 			}
 		});
+		this.resequencer.afterPropertiesSet();
+
 		//Message<?> message0 = MessageBuilder.withPayload("0").setSequenceNumber(0).build();
 		Message<?> message1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setReplyChannel(replyChannel).build();
 		Message<?> message2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setReplyChannel(replyChannel).build();

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml

@@ -11,7 +11,7 @@
 		<queue capacity="5" />
 	</channel>
 
-	<aggregator ref="summer" method="sum" input-channel="input" output-channel="output">
+	<aggregator ref="summer" method="sum" input-channel="input" output-channel="output" expression="">
 		<poller task-executor="executor" max-messages-per-poll="5" fixed-delay="20" />
 	</aggregator>
 
@@ -20,19 +20,19 @@
 	<channel id="output">
 		<queue capacity="5" />
 	</channel>
-	
+
 	<channel id="discard">
 		<queue capacity="5" />
 	</channel>
 
 	<beans:bean id="summer"
 		class="org.springframework.integration.aggregator.integration.AggregatorIntegrationTests$SummingAggregator" />
-		
-	<aggregator id="expiringAggregator" input-channel="expiringAggregatorInput" output-channel="output" 
-				expire-groups-upon-completion="true" discard-channel="discard"/>
-				
-	<aggregator id="nonExpiringAggregator" input-channel="nonExpiringAggregatorInput" output-channel="output" 
-				expire-groups-upon-completion="false" discard-channel="discard"/>
-		
 
-</beans:beans>
+	<aggregator id="expiringAggregator" input-channel="expiringAggregatorInput" output-channel="output"
+				expire-groups-upon-completion="true" discard-channel="discard"/>
+
+	<aggregator id="nonExpiringAggregator" input-channel="nonExpiringAggregatorInput" output-channel="output"
+				expire-groups-upon-completion="false" discard-channel="discard"/>
+
+
+</beans:beans>