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;