From eb84de121f182cedf34615e46b76d95d29be74d4 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 3 Sep 2010 05:40:09 +0000 Subject: [PATCH] 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) --- ...tractAggregatingMessageGroupProcessor.java | 11 +- .../aggregator/CorrelatingMessageHandler.java | 96 +++++++++----- .../aggregator/MessageGroupProcessor.java | 11 +- .../PassThroughMessageGroupProcessor.java | 19 +-- .../ResequencingMessageGroupProcessor.java | 13 +- ...atingMessageGroupProcessorHeaderTests.java | 64 ++++----- .../aggregator/AggregatorTests.java | 19 ++- .../aggregator/ConcurrentAggregatorTests.java | 22 ++-- .../CorrelatingMessageHandlerTests.java | 23 ++-- ...nEvaluatingMessageGroupProcessorTests.java | 124 ++++++------------ ...hodInvokingMessageGroupProcessorTests.java | 100 +++----------- 11 files changed, 208 insertions(+), 294 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index 8ba9645ef8..0253aa7901 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -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 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(); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 114b01623d..ded9624d3e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -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; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java index 45a0fe8c9a..13481f2a81 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java @@ -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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java index f08bf409af..ccb078448e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java @@ -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(); + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java index 75ca81579e..fc89af08c8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java @@ -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> comparator) { this.comparator = comparator; } - public void processAndSend(MessageGroup group, MessagingOperations messagingTemplate, MessageChannel outputChannel) { + public Object processMessageGroup(MessageGroup group) { Collection> messages = group.getUnmarked(); if (messages.size() > 0) { List> sorted = new ArrayList>(messages); - Collections.sort(sorted, comparator); - for (Message message : sorted) { - messagingTemplate.send(outputChannel, message); - } + Collections.sort(sorted, this.comparator); + return sorted; } + return null; } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java index 8248c8e903..43b0eecd2b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java @@ -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> messages = Collections.>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> messages = Arrays.>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> messages = Arrays.>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> messages = Arrays.>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> messages = Arrays.>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, diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index 5376d7d3cc..cda94b6b63 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -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 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; } } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java index c4873d1187..508f0839aa 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java @@ -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; } } + } \ No newline at end of file diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index 68e75857d3..b509425d85 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -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(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java index be7212e957..9bcfded2c1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java @@ -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> messages = new ArrayList>(); + private final List> messages = new ArrayList>(); + @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> { - - 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 Matcher> hasPayload(Matcher payloadMatcher) { - return new MessageMatcher(payloadMatcher, CoreMatchers.anything()); - } - - @Factory - public static Matcher> hasHeader(Matcher headerMatcher) { - return new MessageMatcher(CoreMatchers.anything(), headerMatcher); - } - } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java index 721de2ae66..9d43f6a027 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java @@ -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> messagesUpForProcessing = new ArrayList>(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 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 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 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 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 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 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 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 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 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("foo")); group.add(new GenericMessage("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 names); } + public static class GreetingBean implements GreetingService { private String greeting = "hello";