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

INT-1288, INT-1289: fix some issues with correlation

Browse Source
This commit is contained in:
David Syer
2010-07-28 09:26:43 +00:00
parent 3101611801
commit 6d69c01455
4 changed files with 129 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -154,6 +154,9 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
logger.debug("Handling message with correlationKey ["
+ correlationKey + "]: " + message);
}
if (correlationKey==null) {
throw new IllegalStateException("Null correlation not allowed. Maybe the CorrelationStrategy is failing?");
}
// TODO: INT-1117 - make the lock global?
Object lock = getLock(correlationKey);

1
spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java
View File

@@ -39,7 +39,6 @@ public class CorrelationStrategyAdapter implements CorrelationStrategy {
public CorrelationStrategyAdapter(Object object, Method method) {
Assert.notNull(object, "'object' must not be null");
Assert.notNull(method, "'method' must not be null");
Assert.isTrue(method.getParameterTypes().length == 1, "Method must accept exactly one parameter");
Assert.isTrue(!Void.TYPE.equals(method.getReturnType()), "Method return type must not be void");
this.processor = new MethodInvokingMessageProcessor(object, method);
}

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

@@ -71,8 +71,7 @@ public class CorrelatingMessageHandlerTests {
@Before
public void initializeSubject() {
handler = new CorrelatingMessageHandler(processor, store, correlationStrategy,
ReleaseStrategy);
handler = new CorrelatingMessageHandler(processor, store, correlationStrategy, ReleaseStrategy);
handler.setOutputChannel(outputChannel);
doAnswer(new DoesNothing()).when(processor).processAndSend(isA(SimpleMessageGroup.class),
isA(MessagingTemplate.class), eq(outputChannel));
@@ -94,7 +93,8 @@ public class CorrelatingMessageHandlerTests {
verify(correlationStrategy).getCorrelationKey(message1);
verify(correlationStrategy).getCorrelationKey(message2);
verify(processor).processAndSend(isA(SimpleMessageGroup.class), isA(MessagingTemplate.class), eq(outputChannel));
verify(processor)
.processAndSend(isA(SimpleMessageGroup.class), isA(MessagingTemplate.class), eq(outputChannel));
}
private void verifyLocks(CorrelatingMessageHandler handler, int lockCount) {
@@ -104,8 +104,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).processAndSend(
isA(SimpleMessageGroup.class), isA(MessagingTemplate.class), eq(outputChannel));
String correlationKey = "key";
Message<?> message1 = testMessage(correlationKey, 1, 2);
@@ -114,17 +114,19 @@ public class CorrelatingMessageHandlerTests {
when(correlationStrategy.getCorrelationKey(isA(Message.class))).thenReturn(correlationKey);
handler.handleMessage(message1);
try {
handler.handleMessage(message2);
fail("Expected MessageHandlingException");
} catch (MessageHandlingException e) {
}
catch (MessageHandlingException e) {
assertEquals(0, store.getMessageGroup(correlationKey).size());
}
verify(correlationStrategy).getCorrelationKey(message1);
verify(correlationStrategy).getCorrelationKey(message2);
verify(processor).processAndSend(isA(SimpleMessageGroup.class), isA(MessagingTemplate.class), eq(outputChannel));
verify(processor)
.processAndSend(isA(SimpleMessageGroup.class), isA(MessagingTemplate.class), eq(outputChannel));
}
/*
@@ -161,6 +163,22 @@ public class CorrelatingMessageHandlerTests {
}
@Test
public void testNullCorrelationKey() throws Exception {
final Message<?> message1 = MessageBuilder.withPayload("foo").build();
when(correlationStrategy.getCorrelationKey(isA(Message.class))).thenReturn(null);
try {
handler.handleMessage(message1);
fail("Expected MessageHandlingException");
} catch (MessageHandlingException e) {
Throwable cause = e.getCause();
boolean pass = cause instanceof IllegalStateException && cause.getMessage().toLowerCase().contains("null correlation");
if (!pass) {
throw e;
}
}
}
private Message<?> testMessage(String correlationKey, int sequenceNumber, int sequenceSize) {
return MessageBuilder.withPayload("test" + sequenceNumber).setCorrelationId(correlationKey).setSequenceNumber(
sequenceNumber).setSequenceSize(sequenceSize).build();

100
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java Normal file
View File

@@ -0,0 +1,100 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aggregator;
import static org.junit.Assert.assertEquals;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.util.ReflectionUtils;
/**
* @author Dave Syer
*
*/
public class CorrelationStrategyAdapterTests {
private Message<?> message;
@Before
public void init() {
message = MessageBuilder.withPayload("foo").setHeader("a", "b").setHeader("c", "d").build();
}
@Test
public void testCorrelationStrategyAdapterObjectString() {
CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new SimpleMessageCorrelator(), "getKey");
assertEquals("b", adapter.getCorrelationKey(message));
}
@Test
public void testCorrelationStrategyAdapterObjectMethod() {
CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new SimpleMessageCorrelator(),
ReflectionUtils.findMethod(SimpleMessageCorrelator.class, "getKey", Message.class));
assertEquals("b", adapter.getCorrelationKey(message));
}
@Test
public void testCorrelationStrategyAdapterPojoMethod() {
CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new SimplePojoCorrelator(), "getKey");
assertEquals("foo", adapter.getCorrelationKey(message));
}
@Test
public void testHeaderPojoMethod() {
CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new SimpleHeaderCorrelator(), "getKey");
assertEquals("b", adapter.getCorrelationKey(message));
}
@Test
public void testHeadersPojoMethod() {
CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new MultiHeaderCorrelator(),
ReflectionUtils.findMethod(MultiHeaderCorrelator.class, "getKey", String.class, String.class));
assertEquals("bd", adapter.getCorrelationKey(message));
}
private static class MultiHeaderCorrelator {
@SuppressWarnings("unused")
public String getKey(@Header("a") String header, @Header("c") String other) {
return header + other;
}
}
private static class SimpleHeaderCorrelator {
@SuppressWarnings("unused")
public String getKey(@Header("a") String header) {
return header;
}
}
private static class SimplePojoCorrelator {
@SuppressWarnings("unused")
public String getKey(String message) {
return message;
}
}
private static class SimpleMessageCorrelator {
@SuppressWarnings("unused")
public String getKey(Message<?> message) {
return (String) message.getHeaders().get("a");
}
}
}