diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 9026594b94..63eb5936b9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * 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. @@ -16,13 +16,20 @@ package org.springframework.integration.aggregator; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.context.Lifecycle; import org.springframework.integration.channel.ChannelResolver; +import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.scheduling.TaskScheduler; @@ -47,18 +54,24 @@ import java.util.concurrent.locks.ReentrantLock; * * @author Iwein Fuld */ -public class CorrelatingMessageHandler extends AbstractMessageHandler implements Lifecycle { +public class CorrelatingMessageHandler extends AbstractMessageHandler implements Lifecycle, BeanFactoryAware { + private static final Log logger = LogFactory.getLog(CorrelatingMessageHandler.class); - // TODO: need to support 'resequencer' as well - public static final String COMPONENT_TYPE_LABEL = "aggregator"; + public static final String COMPONENT_TYPE_LABEL = "aggregator"; + + private static final long DEFAULT_SEND_TIMEOUT = 1000l; + private static final long DEFAULT_REAPER_INTERVAL = 1000l; + private static final long DEFAULT_TIMEOUT = 60000l; + private final MessageStore store; + private final MessageGroupProcessor outputProcessor; - private MessageStore store = new SimpleMessageStore(100); - private final CorrelationStrategy correlationStrategy; - private final CompletionStrategy completionStrategy; - private MessageGroupProcessor outputProcessor; + private volatile CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID); + private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); private MessageChannel outputChannel; + private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private volatile MessageChannel discardChannel = new NullChannel(); private ChannelResolver channelResolver; @@ -67,8 +80,8 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements private volatile TaskScheduler taskScheduler; private volatile ScheduledFuture reaperFutureTask; - private volatile long reaperInterval = 1000l; - private volatile long timeout = 60000l; + private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL; + private volatile long timeout = DEFAULT_TIMEOUT; private volatile boolean sendPartialResultOnTimeout; private Object lifecycleMonitor = new Object(); @@ -78,13 +91,14 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements CompletionStrategy completionStrategy, MessageGroupProcessor processor) { Assert.notNull(store); - Assert.notNull(correlationStrategy); - Assert.notNull(completionStrategy); Assert.notNull(processor); + Assert.notNull(correlationStrategy); + Assert.notNull(completionStrategy); this.store = store; + this.outputProcessor = processor; this.correlationStrategy = correlationStrategy; this.completionStrategy = completionStrategy; - this.outputProcessor = processor; + this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT); } public CorrelatingMessageHandler(MessageStore store, @@ -94,6 +108,23 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements new SequenceSizeCompletionStrategy(), processor); } + public CorrelatingMessageHandler( + MessageGroupProcessor processor) { + this(new SimpleMessageStore(100), + new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID), + new SequenceSizeCompletionStrategy(), processor); + } + + public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) { + Assert.notNull(correlationStrategy); + this.correlationStrategy = correlationStrategy; + } + + public void setCompletionStrategy(CompletionStrategy completionStrategy) { + Assert.notNull(completionStrategy); + this.completionStrategy = completionStrategy; + } + public void setTaskScheduler(TaskScheduler taskScheduler) { this.taskScheduler = taskScheduler; } @@ -107,6 +138,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } public void setOutputChannel(MessageChannel outputChannel) { + Assert.notNull(outputChannel, "'outputChannel' must not be null"); this.outputChannel = outputChannel; } @@ -118,13 +150,26 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements this.discardChannel = discardChannel; } + public void setSendTimeout(long sendTimeout) { + this.channelTemplate.setSendTimeout(sendTimeout); + } + public void setSendPartialResultOnTimeout(boolean sendPartialResultOnTimeout) { this.sendPartialResultOnTimeout = sendPartialResultOnTimeout; } + public void setBeanFactory(BeanFactory beanFactory) { + if (this.taskScheduler == null) { + this.taskScheduler = IntegrationContextUtils.getRequiredTaskScheduler(beanFactory); + } + } + @Override protected void handleMessageInternal(Message message) throws Exception { Object correlationKey = correlationStrategy.getCorrelationKey(message); + if (logger.isDebugEnabled()) { + logger.debug("Handling message with correllationKey [" + correlationKey + "]: " + message); + } try { if (tracker.waitForLockIfNotTracked(correlationKey)) { MessageGroup group = @@ -135,8 +180,11 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements group.add(message); if (group.isComplete()) { + if (logger.isDebugEnabled()) { + logger.debug("Completing group with correllationKey [" + correlationKey + "]"); + } outputProcessor.processAndSend(group, - this.resolveReplyChannel(message, this.outputChannel, this.channelResolver)); + channelTemplate, this.resolveReplyChannel(message, this.outputChannel, this.channelResolver)); } } else { discardChannel.send(message); @@ -164,7 +212,12 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } private void store(Message message, Object correlationKey) { - store.put(message); + Message toStore = message; + if (!correlationKey.equals(message.getHeaders().getCorrelationId())) { + toStore = MessageBuilder.fromMessage(message) + .setCorrelationId(correlationKey).build(); + } + store.put(toStore); if (!keysInBuffer.contains(correlationKey)) { keysInBuffer.add(new DelayedKey(correlationKey, timeout)); } @@ -228,7 +281,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageChannel outputChannel = resolveReplyChannel(all.get(0), this.outputChannel, this.channelResolver); boolean processed = false; if (group.isComplete()) { - outputProcessor.processAndSend(group, outputChannel); + outputProcessor.processAndSend(group, channelTemplate, outputChannel); processed = true; } if (!processed) { @@ -236,7 +289,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements if (logger.isInfoEnabled()) { logger.info("Processing partially complete messages for key [" + key + "] to: " + outputChannel); } - outputProcessor.processAndSend(group, outputChannel); + outputProcessor.processAndSend(group, channelTemplate, outputChannel); } else { if (logger.isInfoEnabled()) { logger.info("Discarding partially complete messages for key [" + key + "] to: " + discardChannel); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java new file mode 100644 index 0000000000..040c02031d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java @@ -0,0 +1,57 @@ +/* + * 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 org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.util.Assert; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +/** + * This implementation of MessageGroupProcessor will take the messages from the MessageGroup and pass them on in a + * single message with a Collection as a payload. + * + * @author Iwein Fuld + * @author Alexander Peters + * @since 2.0.0 + */ +public class DefaultAggregatingMessageGroupProcessor implements MessageGroupProcessor { + public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) { + Assert.notNull(group, "Message group must not be null."); + Assert.notNull(outputChannel, "'outputChannel' must not be null."); + + List> messages = group.getMessages(); + + Assert.notEmpty(messages, this.getClass().getSimpleName() + " cannot process empty message groups"); + + channelTemplate.send(aggregateMessages(messages), outputChannel); + + } + + private Message aggregateMessages(List> messages) { + List payloads = new ArrayList(messages.size()); + for (Message message : messages) { + payloads.add(message.getPayload()); + } + return MessageBuilder.withPayload(payloads).build(); + } +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java index d43182e2f3..9e17d8cd09 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java @@ -15,6 +15,7 @@ */ package org.springframework.integration.aggregator; +import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; @@ -25,6 +26,8 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicInteger; /** + * This class implements all the strategy interfaces needed for a default resequencer. + * * @author Iwein Fuld */ public class DefaultResequencerStrategies implements CorrelationStrategy, CompletionStrategy, MessageGroupProcessor { @@ -39,12 +42,12 @@ public class DefaultResequencerStrategies implements CorrelationStrategy, Comple } public boolean isComplete(List> messages) { - return releasePartialSequences|| - messages.get(0).getHeaders().getSequenceSize()==messages.size(); + return releasePartialSequences || + messages.get(0).getHeaders().getSequenceSize() == messages.size(); } - public void processAndSend(MessageGroup group, MessageChannel outputChannel) { - List> all = group.getMessages(); + public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) { + List> all = group.getMessages(); Object correlationKey = group.getCorrelationKey(); if (all.size() > 0) { List sorted = new ArrayList(all); @@ -53,13 +56,13 @@ public class DefaultResequencerStrategies implements CorrelationStrategy, Comple for (Message message : sorted) { final int sequenceNumber = message.getHeaders().getSequenceNumber(); if (sequenceNumber <= nextSequence.get()) { - outputChannel.send(message); + channelTemplate.send(message, outputChannel); nextSequence.compareAndSet(sequenceNumber, sequenceNumber + 1); group.onProcessingOf(message); } } MessageHeaders headers = sorted.get(0).getHeaders(); - if (all.size() == headers.getSequenceSize()){ + if (all.size() == headers.getSequenceSize()) { group.onCompletion(); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java index b3607fe02a..9269454e64 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java @@ -35,8 +35,13 @@ public class MessageGroup { } public boolean hasNoMessageSuperseding(Message message) { + Integer messageSequenceNumber = message.getHeaders().getSequenceNumber(); + if (messageSequenceNumber == null) { + return true; + } for (Message member : messages) { - if (member.getHeaders().getSequenceNumber() == message.getHeaders().getSequenceNumber()) { + Integer memberSequenceNumber = member.getHeaders().getSequenceNumber(); + if (memberSequenceNumber == messageSequenceNumber) { return false; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java index e1403b28d3..0279b3784a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java @@ -1,13 +1,24 @@ package org.springframework.integration.aggregator; +import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.MessageChannel; /** + * A processor for correlated groups of messages. When a message group is complete it is passed to the + * processor by e.g. the CorrelatingMessageHandler. + * * @author Iwein Fuld + * @see org.springframework.integration.aggregator.CorrelatingMessageHandler */ public interface MessageGroupProcessor { + /** + * Processed the given group and sends the resulting message(s) to the output channel using the channelTemplate. + * Implementations are free to send as little or as many messages based on the invocation as needed. For example the + * DefaultAggregatingMessageGroupProcessor will send only a single message containing a collection of all messages + * in the group, where the resequencing equivalent strategy will send all messages in the group individually. + */ void processAndSend(MessageGroup group, - MessageChannel outputChannel + MessageChannelTemplate channelTemplate, MessageChannel outputChannel ); } \ No newline at end of file diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java index a23e788818..656526b59b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java @@ -63,7 +63,7 @@ public class MessageListMethodAdapter { } - protected Method getMethod() { + public Method getMethod() { return method; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java index 2da9bd3f49..85d71835d3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java @@ -2,6 +2,7 @@ package org.springframework.integration.aggregator; import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.Header; +import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; @@ -13,14 +14,14 @@ import java.util.*; public class MethodInvokingMessageGroupProcessor implements MessageGroupProcessor { - private final Object target; - private final Method method; private final MessageListMethodAdapter adapter; public MethodInvokingMessageGroupProcessor(Object target) { - this.target = target; - this.method = selectMethodFrom(target); + this.adapter = new MessageListMethodAdapter(target, selectMethodFrom(target)); + } + + public MethodInvokingMessageGroupProcessor (Object target, String method){ this.adapter = new MessageListMethodAdapter(target, method); } @@ -116,7 +117,7 @@ public class MethodInvokingMessageGroupProcessor implements MessageGroupProcesso } public void processAndSend(MessageGroup group, - MessageChannel outputChannel + MessageChannelTemplate channelTemplate, MessageChannel outputChannel ) { final Collection> messagesUpForProcessing = group.getMessages(); Message reply = MessageBuilder.withPayload( @@ -124,10 +125,10 @@ public class MethodInvokingMessageGroupProcessor implements MessageGroupProcesso group.onCompletion(); group.onProcessingOf(messagesUpForProcessing .toArray(new Message[]{})); - outputChannel.send(reply); + channelTemplate.send(reply, outputChannel); } - public Set removeMethodsMatchingSelector(Set candidates, MethodSelector selector) { + private Set removeMethodsMatchingSelector(Set candidates, MethodSelector selector) { Set removed = new HashSet(); Iterator iterator = candidates.iterator(); while (iterator.hasNext()) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java index 564ace3168..17680000cd 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java @@ -1,13 +1,22 @@ package org.springframework.integration.aggregator; +import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; +/** + * This implementation of MessageGroupProcessor will forward all messages inside the group to the given output channel. + * This is useful if there is no requirement to process the messages, but they should just be blocked as a group until + * their CompletionStrategy lets them pass through. + * + * @author Iwein Fuld + * @since 2.0.0 + */ public class PassThroughMessageGroupProcessor implements MessageGroupProcessor { - public void processAndSend(MessageGroup group, MessageChannel outputChannel) { + public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) { for (Message message : group.getMessages()) { - outputChannel.send(message); + channelTemplate.send(message, outputChannel); group.onProcessingOf(message); } group.onCompletion(); 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 1e9cc98794..95d0b5c277 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 @@ -25,112 +25,119 @@ import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.StringUtils; /** - * Parser for the aggregator element of the integration namespace. - * Registers the annotation-driven post-processors. - * + * Parser for the aggregator element of the integration namespace. Registers the annotation-driven + * post-processors. + * * @author Marius Bogoevici * @author Mark Fisher - * @author Oleg Zhurakousky + * @author Oleg Zhurakousky */ public class AggregatorParser extends AbstractConsumerEndpointParser { - private static final String COMPLETION_STRATEGY_REF_ATTRIBUTE = "completion-strategy"; + private static final String COMPLETION_STRATEGY_REF_ATTRIBUTE = "completion-strategy"; - private static final String COMPLETION_STRATEGY_METHOD_ATTRIBUTE = "completion-strategy-method"; + private static final String COMPLETION_STRATEGY_METHOD_ATTRIBUTE = "completion-strategy-method"; - private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy"; + 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_METHOD_ATTRIBUTE = "correlation-strategy-method"; - private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel"; + private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel"; - private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout"; + private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel"; - private static final String SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE = "send-partial-result-on-timeout"; + private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout"; - private static final String REAPER_INTERVAL_ATTRIBUTE = "reaper-interval"; + private static final String SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE = "send-partial-result-on-timeout"; - private static final String TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE = "tracked-correlation-id-capacity"; + private static final String REAPER_INTERVAL_ATTRIBUTE = "reaper-interval"; - private static final String TIMEOUT_ATTRIBUTE = "timeout"; + private static final String TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE = "tracked-correlation-id-capacity"; - private static final String COMPLETION_STRATEGY_PROPERTY = "completionStrategy"; + private static final String TIMEOUT_ATTRIBUTE = "timeout"; - private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy"; + private static final String COMPLETION_STRATEGY_PROPERTY = "completionStrategy"; + + private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy"; - @Override - protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { - BeanDefinition innerHandlerDefinition = IntegrationNamespaceUtils.parseInnerHandlerDefinition(element, parserContext); - String ref = element.getAttribute(REF_ATTRIBUTE); - BeanDefinitionBuilder builder; - - if (innerHandlerDefinition != null || StringUtils.hasText(ref)){ - builder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.MethodInvokingAggregator"); - } else { - builder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultMessageAggregator"); - } - - if (innerHandlerDefinition != null){ - builder.addConstructorArgValue(innerHandlerDefinition); - } else { - if (StringUtils.hasText(ref)) { - builder.addConstructorArgReference(ref); - } - } - if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) { - String method = element.getAttribute(METHOD_ATTRIBUTE); - builder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String"); - } - - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, - DISCARD_CHANNEL_ATTRIBUTE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - SEND_TIMEOUT_ATTRIBUTE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - REAPER_INTERVAL_ATTRIBUTE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE); - this.injectPropertyWithBean(COMPLETION_STRATEGY_REF_ATTRIBUTE, - COMPLETION_STRATEGY_METHOD_ATTRIBUTE, COMPLETION_STRATEGY_PROPERTY, - "CompletionStrategyAdapter", element, builder, parserContext); - this.injectPropertyWithBean(CORRELATION_STRATEGY_REF_ATTRIBUTE, - CORRELATION_STRATEGY_METHOD_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY, - "CorrelationStrategyAdapter", element, builder, parserContext); - return builder; - } + @Override + protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { + BeanDefinition innerHandlerDefinition = IntegrationNamespaceUtils.parseInnerHandlerDefinition(element, parserContext); + String ref = element.getAttribute(REF_ATTRIBUTE); + BeanDefinitionBuilder builder; - private void injectPropertyWithBean(String beanRefAttribute, String methodRefAttribute, - String beanProperty, String adapterClass, Element element, - BeanDefinitionBuilder builder, ParserContext parserContext) { - final String beanRef = element.getAttribute(beanRefAttribute); - final String beanMethod = element.getAttribute(methodRefAttribute); - if (StringUtils.hasText(beanRef)) { - if (StringUtils.hasText(beanMethod)) { - String adapterBeanName = this.createAdapter(beanRef, beanMethod, adapterClass, - parserContext); - builder.addPropertyReference(beanProperty, adapterBeanName); - } - else { - builder.addPropertyReference(beanProperty, beanRef); - } - } - } + builder = BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelatingMessageHandler"); + BeanDefinitionBuilder processorBuilder = null; - private String createAdapter(String ref, String method, String unqualifiedClassName, - ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator." + unqualifiedClassName); - builder.addConstructorArgReference(ref); - builder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String"); - return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), - parserContext.getRegistry()); - } + if (innerHandlerDefinition != null || StringUtils.hasText(ref)) { + processorBuilder = BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.MethodInvokingMessageGroupProcessor"); + builder.addConstructorArgValue(processorBuilder.getBeanDefinition()); + } else { + builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultAggregatingMessageGroupProcessor").getBeanDefinition()); + } + + if (innerHandlerDefinition != null) { + processorBuilder.addConstructorArgValue(innerHandlerDefinition); + } else { + if (StringUtils.hasText(ref)) { + processorBuilder.addConstructorArgReference(ref); + } + } + if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) { + String method = element.getAttribute(METHOD_ATTRIBUTE); + processorBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String"); + } + + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + DISCARD_CHANNEL_ATTRIBUTE); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + OUTPUT_CHANNEL_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + SEND_TIMEOUT_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + REAPER_INTERVAL_ATTRIBUTE); +// IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE); + this.injectPropertyWithBean(COMPLETION_STRATEGY_REF_ATTRIBUTE, + COMPLETION_STRATEGY_METHOD_ATTRIBUTE, COMPLETION_STRATEGY_PROPERTY, + "CompletionStrategyAdapter", element, builder, parserContext); + this.injectPropertyWithBean(CORRELATION_STRATEGY_REF_ATTRIBUTE, + CORRELATION_STRATEGY_METHOD_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY, + "CorrelationStrategyAdapter", element, builder, parserContext); + return builder; + } + + private void injectPropertyWithBean(String beanRefAttribute, String methodRefAttribute, + String beanProperty, String adapterClass, Element element, + BeanDefinitionBuilder builder, ParserContext parserContext) { + final String beanRef = element.getAttribute(beanRefAttribute); + final String beanMethod = element.getAttribute(methodRefAttribute); + if (StringUtils.hasText(beanRef)) { + if (StringUtils.hasText(beanMethod)) { + String adapterBeanName = this.createAdapter(beanRef, beanMethod, adapterClass, + parserContext); + builder.addPropertyReference(beanProperty, adapterBeanName); + } else { + builder.addPropertyReference(beanProperty, beanRef); + } + } + } + + private String createAdapter(String ref, String method, String unqualifiedClassName, + ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator." + unqualifiedClassName); + builder.addConstructorArgReference(ref); + builder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String"); + return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), + parserContext.getRegistry()); + } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index deccdcfd09..c804fbbea8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -16,15 +16,15 @@ package org.springframework.integration.store; +import org.springframework.integration.core.Message; +import org.springframework.util.Assert; + import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import org.springframework.integration.core.Message; -import org.springframework.util.Assert; - /** * Map-based implementation of {@link MessageStore} that enforces a maximum capacity. * @@ -34,46 +34,47 @@ import org.springframework.util.Assert; */ public class SimpleMessageStore implements MessageStore { - private final Map> map; + private final Map> map; - public SimpleMessageStore(int capacity) { - this.map = new ConcurrentHashMap>(capacity); - } + public SimpleMessageStore(int capacity) { + this.map = new ConcurrentHashMap>(capacity); + } - @SuppressWarnings("unchecked") - public Message put(Message message) { - return (Message) this.map.put(message.getHeaders().getId(), message); - } + @SuppressWarnings("unchecked") + public Message put(Message message) { + return (Message) this.map.put(message.getHeaders().getId(), message); + } - public Message get(Object key) { - return (key != null) ? this.map.get(key) : null; - } + public Message get(Object key) { + return (key != null) ? this.map.get(key) : null; + } - public List> list() { - return new ArrayList>(this.map.values()); - } + public List> list() { + return new ArrayList>(this.map.values()); + } - public Message delete(Object key) { - return (key != null) ? this.map.remove(key) : null; - } + public Message delete(Object key) { + return (key != null) ? this.map.remove(key) : null; + } - public int size() { - return this.map.size(); - } + public int size() { + return this.map.size(); + } - public List> list(Object correlationKey) { + public List> list(Object correlationKey) { Assert.notNull(correlationKey, "'correlationKey' must not be null"); - List> matched = new ArrayList>(); - Collection> values = map.values(); - for (Message message : values) { - if(message.getHeaders().getCorrelationId().equals(correlationKey)){ - matched.add(message); - } - } - return matched; - } + List> matched = new ArrayList>(); + Collection> values = map.values(); + for (Message message : values) { + Object correlationId = message.getHeaders().getCorrelationId(); + if (correlationId != null && correlationId.equals(correlationKey)) { + matched.add(message); + } + } + return matched; + } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java index 43ff47dabe..cf8476f400 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java @@ -38,6 +38,7 @@ public class CorrelatingMessageHandlerIntegrationTest { @Before public void setupHandler(){ defaultHandler.setOutputChannel(outputChannel); + defaultHandler.setSendTimeout(-1); } private Message correlatedMessage(Object correlationId, diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index b30014798d..063d9c7eab 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -23,6 +23,7 @@ import org.mockito.Mock; import org.mockito.invocation.InvocationOnMock; import org.mockito.runners.MockitoJUnitRunner; import org.mockito.stubbing.Answer; +import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; @@ -76,14 +77,14 @@ public class CorrelatingMessageHandlerTests { return null; } }).when(processor).processAndSend(isA(MessageGroup.class), - eq(outputChannel)); + isA(MessageChannelTemplate.class), eq(outputChannel)); } @Test public void bufferCompletesNormally() throws Exception { String correlationKey = "key"; - Message message1 = testMessage(1, 1); - Message message2 = testMessage(2, 2); + Message message1 = testMessage(correlationKey, 1, 1); + Message message2 = testMessage(correlationKey, 2, 2); List> storedMessages = new ArrayList>(); when(store.list(correlationKey)).thenReturn(storedMessages); @@ -108,7 +109,7 @@ public class CorrelatingMessageHandlerTests { verify(completionStrategy).isComplete(Arrays.asList(message1)); verify(completionStrategy).isComplete(Arrays.asList(message1, message2)); verify(processor).processAndSend(isA(MessageGroup.class), - eq(outputChannel) + isA(MessageChannelTemplate.class), eq(outputChannel) ); } @@ -120,8 +121,8 @@ public class CorrelatingMessageHandlerTests { @Test public void shouldNotPruneWhileCompleting() throws Exception { String correlationKey = "key"; - final Message message1 = testMessage(1, 1); - final Message message2 = testMessage(2, 2); + final Message message1 = testMessage(correlationKey, 1, 1); + final Message message2 = testMessage(correlationKey, 2, 2); final List> storedMessages = new ArrayList>(); final CountDownLatch bothMessagesHandled = new CountDownLatch(2); @@ -159,9 +160,10 @@ public class CorrelatingMessageHandlerTests { verify(store).delete(2); } - private Message testMessage(int id, int sequenceNumber) { + private Message testMessage(String correllationKey, int id, int sequenceNumber) { return MessageBuilder.withPayload("test" + id) .setHeader(MessageHeaders.ID, id) + .setCorrelationId(correllationKey) .setSequenceNumber(sequenceNumber).build(); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java index 48f0065eb4..263b47e907 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java @@ -8,6 +8,7 @@ import org.mockito.Mock; import org.mockito.runners.MockitoJUnitRunner; import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.Header; +import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; @@ -18,6 +19,7 @@ import java.util.List; import static org.hamcrest.CoreMatchers.is; import static org.junit.Assert.assertThat; import static org.junit.Assert.fail; +import static org.mockito.Matchers.eq; import static org.mockito.Matchers.isA; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -36,6 +38,9 @@ public class MethodInvokingMessageGroupProcessorTests { @Mock private MessageGroup messageGroupMock; + @Mock + private MessageChannelTemplate channelTemplate; + @Before public void initializeMessagesUpForProcessing() { messagesUpForProcessing.add(MessageBuilder.withPayload(1).build()); @@ -68,10 +73,9 @@ public class MethodInvokingMessageGroupProcessorTests { .forClass(Message.class); when(outputChannel.send(isA(Message.class))).thenReturn(true); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); - processor.processAndSend(messageGroupMock, outputChannel - ); + processor.processAndSend(messageGroupMock, channelTemplate, outputChannel); // verify - verify(outputChannel).send(messageCaptor.capture()); + verify(channelTemplate).send(messageCaptor.capture(), eq(outputChannel)); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); } @@ -94,10 +98,9 @@ public class MethodInvokingMessageGroupProcessorTests { .forClass(Message.class); when(outputChannel.send(isA(Message.class))).thenReturn(true); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); - processor.processAndSend(messageGroupMock, outputChannel - ); + processor.processAndSend(messageGroupMock, channelTemplate, outputChannel); // verify - verify(outputChannel).send(messageCaptor.capture()); + verify(channelTemplate).send(messageCaptor.capture(), eq(outputChannel)); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); } @@ -132,10 +135,10 @@ public class MethodInvokingMessageGroupProcessorTests { when(outputChannel.send(isA(Message.class))).thenReturn(true); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); - processor.processAndSend(messageGroupMock, outputChannel + processor.processAndSend(messageGroupMock, channelTemplate, outputChannel ); // verify - verify(outputChannel).send(messageCaptor.capture()); + verify(channelTemplate).send(messageCaptor.capture(), eq(outputChannel)); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); } @@ -165,10 +168,10 @@ public class MethodInvokingMessageGroupProcessorTests { when(outputChannel.send(isA(Message.class))).thenReturn(true); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); - processor.processAndSend(messageGroupMock, outputChannel + processor.processAndSend(messageGroupMock, channelTemplate, outputChannel ); // verify - verify(outputChannel).send(messageCaptor.capture()); + verify(channelTemplate).send(messageCaptor.capture(), eq(outputChannel)); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java index 5d2d84f320..01d1eececd 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java @@ -19,6 +19,7 @@ package org.springframework.integration.aggregator; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; +import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -251,14 +252,14 @@ public class NewAggregatorEndpointTests { private class MultiplyingProcessor implements MessageGroupProcessor { public void processAndSend(MessageGroup group, - MessageChannel outputChannel + MessageChannelTemplate channelTemplate, MessageChannel outputChannel ) { Integer product = 1; List> messagesUpForProcessing = group.getMessages(); for (Message message : messagesUpForProcessing) { product *= (Integer) message.getPayload(); } - outputChannel.send(MessageBuilder.withPayload(product).build()); + channelTemplate.send(MessageBuilder.withPayload(product).build(), outputChannel); group.onProcessingOf( messagesUpForProcessing.toArray(new Message[messagesUpForProcessing.size()]) @@ -268,7 +269,7 @@ public class NewAggregatorEndpointTests { } private class NullReturningMessageProcessor implements MessageGroupProcessor { - public void processAndSend(MessageGroup group, MessageChannel outputChannel) { + public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) { //noop } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java index ae7958d931..069094a8a4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java @@ -22,6 +22,7 @@ import org.junit.Ignore; import org.junit.Test; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -320,14 +321,14 @@ public class NewConcurrentAggregatorEndpointTests { private class MultiplyingProcessor implements MessageGroupProcessor { public void processAndSend(MessageGroup group, - MessageChannel outputChannel + MessageChannelTemplate channelTemplate, MessageChannel outputChannel ) { Integer product = 1; List> messagesUpForProcessing = group.getMessages(); for (Message message : messagesUpForProcessing) { product *= (Integer) message.getPayload(); } - outputChannel.send(MessageBuilder.withPayload(product).build()); + channelTemplate.send(MessageBuilder.withPayload(product).build(), outputChannel); group.onProcessingOf( messagesUpForProcessing.toArray(new Message[messagesUpForProcessing.size()]) @@ -337,7 +338,7 @@ public class NewConcurrentAggregatorEndpointTests { } private class NullReturningMessageProcessor implements MessageGroupProcessor { - public void processAndSend(MessageGroup group, MessageChannel outputChannel) { + public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) { //noop } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml similarity index 95% rename from org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml rename to org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml index 2c76e28c00..585ca61759 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml @@ -25,6 +25,6 @@ + class="org.springframework.integration.aggregator.integration.AggregatorIntegrationTests$SummingAggregator" /> \ No newline at end of file diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java similarity index 84% rename from org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java rename to org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java index 2e9fc8bfa5..de0d625f5f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java @@ -16,15 +16,10 @@ package org.springframework.integration.aggregator.integration; -import static org.hamcrest.CoreMatchers.is; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertThat; 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.aggregator.AbstractMessageAggregator; -import org.springframework.integration.aggregator.MethodInvokingAggregator; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; @@ -36,6 +31,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import static org.junit.Assert.assertEquals; + /** * @author Iwein Fuld * @author Alex Peters @@ -43,7 +40,7 @@ import java.util.Map; */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration -public class ConcurrentAggregatorIntegrationTests { +public class AggregatorIntegrationTests { @Autowired @Qualifier("input") @@ -53,14 +50,6 @@ public class ConcurrentAggregatorIntegrationTests { @Qualifier("output") private PollableChannel output; - @Autowired - AbstractMessageAggregator aggregator; - - @Test - public void configOk() throws Exception { - assertThat(aggregator, is(MethodInvokingAggregator.class)); - } - @Test(timeout=5000) public void aggregate() throws Exception { for (int i = 0; i < 5; i++) { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java index 633fb2cabe..ade8d019cc 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java @@ -16,22 +16,10 @@ package org.springframework.integration.aggregator.integration; -import static org.hamcrest.CoreMatchers.is; -import static org.junit.Assert.assertThat; -import static org.junit.Assert.assertTrue; - -import java.util.Arrays; -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.aggregator.AbstractMessageAggregator; -import org.springframework.integration.aggregator.DefaultMessageAggregator; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; @@ -39,6 +27,15 @@ import org.springframework.integration.message.GenericMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + /** * @author Alex Peters * @author Iwein Fuld @@ -55,15 +52,6 @@ public class DefaultMessageAggregatorIntegrationTests { @Qualifier("output") private PollableChannel output; - @Autowired - private AbstractMessageAggregator aggregator; - - - @Test - public void configOk() throws Exception { - assertThat(aggregator, is(DefaultMessageAggregator.class)); - } - @SuppressWarnings("unchecked") @Test(timeout = 1000) public void aggregate() throws Exception { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index a2cac5167f..b77f2782fc 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -16,80 +16,81 @@ package org.springframework.integration.config; -import java.lang.reflect.Method; -import java.util.ArrayList; -import java.util.List; - import org.junit.Assert; import org.junit.Before; import org.junit.Test; - import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.aggregator.CompletionStrategy; -import org.springframework.integration.aggregator.CompletionStrategyAdapter; -import org.springframework.integration.aggregator.CorrelationStrategy; -import org.springframework.integration.aggregator.MethodInvokingAggregator; +import org.springframework.integration.aggregator.*; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.util.MethodInvoker; import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.util.MethodInvoker; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; + +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; /** * @author Marius Bogoevici * @author Mark Fisher + * @author Iwein Fuld */ public class AggregatorParserTests { - private ApplicationContext context; + private ApplicationContext context; - @Before - public void setUp() { - this.context = new ClassPathXmlApplicationContext("aggregatorParserTests.xml", this.getClass()); - } + @Before + public void setUp() { + this.context = new ClassPathXmlApplicationContext("aggregatorParserTests.xml", this.getClass()); + } - @Test - public void testAggregation() { - MessageChannel input = (MessageChannel) context.getBean("aggregatorWithReferenceInput"); - TestAggregatorBean aggregatorBean = (TestAggregatorBean) context.getBean("aggregatorBean"); - List> outboundMessages = new ArrayList>(); - outboundMessages.add(createMessage("123", "id1", 3, 1, null)); - outboundMessages.add(createMessage("789", "id1", 3, 3, null)); - outboundMessages.add(createMessage("456", "id1", 3, 2, null)); - for (Message message : outboundMessages) { - input.send(message); - } - Assert.assertEquals("One and only one message must have been aggregated", 1, aggregatorBean - .getAggregatedMessages().size()); - Message aggregatedMessage = aggregatorBean.getAggregatedMessages().get("id1"); - Assert.assertEquals("The aggreggated message payload is not correct", "123456789", aggregatedMessage - .getPayload()); - } + @Test + public void testAggregation() { + MessageChannel input = (MessageChannel) context.getBean("aggregatorWithReferenceInput"); + TestAggregatorBean aggregatorBean = (TestAggregatorBean) context.getBean("aggregatorBean"); + List> outboundMessages = new ArrayList>(); + outboundMessages.add(createMessage("123", "id1", 3, 1, null)); + outboundMessages.add(createMessage("789", "id1", 3, 3, null)); + outboundMessages.add(createMessage("456", "id1", 3, 2, null)); + for (Message message : outboundMessages) { + input.send(message); + } + assertEquals("One and only one message must have been aggregated", 1, aggregatorBean + .getAggregatedMessages().size()); + Message aggregatedMessage = aggregatorBean.getAggregatedMessages().get("id1"); + assertEquals("The aggregated message payload is not correct", "123456789", aggregatedMessage + .getPayload()); + } - @Test - public void testPropertyAssignment() throws Exception { - EventDrivenConsumer endpoint = - (EventDrivenConsumer) context.getBean("completelyDefinedAggregator"); - CompletionStrategy completionStrategy = (CompletionStrategy) context.getBean("completionStrategy"); + @Test + public void testPropertyAssignment() throws Exception { + EventDrivenConsumer endpoint = + (EventDrivenConsumer) context.getBean("completelyDefinedAggregator"); + CompletionStrategy completionStrategy = (CompletionStrategy) context.getBean("completionStrategy"); CorrelationStrategy correlationStrategy = (CorrelationStrategy) context.getBean("correlationStrategy"); - MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel"); - MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel"); - Object consumer = TestUtils.getPropertyValue(endpoint, "handler"); - Assert.assertEquals(MethodInvokingAggregator.class, consumer.getClass()); - DirectFieldAccessor accessor = new DirectFieldAccessor(consumer); - Method expectedMethod = TestAggregatorBean.class.getMethod("createSingleMessageFromGroup", List.class); - Assert.assertEquals("The MethodInvokingAggregator is not injected with the appropriate aggregation method", - expectedMethod, new DirectFieldAccessor(accessor.getPropertyValue("methodInvoker")).getPropertyValue("method")); - Assert.assertEquals( - "The AggregatorEndpoint is not injected with the appropriate CompletionStrategy instance", - completionStrategy, accessor.getPropertyValue("completionStrategy")); - Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate CorrelationStrategy instance", + MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel"); + MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel"); + Object consumer = new DirectFieldAccessor(endpoint).getPropertyValue("handler"); + assertThat(consumer, is(CorrelatingMessageHandler.class)); + DirectFieldAccessor accessor = new DirectFieldAccessor(consumer); + Method expectedMethod = TestAggregatorBean.class.getMethod("createSingleMessageFromGroup", List.class); + assertEquals("The MethodInvokingAggregator is not injected with the appropriate aggregation method", + expectedMethod, ((MessageListMethodAdapter) new DirectFieldAccessor(accessor.getPropertyValue("outputProcessor")).getPropertyValue("adapter")).getMethod()); + assertEquals( + "The AggregatorEndpoint is not injected with the appropriate CompletionStrategy instance", + completionStrategy, accessor.getPropertyValue("completionStrategy")); + assertEquals("The AggregatorEndpoint is not injected with the appropriate CorrelationStrategy instance", correlationStrategy, accessor.getPropertyValue("correlationStrategy")); Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate output channel", outputChannel, accessor.getPropertyValue("outputChannel")); @@ -102,9 +103,6 @@ public class AggregatorParserTests { true, accessor.getPropertyValue("sendPartialResultOnTimeout")); Assert.assertEquals("The AggregatorEndpoint is not configured with the appropriate reaper interval", 135l, accessor.getPropertyValue("reaperInterval")); - Assert.assertEquals( - "The AggregatorEndpoint is not configured with the appropriate tracked correlationId capacity", - 99, accessor.getPropertyValue("trackedCorrelationIdCapacity")); Assert.assertEquals("The AggregatorEndpoint is not configured with the appropriate timeout", 42l, accessor.getPropertyValue("timeout")); } @@ -136,43 +134,43 @@ public class AggregatorParserTests { "completionStrategyMethodWithMissingReference.xml", this.getClass()); } - @Test - public void testAggregatorWithPojoCompletionStrategy() { - MessageChannel input = (MessageChannel) context.getBean("aggregatorWithPojoCompletionStrategyInput"); - EventDrivenConsumer endpoint = - (EventDrivenConsumer) context.getBean("aggregatorWithPojoCompletionStrategy"); - CompletionStrategy completionStrategy = TestUtils.getPropertyValue(endpoint, - "handler.completionStrategy", CompletionStrategy.class); - Assert.assertTrue(completionStrategy instanceof CompletionStrategyAdapter); - DirectFieldAccessor completionStrategyAccessor = new DirectFieldAccessor(completionStrategy); - MethodInvoker invoker = (MethodInvoker) completionStrategyAccessor.getPropertyValue("invoker"); - Assert.assertTrue(new DirectFieldAccessor(invoker).getPropertyValue("object") instanceof MaxValueCompletionStrategy); - Assert.assertTrue(((Method)completionStrategyAccessor.getPropertyValue("method")).getName().equals("checkCompleteness")); - input.send(createMessage(1l, "id1", 0 , 0, null)); - input.send(createMessage(2l, "id1", 0 , 0, null)); - input.send(createMessage(3l, "id1", 0 , 0, null)); - PollableChannel outputChannel = (PollableChannel) context.getBean("outputChannel"); - Message reply = outputChannel.receive(0); - Assert.assertNull(reply); - input.send(createMessage(5l, "id1", 0 , 0, null)); - reply = outputChannel.receive(0); - Assert.assertNotNull(reply); - Assert.assertEquals(11l, reply.getPayload()); - } + @Test + public void testAggregatorWithPojoCompletionStrategy() { + MessageChannel input = (MessageChannel) context.getBean("aggregatorWithPojoCompletionStrategyInput"); + EventDrivenConsumer endpoint = + (EventDrivenConsumer) context.getBean("aggregatorWithPojoCompletionStrategy"); + CompletionStrategy completionStrategy = (CompletionStrategy) new DirectFieldAccessor( + new DirectFieldAccessor(endpoint).getPropertyValue("handler")).getPropertyValue("completionStrategy"); + Assert.assertTrue(completionStrategy instanceof CompletionStrategyAdapter); + DirectFieldAccessor completionStrategyAccessor = new DirectFieldAccessor(completionStrategy); + MethodInvoker invoker = (MethodInvoker) completionStrategyAccessor.getPropertyValue("invoker"); + Assert.assertTrue(new DirectFieldAccessor(invoker).getPropertyValue("object") instanceof MaxValueCompletionStrategy); + Assert.assertTrue(((Method) completionStrategyAccessor.getPropertyValue("method")).getName().equals("checkCompleteness")); + input.send(createMessage(1l, "correllationId", 0, 0, null)); + input.send(createMessage(2l, "correllationId", 0, 1, null)); + input.send(createMessage(3l, "correllationId", 0, 2, null)); + PollableChannel outputChannel = (PollableChannel) context.getBean("outputChannel"); + Message reply = outputChannel.receive(0); + Assert.assertNull(reply); + input.send(createMessage(5l, "correllationId", 0, 3, null)); + reply = outputChannel.receive(0); + Assert.assertNotNull(reply); + assertEquals(11l, reply.getPayload()); + } - @Test(expected=BeanCreationException.class) - public void testAggregatorWithInvalidCompletionStrategyMethod() { - context = new ClassPathXmlApplicationContext("invalidCompletionStrategyMethod.xml", this.getClass()); - } + @Test(expected = BeanCreationException.class) + public void testAggregatorWithInvalidCompletionStrategyMethod() { + context = new ClassPathXmlApplicationContext("invalidCompletionStrategyMethod.xml", this.getClass()); + } - private static Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, - MessageChannel outputChannel) { - return MessageBuilder.withPayload(payload) - .setCorrelationId(correlationId) - .setSequenceSize(sequenceSize) - .setSequenceNumber(sequenceNumber) - .setReplyChannel(outputChannel).build(); - } + private static Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, + MessageChannel outputChannel) { + return MessageBuilder.withPayload(payload) + .setCorrelationId(correlationId) + .setSequenceSize(sequenceSize) + .setSequenceNumber(sequenceNumber) + .setReplyChannel(outputChannel).build(); + } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests-context.xml index 5a0099c9a6..36b546c83c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests-context.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests-context.xml @@ -6,9 +6,11 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"> - + - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java index 55902c8840..b513481b1a 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java @@ -16,12 +16,9 @@ package org.springframework.integration.config; -import java.util.List; - import org.junit.Assert; 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.aggregator.CompletionStrategy; @@ -34,8 +31,14 @@ import org.springframework.integration.message.MessageBuilder; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import java.util.List; + +import static org.junit.Assert.assertThat; +import static org.junit.matchers.JUnitMatchers.containsString; + /** - * @author: Marius Bogoevici + * @author Marius Bogoevici + * @author Iwein Fuld */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @@ -60,42 +63,40 @@ public class AggregatorWithCorrelationStrategyTests { @Test public void testCorrelationAndCompletion() { - inputChannel.send(MessageBuilder.withPayload("A1").build()); - inputChannel.send(MessageBuilder.withPayload("B2").build()); - inputChannel.send(MessageBuilder.withPayload("C3").build()); - inputChannel.send(MessageBuilder.withPayload("A4").build()); - inputChannel.send(MessageBuilder.withPayload("B5").build()); - inputChannel.send(MessageBuilder.withPayload("C6").build()); - inputChannel.send(MessageBuilder.withPayload("A7").build()); - inputChannel.send(MessageBuilder.withPayload("B8").build()); - inputChannel.send(MessageBuilder.withPayload("C9").build()); - receiveAndCompare(outputChannel, "A1A4A7"); - receiveAndCompare(outputChannel, "B2B5B8"); - receiveAndCompare(outputChannel, "C3C6C9"); + inputChannel.send(MessageBuilder.withPayload("A1").setSequenceNumber(0).build()); + inputChannel.send(MessageBuilder.withPayload("B2").setSequenceNumber(0).build()); + inputChannel.send(MessageBuilder.withPayload("C3").setSequenceNumber(0).build()); + inputChannel.send(MessageBuilder.withPayload("A4").setSequenceNumber(1).build()); + inputChannel.send(MessageBuilder.withPayload("B5").setSequenceNumber(1).build()); + inputChannel.send(MessageBuilder.withPayload("C6").setSequenceNumber(1).build()); + inputChannel.send(MessageBuilder.withPayload("A7").setSequenceNumber(2).build()); + inputChannel.send(MessageBuilder.withPayload("B8").setSequenceNumber(2).build()); + inputChannel.send(MessageBuilder.withPayload("C9").setSequenceNumber(2).build()); + receiveAndCompare(outputChannel, "A1","A4","A7"); + receiveAndCompare(outputChannel, "B2","B5","B8"); + receiveAndCompare(outputChannel, "C3","C6","C9"); } @Test public void testCorrelationAndCompletionWithPojo() { // the test verifies how a pojo strategy is applied // Strings are correlated by their first letter, integers are correlated by the last digit - pojoInputChannel.send(MessageBuilder.withPayload("X1").build()); - pojoInputChannel.send(MessageBuilder.withPayload("Y2").build()); - pojoInputChannel.send(MessageBuilder.withPayload(93).build()); - pojoInputChannel.send(MessageBuilder.withPayload("X4").build()); - pojoInputChannel.send(MessageBuilder.withPayload("Y5").build()); - pojoInputChannel.send(MessageBuilder.withPayload(113).build()); - pojoInputChannel.send(MessageBuilder.withPayload("X7").build()); - pojoInputChannel.send(MessageBuilder.withPayload("Y8").build()); - pojoInputChannel.send(MessageBuilder.withPayload(213).build()); - receiveAndCompare(pojoOutputChannel, "X1X4X7"); - receiveAndCompare(pojoOutputChannel, "Y2Y5Y8"); - receiveAndCompare(pojoOutputChannel, "93113213"); + pojoInputChannel.send(MessageBuilder.withPayload("X1").setSequenceNumber(0).build()); + pojoInputChannel.send(MessageBuilder.withPayload(93).setSequenceNumber(0).build()); + pojoInputChannel.send(MessageBuilder.withPayload("X4").setSequenceNumber(1).build()); + pojoInputChannel.send(MessageBuilder.withPayload(113).setSequenceNumber(1).build()); + pojoInputChannel.send(MessageBuilder.withPayload("X7").setSequenceNumber(2).build()); + pojoInputChannel.send(MessageBuilder.withPayload(213).setSequenceNumber(2).build()); + receiveAndCompare(pojoOutputChannel, "X1","X4","X7"); + receiveAndCompare(pojoOutputChannel, "93","113","213"); } - private void receiveAndCompare(PollableChannel outputChannel, String expectedValue) { - Message firstResult = outputChannel.receive(500); - Assert.assertNotNull(firstResult); - Assert.assertEquals(expectedValue, firstResult.getPayload()); + private void receiveAndCompare(PollableChannel outputChannel, String... expectedValues) { + Message message = outputChannel.receive(500); + Assert.assertNotNull(message); + for (String expectedValue : expectedValues) { + assertThat((String)message.getPayload(), containsString(expectedValue)); + } } @@ -128,8 +129,8 @@ public class AggregatorWithCorrelationStrategyTests { return message.substring(0,1); } - public String correlate(Integer mesage) { - return Integer.toString(mesage % 10); + public String correlate(Integer message) { + return Integer.toString(message % 10); } } @@ -137,7 +138,7 @@ public class AggregatorWithCorrelationStrategyTests { public static class SimpleAggregator { @Aggregator - protected String concatenate(List payloads) { + public String concatenate(List payloads) { StringBuffer buffer = new StringBuffer(); for (Object payload: payloads) { buffer.append(payload.toString()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml index 158191439f..b14f9aa125 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml @@ -15,7 +15,8 @@ - + - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/InnerDefinitionHandlerAwareEndpointParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/InnerDefinitionHandlerAwareEndpointParserTests.java index 310a6edda4..fd96d10a4c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/InnerDefinitionHandlerAwareEndpointParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/InnerDefinitionHandlerAwareEndpointParserTests.java @@ -16,20 +16,9 @@ package org.springframework.integration.config.xml; -import static org.junit.Assert.assertEquals; - -import java.io.ByteArrayInputStream; -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; - import junit.framework.Assert; - import org.junit.Test; import org.junit.runner.RunWith; - import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.xml.XmlBeanDefinitionReader; @@ -42,7 +31,6 @@ import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.endpoint.EventDrivenConsumer; -import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.StringMessage; import org.springframework.test.context.ContextConfiguration; @@ -50,6 +38,11 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; +import java.io.ByteArrayInputStream; +import java.util.*; + +import static org.junit.Assert.assertEquals; + /** * @author Oleg Zhurakousky */ @@ -253,11 +246,11 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { MessageChannel inChannel = (MessageChannel) ac.getBean("inChannel"); for (int i = 0; i < 5; i++) { Map headers = stubHeaders(i, 5, 1); - Message message = new GenericMessage(i, headers); + Message message = MessageBuilder.withPayload(i).copyHeaders(headers).build(); inChannel.send(message); } PollableChannel output = (PollableChannel) ac.getBean("outChannel"); - assertEquals(0 + 1 + 2 + 3 + 4, output.receive().getPayload()); + assertEquals(0 + 1 + 2 + 3 + 4, output.receive(100).getPayload()); } private void testFilterDefinitionSuccess(String configProperty){ @@ -284,7 +277,6 @@ public class InnerDefinitionHandlerAwareEndpointParserTests { headers.put(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber); headers.put(MessageHeaders.SEQUENCE_SIZE, sequenceSize); headers.put(MessageHeaders.CORRELATION_ID, correllationId); - headers.put(MessageHeaders.ID, 1); return headers; }