INT-603: added custom comparator to resequencer

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java

@@ -32,12 +32,10 @@ import org.springframework.integration.store.MessageGroup;
  * 
  * @since 2.0
  */
-public class Resequencer implements ReleaseStrategy, MessageGroupProcessor {
+public class ResequencingMessageGroupProcessor implements MessageGroupProcessor {
 
 	private volatile Comparator<Message<?>> comparator = new SequenceNumberComparator();
 
-	private volatile boolean releasePartialSequences;
-
 	/**
 	 * A comparator to use to order messages before processing. The default is to order by sequence number.
 	 * 
@@ -47,27 +45,6 @@ public class Resequencer implements ReleaseStrategy, MessageGroupProcessor {
 		this.comparator = comparator;
 	}
 
-	/**
-	 * Flag that determines if partial sequences are allowed. If true then as soon as enough messages arrive that can be
-	 * ordered they will be released, provided they all have sequence numbers greater than those already released.
-	 * 
-	 * @param releasePartialSequences
-	 */
-	public void setReleasePartialSequences(boolean releasePartialSequences) {
-		this.releasePartialSequences = releasePartialSequences;
-	}
-
-	public boolean canRelease(MessageGroup messages) {
-		if (releasePartialSequences) {
-			List<Message<?>> sorted = new ArrayList<Message<?>>(messages.getUnmarked());
-			Collections.sort(sorted, comparator);
-			int head = sorted.get(sorted.size() - 1).getHeaders().getSequenceNumber();
-			int tail = sorted.get(0).getHeaders().getSequenceNumber() - 1;
-			return tail == messages.getMarked().size() && head - tail == sorted.size();
-		}
-		return messages.isComplete();
-	}
-
 	public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) {
 		Collection<Message<?>> messages = group.getUnmarked();
 		if (messages.size() > 0) {
@@ -79,10 +56,4 @@ public class Resequencer implements ReleaseStrategy, MessageGroupProcessor {
 		}
 	}
 
-	private static class SequenceNumberComparator implements Comparator<Message<?>> {
-		public int compare(Message<?> o1, Message<?> o2) {
-			return o1.getHeaders().getSequenceNumber().compareTo(o2.getHeaders().getSequenceNumber());
-		}
-	}
-
 }

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceNumberComparator.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2002-2010 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.util.Comparator;
+
+import org.springframework.integration.core.Message;
+
+/**
+ * @author Dave Syer
+ * 
+ * @since 2.0
+ * 
+ */
+public class SequenceNumberComparator implements Comparator<Message<?>> {
+
+	/**
+	 * If both messages have a sequence number then compare that, otherwise if one has a sequence number and the other
+	 * doesn't then the numbered message comes first, or finally of neither has a sequence number then they are equal in
+	 * rank.
+	 */
+	public int compare(Message<?> o1, Message<?> o2) {
+		Integer sequenceNumber1 = o1.getHeaders().getSequenceNumber();
+		Integer sequenceNumber2 = o2.getHeaders().getSequenceNumber();
+		if (sequenceNumber1 == sequenceNumber2) {
+			return 0;
+		}
+		if (sequenceNumber1 == null) {
+			return -sequenceNumber2;
+		}
+		if (sequenceNumber2 == null) {
+			return sequenceNumber1;
+		}
+		return sequenceNumber1.compareTo(sequenceNumber2);
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java

@@ -16,6 +16,12 @@
 
 package org.springframework.integration.aggregator;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+
+import org.springframework.integration.core.Message;
 import org.springframework.integration.store.MessageGroup;
 
 /**
@@ -28,7 +34,36 @@ import org.springframework.integration.store.MessageGroup;
  */
 public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
 
+	private volatile Comparator<Message<?>> comparator = new SequenceNumberComparator();
+
+	private volatile boolean releasePartialSequences;
+	
+	public SequenceSizeReleaseStrategy() {
+		this(false);
+	}
+
+	public SequenceSizeReleaseStrategy(boolean releasePartialSequences) {
+		this.releasePartialSequences = releasePartialSequences;
+	}
+
+	/**
+	 * Flag that determines if partial sequences are allowed. If true then as soon as enough messages arrive that can be
+	 * ordered they will be released, provided they all have sequence numbers greater than those already released.
+	 * 
+	 * @param releasePartialSequences
+	 */
+	public void setReleasePartialSequences(boolean releasePartialSequences) {
+		this.releasePartialSequences = releasePartialSequences;
+	}
+
 	public boolean canRelease(MessageGroup messages) {
+		if (releasePartialSequences) {
+			List<Message<?>> sorted = new ArrayList<Message<?>>(messages.getUnmarked());
+			Collections.sort(sorted, comparator);
+			int head = sorted.get(sorted.size() - 1).getHeaders().getSequenceNumber();
+			int tail = sorted.get(0).getHeaders().getSequenceNumber() - 1;
+			return tail == messages.getMarked().size() && head - tail == sorted.size();
+		}
 		return messages.isComplete();
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java

@@ -50,7 +50,7 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
 
     private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout";
 
-    private static final String SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE = "send-partial-result-on-expiry";
+    private static final String SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE = "send-partial-result-on-expiry";
 
     private static final String RELEASE_STRATEGY_PROPERTY = "releaseStrategy";
 
@@ -97,7 +97,7 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
         IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
                 SEND_TIMEOUT_ATTRIBUTE);
         IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
-                SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
+                SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE);
         IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
         this.injectPropertyWithBean(RELEASE_STRATEGY_REF_ATTRIBUTE,
                 RELEASE_STRATEGY_METHOD_ATTRIBUTE, RELEASE_STRATEGY_PROPERTY,

org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java

@@ -1,21 +1,19 @@
 /*
  * 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.
+ * 
+ * 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.xml;
 
+import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.ParserContext;
@@ -30,18 +28,39 @@ import org.w3c.dom.Element;
  */
 public class ResequencerParser extends AbstractConsumerEndpointParser {
 
+	private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method";
+
+	private static final String CORRELATION_STRATEGY_ATTRIBUTE = "correlation-strategy";
+
+	private static final String SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE = "send-partial-result-on-expiry";
+
+	private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout";
+
+	private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
+
+	private static final String MESSAGE_STORE_ATTRIBUTE = "message-store";
+
+	private static final String COMPARATOR_ATTRIBUTE = "comparator";
+
+	private static final String RELEASE_STRATEGY_REF_ATTRIBUTE = "release-strategy";
+
+	private static final String RELEASE_STRATEGY_METHOD_ATTRIBUTE = "release-strategy-method";
+
+	private static final String RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE = "release-partial-sequences";
+
 	@Override
 	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
 
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder
 				.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelatingMessageHandler");
-		BeanDefinitionBuilder processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-				IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.Resequencer");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(processorBuilder, element, "release-partial-sequences");
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(processorBuilder, element, "comparator");
+		BeanDefinitionBuilder processorBuilder = BeanDefinitionBuilder
+				.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+						+ ".aggregator.ResequencingMessageGroupProcessor");
 
-		String processorRef = BeanDefinitionReaderUtils.registerWithGeneratedName(processorBuilder
-				.getBeanDefinition(), parserContext.getRegistry());
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(processorBuilder, element, COMPARATOR_ATTRIBUTE);
+
+		String processorRef = BeanDefinitionReaderUtils.registerWithGeneratedName(processorBuilder.getBeanDefinition(),
+				parserContext.getRegistry());
 
 		// Message group processor
 		builder.addConstructorArgReference(processorRef);
@@ -53,25 +72,42 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
 		String correlationStrategyRef = getCorrelationStrategyRef(element, parserContext);
 		if (correlationStrategyRef != null) {
 			builder.addConstructorArgReference(correlationStrategyRef);
-		}
-		else {
+		} else {
 			// Correlation strategy
 			builder.addConstructorArgValue(null);
 		}
-		// Release strategy
-		builder.addConstructorArgReference(processorRef);
 
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-store");
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "discard-channel");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-partial-result-on-expiry");
+		// Release strategy
+		builder.addConstructorArgValue(getReleaseStrategy(element, parserContext));
+
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, MESSAGE_STORE_ATTRIBUTE);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
 		return builder;
 	}
 
+	private Object getReleaseStrategy(Element element, ParserContext parserContext) {
+		String releaseStrategyRef = getReleasenStrategyRef(element, parserContext);
+		if (releaseStrategyRef == null) {
+			BeanDefinitionBuilder builder = BeanDefinitionBuilder
+					.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+							+ ".aggregator.SequenceSizeReleaseStrategy");
+			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE);
+			return builder.getBeanDefinition();
+		}
+		if (StringUtils.hasText(element.getAttribute(RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE))) {
+			parserContext.getReaderContext().error(
+					"Only one of " + RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE + " and " + RELEASE_STRATEGY_REF_ATTRIBUTE
+							+ " can be specified at once", element);
+		}
+		return new RuntimeBeanReference(releaseStrategyRef);
+	}
+
 	private String getCorrelationStrategyRef(Element element, ParserContext parserContext) {
-		String ref = element.getAttribute("correlation-strategy");
-		String method = element.getAttribute("correlation-strategy-method");
+		String ref = element.getAttribute(CORRELATION_STRATEGY_ATTRIBUTE);
+		String method = element.getAttribute(CORRELATION_STRATEGY_METHOD_ATTRIBUTE);
 		if (StringUtils.hasText(ref)) {
 			if (StringUtils.hasText(method)) {
 				BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
@@ -83,8 +119,28 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
 				String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder
 						.getBeanDefinition(), parserContext.getRegistry());
 				return adapterBeanName;
+			} else {
+				return ref;
 			}
-			else {
+		}
+		return null;
+	}
+
+	private String getReleasenStrategyRef(Element element, ParserContext parserContext) {
+		String ref = element.getAttribute(RELEASE_STRATEGY_REF_ATTRIBUTE);
+		String method = element.getAttribute(RELEASE_STRATEGY_METHOD_ATTRIBUTE);
+		if (StringUtils.hasText(ref)) {
+			if (StringUtils.hasText(method)) {
+				BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
+						.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+								+ ".aggregator.ReleaseStrategyAdapter");
+				adapterBuilder.addConstructorArgReference(ref);
+				adapterBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method,
+						"java.lang.String");
+				String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder
+						.getBeanDefinition(), parserContext.getRegistry());
+				return adapterBeanName;
+			} else {
 				return ref;
 			}
 		}