From e378af900609cdd0a46cfd198def36777d94e2a3 Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Fri, 15 Oct 2010 17:50:03 +0200 Subject: [PATCH] INT-1339: Ensure sequences with gaps work in integrated scenario. Improve logging around correlated messages. --- .../SequenceSizeReleaseStrategy.java | 14 +++- .../integration/store/SimpleMessageGroup.java | 10 +++ .../src/test/java/log4j.properties | 8 -- .../PartialSequencesWithGapsTests-context.xml | 16 ++++ .../PartialSequencesWithGapsTests.java | 81 +++++++++++++++++++ 5 files changed, 120 insertions(+), 9 deletions(-) delete mode 100644 spring-integration-core/src/test/java/log4j.properties create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/PartialSequencesWithGapsTests-context.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/PartialSequencesWithGapsTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java index b98eeaa71d..bc4075fc76 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java +++ b/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> 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> sorted = new ArrayList>(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(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index 59a6676b0d..294ddf5dd6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java +++ b/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 + + '}'; + } } diff --git a/spring-integration-core/src/test/java/log4j.properties b/spring-integration-core/src/test/java/log4j.properties deleted file mode 100644 index 941cbe4822..0000000000 --- a/spring-integration-core/src/test/java/log4j.properties +++ /dev/null @@ -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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/PartialSequencesWithGapsTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/PartialSequencesWithGapsTests-context.xml new file mode 100644 index 0000000000..bbfce39234 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/PartialSequencesWithGapsTests-context.xml @@ -0,0 +1,16 @@ + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/PartialSequencesWithGapsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/PartialSequencesWithGapsTests.java new file mode 100644 index 0000000000..4949afbe6b --- /dev/null +++ b/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 received = new ArrayBlockingQueue(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(); + } +}