INT-1339: Ensure sequences with gaps work in integrated scenario. Improve logging around correlated messages.

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

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.aggregator;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.integration.Message;
 import org.springframework.integration.store.MessageGroup;
 
@@ -31,9 +33,12 @@ import java.util.List;
  * @author Mark Fisher
  * @author Marius Bogoevici
  * @author Dave Syer
+ * @author Iwein Fuld
  */
 public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
 
+	private static final Log logger = LogFactory.getLog(SequenceSizeReleaseStrategy.class);
+
 	private volatile Comparator<Message<?>> comparator = new SequenceNumberComparator();
 
 	private volatile boolean releasePartialSequences;
@@ -58,10 +63,17 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
 
 	public boolean canRelease(MessageGroup messages) {
 		if (releasePartialSequences) {
+			if(logger.isTraceEnabled()){
+				logger.trace("Considering partial release of group [" + messages + "]");
+			}
 			List<Message<?>> sorted = new ArrayList<Message<?>>(messages.getUnmarked());
 			Collections.sort(sorted, comparator);
 			int tail = sorted.get(0).getHeaders().getSequenceNumber() - 1;
-			return tail == messages.getMarked().size();
+			boolean release = tail == messages.getMarked().size();
+			if (logger.isTraceEnabled() && release) {
+				logger.trace("Release imminent because tail [" + tail + "] is next in line.");
+			}
+			return release;
 		}
 		return messages.isComplete();
 	}

spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java

@@ -221,4 +221,14 @@ public class SimpleMessageGroup implements MessageGroup {
 		return false;
 	}
 
+	@Override
+	public String toString() {
+		return "SimpleMessageGroup{" +
+				"groupId=" + groupId +
+				", lock=" + lock +
+				", marked=" + marked +
+				", unmarked=" + unmarked +
+				", timestamp=" + timestamp +
+				'}';
+	}
 }

spring-integration-core/src/test/java/log4j.properties

@@ -1,8 +0,0 @@
-log4j.rootCategory=WARN, stdout
-
-log4j.appender.stdout=org.apache.log4j.ConsoleAppender
-log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%c{1}: %m%n
-
-log4j.category.org.springframework.integration=WARN
-log4j.category.org.springframework.integration.file=WARN

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

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns:beans="http://www.springframework.org/schema/beans"
+			 xmlns="http://www.springframework.org/schema/integration"
+			 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+			 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="in"/>
+
+	<resequencer input-channel="in" output-channel="out" release-partial-sequences="true"/>
+
+	<channel id="out"/>
+	
+</beans:beans>

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

@@ -0,0 +1,81 @@
+/*
+ * 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.scenarios;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.core.SubscribableChannel;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+import static org.hamcrest.CoreMatchers.*;
+import static org.junit.Assert.assertThat;
+
+/**
+ * @author Iwein Fuld
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class PartialSequencesWithGapsTests {
+
+	@Autowired
+	MessageChannel in;
+
+	@Autowired
+	SubscribableChannel out;
+
+	Queue<Message> received = new ArrayBlockingQueue<Message>(10);
+
+	@Before
+	public void collectOutput() {
+		out.subscribe(new MessageHandler() {
+			public void handleMessage(Message<?> message) throws MessagingException {
+				received.add(message);
+			}
+		});
+	}
+
+	@Test
+	public void shouldNotReleaseAfterGap() {
+		in.send(message(6, 6));
+		in.send(message(2, 6));
+		in.send(message(1, 6));
+		assertThat(received.poll().getHeaders().getSequenceNumber(), is(1));
+		assertThat(received.poll().getHeaders().getSequenceNumber(), is(2));
+		received.poll();
+		received.poll();
+		in.send(message(5, 6));
+		assertThat(received.poll(), is(nullValue()));
+		in.send(message(4, 6));
+		assertThat(received.poll(), is(nullValue()));
+	}
+
+	private Message<?> message(int sequenceNumber, int sequenceSize) {
+		return MessageBuilder.withPayload("foo")
+				.setSequenceNumber(sequenceNumber)
+				.setSequenceSize(sequenceSize)
+				.setCorrelationId("foo").build();
+	}
+}