diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index 8b18198c50..80ef182b8b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -58,7 +58,6 @@ import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerChain; import org.springframework.integration.handler.config.DefaultMessageHandlerCreator; import org.springframework.integration.handler.config.MessageHandlerCreator; -import org.springframework.integration.message.Message; import org.springframework.integration.router.AggregatingMessageHandler; import org.springframework.integration.router.CompletionStrategyAdapter; import org.springframework.integration.router.config.AggregatorMessageHandlerCreator; @@ -120,6 +119,9 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor } String defaultOutputChannelName = endpointAnnotation.defaultOutput(); MessageHandlerChain handlerChain = this.createHandlerChain(bean, defaultOutputChannelName); + if (handlerChain == null) { + throw new ConfigurationException("@MessageEndpoint has no handler method"); + } DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handlerChain); this.configureInput(bean, beanName, endpointAnnotation, endpoint); if (StringUtils.hasText(defaultOutputChannelName)) { @@ -136,13 +138,6 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity()); endpoint.setConcurrencyPolicy(concurrencyPolicy); } - if (endpoint.getHandler() == null) { - endpoint.setHandler(new MessageHandler() { - public Message handle(Message message) { - return message; - } - }); - } this.configureCompletionStrategy(bean, endpoint); this.messageBus.registerEndpoint(beanName + "-endpoint", endpoint); return bean; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index 9da66a94c2..3a5e319e8e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -238,6 +238,9 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA } public final Message handle(Message message) { + if (this.handler == null) { + throw new ConfigurationException("endpoint has no 'handler'"); + } if (logger.isDebugEnabled()) { logger.debug("endpoint '" + this + "' handling message: " + message); } @@ -249,21 +252,6 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA throw new MessageSelectorRejectedException(message); } } - if (this.handler == null) { - if (this.defaultOutputChannelName == null) { - throw new ConfigurationException( - "endpoint must have either a 'handler' or 'defaultOutputChannelName'"); - } - MessageChannel outputChannel = this.channelRegistry.lookupChannel(this.defaultOutputChannelName); - if (logger.isDebugEnabled()) { - logger.debug("endpoint '" + this + "' sending to output channel '" + outputChannel + "', message: " + message); - } - if (!outputChannel.send(message, this.replyTimeout)) { - this.errorHandler.handle(new MessageDeliveryException(message, - "unable to send output message within alloted timeout of " + replyTimeout + " milliseconds")); - } - return null; - } try { Message replyMessage = this.handler.handle(message); if (replyMessage != null) { @@ -317,8 +305,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA MessageChannel replyChannel = resolveReplyChannel(originalMessageHeader); if (replyChannel == null) { throw new MessageHandlingException(replyMessage, - "Unable to determine reply channel for message. " + - "Provide a 'replyChannel' or 'replyChannelName' in the message header " + + "Unable to determine reply channel for message. Provide a 'returnAddress' in the message header " + "or a 'defaultOutputChannelName' on the message endpoint."); } if (logger.isDebugEnabled()) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml index 96eec93029..87d9349f94 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml @@ -8,9 +8,7 @@ - - - + @@ -21,7 +19,7 @@ - + @@ -37,20 +35,11 @@ - + - - - - - - - - - diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTestsWithNamespace.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTestsWithNamespace.xml index a8125fa0e8..e7700d7a05 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTestsWithNamespace.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTestsWithNamespace.xml @@ -9,15 +9,11 @@ - + - + - - - - - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml index 69b77f260d..0c73be72b2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml @@ -11,6 +11,7 @@ + @@ -18,5 +19,7 @@ + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java index 151030bb60..1a5ab81d2f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java @@ -28,6 +28,7 @@ import org.junit.Test; import org.springframework.aop.framework.ProxyFactory; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.ConfigurationException; import org.springframework.integration.annotation.Concurrency; import org.springframework.integration.annotation.DefaultOutput; import org.springframework.integration.annotation.Handler; @@ -93,6 +94,7 @@ public class MessageEndpointAnnotationPostProcessorTests { messageBus.registerChannel("testChannel", testChannel); MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); PolledAnnotationTestBean testBean = new PolledAnnotationTestBean(); postProcessor.postProcessAfterInitialization(testBean, "testBean"); messageBus.start(); @@ -108,6 +110,7 @@ public class MessageEndpointAnnotationPostProcessorTests { messageBus.registerChannel("testChannel", testChannel); MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); CountDownLatch latch = new CountDownLatch(1); DefaultOutputAnnotationTestBean testBean = new DefaultOutputAnnotationTestBean(latch); postProcessor.postProcessAfterInitialization(testBean, "testBean"); @@ -124,6 +127,7 @@ public class MessageEndpointAnnotationPostProcessorTests { MessageBus messageBus = new MessageBus(); MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); ConcurrencyAnnotationTestBean testBean = new ConcurrencyAnnotationTestBean(); postProcessor.postProcessAfterInitialization(testBean, "testBean"); DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) messageBus.lookupEndpoint("testBean-endpoint"); @@ -144,6 +148,7 @@ public class MessageEndpointAnnotationPostProcessorTests { MessageBus messageBus = new MessageBus(); MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); ChannelRegistryAwareTestBean testBean = new ChannelRegistryAwareTestBean(); assertNull(testBean.getChannelRegistry()); postProcessor.postProcessAfterInitialization(testBean, "testBean"); @@ -285,6 +290,18 @@ public class MessageEndpointAnnotationPostProcessorTests { assertNull(output.receive(500)); } + @Test(expected=ConfigurationException.class) + public void testEndpointWithNoHandlerMethod() { + MessageBus messageBus = new MessageBus(); + SimpleChannel testChannel = new SimpleChannel(); + messageBus.registerChannel("testChannel", testChannel); + MessageEndpointAnnotationPostProcessor postProcessor = + new MessageEndpointAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + AnnotatedEndpointWithNoHandlerMethod endpoint = new AnnotatedEndpointWithNoHandlerMethod(); + postProcessor.postProcessAfterInitialization(endpoint, "endpoint"); + } + @MessageEndpoint(defaultOutput="testChannel") private static class PolledAnnotationTestBean { @@ -293,6 +310,11 @@ public class MessageEndpointAnnotationPostProcessorTests { public String poller() { return "test"; } + + @Handler + public Message handle(Message message) { + return message; + } } @@ -312,6 +334,11 @@ public class MessageEndpointAnnotationPostProcessorTests { return this.messageText; } + @Handler + public Message handle(Message message) { + return message; + } + @DefaultOutput public void countdown(String input) { this.messageText = input; @@ -323,6 +350,11 @@ public class MessageEndpointAnnotationPostProcessorTests { @MessageEndpoint(input="inputChannel") @Concurrency(coreSize=17, maxSize=42, keepAliveSeconds=123, queueCapacity=11) private static class ConcurrencyAnnotationTestBean { + + @Handler + public Message handle(Message message) { + return null; + } } @@ -338,6 +370,11 @@ public class MessageEndpointAnnotationPostProcessorTests { public ChannelRegistry getChannelRegistry() { return this.channelRegistry; } + + @Handler + public Message handle(Message message) { + return null; + } } @@ -369,4 +406,9 @@ public class MessageEndpointAnnotationPostProcessorTests { } } + + @MessageEndpoint(input="testChannel") + private static class AnnotatedEndpointWithNoHandlerMethod { + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java index dde3ff3419..996759dd00 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java @@ -40,6 +40,17 @@ public abstract class TestHandlers { }; } + /** + * Create a {@link MessageHandler} that simply returns the {@link Message} it receives. + */ + public final static MessageHandler echoHandler() { + return new MessageHandler() { + public Message handle(Message message) { + return message; + } + }; + } + /** * Create a {@link MessageHandler} that throws a {@link MessageHandlerRejectedExecutionException}. */ diff --git a/spring-integration-samples/src/main/java/org/springframework/integration/samples/quote/QuotePublisher.java b/spring-integration-samples/src/main/java/org/springframework/integration/samples/quote/QuotePublisher.java index 69f787ee22..9b094c1226 100644 --- a/spring-integration-samples/src/main/java/org/springframework/integration/samples/quote/QuotePublisher.java +++ b/spring-integration-samples/src/main/java/org/springframework/integration/samples/quote/QuotePublisher.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * Copyright 2002-2008 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. @@ -20,6 +20,7 @@ import java.math.BigDecimal; import java.math.RoundingMode; import java.util.Random; +import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Polled; @@ -30,17 +31,18 @@ import org.springframework.integration.annotation.Polled; public class QuotePublisher { @Polled(period=300) - public Quote getQuote() { - BigDecimal price = new BigDecimal(new Random().nextDouble() * 100); - return new Quote(generateTicker(), price.setScale(2, RoundingMode.HALF_EVEN)); - } - - private String generateTicker() { + public String generateTicker() { char[] chars = new char[3]; for (int i = 0; i < 3; i++) { chars[i] = (char) (new Random().nextInt(25) + 65); } - return new String(chars); + return new String(chars); + } + + @Handler + public Quote getQuote(String ticker) { + BigDecimal price = new BigDecimal(new Random().nextDouble() * 100); + return new Quote(ticker, price.setScale(2, RoundingMode.HALF_EVEN)); } }