Now enforcing that a message endpoint must contain a handler method (INT-177).

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;

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()) {

spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml

@@ -8,9 +8,7 @@
 		<property name="autoStartup" value="false"/>
 	</bean>
 
-	<bean id="inputChannel" class="org.springframework.integration.channel.SimpleChannel"/>
-
-	<bean id="outputChannel" class="org.springframework.integration.channel.SimpleChannel"/>
+	<bean id="channel" class="org.springframework.integration.channel.SimpleChannel"/>
 
 	<bean id="sourceAdapter" class="org.springframework.integration.adapter.PollingSourceAdapter">
 		<constructor-arg>
@@ -21,7 +19,7 @@
 				<property name="method" value="foo"/>
 			</bean>
 		</constructor-arg>
-		<property name="channel" ref="inputChannel"/>
+		<property name="channel" ref="channel"/>
 	</bean>
 
 	<bean id="targetAdapter" class="org.springframework.integration.adapter.DefaultTargetAdapter">
@@ -37,20 +35,11 @@
 		<constructor-arg ref="targetAdapter"/>
 		<property name="subscription">
 			<bean class="org.springframework.integration.scheduling.Subscription">
-				<constructor-arg ref="outputChannel"/>
+				<constructor-arg ref="channel"/>
 			</bean>
 		</property>
 	</bean>
 
 	<bean id="sink" class="org.springframework.integration.adapter.TestSink"/>
 
-	<bean class="org.springframework.integration.endpoint.DefaultMessageEndpoint">
-		<property name="subscription">
-			<bean class="org.springframework.integration.scheduling.Subscription">
-				<constructor-arg ref="inputChannel"/>
-			</bean>
-		</property>
-		<property name="defaultOutputChannelName" value="outputChannel"/>
-	</bean>
-
 </beans>

spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTestsWithNamespace.xml

@@ -9,15 +9,11 @@
 
 	<si:message-bus auto-startup="false"/>
 
-	<si:channel id="inputChannel"/>
+	<si:channel id="channel"/>
 
-	<si:channel id="outputChannel"/>
+	<si:source-adapter ref="source" method="foo" channel="channel" period="100"/>
 
-	<si:source-adapter ref="source" method="foo" channel="inputChannel" period="100"/>
-
-	<si:target-adapter ref="sink" method="store" channel="outputChannel"/>
-
-	<si:endpoint input-channel="inputChannel" default-output-channel="outputChannel"/>
+	<si:target-adapter ref="sink" method="store" channel="channel"/>
 
 	<bean id="source" class="org.springframework.integration.adapter.TestSource"/>
 

spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml

@@ -11,6 +11,7 @@
 	<bean id="targetChannel" class="org.springframework.integration.channel.SimpleChannel"/>
 
 	<bean id="endpoint" class="org.springframework.integration.endpoint.DefaultMessageEndpoint">
+		<constructor-arg ref="handler"/>
 		<property name="subscription">
 			<bean class="org.springframework.integration.scheduling.Subscription">
 				<constructor-arg ref="sourceChannel"/>
@@ -18,5 +19,7 @@
 		</property>
 		<property name="defaultOutputChannelName" value="targetChannel"/>
 	</bean>
+
+	<bean id="handler" class="org.springframework.integration.handler.TestHandlers" factory-method="echoHandler"/>
 
 </beans>

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 {
+	}
+
 }

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}.
 	 */

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