From 7991ef889c1fc8862db49623cd6c82449e0e6437 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 7 Oct 2008 02:10:20 +0000 Subject: [PATCH] AbstractMessageBarrierEndpoint is now AbstractMessageBarrierConsumer. --- ...nt.java => AbstractMessageBarrierConsumer.java} | 14 ++++++++------ .../integration/aggregator/AggregatorEndpoint.java | 8 ++++---- .../integration/aggregator/Resequencer.java | 8 ++++---- 3 files changed, 16 insertions(+), 14 deletions(-) rename org.springframework.integration/src/main/java/org/springframework/integration/aggregator/{AbstractMessageBarrierEndpoint.java => AbstractMessageBarrierConsumer.java} (94%) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java similarity index 94% rename from org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java rename to org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java index 749aad4265..ccf7324dcf 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java @@ -58,12 +58,14 @@ import org.springframework.util.ObjectUtils; * complete group after the arrival of the first {@link Message} of the group. * The default value is 1 minute. If the timeout elapses prior to completion, * then Messages with that timed-out 'correlationId' will be sent to the - * 'discardChannel' if provided. + * 'discardChannel' if provided unless 'sendPartialResultsOnTimeout' is set to + * true in which case the incomplete group will be sent to the output channel. * * @author Mark Fisher * @author Marius Bogoevici */ -public abstract class AbstractMessageBarrierEndpoint extends AbstractReplyProducingMessageConsumer implements TaskSchedulerAware, InitializingBean { +public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProducingMessageConsumer + implements TaskSchedulerAware, InitializingBean { public final static long DEFAULT_SEND_TIMEOUT = 1000; @@ -116,7 +118,7 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractReplyProduc /** * Specify whether to aggregate and send the resulting Message when the - * timeout elapses prior to the CompletionStrategy. + * timeout elapses prior to the CompletionStrategy returning true. */ public void setSendPartialResultOnTimeout(boolean sendPartialResultOnTimeout) { this.sendPartialResultOnTimeout = sendPartialResultOnTimeout; @@ -187,8 +189,8 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractReplyProduc } if (this.trackedCorrelationIds.contains(correlationId)) { if (logger.isDebugEnabled()) { - logger.debug("Handling for correlationId '" + correlationId + - "' has already completed or timed out."); + logger.debug("Handling of Message group with correlationId '" + + correlationId + "' has already completed or timed out."); } this.sendToDiscardChannelIfAvailable(message); return null; @@ -208,7 +210,7 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractReplyProduc if (ObjectUtils.isEmpty(processedMessages)) { return null; } - afterRelease(correlationId, releasedMessages); + this.afterRelease(correlationId, releasedMessages); return null; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorEndpoint.java index e6c3344094..87a7d3c639 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorEndpoint.java @@ -24,7 +24,7 @@ import org.springframework.integration.scheduling.TaskScheduler; import org.springframework.util.Assert; /** - * An {@link AbstractMessageBarrierEndpoint} that waits for a complete + * An {@link AbstractMessageBarrierConsumer} that waits for a complete * group of {@link Message Messages} to arrive and then delegates to an * {@link Aggregator} to combine them into a single {@link Message}. *

@@ -33,13 +33,13 @@ import org.springframework.util.Assert; * custom implementation of the {@link CompletionStrategy} may be provided. *

* All considerations regarding timeout and grouping by ' - * correlationId' from {@link AbstractMessageBarrierEndpoint} apply - * here as well. + * correlationId' from {@link AbstractMessageBarrierConsumer} + * apply here as well. * * @author Mark Fisher * @author Marius Bogoevici */ -public class AggregatorEndpoint extends AbstractMessageBarrierEndpoint { +public class AggregatorEndpoint extends AbstractMessageBarrierConsumer { private final Aggregator aggregator; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java index 426085d10b..044d4121a7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java @@ -22,7 +22,7 @@ import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageHeaders; /** - * An {@link AbstractMessageBarrierEndpoint} that waits for a group of + * An {@link AbstractMessageBarrierConsumer} that waits for a group of * {@link Message Messages} to arrive and re-sends them in order, sorted * by their sequenceNumber. *

@@ -30,12 +30,12 @@ import org.springframework.integration.message.MessageHeaders; * wait for the whole sequence to arrive before re-sending them. *

* All considerations regarding timeout and grouping by - * 'correlationId' from {@link AbstractMessageBarrierEndpoint} apply - * here as well. + * 'correlationId' from {@link AbstractMessageBarrierConsumer} + * apply here as well. * * @author Marius Bogoevici */ -public class Resequencer extends AbstractMessageBarrierEndpoint { +public class Resequencer extends AbstractMessageBarrierConsumer { private volatile boolean releasePartialSequences = true;