diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
index 8438a23696..a00b2725f7 100644
--- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
+++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
@@ -19,7 +19,7 @@ package org.springframework.integration.file;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
-import org.springframework.integration.aggregator.Resequencer;
+import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.file.locking.FileLocker;
@@ -43,7 +43,7 @@ import java.util.concurrent.PriorityBlockingQueue;
* this. See {@link org.springframework.integration.file.CompositeFileListFilter} for a way to do this.
*
* A {@link Comparator} can be used to ensure internal ordering of the Files in a {@link PriorityBlockingQueue}. This
- * does not provide the same guarantees as a {@link Resequencer}, but in cases where writing files and failure
+ * does not provide the same guarantees as a {@link ResequencingMessageGroupProcessor}, but in cases where writing files and failure
* downstream are rare it might be sufficient.
*
* FileReadingMessageSource is fully thread-safe under concurrent receive() invocations and message
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/ResequencingMessageGroupProcessor.java
similarity index 61%
rename from org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java
rename to org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java
index 055fe6f1a2..01be8a2510 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/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> 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> sorted = new ArrayList>(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> messages = group.getUnmarked();
if (messages.size() > 0) {
@@ -79,10 +56,4 @@ public class Resequencer implements ReleaseStrategy, MessageGroupProcessor {
}
}
- private static class SequenceNumberComparator implements Comparator> {
- public int compare(Message> o1, Message> o2) {
- return o1.getHeaders().getSequenceNumber().compareTo(o2.getHeaders().getSequenceNumber());
- }
- }
-
}
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceNumberComparator.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceNumberComparator.java
new file mode 100644
index 0000000000..6854d73bdf
--- /dev/null
+++ b/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> {
+
+ /**
+ * 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);
+ }
+
+}
\ No newline at end of file
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java
index 28c18e4f2a..dfcdbd74ff 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java
+++ b/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> 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> sorted = new ArrayList>(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();
}
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
index 8ba4cfe85d..79eda24103 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
+++ b/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,
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java
index d528e87bf5..d720ca20b0 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java
+++ b/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;
}
}
diff --git a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
index 5243a32eef..cc7c0458dc 100644
--- a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
+++ b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
@@ -1,7 +1,6 @@
@@ -18,7 +17,7 @@
- Enables annotation support for Message Endpoints.
+ Enables annotation support for Message Endpoints.
@@ -27,9 +26,11 @@
- Defines the ApplicationEventMulticaster to use for this ApplicationContext.
- The "task-executor" reference is optional. If not provided, an instance of
- ThreadPoolTaskExecutor will be created by default.
+ Defines the ApplicationEventMulticaster to use for this
+ ApplicationContext.
+ The "task-executor" reference is optional. If not provided, an
+ instance of
+ ThreadPoolTaskExecutor will be created by default.
@@ -54,7 +55,7 @@
- Defines a Point-to-Point MessageChannel.
+ Defines a Point-to-Point MessageChannel.
@@ -70,14 +71,16 @@
- Identifies this channel as a Queue style channel
+ Identifies this channel as a Queue style
+ channel
- Identifies this channel as a Queue style channel where messages could be prioritized
+ Identifies this channel as a Queue style
+ channel where messages could be prioritized
based on custom logic
@@ -102,15 +105,17 @@
- Defines a queue for messages. If 'capacity' is specified, it will be a bounded queue.
- A custom Queue implementation can be injected using the 'ref' attribute.
+ Defines a queue for messages. If 'capacity' is specified, it will be a
+ bounded queue.
+ A custom Queue implementation can be injected using the 'ref'
+ attribute.
- Capacity for this queue. Default capacity is 0 which means
- this queue will accumulate as many messages as memory allows.
+ Capacity for this queue. Default capacity is 0 which means
+ this queue will accumulate as many messages as memory allows.
@@ -128,14 +133,14 @@
- Defines a queue with priority-ordering for message reception.
+ Defines a queue with priority-ordering for message reception.
- Capacity for this queue. Default capacity is 0 which means
- this queue will accumulate as many messages as memory allows.
+ Capacity for this queue. Default capacity is 0 which means
+ this queue will accumulate as many messages as memory allows.
@@ -154,7 +159,8 @@
- Defines a rendezvous queue where a sender will block until the receiver arrives or vice-versa.
+ Defines a rendezvous queue where a sender will block until the receiver
+ arrives or vice-versa.
@@ -162,33 +168,35 @@
- Defines the dispatching configuration for a non-buffering channel
- (i.e. one without a queue).
+ Defines the dispatching configuration for a non-buffering channel
+ (i.e. one without a queue).
- Defines a load-balancing strategy for the channel's dispatcher.
- The default is a round-robin load balancer.
+ Defines a load-balancing strategy for the channel's dispatcher.
+ The default is a round-robin load balancer.
- [DEFAULT] Defines a Round Robin dispatching strategy which allows
- load balancing of messages between multiple Message Handlers. Which message
- handler receives the message first is determined by the 'order' attribute
- of such Message Handler.
+ [DEFAULT] Defines a Round Robin dispatching strategy which allows
+ load balancing of messages between multiple Message Handlers. Which
+ message
+ handler receives the message first is determined by the 'order'
+ attribute
+ of such Message Handler.
- No LoadBalancingStrategy will be used.
+ No LoadBalancingStrategy will be used.
@@ -198,14 +206,18 @@
- Specifies whether this dispatcher has failover enabled. By default,
- failover will be enabled. Set this to 'false' to disable it.
- When enabled and message delivery to the primary Message Handler fails,
- an attempt will be made to deliver the message to the next handler and so on...
- Primary, secondary etc... is determined by the load-balancing strategy in use
- (e.g. round-robin). If no load-balancer strategy is configured, the order will
- be fixed in a sequence determined by the 'order' attribute on the Message Handlers
- (or the @Ordered annotation on adapted methods).
+ Specifies whether this dispatcher has failover enabled. By default,
+ failover will be enabled. Set this to 'false' to disable it.
+ When enabled and message delivery to the primary Message Handler fails,
+ an attempt will be made to deliver the message to the next handler
+ and so on...
+ Primary, secondary etc... is determined by the load-balancing strategy in
+ use
+ (e.g. round-robin). If no load-balancer strategy is configured, the
+ order will
+ be fixed in a sequence determined by the 'order' attribute on the
+ Message Handlers
+ (or the @Ordered annotation on adapted methods).
@@ -230,7 +242,8 @@
- Defines a Publish-Subscribe channel that broadcasts messages to its subscribers.
+ Defines a Publish-Subscribe channel that broadcasts messages to its
+ subscribers.
@@ -263,7 +276,7 @@
TaskExecutor dispatches to the handler on a separate Thread.
Usually configured using 'task' namespace support provided by Spring (e.g., <task:executor/>)
]]>
-
+
-
+
-
+
-
+
- Specify whether Exceptions thrown by any subscribed handler should be ignored (only logged).
+ Specify whether Exceptions thrown by any subscribed handler should be
+ ignored (only logged).
-
+
- Specify whether the sequence size, sequence number, and correlation id headers should be set on
- Messages that are sent through this channel.
+ Specify whether the sequence size, sequence number, and correlation id
+ headers should be set on
+ Messages that are sent through this channel.
@@ -312,7 +328,7 @@
- Defines a channel that maintains its Messages on a thread-bound queue.
+ Defines a channel that maintains its Messages on a thread-bound queue.
@@ -335,7 +351,7 @@
- Defines a message channel.
+ Defines a message channel.
@@ -348,21 +364,22 @@
is a channel that accepts messages cotaining payload of certain type.
]]>
-
+
- This attribute is DEPRECATED. Please use the dispatcher sub-element instead.
+ This attribute is DEPRECATED. Please use the dispatcher sub-element
+ instead.
- Enables failover, but disables load-balancing.
- See the dispatcher sub-element for more information.
+ Enables failover, but disables load-balancing.
+ See the dispatcher sub-element for more information.
@@ -375,7 +392,7 @@
- Defines a Messaging Gateway.
+ Defines a Messaging Gateway.
@@ -398,7 +415,7 @@
]]>
-
+
@@ -408,7 +425,8 @@
-
+
-
+
@@ -443,7 +461,7 @@
-
+
-
+
- Defines a Messaging Gateway to be used within a chain.
+ Defines a Messaging Gateway to be used within a chain.
@@ -567,21 +586,21 @@
+ type="org.springframework.integration.core.MessageChannel" />
-
+
-
+
@@ -589,7 +608,8 @@
- Defines a Channel Adapter that receives from a MessageSource and sends to a MessageChannel.
+ Defines a Channel Adapter that receives from a MessageSource and sends to a
+ MessageChannel.
@@ -597,8 +617,8 @@
- Defines a Channel Adapter that receives from a MessageChannel and passes to
- a method-invoking MessageHandler.
+ Defines a Channel Adapter that receives from a MessageChannel and passes to
+ a method-invoking MessageHandler.
@@ -607,8 +627,8 @@
- Specifies the order for invocation when this endpoint is connected as a
- subscriber to a SubscribableChannel.
+ Specifies the order for invocation when this endpoint is connected as a
+ subscriber to a SubscribableChannel.
@@ -680,7 +700,7 @@
-
+
@@ -693,7 +713,7 @@
-
+
@@ -713,12 +733,12 @@
- Defines an endpoint for exposing any bean reference as a service that
- receives request Messages from an 'input-channel' and may send reply
- Messages to an 'output-channel'. The 'ref' may point to an instance
- that has either a single public method or a method with the
- @ServiceActivator annotation. Otherwise, the 'method' attribute
- should be provided along with 'ref'.
+ Defines an endpoint for exposing any bean reference as a service that
+ receives request Messages from an 'input-channel' and may send reply
+ Messages to an 'output-channel'. The 'ref' may point to an instance
+ that has either a single public method or a method with the
+ @ServiceActivator annotation. Otherwise, the 'method' attribute
+ should be provided along with 'ref'.
@@ -726,7 +746,7 @@
- Base type for Message-handling endpoints.
+ Base type for Message-handling endpoints.
@@ -743,8 +763,10 @@
- Specify the maximum amount of time in milliseconds to wait when sending a reply
- Message to the output channel. By default the send will block for one second.
+ Specify the maximum amount of time in milliseconds to wait when sending
+ a reply
+ Message to the output channel. By default the send will block for one
+ second.
@@ -752,7 +774,7 @@
-
+
@@ -764,8 +786,9 @@
- Base type for Message Endpoint elements that accept Messages from an
- input-channel and also may produce reply Messages to be sent to an output-channel.
+ Base type for Message Endpoint elements that accept Messages from an
+ input-channel and also may produce reply Messages to be sent to an
+ output-channel.
@@ -787,7 +810,8 @@
- Base type for Message Endpoint elements that accept Messages from an
+ Base type for Message Endpoint elements that
+ accept Messages from an
input-channel.
@@ -839,7 +863,7 @@
-
+
@@ -851,8 +875,10 @@
- Defines an endpoint that passes a Message to the output-channel after a delay. The delay may
- be retrieved from a Message header or else fallback to the 'default-delay' of this endpoint.
+ Defines an endpoint that passes a Message to the output-channel after a
+ delay. The delay may
+ be retrieved from a Message header or else fallback to the
+ 'default-delay' of this endpoint.
@@ -860,37 +886,44 @@
+ minOccurs="0" maxOccurs="1" />
-
+
- Specify the default delay in milliseconds. This value can be set to 0 if the only Messages
- that should be delayed are those with a particular header (in that case, be sure to provide
- a value for the 'delay-header-name' attribute).
+ Specify the default delay in milliseconds. This value can be set to 0
+ if the only Messages
+ that should be delayed are those with a particular header (in that
+ case, be sure to provide
+ a value for the 'delay-header-name' attribute).
- Specify the name of the header that should contain the delay value. This value can either
- represent the number of milliseconds to delay counting from the current time or it can be an
- absolute Date until which the Message should be delayed.
+ Specify the name of the header that should contain the delay value.
+ This value can either
+ represent the number of milliseconds to delay counting from the current
+ time or it can be an
+ absolute Date until which the Message should be delayed.
- Provide a reference to the ScheduledExecutorService instance to which this endpoint should
- delegate when scheduling the sending of delayed Messages. If not provided, the default
- will use a thread pool of size 1.
+ Provide a reference to the ScheduledExecutorService instance to which
+ this endpoint should
+ delegate when scheduling the sending of delayed Messages. If not
+ provided, the default
+ will use a thread pool of size 1.
+ type="java.util.concurrent.ScheduledExecutorService" />
@@ -898,15 +931,19 @@
- Specify the maximum amount of time in milliseconds to wait when sending the released
- Messages (after delay) to the output channel. By default the send will block indefinitely.
+ Specify the maximum amount of time in milliseconds to wait when sending
+ the released
+ Messages (after delay) to the output channel. By default the send will
+ block indefinitely.
-
+
- Specify whether tasks should be able to complete on shutdown. By default this is 'false'.
+ Specify whether tasks should be able to complete on shutdown. By
+ default this is 'false'.
@@ -1101,7 +1138,7 @@
- Defines a MessageSelector chain.
+ Defines a MessageSelector chain.
@@ -1140,8 +1177,9 @@
- Provides a MessageSelector reference. If a method attribute is set the
- referred bean doesn't need to implement the MessageSelector interface.
+ Provides a MessageSelector reference. If a method attribute is set the
+ referred bean doesn't need to implement the MessageSelector
+ interface.
@@ -1157,7 +1195,7 @@
-
+
@@ -1167,32 +1205,33 @@
- Defines a HeaderEnricher endpoint for values defined in the MessageHeader.
+ Defines a HeaderEnricher endpoint for values defined in the MessageHeader.
-
-
-
-
+
+
+
+
-
+
- Boolean value to indicate whether this header value should overwrite an existing header value for the same name.
+ Boolean value to indicate whether this header value should overwrite
+ an existing header value for the same name.
-
+
@@ -1200,23 +1239,27 @@
- Element that accepts any user-defined header name/value pair.
+ Element that accepts any user-defined header name/value pair.
-
+
- Specify the default boolean value for whether to overwrite existing header values. This will only take effect for
- sub-elements that do not provide their own 'overwrite' attribute. If the 'default-overwrite' attribute is not
- provided, then the specified header values will NOT overwrite any existing ones with the same header names.
+ Specify the default boolean value for whether to overwrite existing
+ header values. This will only take effect for
+ sub-elements that do not provide their own 'overwrite' attribute. If the
+ 'default-overwrite' attribute is not
+ provided, then the specified header values will NOT overwrite any
+ existing ones with the same header names.
-
-
+
+
@@ -1225,18 +1268,18 @@
-
-
-
-
-
+
+
+
+
+
- Defines a Message Header with a literal value or object reference.
+ Defines a Message Header with a literal value or object reference.
@@ -1244,7 +1287,7 @@
- Name of the header to be added.
+ Name of the header to be added.
@@ -1255,10 +1298,10 @@
- Provides a header value for the given header name. Requires
- exactly one of the 'ref', 'value', or 'expression' attributes.
- The 'type' attribute allows for the specification of the expected
- type when using a 'value' or 'expression', but it is optional.
+ Provides a header value for the given header name. Requires
+ exactly one of the 'ref', 'value', or 'expression' attributes.
+ The 'type' attribute allows for the specification of the expected
+ type when using a 'value' or 'expression', but it is optional.
@@ -1266,15 +1309,16 @@
- Literal value to be associated with the given header name.
+ Literal value to be associated with the given header name.
- Expression to be evaulated at runtime to determine the header value.
- The EvaluationContext will include variables for 'payload' and 'headers'.
+ Expression to be evaulated at runtime to determine the header value.
+ The EvaluationContext will include variables for 'payload' and
+ 'headers'.
@@ -1293,21 +1337,21 @@
- Reference to be associated with the given header name.
+ Reference to be associated with the given header name.
-
+
- Name of a method to be invoked on the referenced target object.
+ Name of a method to be invoked on the referenced target object.
-
+
@@ -1315,19 +1359,21 @@
- Boolean value to indicate whether this header value should overwrite an existing header value for the same name.
+ Boolean value to indicate whether this header value should overwrite an
+ existing header value for the same name.
-
+
-
-
+
+
- Defines a Transformer.
+ Defines a Transformer.
@@ -1335,28 +1381,28 @@
- Defines a Transformer that converts any Object payload to a String by
- invoking its toString() method.
+ Defines a Transformer that converts any Object payload to a String by
+ invoking its toString() method.
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
- Defines a Transformer that stores a Message and returns a new Message whose
- payload is the id of the stored Message.
+ Defines a Transformer that stores a Message and returns a new Message whose
+ payload is the id of the stored Message.
@@ -1364,8 +1410,10 @@
- Defines a Transformer that accepts a Message whose payload is a UUID and retrieves
- the Message associated with that id from a MessageStore if available (else null).
+ Defines a Transformer that accepts a Message whose payload is a UUID and
+ retrieves
+ the Message associated with that id from a MessageStore if available
+ (else null).
@@ -1374,16 +1422,17 @@
-
+
- Reference to the MessageStore to be used by this Claim Check transformer.
+ Reference to the MessageStore to be used by this Claim Check transformer.
-
+
@@ -1395,46 +1444,47 @@
- Defines a Transformer that serializes any Object payload that implements
- Serializable into a byte array.
+ Defines a Transformer that serializes any Object payload that implements
+ Serializable into a byte array.
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
- Defines a Transformer that deserializes a byte array payload into an Object.
+ Defines a Transformer that deserializes a byte array payload into an
+ Object.
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
- Defines a Filter.
+ Defines a Filter.
@@ -1445,12 +1495,13 @@
+ type="org.springframework.integration.core.MessageChannel" />
-
+
-
+
@@ -1477,7 +1528,7 @@
+ type="org.springframework.integration.core.MessageChannel" />
@@ -1514,7 +1565,7 @@
+ type="org.springframework.integration.core.MessageChannel" />
@@ -1546,7 +1597,7 @@
+ type="org.springframework.integration.core.MessageChannel" />
@@ -1569,10 +1620,11 @@
-
-
-
-
+
+
+
+
@@ -1586,12 +1638,12 @@
-
+
-
+
-
+
- Specify whether a failure to resolve a channel name returned by this router should be ignored.
+ Specify whether a failure to resolve a channel name returned by this
+ router should be ignored.
@@ -1637,10 +1691,11 @@
type="org.springframework.integration.core.MessageChannel" />
- Reference to the default channel where Messages should be sent if channel
- resolution fails to return any channels. If no default channel is
- provided, the router will either drop the Message or throw an Exception
- depending on the value of the "resolution-required" attribute.
+ Reference to the default channel where Messages should be sent if channel
+ resolution fails to return any channels. If no default channel
+ is
+ provided, the router will either drop the Message or throw an Exception
+ depending on the value of the "resolution-required" attribute.
@@ -1648,7 +1703,8 @@
- Specify whether this router should always be required to return at least one channel or name.
+ Specify whether this router should always be required to return at least
+ one channel or name.
@@ -1662,23 +1718,26 @@
- Specify the maximum amount of time in milliseconds to wait when sending Messages
- to the target MessageChannels. By default the send will block indefinitely.
+ Specify the maximum amount of time in milliseconds to wait when sending
+ Messages
+ to the target MessageChannels. By default the send will block
+ indefinitely.
- Specify whether a failure to send to a single channel should be ignored.
- Otherwise MessageDeliveryExceptions will be thrown.
+ Specify whether a failure to send to a single channel should be ignored.
+ Otherwise MessageDeliveryExceptions will be thrown.
-
+
- Specify whether sequence number and size headers should be added to each Message.
+ Specify whether sequence number and size headers should be added to each
+ Message.
@@ -1686,7 +1745,8 @@
-
+
Defines a Splitter.
@@ -1700,7 +1760,7 @@
Defines an aggregating message endpoint.
-
+
@@ -1716,7 +1776,7 @@
-
+
@@ -1730,11 +1790,12 @@
-
+
-
+
@@ -1752,9 +1813,13 @@
- Reference to a MessageGroupStore for holding state in between message processing. The default
- is to use a volatile in-memory store, which means that unprocessed messages will be lost if the
- JVM exits. To customize the expiry of incomplete message groups configure the message store.
+ Reference to a MessageGroupStore for holding
+ state in between message processing. The default
+ is to use a
+ volatile in-memory store, which means that unprocessed messages
+ will be lost if the
+ JVM exits. To customize the expiry of incomplete message groups
+ configure the message store.
@@ -1764,7 +1829,8 @@
-
+
@@ -1774,7 +1840,7 @@
- Defines a resequencing message endpoint.
+ Defines a resequencing message endpoint.
@@ -1788,11 +1854,37 @@
-
+
-
+
+
+
+
+
+
+
+
+ The release strategy to use to decide when
+ messages can be processed. Defaults to a SequenceSizeReleaseStrategy,
+ releasing all messages once the sequence is complete.
+ This is mutually exclusive with the release-partial-sequences
+ attribute (either or none can be specified , but not both).
+
+
+
+
+
+
+
+
+
+
+
+
+
@@ -1810,9 +1902,13 @@
- Reference to a MessageGroupStore for holding state in between message processing. The default
- is to use a volatile in-memory store, which means that unprocessed messages will be lost if the
- JVM exits. To customize the expiry of incomplete message groups configure the message store.
+ Reference to a MessageGroupStore for holding
+ state in between message processing. The default
+ is to use a
+ volatile in-memory store, which means that unprocessed messages
+ will be lost if the
+ JVM exits. To customize the expiry of incomplete message groups
+ configure the message store.
@@ -1831,8 +1927,19 @@
-
-
+
+
+
+ Flag to say that partial sequences can be released (e.g. 1-4 of 10).
+ Defaults to true, so the sequence has to be complete before any messages
+ are released.
+ This is mutually exclusive with the release-strategy
+ attribute (either or none can be specified , but not both).
+
+
+
+
@@ -1841,7 +1948,8 @@
- Defines a list of interceptors. Each element may be a ChannelInterceptor, ref, or inner-bean.
+ Defines a list of interceptors. Each element may be a ChannelInterceptor,
+ ref, or inner-bean.
@@ -1870,7 +1978,7 @@
- Defines a Wire Tap Channel Interceptor.
+ Defines a Wire Tap Channel Interceptor.
@@ -1888,7 +1996,8 @@
-
+
+ type="org.springframework.transaction.PlatformTransactionManager" />
@@ -1944,7 +2053,8 @@
]]>
-
+
-
+
@@ -1964,22 +2074,23 @@
-
-
-
-
+
+
+
+
-
+
- Defines a MessagePublishingInterceptor which allows you to generate messages
- as a by-product of method invocations on Spring configured components.
+ Defines a MessagePublishingInterceptor which allows you to generate
+ messages
+ as a by-product of method invocations on Spring configured components.
-
+
@@ -1992,7 +2103,7 @@
+ type="org.springframework.integration.core.MessageChannel" />
@@ -2001,55 +2112,59 @@
-
-
+
+
+ type="org.springframework.integration.core.MessageChannel" />
-
+
- Allows you to define channel interceptors to be applied globally.
+ Allows you to define channel interceptors to be applied globally.
-
-
-
-
-
-
-
+
+
+
+
+
+
+
- [REQUIRED] Channel name(s) or patterns. To specify more than one channel use ','
- (e.g., channel-name-pattern="input*, foo, bar")
+ [REQUIRED] Channel name(s) or patterns. To specify more than one channel use
+ ','
+ (e.g., channel-name-pattern="input*, foo, bar")
- [OPTIONAL] Specifies the order in which these interceptors will be
- added to the existing channel interceptors (if any).
- Negative value (e.g., -2) will signify AFTER, but BEFORE the
- the chain that might specify -1 (if any). Positive value (e.g., 2)
- will signify BEFORE, but AFTER the chain that might specify 1 (if any).
+ [OPTIONAL] Specifies the order in which these interceptors will be
+ added to the existing channel interceptors (if any).
+ Negative value (e.g., -2) will signify AFTER, but BEFORE the
+ the chain that might specify -1 (if any). Positive value (e.g., 2)
+ will signify BEFORE, but AFTER the chain that might specify 1 (if
+ any).
-
+
\ No newline at end of file
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
index c90502b625..1b49dafcb5 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
@@ -43,18 +43,17 @@ public class ResequencerTests {
private CorrelatingMessageHandler resequencer;
- private Resequencer processor = new Resequencer();
+ private ResequencingMessageGroupProcessor processor = new ResequencingMessageGroupProcessor();
private MessageGroupStore store = new SimpleMessageStore();
@Before
public void configureResequencer() {
- this.resequencer = new CorrelatingMessageHandler(processor, store, null, processor);
+ this.resequencer = new CorrelatingMessageHandler(processor, store, null, null);
}
@Test
public void testBasicResequencing() throws InterruptedException {
- this.processor.setReleasePartialSequences(false);
QueueChannel replyChannel = new QueueChannel();
Message> message1 = createMessage("123", "ABC", 3, 3, replyChannel);
Message> message2 = createMessage("456", "ABC", 3, 1, replyChannel);
@@ -75,7 +74,6 @@ public class ResequencerTests {
@Test
public void testBasicResequencingWithCustomComparator() throws InterruptedException {
- this.processor.setReleasePartialSequences(false);
this.processor.setComparator(new Comparator>() {
@SuppressWarnings("unchecked")
public int compare(Message> o1, Message> o2) {
@@ -102,7 +100,6 @@ public class ResequencerTests {
@Test
public void testResequencingWithDuplicateMessages() {
- this.processor.setReleasePartialSequences(false);
QueueChannel replyChannel = new QueueChannel();
Message> message1 = createMessage("123", "ABC", 3, 3, replyChannel);
Message> message2 = createMessage("456", "ABC", 3, 1, replyChannel);
@@ -124,7 +121,7 @@ public class ResequencerTests {
@Test
public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException {
- this.processor.setReleasePartialSequences(true);
+ this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true));
QueueChannel replyChannel = new QueueChannel();
Message> message1 = createMessage("123", "ABC", 4, 2, replyChannel);
Message> message2 = createMessage("456", "ABC", 4, 1, replyChannel);
@@ -144,6 +141,41 @@ public class ResequencerTests {
assertNull(reply3);
// when sending the last message, the whole sequence must have been sent
this.resequencer.handleMessage(message4);
+ reply3 = replyChannel.receive(0); Message> reply4 = replyChannel.receive(0);
+ assertNotNull(reply3);
+ assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber());
+ assertNotNull(reply4);
+ assertEquals(new Integer(4), reply4.getHeaders().getSequenceNumber());
+ }
+
+ @Test
+ public void testResequencingWithPartialSequenceAndComparator() throws InterruptedException {
+ this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true));
+ this.processor.setComparator(new Comparator>() {
+ @SuppressWarnings("unchecked")
+ public int compare(Message> o1, Message> o2) {
+ return ((Comparable)o1.getPayload()).compareTo(o2.getPayload());
+ }
+ });
+ QueueChannel replyChannel = new QueueChannel();
+ Message> message1 = createMessage("456", "ABC", 4, 2, replyChannel);
+ Message> message2 = createMessage("123", "ABC", 4, 1, replyChannel);
+ Message> message3 = createMessage("XYZ", "ABC", 4, 4, replyChannel);
+ Message> message4 = createMessage("789", "ABC", 4, 3, replyChannel);
+ this.resequencer.handleMessage(message1);
+ this.resequencer.handleMessage(message2);
+ this.resequencer.handleMessage(message3);
+ Message> reply1 = replyChannel.receive(0);
+ Message> reply2 = replyChannel.receive(0);
+ Message> reply3 = replyChannel.receive(0);
+ // only messages 1 and 2 should have been received by now
+ assertNotNull(reply1);
+ assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber());
+ assertNotNull(reply2);
+ assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber());
+ assertNull(reply3);
+ // when sending the last message, the whole sequence must have been sent
+ this.resequencer.handleMessage(message4);
reply3 = replyChannel.receive(0);
Message> reply4 = replyChannel.receive(0);
assertNotNull(reply3);
@@ -159,7 +191,6 @@ public class ResequencerTests {
Message> message2 = createMessage("456", "ABC", 4, 1, null);
Message> message3 = createMessage("789", "ABC", 4, 4, null);
this.resequencer.setSendPartialResultOnExpiry(false);
- this.processor.setReleasePartialSequences(false);
this.resequencer.setDiscardChannel(discardChannel);
this.resequencer.handleMessage(message1);
this.resequencer.handleMessage(message2);
@@ -187,7 +218,6 @@ public class ResequencerTests {
Message> message1 = createMessage("123", "ABC", 4, 2, null);
Message> message2 = createMessage("456", "ABC", 5, 1, null);
this.resequencer.setSendPartialResultOnExpiry(false);
- this.processor.setReleasePartialSequences(false);
this.resequencer.setDiscardChannel(discardChannel);
this.resequencer.handleMessage(message1);
this.resequencer.handleMessage(message2);
@@ -205,7 +235,6 @@ public class ResequencerTests {
QueueChannel discardChannel = new QueueChannel();
Message> message1 = createMessage("123", "ABC", 2, 4, null);
this.resequencer.setSendPartialResultOnExpiry(false);
- this.processor.setReleasePartialSequences(false);
this.resequencer.setDiscardChannel(discardChannel);
this.resequencer.handleMessage(message1);
// this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC"));
@@ -216,7 +245,6 @@ public class ResequencerTests {
@Test
public void testResequencingWithCompleteSequenceRelease() throws InterruptedException {
- this.processor.setReleasePartialSequences(false);
QueueChannel replyChannel = new QueueChannel();
Message> message1 = createMessage("123", "ABC", 4, 2, replyChannel);
Message> message2 = createMessage("456", "ABC", 4, 1, replyChannel);
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
index 4768a49a9f..3bb7cf68b6 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
@@ -30,7 +30,8 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.aggregator.CorrelatingMessageHandler;
import org.springframework.integration.aggregator.CorrelationStrategy;
import org.springframework.integration.aggregator.CorrelationStrategyAdapter;
-import org.springframework.integration.aggregator.Resequencer;
+import org.springframework.integration.aggregator.ReleaseStrategyAdapter;
+import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
@@ -88,7 +89,7 @@ public class ResequencerParserTests {
"The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag",
false, getPropertyValue(resequencer, "sendPartialResultOnExpiry"));
assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag",
- false, getPropertyValue(getPropertyValue(resequencer, "outputProcessor"), "releasePartialSequences"));
+ false, getPropertyValue(getPropertyValue(resequencer, "releaseStrategy"), "releasePartialSequences"));
}
@Test
@@ -108,7 +109,7 @@ public class ResequencerParserTests {
"The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag",
true, getPropertyValue(resequencer, "sendPartialResultOnExpiry"));
assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag",
- false, getPropertyValue(getPropertyValue(resequencer, "outputProcessor"), "releasePartialSequences"));
+ false, getPropertyValue(getPropertyValue(resequencer, "releaseStrategy"), "releasePartialSequences"));
}
@Test
@@ -139,12 +140,23 @@ public class ResequencerParserTests {
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithComparator");
CorrelatingMessageHandler handler = TestUtils.getPropertyValue(endpoint, "handler",
CorrelatingMessageHandler.class);
- Resequencer resequencer = TestUtils.getPropertyValue(handler, "outputProcessor", Resequencer.class);
+ ResequencingMessageGroupProcessor resequencer = TestUtils.getPropertyValue(handler, "outputProcessor",
+ ResequencingMessageGroupProcessor.class);
Object comparator = getPropertyValue(resequencer, "comparator");
assertEquals("The Resequencer is not configured with a TestComparator", TestComparator.class, comparator
.getClass());
}
+ @Test
+ public void testReleaseStrategy() throws Exception {
+ EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithReleaseStrategy");
+ CorrelatingMessageHandler handler = TestUtils.getPropertyValue(endpoint, "handler",
+ CorrelatingMessageHandler.class);
+ Object releaseStrategy = getPropertyValue(handler, "releaseStrategy");
+ assertEquals("The Resequencer is not configured with an adapter", ReleaseStrategyAdapter.class, releaseStrategy
+ .getClass());
+ }
+
private static Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel outputChannel) {
return MessageBuilder.withPayload(payload).setCorrelationId(correlationId).setSequenceSize(sequenceSize)
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml
index ec075f060c..7a766c4656 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml
@@ -27,6 +27,8 @@
+
+
+
+
@@ -57,4 +64,9 @@
+
+
+
+