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

INT-1305 Refactored CorrelatingMessageHandler to rely on MessagingTemplate rather than using ChannelResolver directly. More importantly, the MessageGroupProcessor now defines a simpler method: Object processMessageGroup(MessageGroup) which is consistent with the MessageProcessor method: Object processMessage(Message)

Browse Source
This commit is contained in:
Mark Fisher
2010-09-03 05:40:09 +00:00
parent 2f5dc0c202
commit eb84de121f
11 changed files with 208 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
View File

@@ -23,10 +23,9 @@ import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.splitter.AbstractMessageSplitter;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.support.MessageBuilder;
@@ -39,16 +38,15 @@ import org.springframework.util.Assert;
* @author Alexander Peters
* @author Mark Fisher
* @author Dave Syer
*
* @since 2.0
*/
public abstract class AbstractAggregatingMessageGroupProcessor implements MessageGroupProcessor {
private final Log logger = LogFactory.getLog(this.getClass());
public final void processAndSend(MessageGroup group, MessagingOperations messagingTemplate, MessageChannel outputChannel) {
public final Object processMessageGroup(MessageGroup group) {
Assert.notNull(group, "MessageGroup must not be null");
Assert.notNull(outputChannel, "'outputChannel' must not be null");
Map<String, Object> headers = this.aggregateHeaders(group);
Object payload = this.aggregatePayloads(group, headers);
MessageBuilder<?> builder;
@@ -58,8 +56,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
else {
builder = MessageBuilder.withPayload(payload).copyHeadersIfAbsent(headers);
}
Message<?> message = builder.build();
messagingTemplate.send(outputChannel, message);
return builder.build();
}
/**

96
spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.core.MessagingTemplate;
@@ -32,9 +33,6 @@ import org.springframework.integration.store.MessageGroupCallback;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.support.channel.ChannelResolutionException;
import org.springframework.integration.support.channel.ChannelResolver;
import org.springframework.util.Assert;
/**
@@ -78,8 +76,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
private final MessagingTemplate messagingTemplate = new MessagingTemplate();
private volatile ChannelResolver channelResolver;
private volatile MessageChannel discardChannel = new NullChannel();
private boolean sendPartialResultOnExpiry = false;
@@ -136,8 +132,8 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
protected void onInit() throws Exception {
super.onInit();
BeanFactory beanFactory = this.getBeanFactory();
if (this.channelResolver == null && beanFactory != null) {
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
if (beanFactory != null) {
this.messagingTemplate.setBeanFactory(beanFactory);
}
}
@@ -160,7 +156,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
Object correlationKey = correlationStrategy.getCorrelationKey(message);
if (logger.isDebugEnabled()) {
logger.debug("Handling message with correlationKey ["
@@ -182,11 +177,10 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
+ correlationKey + "]");
}
try {
outputProcessor.processAndSend(group, messagingTemplate,
this.resolveReplyChannel(message, this.outputChannel));
Object result = outputProcessor.processMessageGroup(group);
this.sendReplies(result, message.getHeaders().getReplyChannel());
}
finally {
// Always clean up even if there was an exception
// processing messages
if (group.isComplete() || group.getSequenceSize() == 0) {
@@ -230,8 +224,8 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
// last chance for normal completion
try {
if (releaseStrategy.canRelease(group)) {
outputProcessor.processAndSend(group, messagingTemplate,
resolveReplyChannel(group.getOne(), this.outputChannel));
Object result = outputProcessor.processMessageGroup(group);
this.sendRepliesForGroup(result, group);
}
else {
if (sendPartialResultOnExpiry) {
@@ -239,8 +233,8 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
logger.info("Processing partially complete messages for key ["
+ correlationKey + "] to: " + outputChannel);
}
outputProcessor.processAndSend(group, messagingTemplate,
resolveReplyChannel(group.getOne(), this.outputChannel));
Object result = outputProcessor.processMessageGroup(group);
this.sendRepliesForGroup(result, group);
}
else {
if (logger.isInfoEnabled()) {
@@ -281,28 +275,62 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
return messageStore.addMessageToGroup(correlationKey, message);
}
private MessageChannel resolveReplyChannel(Message<?> requestMessage, MessageChannel defaultOutputChannel) {
MessageChannel replyChannel = defaultOutputChannel;
if (replyChannel == null) {
Object replyChannelHeader = requestMessage.getHeaders().getReplyChannel();
if (replyChannelHeader instanceof MessageChannel) {
replyChannel = (MessageChannel) replyChannelHeader;
}
else if (replyChannelHeader instanceof String) {
Assert.state(channelResolver != null,
"ChannelResolver is required for resolving a reply channel by name");
replyChannel = this.channelResolver.resolveChannelName((String) replyChannelHeader);
}
else if (replyChannelHeader != null) {
throw new ChannelResolutionException(
"expected a MessageChannel or String for 'replyChannel' header, " +
"but type is [" + replyChannelHeader.getClass() + "]");
private void sendRepliesForGroup(Object processorResult, MessageGroup group) {
Object replyChannelHeader = null;
if (group != null) {
Message<?> first = group.getOne();
if (first != null) {
replyChannelHeader = first.getHeaders().getReplyChannel();
}
}
if (replyChannel == null) {
throw new ChannelResolutionException("unable to resolve reply channel for message: " + requestMessage);
this.sendReplies(processorResult, replyChannelHeader);
}
private void sendReplies(Object processorResult, Object replyChannelHeader) {
Object replyChannel = this.outputChannel;
if (this.outputChannel == null) {
replyChannel = replyChannelHeader;
}
return replyChannel;
Assert.notNull(replyChannel, "no outputChannel or replyChannel header available");
if (processorResult instanceof Iterable<?> && shouldSendMultipleReplies((Iterable<?>) processorResult)) {
for (Object next : (Iterable<?>) processorResult) {
this.sendReplyMessage(next, replyChannel);
}
}
else {
this.sendReplyMessage(processorResult, replyChannel);
}
}
private void sendReplyMessage(Object reply, Object replyChannel) {
if (replyChannel instanceof MessageChannel) {
if (reply instanceof Message<?>) {
this.messagingTemplate.send((MessageChannel) replyChannel, (Message<?>) reply);
}
else {
this.messagingTemplate.convertAndSend((MessageChannel) replyChannel, reply);
}
}
else if (replyChannel instanceof String) {
if (reply instanceof Message<?>) {
this.messagingTemplate.send((String) replyChannel, (Message<?>) reply);
}
else {
this.messagingTemplate.convertAndSend((String) replyChannel, reply);
}
}
else {
throw new MessagingException("replyChannel must be a MessageChannel or String");
}
}
private boolean shouldSendMultipleReplies(Iterable<?> iter) {
for (Object next : iter) {
if (next instanceof Message<?>) {
return true;
}
}
return false;
}
}

11
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java
View File

@@ -13,8 +13,6 @@
package org.springframework.integration.aggregator;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.store.MessageGroup;
/**
@@ -26,11 +24,10 @@ import org.springframework.integration.store.MessageGroup;
public interface MessageGroupProcessor {
/**
* Process the given group and send the resulting message(s) to the output channel using the messaging operations.
* Implementations are free to send as few or as many messages based on the invocation as needed. For example an
* aggregating processor will send only a single message representing the group, where a resequencing strategy will
* send all messages in the group individually.
* Process the given MessageGroup. Implementations are free to return as few or as many messages based on the
* invocation as needed. For example an aggregating processor will return only a single message representing the
* group, while a resequencing processor will return all messages whose preceding sequence has been satisfied.
*/
void processAndSend(MessageGroup group, MessagingOperations messagingTemplate, MessageChannel outputChannel);
Object processMessageGroup(MessageGroup group);
}

19
spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java
View File

@@ -13,25 +13,20 @@
package org.springframework.integration.aggregator;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.store.MessageGroup;
/**
* 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 ReleaseStrategy lets them pass through.
*
* This implementation of MessageGroupProcessor will return all unmarked messages inside the group.
* This is useful if there is no requirement to process the messages, but they should just be
* blocked as a group until their ReleaseStrategy lets them pass through.
*
* @author Iwein Fuld
* @since 2.0.0
*/
public class PassThroughMessageGroupProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group, MessagingOperations messagingTemplate, MessageChannel outputChannel) {
for (Message<?> message : group.getUnmarked()) {
messagingTemplate.send(outputChannel, message);
}
}
public Object processMessageGroup(MessageGroup group) {
return group.getUnmarked();
}
}

13
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java
View File

@@ -20,8 +20,6 @@ import java.util.Comparator;
import java.util.List;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.store.MessageGroup;
/**
@@ -29,7 +27,6 @@ import org.springframework.integration.store.MessageGroup;
*
* @author Iwein Fuld
* @author Dave Syer
*
* @since 2.0
*/
public class ResequencingMessageGroupProcessor implements MessageGroupProcessor {
@@ -38,22 +35,20 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor
/**
* A comparator to use to order messages before processing. The default is to order by sequence number.
*
* @param comparator the comparator to use to order messages
*/
public void setComparator(Comparator<Message<?>> comparator) {
this.comparator = comparator;
}
public void processAndSend(MessageGroup group, MessagingOperations messagingTemplate, MessageChannel outputChannel) {
public Object processMessageGroup(MessageGroup group) {
Collection<Message<?>> messages = group.getUnmarked();
if (messages.size() > 0) {
List<Message<?>> sorted = new ArrayList<Message<?>>(messages);
Collections.sort(sorted, comparator);
for (Message<?> message : sorted) {
messagingTemplate.send(outputChannel, message);
}
Collections.sort(sorted, this.comparator);
return sorted;
}
return null;
}
}

64
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.aggregator;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.Arrays;
import java.util.Collections;
@@ -28,8 +29,6 @@ import java.util.Map;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.support.MessageBuilder;
@@ -40,10 +39,6 @@ import org.springframework.integration.support.MessageBuilder;
*/
public class AggregatingMessageGroupProcessorHeaderTests {
private final QueueChannel outputChannel = new QueueChannel(1);
private final MessagingTemplate messagingTemplate = new MessagingTemplate();
private final DefaultAggregatingMessageGroupProcessor defaultProcessor = new DefaultAggregatingMessageGroupProcessor();
private final MethodInvokingMessageGroupProcessor methodInvokingProcessor =
@@ -107,11 +102,12 @@ public class AggregatingMessageGroupProcessorHeaderTests {
Message<?> message = correlatedMessage(1, 1, 1, headers);
List<Message<?>> messages = Collections.<Message<?>>singletonList(message);
MessageGroup group = new SimpleMessageGroup(messages, 1);
processor.processAndSend(group, messagingTemplate, outputChannel);
Message<?> result = outputChannel.receive(0);
Object result = processor.processMessageGroup(group);
assertNotNull(result);
assertEquals("value1", result.getHeaders().get("k1"));
assertEquals(2, result.getHeaders().get("k2"));
assertTrue(result instanceof Message<?>);
Message<?> resultMessage = (Message<?>) result;
assertEquals("value1", resultMessage.getHeaders().get("k1"));
assertEquals(2, resultMessage.getHeaders().get("k2"));
}
private void twoMessagesWithoutConflicts(MessageGroupProcessor processor) {
@@ -122,11 +118,12 @@ public class AggregatingMessageGroupProcessorHeaderTests {
Message<?> message2 = correlatedMessage(1, 2, 2, headers);
List<Message<?>> messages = Arrays.<Message<?>>asList(message1, message2);
MessageGroup group = new SimpleMessageGroup(messages, 1);
processor.processAndSend(group, messagingTemplate, outputChannel);
Message<?> result = outputChannel.receive(0);
Object result = processor.processMessageGroup(group);
assertNotNull(result);
assertEquals("value1", result.getHeaders().get("k1"));
assertEquals(2, result.getHeaders().get("k2"));
assertTrue(result instanceof Message<?>);
Message<?> resultMessage = (Message<?>) result;
assertEquals("value1", resultMessage.getHeaders().get("k1"));
assertEquals(2, resultMessage.getHeaders().get("k2"));
}
private void twoMessagesWithConflicts(MessageGroupProcessor processor) {
@@ -140,11 +137,12 @@ public class AggregatingMessageGroupProcessorHeaderTests {
Message<?> message2 = correlatedMessage(1, 2, 2, headers2);
List<Message<?>> messages = Arrays.<Message<?>>asList(message1, message2);
MessageGroup group = new SimpleMessageGroup(messages, 1);
processor.processAndSend(group, messagingTemplate, outputChannel);
Message<?> result = outputChannel.receive(0);
Object result = processor.processMessageGroup(group);
assertNotNull(result);
assertNull(result.getHeaders().get("k1"));
assertEquals(123, result.getHeaders().get("k2"));
assertTrue(result instanceof Message<?>);
Message<?> resultMessage = (Message<?>) result;
assertNull(resultMessage.getHeaders().get("k1"));
assertEquals(123, resultMessage.getHeaders().get("k2"));
}
private void missingValuesDoNotConflict(MessageGroupProcessor processor) {
@@ -170,17 +168,18 @@ public class AggregatingMessageGroupProcessorHeaderTests {
Message<?> message3 = correlatedMessage(1, 3, 3, headers3);
List<Message<?>> messages = Arrays.<Message<?>>asList(message1, message2, message3);
MessageGroup group = new SimpleMessageGroup(messages, 1);
processor.processAndSend(group, messagingTemplate, outputChannel);
Message<?> result = outputChannel.receive(0);
Object result = processor.processMessageGroup(group);
assertNotNull(result);
assertEquals("value1", result.getHeaders().get("only1"));
assertEquals("value2", result.getHeaders().get("only2"));
assertEquals("value3", result.getHeaders().get("only3"));
assertEquals("foo", result.getHeaders().get("commonTo1And2"));
assertEquals("bar", result.getHeaders().get("commonTo2And3"));
assertEquals(123, result.getHeaders().get("commonToAll"));
assertNull(result.getHeaders().get("conflictBetween1And2"));
assertNull(result.getHeaders().get("conflictBetween2And3"));
assertTrue(result instanceof Message<?>);
Message<?> resultMessage = (Message<?>) result;
assertEquals("value1", resultMessage.getHeaders().get("only1"));
assertEquals("value2", resultMessage.getHeaders().get("only2"));
assertEquals("value3", resultMessage.getHeaders().get("only3"));
assertEquals("foo", resultMessage.getHeaders().get("commonTo1And2"));
assertEquals("bar", resultMessage.getHeaders().get("commonTo2And3"));
assertEquals(123, resultMessage.getHeaders().get("commonToAll"));
assertNull(resultMessage.getHeaders().get("conflictBetween1And2"));
assertNull(resultMessage.getHeaders().get("conflictBetween2And3"));
}
private void multipleValuesConflict(MessageGroupProcessor processor) {
@@ -198,11 +197,12 @@ public class AggregatingMessageGroupProcessorHeaderTests {
Message<?> message3 = correlatedMessage(1, 3, 3, headers3);
List<Message<?>> messages = Arrays.<Message<?>>asList(message1, message2, message3);
MessageGroup group = new SimpleMessageGroup(messages, 1);
processor.processAndSend(group, messagingTemplate, outputChannel);
Message<?> result = outputChannel.receive(0);
Object result = processor.processMessageGroup(group);
assertNotNull(result);
assertEquals("valueForAll", result.getHeaders().get("common"));
assertNull(result.getHeaders().get("conflict"));
assertTrue(result instanceof Message<?>);
Message<?> resultMessage = (Message<?>) result;
assertEquals("valueForAll", resultMessage.getHeaders().get("common"));
assertNull(resultMessage.getHeaders().get("conflict"));
}
private static Message<?> correlatedMessage(Object correlationId, Integer sequenceSize,

19
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java
View File

@@ -25,12 +25,12 @@ import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MessageBuilder;
@@ -43,13 +43,16 @@ import org.springframework.integration.support.MessageBuilder;
public class AggregatorTests {
private CorrelatingMessageHandler aggregator;
private SimpleMessageStore store = new SimpleMessageStore(50);
@Before
public void configureAggregator() {
this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), store);
}
@Test
public void testCompleteGroupWithinTimeout() throws InterruptedException {
QueueChannel replyChannel = new QueueChannel();
@@ -220,6 +223,7 @@ public class AggregatorTests {
assertNull(reply);
}
private static Message<?> createMessage(Object payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel replyChannel, String predefinedId) {
MessageBuilder<Object> builder = MessageBuilder.withPayload(payload).setCorrelationId(correlationId)
@@ -230,21 +234,22 @@ public class AggregatorTests {
return builder.build();
}
private class MultiplyingProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group, MessagingOperations messagingTemplate,
MessageChannel outputChannel) {
public Object processMessageGroup(MessageGroup group) {
Integer product = 1;
for (Message<?> message : group.getUnmarked()) {
product *= (Integer) message.getPayload();
}
messagingTemplate.send(outputChannel, MessageBuilder.withPayload(product).build());
return product;
}
}
private class NullReturningMessageProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group, MessagingOperations messagingTemplate,
MessageChannel outputChannel) {
// noop
public Object processMessageGroup(MessageGroup group) {
return null;
}
}
}

22
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java
View File

@@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.Message;
@@ -36,7 +37,6 @@ import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
@@ -55,12 +55,14 @@ public class ConcurrentAggregatorTests {
private MessageGroupStore store = new SimpleMessageStore();
@Before
public void configureAggregator() {
this.taskExecutor = new SimpleAsyncTaskExecutor();
this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), store);
}
@Test
public void testCompleteGroupWithinTimeout() throws InterruptedException {
QueueChannel replyChannel = new QueueChannel();
@@ -296,6 +298,7 @@ public class ConcurrentAggregatorTests {
assertNull(reply);
}
private static Message<?> createMessage(Object payload,
Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel replyChannel, String predefinedId) {
@@ -309,6 +312,7 @@ public class ConcurrentAggregatorTests {
return builder.build();
}
private static class AggregatorTestTask implements Runnable {
private MessageHandler aggregator;
@@ -342,23 +346,23 @@ public class ConcurrentAggregatorTests {
}
}
private class MultiplyingProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group,
MessagingOperations messagingTemplate,
MessageChannel outputChannel) {
public Object processMessageGroup(MessageGroup group) {
Integer product = 1;
for (Message<?> message : group.getUnmarked()) {
product *= (Integer) message.getPayload();
}
messagingTemplate.send(outputChannel, MessageBuilder.withPayload(product).build());
return product;
//messagingTemplate.send(outputChannel, MessageBuilder.withPayload(product).build());
}
}
private class NullReturningMessageProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group,
MessagingOperations messagingTemplate,
MessageChannel outputChannel) {
// noop
public Object processMessageGroup(MessageGroup group) {
return null;
}
}
}

23
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
View File

@@ -18,7 +18,6 @@ package org.springframework.integration.aggregator;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.isA;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.verify;
@@ -34,13 +33,12 @@ import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.internal.stubbing.answers.DoesNothing;
import org.mockito.internal.stubbing.answers.ThrowsException;
import org.mockito.runners.MockitoJUnitRunner;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.store.SimpleMessageStore;
@@ -69,14 +67,14 @@ public class CorrelatingMessageHandlerTests {
private MessageGroupStore store = new SimpleMessageStore();
@Before
public void initializeSubject() {
handler = new CorrelatingMessageHandler(processor, store, correlationStrategy, ReleaseStrategy);
handler.setOutputChannel(outputChannel);
doAnswer(new DoesNothing()).when(processor).processAndSend(isA(SimpleMessageGroup.class),
isA(MessagingTemplate.class), eq(outputChannel));
}
@Test
public void bufferCompletesNormally() throws Exception {
String correlationKey = "key";
@@ -93,8 +91,7 @@ public class CorrelatingMessageHandlerTests {
verify(correlationStrategy).getCorrelationKey(message1);
verify(correlationStrategy).getCorrelationKey(message2);
verify(processor)
.processAndSend(isA(SimpleMessageGroup.class), isA(MessagingTemplate.class), eq(outputChannel));
verify(processor).processMessageGroup(isA(SimpleMessageGroup.class));
}
private void verifyLocks(CorrelatingMessageHandler handler, int lockCount) {
@@ -104,8 +101,8 @@ public class CorrelatingMessageHandlerTests {
@Test
public void bufferCompletesWithException() throws Exception {
doAnswer(new ThrowsException(new RuntimeException("Planned test exception"))).when(processor).processAndSend(
isA(SimpleMessageGroup.class), isA(MessagingTemplate.class), eq(outputChannel));
doAnswer(new ThrowsException(new RuntimeException("Planned test exception")))
.when(processor).processMessageGroup(isA(SimpleMessageGroup.class));
String correlationKey = "key";
Message<?> message1 = testMessage(correlationKey, 1, 2);
@@ -125,8 +122,7 @@ public class CorrelatingMessageHandlerTests {
verify(correlationStrategy).getCorrelationKey(message1);
verify(correlationStrategy).getCorrelationKey(message2);
verify(processor)
.processAndSend(isA(SimpleMessageGroup.class), isA(MessagingTemplate.class), eq(outputChannel));
verify(processor).processMessageGroup(isA(SimpleMessageGroup.class));
}
/*
@@ -160,7 +156,6 @@ public class CorrelatingMessageHandlerTests {
assertEquals(0, store.expireMessageGroups(10000));
bothMessagesHandled.await();
}
@Test
@@ -170,7 +165,8 @@ public class CorrelatingMessageHandlerTests {
try {
handler.handleMessage(message1);
fail("Expected MessageHandlingException");
} catch (MessageHandlingException e) {
}
catch (MessageHandlingException e) {
Throwable cause = e.getCause();
boolean pass = cause instanceof IllegalStateException && cause.getMessage().toLowerCase().contains("null correlation");
if (!pass) {
@@ -179,6 +175,7 @@ public class CorrelatingMessageHandlerTests {
}
}
private Message<?> testMessage(String correlationKey, int sequenceNumber, int sequenceSize) {
return MessageBuilder.withPayload("test" + sequenceNumber).setCorrelationId(correlationKey).setSequenceNumber(
sequenceNumber).setSequenceSize(sequenceSize).build();

124
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java
View File

@@ -16,92 +16,98 @@
package org.springframework.integration.aggregator;
import static org.junit.matchers.JUnitMatchers.hasItems;
import static org.mockito.Matchers.isA;
import static org.mockito.Mockito.verify;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.hamcrest.CoreMatchers;
import org.hamcrest.Description;
import org.hamcrest.Factory;
import org.hamcrest.Matcher;
import org.hamcrest.TypeSafeMatcher;
import org.hamcrest.collection.IsMapContaining;
import org.hamcrest.core.IsEqual;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Matchers;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.support.MessageBuilder;
/**
* @author Alex Peters
*
* @author Mark Fisher
*/
@RunWith(MockitoJUnitRunner.class)
public class ExpressionEvaluatingMessageGroupProcessorTests {
private ExpressionEvaluatingMessageGroupProcessor processor;
private MessagingTemplate template = new MessagingTemplate();
@Mock
private MessageChannel outputChannel;
@Mock
private MessageGroup group;
List<Message<?>> messages = new ArrayList<Message<?>>();
private final List<Message<?>> messages = new ArrayList<Message<?>>();
@Before
public void setup() {
when(outputChannel.send(isA(Message.class))).thenReturn(true);
messages.clear();
for (int i = 0; i < 5; i++) {
messages.add(MessageBuilder.withPayload(i + 1).setHeader("foo", "bar").build());
}
}
@Test
public void testProcessAndSendWithSizeExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("#root.size()");
processor.processAndSend(group, template, outputChannel);
verify(outputChannel).send(messageWithPayload(5));
Object result = processor.processMessageGroup(group);
assertTrue(result instanceof Message<?>);
Message<?> resultMessage = (Message<?>) result;
assertEquals(5, resultMessage.getPayload());
}
@Test
public void testProcessAndCheckHeaders() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("#root");
processor.processAndSend(group, template, outputChannel);
verify(outputChannel).send(messageWithHeader("foo", "bar"));
Object result = processor.processMessageGroup(group);
assertTrue(result instanceof Message<?>);
Message<?> resultMessage = (Message<?>) result;
assertEquals("bar", resultMessage.getHeaders().get("foo"));
}
@Test
public void testProcessAndSendWithProjectionExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("![payload]");
processor.processAndSend(group, template, outputChannel);
verify(outputChannel).send(messageWithPayload(hasItems(1, 2, 3, 4, 5)));
Object result = processor.processMessageGroup(group);
assertTrue(result instanceof Message<?>);
Message<?> resultMessage = (Message<?>) result;
assertTrue(resultMessage.getPayload() instanceof Collection<?>);
Collection<?> list = (Collection<?>) resultMessage.getPayload();
assertEquals(5, list.size());
assertTrue(list.contains(1));
assertTrue(list.contains(2));
assertTrue(list.contains(3));
assertTrue(list.contains(4));
assertTrue(list.contains(5));
}
@Test
public void testProcessAndSendWithFilterAndProjectionExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("?[payload>2].![payload]");
processor.processAndSend(group, template, outputChannel);
verify(outputChannel).send(messageWithPayload(hasItems(3, 4, 5)));
Object result = processor.processMessageGroup(group);
assertTrue(result instanceof Message<?>);
Message<?> resultMessage = (Message<?>) result;
assertTrue(resultMessage.getPayload() instanceof Collection<?>);
Collection<?> list = (Collection<?>) resultMessage.getPayload();
assertEquals(3, list.size());
assertTrue(list.contains(3));
assertTrue(list.contains(4));
assertTrue(list.contains(5));
}
@Test
@@ -109,21 +115,12 @@ public class ExpressionEvaluatingMessageGroupProcessorTests {
when(group.getUnmarked()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor(String.format("T(%s).sum(?[payload>2].![payload])",
getClass().getName()));
processor.processAndSend(group, template, outputChannel);
verify(outputChannel).send(messageWithPayload(3 + 4 + 5));
Object result = processor.processMessageGroup(group);
assertTrue(result instanceof Message<?>);
Message<?> resultMessage = (Message<?>) result;
assertEquals(3 + 4 + 5, resultMessage.getPayload());
}
private Message<?> messageWithHeader(String key, Object value) {
return Matchers.argThat(MessageMatcher.hasHeader(IsMapContaining.hasEntry(key, value)));
}
private Message<?> messageWithPayload(Matcher<?> matcher) {
return Matchers.argThat(MessageMatcher.hasPayload(matcher));
}
private Message<?> messageWithPayload(int i) {
return Matchers.argThat(MessageMatcher.hasPayload(IsEqual.equalTo(i)));
}
/*
* sample static method invoked by SpEL
@@ -135,48 +132,5 @@ public class ExpressionEvaluatingMessageGroupProcessorTests {
}
return result;
}
private static class MessageMatcher extends TypeSafeMatcher<Message<?>> {
private final Matcher<?> payloadMatcher;
private final Matcher<?> headerMatcher;
/**
* @param matcher
*/
MessageMatcher(Matcher<?> matcher, Matcher<?> headerMatcher) {
super();
this.payloadMatcher = matcher;
this.headerMatcher = headerMatcher;
}
/**
* {@inheritDoc}
*/
@Override
public boolean matchesSafely(Message<?> message) {
return payloadMatcher.matches(message.getPayload()) && headerMatcher.matches(message.getHeaders());
}
/**
* {@inheritDoc}
*/
//@Override
public void describeTo(Description description) {
description.appendText("a Message with payload: ").appendDescriptionOf(payloadMatcher);
description.appendText(" and headers: ").appendDescriptionOf(headerMatcher);
}
@Factory
public static <T> Matcher<Message<?>> hasPayload(Matcher<T> payloadMatcher) {
return new MessageMatcher(payloadMatcher, CoreMatchers.anything());
}
@Factory
public static <T> Matcher<Message<?>> hasHeader(Matcher<T> headerMatcher) {
return new MessageMatcher(CoreMatchers.anything(), headerMatcher);
}
}
}

100
spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java
View File

@@ -20,9 +20,6 @@ import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
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;
import java.util.ArrayList;
@@ -35,7 +32,6 @@ import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
@@ -44,14 +40,12 @@ import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.support.ConversionServiceFactory;
import org.springframework.core.convert.support.GenericConversionService;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.annotation.Headers;
import org.springframework.integration.annotation.Payloads;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.store.MessageGroup;
@@ -61,16 +55,11 @@ import org.springframework.integration.support.MessageBuilder;
@RunWith(MockitoJUnitRunner.class)
public class MethodInvokingMessageGroupProcessorTests {
@Mock
private MessageChannel outputChannel;
private List<Message<?>> messagesUpForProcessing = new ArrayList<Message<?>>(3);
@Mock
private MessageGroup messageGroupMock;
@Mock
private MessagingTemplate messagingTemplate;
@Before
public void initializeMessagesUpForProcessing() {
@@ -79,7 +68,7 @@ public class MethodInvokingMessageGroupProcessorTests {
messagesUpForProcessing.add(MessageBuilder.withPayload(4).build());
}
@SuppressWarnings("unchecked")
@Test
public void shouldFindAnnotatedAggregatorMethod() throws Exception {
@@ -101,17 +90,12 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new AnnotatedAggregatorMethod());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindSimpleAggregatorMethod() throws Exception {
@SuppressWarnings("unused")
@@ -126,17 +110,12 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindSimpleAggregatorMethodForMessages() throws Exception {
@SuppressWarnings("unused")
@@ -151,17 +130,12 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindAnnotatedPayloads() throws Exception {
@SuppressWarnings("unused")
@@ -179,18 +153,13 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
messagesUpForProcessing.add(MessageBuilder.withPayload(3).setHeader("foo", Arrays.asList(101, 102)).build());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((String) messageCaptor.getValue().getPayload(), is("[1, 2, 4, 3, 101, 102]"));
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((String) ((Message<?>) result).getPayload(), is("[1, 2, 4, 3, 101, 102]"));
}
@Test
@SuppressWarnings("unchecked")
public void shouldUseAnnotatedPayloads() throws Exception {
@SuppressWarnings("unused")
@@ -209,17 +178,12 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((String) messageCaptor.getValue().getPayload(), is("[1, 2, 4]"));
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((String) ((Message<?>) result).getPayload(), is("[1, 2, 4]"));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindSimpleAggregatorMethodWithCollection() throws Exception {
@SuppressWarnings("unused")
@@ -234,17 +198,12 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindSimpleAggregatorMethodWithArray() throws Exception {
@SuppressWarnings("unused")
@@ -259,17 +218,12 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindSimpleAggregatorMethodWithIterator() throws Exception {
@SuppressWarnings("unused")
@@ -291,17 +245,12 @@ public class MethodInvokingMessageGroupProcessorTests {
}
});
processor.setConversionService(conversionService);
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindFittingMethodAmongMultipleUnannotated() {
@SuppressWarnings("unused")
@@ -325,15 +274,9 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new UnannotatedAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@Test(expected = IllegalArgumentException.class)
@@ -381,7 +324,6 @@ public class MethodInvokingMessageGroupProcessorTests {
group.add(new GenericMessage<String>("foo"));
group.add(new GenericMessage<String>("bar"));
assertEquals("foo", aggregator.aggregatePayloads(group, null));
}
@Test
@@ -401,7 +343,6 @@ public class MethodInvokingMessageGroupProcessorTests {
group.add(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build());
group.add(MessageBuilder.withPayload("bar").setHeader("foo", "bar").build());
assertEquals("foobar", aggregator.aggregatePayloads(group, aggregator.aggregateHeaders(group)));
}
@Test
@@ -421,7 +362,6 @@ public class MethodInvokingMessageGroupProcessorTests {
group.add(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build());
group.add(MessageBuilder.withPayload("bar").setHeader("foo", "bar").build());
assertEquals("foobar", aggregator.aggregatePayloads(group, aggregator.aggregateHeaders(group)));
}
@Test(expected = IllegalArgumentException.class)
@@ -540,10 +480,12 @@ public class MethodInvokingMessageGroupProcessorTests {
assertEquals("hello proxy", output.receive(0).getPayload());
}
public interface GreetingService {
String sayHello(List<String> names);
}
public static class GreetingBean implements GreetingService {
private String greeting = "hello";