Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-330: switched namespace support for aggregator

Browse Source
This commit is contained in:
Iwein Fuld
2010-03-02 16:25:24 +00:00
parent 7a79b956d6
commit 1a38553402
24 changed files with 482 additions and 355 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -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);

57
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java Normal file
View File

@@ -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<Message<?>> messages = group.getMessages();
Assert.notEmpty(messages, this.getClass().getSimpleName() + " cannot process empty message groups");
channelTemplate.send(aggregateMessages(messages), outputChannel);
}
private Message<? extends Collection> aggregateMessages(List<Message<?>> messages) {
List payloads = new ArrayList(messages.size());
for (Message<?> message : messages) {
payloads.add(message.getPayload());
}
return MessageBuilder.withPayload(payloads).build();
}
}

15
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java
View File

@@ -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<? extends Message<?>> 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<Message<?>> all = group.getMessages();
public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) {
List<Message<?>> all = group.getMessages();
Object correlationKey = group.getCorrelationKey();
if (all.size() > 0) {
List<Message> 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();
}
}

7
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java
View File

@@ -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;
}
}

13
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java
View File

@@ -1,13 +1,24 @@
package org.springframework.integration.aggregator;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.MessageChannel;
/**
* A processor for <i>correlated</i> groups of messages. When a message group is <i>complete</i> 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
);
}

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java
View File

@@ -63,7 +63,7 @@ public class MessageListMethodAdapter {
}
protected Method getMethod() {
public Method getMethod() {
return method;
}

15
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java
View File

@@ -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<Message<?>> 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<Method> removeMethodsMatchingSelector(Set<Method> candidates, MethodSelector selector) {
private Set<Method> removeMethodsMatchingSelector(Set<Method> candidates, MethodSelector selector) {
Set<Method> removed = new HashSet<Method>();
Iterator<Method> iterator = candidates.iterator();
while (iterator.hasNext()) {

13
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java
View File

@@ -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();

181
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
View File

@@ -25,112 +25,119 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.StringUtils;
/**
* Parser for the <em>aggregator</em> element of the integration namespace.
* Registers the annotation-driven post-processors.
*
* Parser for the <em>aggregator</em> 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());
}
}

67
org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -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<Object, Message<?>> map;
private final Map<Object, Message<?>> map;
public SimpleMessageStore(int capacity) {
this.map = new ConcurrentHashMap<Object, Message<?>>(capacity);
}
public SimpleMessageStore(int capacity) {
this.map = new ConcurrentHashMap<Object, Message<?>>(capacity);
}
@SuppressWarnings("unchecked")
public <T> Message<T> put(Message<T> message) {
return (Message<T>) this.map.put(message.getHeaders().getId(), message);
}
@SuppressWarnings("unchecked")
public <T> Message<T> put(Message<T> message) {
return (Message<T>) 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<Message<?>> list() {
return new ArrayList<Message<?>>(this.map.values());
}
public List<Message<?>> list() {
return new ArrayList<Message<?>>(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<Message<?>> list(Object correlationKey) {
public List<Message<?>> list(Object correlationKey) {
Assert.notNull(correlationKey, "'correlationKey' must not be null");
List<Message<?>> matched = new ArrayList<Message<?>>();
Collection<Message<?>> values = map.values();
for (Message<?> message : values) {
if(message.getHeaders().getCorrelationId().equals(correlationKey)){
matched.add(message);
}
}
return matched;
}
List<Message<?>> matched = new ArrayList<Message<?>>();
Collection<Message<?>> values = map.values();
for (Message<?> message : values) {
Object correlationId = message.getHeaders().getCorrelationId();
if (correlationId != null && correlationId.equals(correlationKey)) {
matched.add(message);
}
}
return matched;
}
}

1
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java
View File

@@ -38,6 +38,7 @@ public class CorrelatingMessageHandlerIntegrationTest {
@Before public void setupHandler(){
defaultHandler.setOutputChannel(outputChannel);
defaultHandler.setSendTimeout(-1);
}
private Message<?> correlatedMessage(Object correlationId,

16
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
View File

@@ -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<Message<?>> storedMessages = new ArrayList<Message<?>>();
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<Message<?>> storedMessages = new ArrayList<Message<?>>();
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();
}

23
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java
View File

@@ -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));
}
}

7
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java
View File

@@ -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<Message<?>> 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
}
}

7
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java
View File

@@ -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<Message<?>> 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
}
}

2
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml → org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml
View File

@@ -25,6 +25,6 @@
</channel>
<beans:bean id="summer"
class="org.springframework.integration.aggregator.integration.ConcurrentAggregatorIntegrationTests$SummingAggregator" />
class="org.springframework.integration.aggregator.integration.AggregatorIntegrationTests$SummingAggregator" />
</beans:beans>

17
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java → org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java
View File

@@ -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++) {

30
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java
View File

@@ -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 {

172
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
View File

@@ -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<Message<?>> outboundMessages = new ArrayList<Message<?>>();
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<Message<?>> outboundMessages = new ArrayList<Message<?>>();
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 <T> Message<T> 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 <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel outputChannel) {
return MessageBuilder.withPayload(payload)
.setCorrelationId(correlationId)
.setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber)
.setReplyChannel(outputChannel).build();
}
}

6
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests-context.xml
View File

@@ -6,9 +6,11 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<beans:bean id="correlationStrategy" class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$FirstLetterCorrelationStrategy"/>
<beans:bean id="correlationStrategy"
class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$FirstLetterCorrelationStrategy"/>
<beans:bean id="pojoCorrelationStrategy" class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$PojoCorrelationStrategy"/>
<beans:bean id="pojoCorrelationStrategy"
class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$PojoCorrelationStrategy"/>
<beans:bean id="completionStrategy" class="org.springframework.integration.config.AggregatorWithCorrelationStrategyTests$MessageCountCompletionStrategy">
<beans:constructor-arg value="3"/>

71
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java
View File

@@ -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<Object> payloads) {
public String concatenate(List<Object> payloads) {
StringBuffer buffer = new StringBuffer();
for (Object payload: payloads) {
buffer.append(payload.toString());

3
org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml
View File

@@ -15,7 +15,8 @@
</channel>
<channel id="aggregatorWithReferenceInput"/>
<aggregator id="aggregatorWithReference" ref="aggregatorBean" input-channel="aggregatorWithReferenceInput"/>
<aggregator id="aggregatorWithReference" ref="aggregatorBean"
input-channel="aggregatorWithReferenceInput" output-channel="outputChannel"/>
<channel id="completelyDefinedAggregatorInput"/>
<aggregator id="completelyDefinedAggregator"

3
org.springframework.integration/src/test/java/org/springframework/integration/config/xml/InnerDefinitionHandlerAwareEndpointParserTests-context.xml
View File

@@ -10,6 +10,7 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<util:properties id="testConfigurations" location="classpath:org/springframework/integration/config/xml/innerdefaware.properties"/>
<util:properties id="testConfigurations"
location="classpath:org/springframework/integration/config/xml/innerdefaware.properties"/>
</beans:beans>

22
org.springframework.integration/src/test/java/org/springframework/integration/config/xml/InnerDefinitionHandlerAwareEndpointParserTests.java
View File

@@ -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<String, Object> headers = stubHeaders(i, 5, 1);
Message<Integer> message = new GenericMessage<Integer>(i, headers);
Message<Integer> 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;
}