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

INT-1075: add finally block to CMH

Browse Source
This commit is contained in:
David Syer
2010-05-07 06:55:47 +00:00
parent 5fcaaf246b
commit 326a23d6fe
5 changed files with 75 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
View File

@@ -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<String, Object> 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<String, Object> aggregateHeaders(MessageGroup group) {
Map<String, Object> aggregatedHeaders = new HashMap<String, Object>();
@@ -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);

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

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

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

@@ -5,21 +5,18 @@ import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.store.MessageGroup;
/**
* A processor for <i>correlated</i> groups of messages. When a message group is <i>complete</i> it is passed to the
* processor by e.g. the CorrelatingMessageHandler.
*
* A processor for <i>correlated</i> 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);
}

2
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java → org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java
View File

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

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

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