From 6d69c01455a1a3968985985389d1b6700da1a403 Mon Sep 17 00:00:00 2001 From: David Syer Date: Wed, 28 Jul 2010 09:26:43 +0000 Subject: [PATCH] INT-1288, INT-1289: fix some issues with correlation --- .../aggregator/CorrelatingMessageHandler.java | 3 + .../CorrelationStrategyAdapter.java | 1 - .../CorrelatingMessageHandlerTests.java | 34 ++++-- .../CorrelationStrategyAdapterTests.java | 100 ++++++++++++++++++ 4 files changed, 129 insertions(+), 9 deletions(-) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java 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 b975adca1c..4b759a960b 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 @@ -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); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java index 1297d269fa..bed28d78f8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java @@ -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); } 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 aaeaff5236..b819968853 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 @@ -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(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java new file mode 100644 index 0000000000..958415bbfc --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java @@ -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"); + } + } + +}