diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java index 53686e4261..8f6b212bd6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java @@ -13,6 +13,7 @@ package org.springframework.integration.config.xml; +import org.springframework.beans.BeanMetadataElement; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; @@ -28,9 +29,11 @@ import org.w3c.dom.Element; */ public class ResequencerParser extends AbstractConsumerEndpointParser { + private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy"; + private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method"; - private static final String CORRELATION_STRATEGY_ATTRIBUTE = "correlation-strategy"; + private static final String CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE = "correlation-strategy-expression"; private static final String SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE = "send-partial-result-on-expiry"; @@ -40,12 +43,14 @@ public class ResequencerParser extends AbstractConsumerEndpointParser { private static final String MESSAGE_STORE_ATTRIBUTE = "message-store"; - private static final String COMPARATOR_ATTRIBUTE = "comparator"; + private static final String COMPARATOR_REF_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_STRATEGY_EXPRESSION_ATTRIBUTE = "release-strategy-expression"; + private static final String RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE = "release-partial-sequences"; @Override @@ -57,7 +62,8 @@ public class ResequencerParser extends AbstractConsumerEndpointParser { .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.ResequencingMessageGroupProcessor"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(processorBuilder, element, COMPARATOR_ATTRIBUTE); + // Comparator + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, COMPARATOR_REF_ATTRIBUTE); String processorRef = BeanDefinitionReaderUtils.registerWithGeneratedName(processorBuilder.getBeanDefinition(), parserContext.getRegistry()); @@ -69,14 +75,8 @@ public class ResequencerParser extends AbstractConsumerEndpointParser { builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition( IntegrationNamespaceUtils.BASE_PACKAGE + ".store.SimpleMessageStore").getBeanDefinition()); - String correlationStrategyRef = getCorrelationStrategyRef(element, parserContext); - if (correlationStrategyRef != null) { - builder.addConstructorArgReference(correlationStrategyRef); - } else { - // Correlation strategy - builder.addConstructorArgValue(null); - } - + // Correlation strategy + builder.addConstructorArgValue(getCorrelationStrategy(element, parserContext)); // Release strategy builder.addConstructorArgValue(getReleaseStrategy(element, parserContext)); @@ -88,27 +88,16 @@ public class ResequencerParser extends AbstractConsumerEndpointParser { 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_ATTRIBUTE); + private BeanMetadataElement getCorrelationStrategy(Element element, ParserContext parserContext) { + String ref = element.getAttribute(CORRELATION_STRATEGY_REF_ATTRIBUTE); + String expression = element.getAttribute(CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE); String method = element.getAttribute(CORRELATION_STRATEGY_METHOD_ATTRIBUTE); if (StringUtils.hasText(ref)) { + if (StringUtils.hasText(expression)) { + parserContext.getReaderContext().error( + "Only one of correlation strategy expression and bean reference must be specified", element); + return null; + } if (StringUtils.hasText(method)) { BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE @@ -118,18 +107,42 @@ public class ResequencerParser extends AbstractConsumerEndpointParser { "java.lang.String"); String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder .getBeanDefinition(), parserContext.getRegistry()); - return adapterBeanName; - } else { - return ref; + return new RuntimeBeanReference(adapterBeanName); + } + else { + return new RuntimeBeanReference(ref); } } - return null; + else { + if (!StringUtils.hasText(expression)) { + return null; + } + BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder + .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + + ".aggregator.ExpressionEvaluatingCorrelationStrategy"); + adapterBuilder.addConstructorArgValue(expression); + String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder + .getBeanDefinition(), parserContext.getRegistry()); + return new RuntimeBeanReference(adapterBeanName); + } } - private String getReleasenStrategyRef(Element element, ParserContext parserContext) { + private BeanMetadataElement getReleaseStrategy(Element element, ParserContext parserContext) { String ref = element.getAttribute(RELEASE_STRATEGY_REF_ATTRIBUTE); String method = element.getAttribute(RELEASE_STRATEGY_METHOD_ATTRIBUTE); + String expression = element.getAttribute(RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE); if (StringUtils.hasText(ref)) { + if (StringUtils.hasText(expression)) { + parserContext.getReaderContext().error( + "Only one of release strategy expression and bean reference must be specified", element); + return null; + } + 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 null; + } if (StringUtils.hasText(method)) { BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE @@ -139,12 +152,30 @@ public class ResequencerParser extends AbstractConsumerEndpointParser { "java.lang.String"); String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder .getBeanDefinition(), parserContext.getRegistry()); - return adapterBeanName; - } else { - return ref; + return new RuntimeBeanReference(adapterBeanName); + } + else { + return new RuntimeBeanReference(ref); } } - return null; + else { + if (!StringUtils.hasText(expression)) { + return null; + } + if (StringUtils.hasText(element.getAttribute(RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE))) { + parserContext.getReaderContext().error( + "Only one of " + RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE + " and " + + RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE + " can be specified at once", element); + return null; + } + BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder + .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + + ".aggregator.ExpressionEvaluatingReleaseStrategy"); + adapterBuilder.addConstructorArgValue(expression); + String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder + .getBeanDefinition(), parserContext.getRegistry()); + return new RuntimeBeanReference(adapterBeanName); + } } } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index 273a04ad1f..1c02036848 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -23,11 +23,11 @@ - Default output channel for the @Publisher annotation support. + Default output channel for the @Publisher annotation support. - + @@ -140,12 +140,15 @@ Reference to a MessageGroupStore that can be used to buffer the messages. If a message store is specified then it will store messages for this channel with a correlation key equal to the - channel name. If you need more control over the correlation key (e.g. two channels in the same - application share a name), then you need to look to the queue implementation itself and provide an - explicit instance via the "ref" attribute, or else maybe the + channel name. If you need + more control over the correlation key (e.g. two channels in the same + application share a name), then you need to + look to the queue implementation itself and provide an + explicit instance via the "ref" attribute, or else maybe the message store has a way to specify a region or similar additional - tag for messages. This attribute is - mutually exclusive with the "ref" attribute (only one can be specified). + tag for messages. This attribute is + mutually + exclusive with the "ref" attribute (only one can be specified). @@ -158,7 +161,8 @@ Reference to a BlockingQueue that can be used to buffer the messages. This attribute is - mutually exclusive with the "message-store" attribute (only one can be specified). + mutually + exclusive with the "message-store" attribute (only one can be specified). @@ -523,7 +527,7 @@ - + - + @@ -868,81 +872,81 @@ - + - + - - - - - - - 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. - - - - - - - - - - - Provide a reference to the MessageStore instance that should be used - to store Messages while - awaiting the delay. + + + + + + + + + + + Provide a reference to the MessageStore instance that should be used + to store Messages while + awaiting the delay. - - - - - - - - - - - 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'. - - - + + + @@ -957,7 +961,7 @@ - + @@ -971,38 +975,38 @@ - + - + - - - - - - - - - - - - - - + + + + + + + + + + + + + + - - - - - - - - + + + + + + + + @@ -1037,12 +1041,12 @@ - + - + @@ -1067,30 +1071,30 @@ - - - Fixed delay trigger (in milliseconds). - - - - - Fixed rate trigger (in milliseconds). - - - - - Cron trigger. - - - - + + + Fixed delay trigger (in milliseconds). + + + + + Fixed rate trigger (in milliseconds). + + + + + Cron trigger. + + + + - + @@ -1218,104 +1222,104 @@ - + - + - - - - - - - - - - - - - - - - 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. - - - - - - - - - - - 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 whether null values, such as might be returned from an expression evaluation, - should be - skipped. The default value is true. Set this to false if a null value should - trigger removal of the corresponding - header instead. + + + + + + + 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. - - - - - - - - - Reference to an Object to be invoked for header values. - The 'method' attribute is required - along with this. + + + + + + + + + Specify whether null values, such as might be returned from an expression evaluation, + should be + skipped. The default value is true. Set this to false if a null value should + trigger removal of the corresponding + header instead. - - - - - - - - - Method to be invoked on the referenced Object as specified by the - 'ref' attribute. The method - should return a Map with String-typed keys. + + + + + + + + + Reference to an Object to be invoked for header values. + The 'method' attribute is required + along with this. - - - - - + + + + + + + + + Method to be invoked on the referenced Object as specified by the + 'ref' attribute. The method + should return a Map with String-typed keys. + + + + + + @@ -1434,19 +1438,20 @@ - + - + Specify one or more header names (as a comma separated list) to be removed from the - MessageHeaders of the Message being handled. + MessageHeaders + of the Message being handled. @@ -1461,7 +1466,7 @@ - + @@ -1478,7 +1483,7 @@ - + @@ -1492,7 +1497,7 @@ - + @@ -1506,12 +1511,12 @@ - + - + @@ -1577,7 +1582,7 @@ - + @@ -1591,7 +1596,7 @@ - + @@ -1607,12 +1612,12 @@ - + - + @@ -1629,12 +1634,12 @@ - + - + @@ -1664,12 +1669,12 @@ - + - + @@ -1693,42 +1698,42 @@ Name of the header whose value to use. - + - + - - - - - - - - - An expression to be evaluated to determine if this recipient should be included in the recipient - list for a given input Message. The evaluation result of the expression must be a boolean. - If this attribute is not defined, the channel will always be among the list of recipients. + + + + + + + + + An expression to be evaluated to determine if this recipient should be included in the recipient + list for a given input Message. The evaluation result of the expression must be a boolean. + If this attribute is not defined, the channel will always be among the list of recipients. - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + @@ -1740,12 +1745,12 @@ Name of the header whose value to use. - + - + @@ -1779,55 +1784,55 @@ Name of the header whose value to use. - + - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + - - - - - - - - - - + + + + + + + + + + @@ -1933,7 +1938,7 @@ Name of the header whose value to use. - + @@ -1948,105 +1953,106 @@ Name of the header whose value to use. - + - + - - - + + + + + + A SpEL expression to be evaluated against the input message list as its root object. + + + + + - - - - - - A SpEL expression to be evaluated against the input message list as its root object. + + + + + + + + + + + + + + + + + + + + + + + An expression to apply to the message group + + + + + + + + + + + + + + + + + + + + + + + An expression to apply to the message group + + + + + + + + + + + + + + + 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. - - - - - - - - - - - - - - - - - - - - - - - An expression to apply to the message group - - - - - - - - - - - - - - - - - - - - - - - An expression to apply to the message group - - - - - - - - - - - - - - - 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. - - - - - - - - - - - + + + + + + + + + + @@ -2058,38 +2064,42 @@ Name of the header whose value to use. - + - + - - - - - - - - - - - - - - - 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). + + + + + + Comparator for messages used to sort the sequence when released. Defaults to comparing + the sequence number header. + + + + + + + + + + + + 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). - - - - + + + + @@ -2202,7 +2212,7 @@ Name of the header whose value to use. - + @@ -2231,17 +2241,20 @@ Name of the header whose value to use. Defines a component that evaluates an expression to generate a Message payload - (as well as optional expression evaluation for headers). The resulting Message - is then sent to a MessageChannel. Each execution is driven by a Trigger. + (as well as optional + expression evaluation for headers). The resulting Message + is then sent to a MessageChannel. Each execution is driven + by a Trigger. Exactly one of the trigger type attributes must be provided. The options are: - fixed-delay, fixed-rate, cron, or trigger (reference). + fixed-delay, fixed-rate, + cron, or trigger (reference). - + - + Fixed delay trigger (in milliseconds). @@ -2264,7 +2277,7 @@ Name of the header whose value to use. - + @@ -2273,7 +2286,8 @@ Name of the header whose value to use. SpEL expression to be evaluated for each triggered execution. - The result of the evaluation will be passed as the payload of + The result of the evaluation will + be passed as the payload of the Message that is sent to the MessageChannel. @@ -2285,7 +2299,7 @@ Name of the header whose value to use. - + @@ -2294,7 +2308,8 @@ Name of the header whose value to use. Specify whether this producer should start automatically. - By default it will. Set this to 'false' to require a manual start. + By default it will. Set this to 'false' + to require a manual start. @@ -2306,7 +2321,8 @@ Name of the header whose value to use. Defines a MessagePublishingInterceptor which allows you to generate messages - as a by-product of method invocations on Spring configured components. + as a by-product of + method invocations on Spring configured components. @@ -2316,9 +2332,9 @@ Name of the header whose value to use. - - - + + + @@ -2397,7 +2413,7 @@ Name of the header whose value to use. - + @@ -2407,7 +2423,7 @@ Name of the header whose value to use. - + @@ -2419,7 +2435,7 @@ Name of the header whose value to use. - + @@ -2430,7 +2446,7 @@ only be one Message History writer per ApplicationContext hierarchy. - + @@ -2446,7 +2462,8 @@ only be one Message History writer per ApplicationContext hierarchy. 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. + Message to the + output channel. By default the send will block for one second. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorExpressionIntegrationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorExpressionIntegrationTests-context.xml new file mode 100644 index 0000000000..0b5bfb0d5c --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorExpressionIntegrationTests-context.xml @@ -0,0 +1,19 @@ + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorExpressionIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorExpressionIntegrationTests.java new file mode 100644 index 0000000000..e43554b0ef --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorExpressionIntegrationTests.java @@ -0,0 +1,70 @@ +/* + * Copyright 2002-2008 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.integration; + +import static org.junit.Assert.assertEquals; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.MessageHeaders; +import org.springframework.integration.core.GenericMessage; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Iwein Fuld + * @author Alex Peters + * @author Oleg Zhurakousky + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class AggregatorExpressionIntegrationTests { + + @Autowired + @Qualifier("input") + private MessageChannel input; + + @Autowired + @Qualifier("output") + private PollableChannel output; + + @Test//(timeout=5000) + public void testVanillaAggregation() throws Exception { + for (int i = 0; i < 5; i++) { + Map headers = stubHeaders(i, 5, 1); + input.send(new GenericMessage(i, headers)); + } + assertEquals("[0, 1, 2, 3, 4]", output.receive().getPayload()); + } + + private Map stubHeaders(int sequenceNumber, int sequenceSize, int correllationId) { + Map headers = new HashMap(); + headers.put(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber); + headers.put(MessageHeaders.SEQUENCE_SIZE, sequenceSize); + headers.put(MessageHeaders.CORRELATION_ID, correllationId); + return headers; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerExpressionIntegrationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerExpressionIntegrationTests-context.xml new file mode 100644 index 0000000000..c65634174a --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerExpressionIntegrationTests-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/integration/ResequencerExpressionIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerExpressionIntegrationTests.java new file mode 100644 index 0000000000..605fc8d71a --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerExpressionIntegrationTests.java @@ -0,0 +1,78 @@ +/* + * Copyright 2002-2008 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.integration; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.Message; +import org.springframework.integration.MessageHeaders; +import org.springframework.integration.core.GenericMessage; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Iwein Fuld + * @author Alex Peters + * @author Oleg Zhurakousky + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class ResequencerExpressionIntegrationTests { + + @Autowired + @Qualifier("input") + private MessageChannel input; + + @Autowired + @Qualifier("output") + private PollableChannel output; + + @Test//(timeout=5000) + public void testVanillaAggregation() throws Exception { + List> messages = new ArrayList>(); + for (int i = 0; i < 5; i++) { + Map headers = stubHeaders(i, 5, 1); + messages.add(new GenericMessage(i, headers)); + } + input.send(messages.get(2)); + input.send(messages.get(1)); + input.send(messages.get(0)); + assertEquals(0, output.receive().getPayload()); + assertEquals(1, output.receive().getPayload()); + assertEquals(2, output.receive().getPayload()); + } + + private Map stubHeaders(int sequenceNumber, int sequenceSize, int correllationId) { + Map headers = new HashMap(); + headers.put(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber); + headers.put(MessageHeaders.SEQUENCE_SIZE, sequenceSize); + headers.put("foo", correllationId); + return headers; + } + +}