diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index aa78446806..aff3971a6d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -1,17 +1,14 @@ /* * Copyright 2002-2010 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. */ package org.springframework.integration.aggregator; @@ -29,8 +26,7 @@ import org.springframework.util.Assert; import java.util.*; /** - * Base class for MessageGroupProcessor implementations that aggregate the group - * of Messages into a single Message. + * Base class for MessageGroupProcessor implementations that aggregate the group of Messages into a single Message. * * @author Iwein Fuld * @author Alexander Peters @@ -48,19 +44,16 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag Assert.notNull(outputChannel, "'outputChannel' must not be null"); Object payload = this.aggregatePayloads(group); Map headers = this.aggregateHeaders(group); - MessageBuilder builder = (payload instanceof Message) - ? MessageBuilder.fromMessage((Message) payload) + MessageBuilder builder = (payload instanceof Message) ? MessageBuilder.fromMessage((Message) payload) : MessageBuilder.withPayload(payload); Message message = builder.copyHeadersIfAbsent(headers).build(); channelTemplate.send(message, outputChannel); } /** - * This default implementation simply returns all headers that have no - * conflicts among the group. An absent header on one or more Messages - * within the group is not considered a conflict. Subclasses may override - * this method with more advanced conflict-resolution strategies if - * necessary. + * This default implementation simply returns all headers that have no conflicts among the group. An absent header + * on one or more Messages within the group is not considered a conflict. Subclasses may override this method with + * more advanced conflict-resolution strategies if necessary. */ protected Map aggregateHeaders(MessageGroup group) { Map aggregatedHeaders = new HashMap(); @@ -75,15 +68,14 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag Object value = currentHeaders.get(key); if (!aggregatedHeaders.containsKey(key)) { aggregatedHeaders.put(key, value); - } - else if (!value.equals(aggregatedHeaders.get(key))) { + } else if (!value.equals(aggregatedHeaders.get(key))) { conflictKeys.add(key); } } } for (String keyToRemove : conflictKeys) { - if (logger.isInfoEnabled()) { - logger.info("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) " + if (logger.isDebugEnabled()) { + logger.debug("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) " + "in MessageGroup with correlation key: " + group.getCorrelationKey()); } aggregatedHeaders.remove(keyToRemove); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 0c979f4beb..fb9ac96fe6 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -100,7 +100,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements public CorrelatingMessageHandler(MessageGroupProcessor processor) { this(processor, new SimpleMessageStore(0), null, null); } - + public void setMessageStore(MessageGroupStore messageStore) { this.messageStore = messageStore; } @@ -164,19 +164,23 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements if (logger.isDebugEnabled()) { logger.debug("Completing group with correlationKey [" + correlationKey + "]"); } - outputProcessor.processAndSend(group, channelTemplate, this.resolveReplyChannel(message, - this.outputChannel)); - if (group.isComplete() || group.getSequenceSize() == 0) { - // The group is complete or else there is no sequence so there is no more state to track - remove(group); - } - else { - // Mark these messages as processed, but do not remove the group from store - mark(group); + try { + outputProcessor.processAndSend(group, channelTemplate, this.resolveReplyChannel(message, + this.outputChannel)); + } finally { + + // Always clean up even if there was an exception processing messages + if (group.isComplete() || group.getSequenceSize() == 0) { + // The group is complete or else there is no sequence so there is no more state to track + remove(group); + } else { + // Mark these messages as processed, but do not remove the group from store + mark(group); + } + } - } - else if (group.isComplete()) { + } else if (group.isComplete()) { // If not releasing any messages the group might still be complete for (Message discard : group.getUnmarked()) { @@ -186,8 +190,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } - } - else { + } else { discardChannel.send(message); } @@ -207,8 +210,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(group.getOne(), this.outputChannel)); remove(group); - } - else { + } else { if (sendPartialResultOnExpiry) { if (logger.isInfoEnabled()) { logger.info("Processing partially complete messages for key [" + correlationKey + "] to: " @@ -216,8 +218,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(group.getOne(), this.outputChannel)); - } - else { + } else { if (logger.isInfoEnabled()) { logger.info("Discarding partially complete messages for key [" + correlationKey + "] to: " + discardChannel); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java index 0d777eb4e4..952eeb26cc 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java @@ -5,21 +5,18 @@ import org.springframework.integration.core.MessageChannel; import org.springframework.integration.store.MessageGroup; /** - * A processor for correlated groups of messages. When a message group is complete it is passed to the - * processor by e.g. the CorrelatingMessageHandler. - * + * A processor for correlated groups of messages. + * * @author Iwein Fuld * @see org.springframework.integration.aggregator.CorrelatingMessageHandler */ public interface MessageGroupProcessor { - /** - * Processed the given group and sends the resulting message(s) to the output channel using the channelTemplate. - * Implementations are free to send as little or as many messages based on the invocation as needed. For example the - * DefaultAggregatingMessageGroupProcessor will send only a single message containing a collection of all messages - * in the group, where the resequencing equivalent strategy will send all messages in the group individually. - */ - void processAndSend(MessageGroup group, - MessageChannelTemplate channelTemplate, MessageChannel outputChannel - ); + /** + * Process the given group and send the resulting message(s) to the output channel using the channel template. + * Implementations are free to send as little 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. + */ + void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel); } \ No newline at end of file diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java similarity index 98% rename from org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java rename to org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java index e0759119f0..fd8ff92924 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java @@ -28,7 +28,7 @@ import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; -public class CorrelatingMessageHandlerIntegrationTest { +public class CorrelatingMessageHandlerIntegrationTests { private MessageGroupStore store = new SimpleMessageStore(100); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index e7f4abba10..7f340cdbcb 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -17,6 +17,7 @@ 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; @@ -34,11 +35,13 @@ 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.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.store.SimpleMessageStore; @@ -80,16 +83,13 @@ public class CorrelatingMessageHandlerTests { String correlationKey = "key"; Message message1 = testMessage(correlationKey, 1, 2); Message message2 = testMessage(correlationKey, 2, 2); - List> storedMessages = new ArrayList>(); when(correlationStrategy.getCorrelationKey(isA(Message.class))).thenReturn(correlationKey); handler.handleMessage(message1); - storedMessages.add(message1); verifyLocks(handler, 1); handler.handleMessage(message2); - storedMessages.add(message2); verifyLocks(handler, 0); // lock is removed when group is complete verify(correlationStrategy).getCorrelationKey(message1); @@ -101,6 +101,32 @@ public class CorrelatingMessageHandlerTests { assertEquals(lockCount, ((Map) ReflectionTestUtils.getField(handler, "locks")).size()); } + @Test + public void bufferCompletesWithException() throws Exception { + + doAnswer(new ThrowsException(new RuntimeException("Planned test exception"))).when(processor).processAndSend(isA(SimpleMessageGroup.class), + isA(MessageChannelTemplate.class), eq(outputChannel)); + + String correlationKey = "key"; + Message message1 = testMessage(correlationKey, 1, 2); + Message message2 = testMessage(correlationKey, 2, 2); + + when(correlationStrategy.getCorrelationKey(isA(Message.class))).thenReturn(correlationKey); + + handler.handleMessage(message1); + + try { + handler.handleMessage(message2); + fail("Expected MessageHandlingException"); + } catch (MessageHandlingException e) { + assertEquals(0, store.getMessageGroup(correlationKey).size()); + } + + verify(correlationStrategy).getCorrelationKey(message1); + verify(correlationStrategy).getCorrelationKey(message2); + verify(processor).processAndSend(isA(SimpleMessageGroup.class), isA(MessageChannelTemplate.class), eq(outputChannel)); + } + /* * The next test verifies that when pruning happens after the completing message arrived, but before the group was * processed locking prevents forced completion and the group completes normally.