Enhanced annotation and namespace support.

spring-eai-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java

@@ -50,7 +50,7 @@ public class ConsumerPolicy {
 
 	private int initialDelay = 0;
 
-	private int period = -1;
+	private int period = 5;
 
 	private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
 

spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java

@@ -58,6 +58,8 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life
 
 	private ScheduledThreadPoolExecutor dispatcherExecutor;
 
+	private MessageChannel invalidMessageChannel;
+
 	private boolean autoCreateChannels;
 
 	private boolean running;
@@ -76,6 +78,10 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life
 		this.autoCreateChannels = autoCreateChannels;
 	}
 
+	public void setInvalidMessageChannel(MessageChannel invalidMessageChannel) {
+		this.invalidMessageChannel = invalidMessageChannel;
+	}
+
 	@SuppressWarnings("unchecked")
 	private void registerChannels(ApplicationContext context) {
 		Map<String, MessageChannel> channelBeans =
@@ -121,6 +127,10 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life
 		return this.channels.get(channelName);
 	}
 
+	public MessageChannel getInvalidMessageChannel() {
+		return this.invalidMessageChannel;
+	}
+
 	public void registerChannel(String name, MessageChannel channel) {
 		this.channels.put(name, channel);
 	}
@@ -162,6 +172,10 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life
 		this.dispatcherTasks.add(dispatcherTask);
 		if (this.isRunning()) {
 			scheduleDispatcherTask(dispatcherTask);
+			if (this.logger.isInfoEnabled()) {
+				logger.info("scheduled dispatcher task: channel='" +
+						channelName + "' endpoint='" + endpointName + "'");
+			}
 		}
 	}
 

spring-eai-core/src/main/java/org/springframework/integration/channel/ChannelMapping.java

@@ -25,4 +25,6 @@ public interface ChannelMapping {
 
 	MessageChannel getChannel(String channelName);
 
+	MessageChannel getInvalidMessageChannel();
+
 }

spring-eai-core/src/main/java/org/springframework/integration/config/AbstractChannelAdapterParser.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2002-2007 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.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.RuntimeBeanReference;
+import org.springframework.beans.factory.parsing.BeanComponentDefinition;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.beans.factory.xml.BeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter;
+import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter;
+import org.springframework.util.StringUtils;
+
+/**
+ * Base parser for inbound and outbound channel adapters.
+ * 
+ * @author Mark Fisher
+ */
+public abstract class AbstractChannelAdapterParser implements BeanDefinitionParser {
+
+	private static final String ID_ATTRIBUTE = "id";
+
+	private static final String REF_ATTRIBUTE = "ref";
+
+	private static final String METHOD_ATTRIBUTE = "method";
+
+
+	public BeanDefinition parse(Element element, ParserContext parserContext) {
+		RootBeanDefinition adapterDef = null;
+		if (this.isInbound()) {
+			adapterDef = new RootBeanDefinition(InboundMethodInvokingChannelAdapter.class);
+		}
+		else {
+			adapterDef = new RootBeanDefinition(OutboundMethodInvokingChannelAdapter.class);
+		}
+		adapterDef.setSource(parserContext.extractSource(element));
+		String ref = element.getAttribute(REF_ATTRIBUTE);
+		String method = element.getAttribute(METHOD_ATTRIBUTE);
+		if (!StringUtils.hasText(ref) || !StringUtils.hasText(method)) {
+		}
+		adapterDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref));
+		adapterDef.getPropertyValues().addPropertyValue("method", method);
+		String beanName = element.getAttribute(ID_ATTRIBUTE);
+		if (!StringUtils.hasText(beanName)) {
+			beanName = parserContext.getReaderContext().generateBeanName(adapterDef);
+		}
+		parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDef, beanName));
+		return adapterDef;
+	}
+
+	protected abstract boolean isInbound();
+
+}

spring-eai-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2002-2007 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.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.RuntimeBeanReference;
+import org.springframework.beans.factory.parsing.BeanComponentDefinition;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.beans.factory.xml.BeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+
+/**
+ * @author Mark Fisher
+ */
+public class AnnotationDrivenParser implements BeanDefinitionParser {
+
+	private static final String PUBLISHER_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
+			"internal.PublisherAnnotationPostProcessor";
+
+	private static final String SUBSCRIBER_ANNOTATION_POST_PROCESSOR_BEAN_NAME = 
+			"internal.SubscriberAnnotationPostProcessor";
+
+
+	public BeanDefinition parse(Element element, ParserContext parserContext) {
+		this.createPublisherPostProcessor(parserContext);
+		this.createSubscriberPostProcessor(parserContext);
+		return null;
+	}
+
+	private void createPublisherPostProcessor(ParserContext parserContext) {
+		BeanDefinition bd = new RootBeanDefinition(PublisherAnnotationPostProcessor.class);
+		bd.getPropertyValues().addPropertyValue("channelMapping",
+				new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
+		BeanComponentDefinition bcd = new BeanComponentDefinition(
+				bd, PUBLISHER_ANNOTATION_POST_PROCESSOR_BEAN_NAME);
+		parserContext.registerBeanComponent(bcd);
+	}
+
+	private void createSubscriberPostProcessor(ParserContext parserContext) {
+		BeanDefinition bd = new RootBeanDefinition(SubscriberAnnotationPostProcessor.class);
+		bd.getPropertyValues().addPropertyValue("messageBus",
+				new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
+		BeanComponentDefinition bcd = new BeanComponentDefinition(
+				bd, SUBSCRIBER_ANNOTATION_POST_PROCESSOR_BEAN_NAME);
+		parserContext.registerBeanComponent(bcd);
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/config/InboundChannelAdapterParser.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2002-2007 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.config;
+
+/**
+ * @author Mark Fisher
+ */
+public class InboundChannelAdapterParser extends AbstractChannelAdapterParser {
+
+	@Override
+	protected boolean isInbound() {
+		return true;
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java

@@ -19,14 +19,18 @@ package org.springframework.integration.config;
 import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
 
 /**
- * Handler for the integration namespace.
+ * Namespace handler for the integration namespace.
  * 
  * @author Mark Fisher
  */
 public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {
 
 	public void init() {
+		registerBeanDefinitionParser("message-bus", new MessageBusParser());
+		registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenParser());
 		registerBeanDefinitionParser("channel", new ChannelParser());
+		registerBeanDefinitionParser("inbound-channel-adapter", new InboundChannelAdapterParser());
+		registerBeanDefinitionParser("outbound-channel-adapter", new OutboundChannelAdapterParser());
 		registerBeanDefinitionParser("endpoint", new EndpointParser());
 	}
 

spring-eai-core/src/main/java/org/springframework/integration/config/MessageBusParser.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2002-2007 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.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.BeanDefinitionStoreException;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.bus.MessageBus;
+
+/**
+ * Parser for the <em>message-bus</em> element of the integration namespace.
+ * 
+ * @author Mark Fisher
+ */
+public class MessageBusParser extends AbstractSingleBeanDefinitionParser {
+
+	public static final String MESSAGE_BUS_BEAN_NAME = "internal.messageBus";
+
+	@Override
+	protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
+			throws BeanDefinitionStoreException {
+		return MESSAGE_BUS_BEAN_NAME;
+	}
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return MessageBus.class;
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/config/OutboundChannelAdapterParser.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2002-2007 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.config;
+
+/**
+ * @author Mark Fisher
+ */
+public class OutboundChannelAdapterParser extends AbstractChannelAdapterParser {
+
+	@Override
+	protected boolean isInbound() {
+		return false;
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.aop;
+package org.springframework.integration.config;
 
 import java.lang.annotation.Annotation;
 
@@ -25,6 +25,8 @@ import org.springframework.aop.support.AopUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.integration.aop.Publisher;
+import org.springframework.integration.aop.PublisherAnnotationAdvisor;
 import org.springframework.integration.channel.ChannelMapping;
 import org.springframework.util.Assert;
 

spring-eai-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2002-2007 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.config;
+
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Method;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.aop.framework.Advised;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.endpoint.GenericMessageEndpoint;
+import org.springframework.integration.endpoint.MessageHandlerAdapter;
+import org.springframework.integration.endpoint.annotation.Subscriber;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * A {@link BeanPostProcessor} that creates a method-invoking handler adapter
+ * when it discovers methods annotated with {@link Subscriber @Subscriber}.
+ * 
+ * @author Mark Fisher
+ */
+public class SubscriberAnnotationPostProcessor implements BeanPostProcessor {
+
+	private Log logger = LogFactory.getLog(this.getClass());
+
+	private Class<? extends Annotation> subscriberAnnotationType = Subscriber.class;
+
+	private String channelNameAttribute = "channel";
+
+	private MessageBus messageBus;
+
+
+	public void setSubscriberAnnotationType(Class<? extends Annotation> subscriberAnnotationType) {
+		Assert.notNull(subscriberAnnotationType, "subscriberAnnotationType must not be null");
+		this.subscriberAnnotationType = subscriberAnnotationType;
+	}
+
+	public void setChannelNameAttribute(String channelNameAttribute) {
+		Assert.notNull(channelNameAttribute, "channelNameAttribute must not be null");
+		this.channelNameAttribute = channelNameAttribute;
+	}
+
+	public void setMessageBus(MessageBus messageBus) {
+		Assert.notNull(messageBus, "messageBus must not be null");
+		this.messageBus = messageBus;
+	}
+
+	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+		return bean;
+	}
+
+	public Object postProcessAfterInitialization(final Object bean, String beanName) throws BeansException {
+		final Class<?> targetClass = bean instanceof Advised ? 
+				((Advised) bean).getTargetSource().getTargetClass() : bean.getClass();
+		if (targetClass == null) {
+			return bean;
+		}
+		ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() {
+			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+				Annotation annotation = method.getAnnotation(subscriberAnnotationType);
+				if (annotation != null) {
+					String channelName = (String) AnnotationUtils.getValue(annotation, channelNameAttribute);
+					MessageHandlerAdapter adapter = new MessageHandlerAdapter();
+					adapter.setMethod(method.getName());
+					adapter.setObject(bean);
+					adapter.afterPropertiesSet();
+					GenericMessageEndpoint endpoint = new GenericMessageEndpoint();
+					endpoint.setInputChannelName(channelName);
+					endpoint.setChannelMapping(messageBus);
+					endpoint.setHandler(adapter);
+					String endpointName = ClassUtils.getShortNameAsProperty(targetClass) + 
+							"-" + method.getName() + "-endpoint";
+					messageBus.registerEndpoint(endpointName, endpoint);
+					if (logger.isInfoEnabled()) {
+						logger.info("registered endpoint: " + endpointName);
+					}
+				}
+			}
+		});
+		return bean;
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd

@@ -17,6 +17,28 @@
 		]]></xsd:documentation>
 	</xsd:annotation>
 
+	<xsd:element name="message-bus">
+		<xsd:complexType>
+			<xsd:annotation>
+				<xsd:documentation>
+	Defines a message bus.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:attribute name="auto-create-channels" type="xsd:boolean"/>
+			<xsd:attribute name="invalid-message-channel" type="xsd:string"/>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="annotation-driven">
+		<xsd:complexType>
+			<xsd:annotation>
+				<xsd:documentation>
+	Enables the publisher annotation post-processor.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:complexType>
+	</xsd:element>
+
 	<xsd:element name="channel">
 		<xsd:complexType>
 			<xsd:annotation>
@@ -32,6 +54,38 @@
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="inbound-channel-adapter">
+		<xsd:complexType>
+			<xsd:annotation>
+				<xsd:documentation>
+	Defines an inbound channel adapter.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:complexContent>
+				<xsd:extension base="beans:identifiedType">
+					<xsd:attribute name="ref" type="xsd:string" use="required"/>
+					<xsd:attribute name="method" type="xsd:string" use="required"/>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="outbound-channel-adapter">
+		<xsd:complexType>
+			<xsd:annotation>
+				<xsd:documentation>
+	Defines an outbound channel adapter.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:complexContent>
+				<xsd:extension base="beans:identifiedType">
+					<xsd:attribute name="ref" type="xsd:string" use="required"/>
+					<xsd:attribute name="method" type="xsd:string" use="required"/>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
 	<xsd:element name="endpoint">
 		<xsd:complexType>
 			<xsd:annotation>

spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java

@@ -47,7 +47,7 @@ public class GenericMessageEndpoint implements MessageEndpoint {
 
 	private ChannelMapping channelMapping;
 
-	private ConsumerPolicy consumerPolicy;
+	private ConsumerPolicy consumerPolicy = new ConsumerPolicy();
 
 
 	/**
@@ -94,7 +94,7 @@ public class GenericMessageEndpoint implements MessageEndpoint {
 	}
 
 
-	public void messageReceived(Message message) {
+	public void messageReceived(Message<?> message) {
 		if (this.handler == null) {
 			if (this.defaultOutputChannelName == null) {
 				throw new MessagingConfigurationException(
@@ -104,7 +104,7 @@ public class GenericMessageEndpoint implements MessageEndpoint {
 			replyChannel.send(message);
 			return;
 		}
-		Message replyMessage = handler.handle(message);
+		Message<?> replyMessage = handler.handle(message);
 		if (replyMessage != null) {
 			MessageChannel replyChannel = this.resolveReplyChannel(message);
 			if (replyChannel == null) {
@@ -116,7 +116,7 @@ public class GenericMessageEndpoint implements MessageEndpoint {
 		}
 	}
 
-	private MessageChannel resolveReplyChannel(Message message) {
+	private MessageChannel resolveReplyChannel(Message<?> message) {
 		if (this.channelMapping == null) {
 			return null;
 		}

spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java

@@ -37,6 +37,6 @@ public interface MessageEndpoint {
 
 	void setChannelMapping(ChannelMapping channelMapping);
 
-	void messageReceived(Message message);
+	void messageReceived(Message<?> message);
 
 }

spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/AnnotationHandlerCreator.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2002-2007 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.endpoint.annotation;
+
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Method;
+
+import org.springframework.integration.handler.MessageHandler;
+
+/**
+ * @author Mark Fisher
+ */
+public interface AnnotationHandlerCreator {
+
+	MessageHandler createHandler(Object object, Method method, Annotation annotation);
+
+}

spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/DefaultAnnotationHandlerCreator.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2002-2007 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.endpoint.annotation;
+
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Method;
+
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.annotation.Order;
+import org.springframework.integration.endpoint.MessageHandlerAdapter;
+import org.springframework.integration.handler.MessageHandler;
+
+/**
+ * @author Mark Fisher
+ */
+public class DefaultAnnotationHandlerCreator implements AnnotationHandlerCreator {
+
+	public MessageHandler createHandler(Object object, Method method, Annotation annotation) {
+		MessageHandlerAdapter<Object> adapter = new MessageHandlerAdapter<Object>();
+		adapter.setObject(object);
+		adapter.setMethod(method.getName());
+		Order orderAnnotation = (Order) AnnotationUtils.getAnnotation(method, Order.class);
+		if (orderAnnotation != null) {
+			adapter.setOrder(orderAnnotation.value());
+		}
+		adapter.afterPropertiesSet();
+		return adapter;
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/EndpointAnnotationPostProcessor.java

@@ -0,0 +1,172 @@
+/*
+ * Copyright 2002-2007 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.endpoint.annotation;
+
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.core.OrderComparator;
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.integration.bus.ConsumerPolicy;
+import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.endpoint.GenericMessageEndpoint;
+import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter;
+import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter;
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.handler.MessageHandlerChain;
+import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * A {@link BeanPostProcessor} implementation that generates endpoints for
+ * classes annotated with {@link MessageEndpoint @MessageEndpoint}.
+ * 
+ * @author Mark Fisher
+ */
+public class EndpointAnnotationPostProcessor implements BeanPostProcessor, InitializingBean {
+
+	private Map<Class<? extends Annotation>, AnnotationHandlerCreator> handlerCreators =
+			new ConcurrentHashMap<Class<? extends Annotation>, AnnotationHandlerCreator>();
+
+	private MessageBus messageBus;
+
+
+	public void setMessageBus(MessageBus messageBus) {
+		Assert.notNull(messageBus, "messageBus must not be null");
+		this.messageBus = messageBus;
+	}
+
+	public void setCustomHandlerCreators(
+			Map<Class<? extends Annotation>, AnnotationHandlerCreator> customHandlerCreators) {
+		for (Map.Entry<Class<? extends Annotation>, AnnotationHandlerCreator> entry : customHandlerCreators.entrySet()) {
+			this.handlerCreators.put(entry.getKey(), entry.getValue());
+		}
+	}
+
+	public void afterPropertiesSet() {
+		Assert.notNull(this.messageBus, "messageBus is required");
+		this.handlerCreators.put(Handler.class, new DefaultAnnotationHandlerCreator());
+	}
+
+	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+		return bean;
+	}
+
+	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+		MessageEndpoint endpointAnnotation = bean.getClass().getAnnotation(MessageEndpoint.class);
+		if (endpointAnnotation == null) {
+			return bean;
+		}
+		GenericMessageEndpoint endpoint = new GenericMessageEndpoint();
+		this.configureInputChannel(bean, beanName, endpointAnnotation, endpoint);
+		this.configureDefaultOutputChannel(bean, beanName, endpointAnnotation, endpoint);
+		MessageHandlerChain handlerChain = this.createHandlerChain(bean);
+		if (handlerChain != null) {
+			endpoint.setHandler(handlerChain);
+		}
+		this.messageBus.registerEndpoint(beanName, endpoint);
+		return endpoint;
+	}
+
+	private void configureInputChannel(final Object bean, final String beanName,
+			MessageEndpoint annotation, final GenericMessageEndpoint endpoint) {
+		String channelName = annotation.input();
+		if (StringUtils.hasText(channelName)) {
+			endpoint.setInputChannelName(channelName);
+			ConsumerPolicy consumerPolicy = new ConsumerPolicy();
+			consumerPolicy.setPeriod(annotation.pollPeriod());
+			endpoint.setConsumerPolicy(consumerPolicy);
+			return;
+		}
+		ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
+			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+				Annotation annotation = AnnotationUtils.getAnnotation(method, Polled.class);
+				if (annotation != null) {
+					InboundMethodInvokingChannelAdapter<Object> adapter = new InboundMethodInvokingChannelAdapter<Object>();
+					adapter.setObject(bean);
+					adapter.setMethod(method.getName());
+					adapter.afterPropertiesSet();
+					String channelName = beanName + "-inputChannel";
+					messageBus.registerChannel(channelName, adapter);
+					endpoint.setInputChannelName(channelName);
+					return;
+				}
+			}
+		});
+	}
+
+	private void configureDefaultOutputChannel(final Object bean, final String beanName,
+			final MessageEndpoint annotation, final GenericMessageEndpoint endpoint) {
+		String channelName = annotation.defaultOutput();
+		if (StringUtils.hasText(channelName)) {
+			endpoint.setDefaultOutputChannelName(channelName);
+			return;
+		}
+		ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
+			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+				Annotation annotation = AnnotationUtils.getAnnotation(method, DefaultOutput.class);
+				if (annotation != null) {
+					OutboundMethodInvokingChannelAdapter<Object> adapter = new OutboundMethodInvokingChannelAdapter<Object>();
+					adapter.setObject(bean);
+					adapter.setMethod(method.getName());
+					adapter.afterPropertiesSet();
+					String channelName = beanName + "-defaultOutputChannel";
+					messageBus.registerChannel(channelName, adapter);
+					endpoint.setDefaultOutputChannelName(channelName);
+					return;
+				}
+			}
+		});
+	}
+
+	@SuppressWarnings("unchecked")
+	private MessageHandlerChain createHandlerChain(final Object bean) {
+		final List<MessageHandler> handlers = new ArrayList<MessageHandler>();
+		ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
+			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+				for (Class<? extends Annotation> annotationType : handlerCreators.keySet()) {
+					Annotation annotation = AnnotationUtils.getAnnotation(method, annotationType);
+					if (annotation != null) {
+						MessageHandler handler = handlerCreators.get(annotationType).createHandler(bean, method, annotation);
+						if (handler != null) {
+							handlers.add(handler);
+						}
+					}
+				}
+			}
+		});
+		if (handlers.size() > 0) {
+			MessageHandlerChain handlerChain = new MessageHandlerChain();
+			Collections.sort(handlers, new OrderComparator());
+			for (MessageHandler handler : handlers) {
+				handlerChain.add(handler);
+			}
+			return handlerChain;
+		}
+		return null;
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/Subscriber.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2002-2007 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.endpoint.annotation;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Indicates that a method-invoking handler adapter should delegate to this
+ * method.
+ * 
+ * @author Mark Fisher
+ */
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+@Inherited
+@Documented
+public @interface Subscriber {
+
+	String channel();
+
+}

spring-eai-core/src/main/java/org/springframework/integration/handler/MessageHandler.java

@@ -27,6 +27,6 @@ import org.springframework.integration.message.Message;
  */
 public interface MessageHandler {
 
-	Message handle(Message message);
+	Message<?> handle(Message<?> message);
 
 }

spring-eai-core/src/main/java/org/springframework/integration/message/GenericMessage.java

@@ -27,18 +27,18 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public abstract class AbstractMessage implements Message {
+public class GenericMessage<T> implements Message<T> {
 
 	private Object id;
 
 	private MessageHeader header = new MessageHeader();
 
-	private Object payload;
+	private T payload;
 
 	private ReentrantLock lock;
 
 
-	protected AbstractMessage(Object id, Object payload) {
+	public GenericMessage(Object id, T payload) {
 		Assert.notNull(id, "id must not be null");
 		Assert.notNull(payload, "payload must not be null");
 		this.id = id;
@@ -57,11 +57,11 @@ public abstract class AbstractMessage implements Message {
 		this.header = header;
 	}
 
-	public Object getPayload() {
+	public T getPayload() {
 		return this.payload;
 	}
 
-	protected void setPayload(Object newPayload) {
+	protected void setPayload(T newPayload) {
 		this.payload = newPayload;
 	}
 
@@ -85,7 +85,8 @@ public abstract class AbstractMessage implements Message {
 	public void transformPayload(ObjectTransformer transformer) {
 		this.lock();
 		try {
-			this.setPayload(transformer.transform(this.getPayload()));
+			// TODO: remove this method (probably) or parameterize transformer
+			this.setPayload((T) transformer.transform(this.getPayload()));
 		}
 		finally {
 			this.unlock();

spring-eai-core/src/main/java/org/springframework/integration/message/Message.java

@@ -23,13 +23,13 @@ import org.springframework.integration.transformer.ObjectTransformer;
  * 
  * @author Mark Fisher
  */
-public interface Message {
+public interface Message<T> {
 
 	Object getId();
 
 	MessageHeader getHeader();
 
-	Object getPayload();
+	T getPayload();
 
 	void transformPayload(ObjectTransformer transformer);
 

spring-eai-core/src/main/java/org/springframework/integration/message/MessageMapper.java

@@ -21,16 +21,16 @@ package org.springframework.integration.message;
  * 
  * @author Mark Fisher
  */
-public interface MessageMapper {
+public interface MessageMapper<M,O> {
 
 	/**
 	 * Map to a {@link Message} from the given object.
 	 */
-	Message toMessage(Object source);
+	Message<M> toMessage(O source);
 
 	/**
 	 * Map from the given {@link Message} to an object.
 	 */
-	Object fromMessage(Message message);
+	O fromMessage(Message<M> message);
 
 }

spring-eai-core/src/main/java/org/springframework/integration/message/SimplePayloadMessageMapper.java

@@ -26,7 +26,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class SimplePayloadMessageMapper implements MessageMapper {
+public class SimplePayloadMessageMapper<T> implements MessageMapper<T,T> {
 
 	private UidGenerator uidGenerator;
 
@@ -54,15 +54,15 @@ public class SimplePayloadMessageMapper implements MessageMapper {
 	/**
 	 * Return the payload of the given Message.
 	 */
-	public Object fromMessage(Message message) {
+	public T fromMessage(Message<T> message) {
 		return message.getPayload();
 	}
 
 	/**
 	 * Return a {@link DocumentMessage} with the given object as its payload.
 	 */
-	public Message toMessage(Object source) {
-		return new DocumentMessage(uidGenerator.generateUid(), source);
+	public Message<T> toMessage(T source) {
+		return new GenericMessage<T>(uidGenerator.generateUid(), source);
 	}
 
 }

spring-eai-core/src/main/java/org/springframework/integration/message/StringMessage.java

@@ -17,13 +17,11 @@
 package org.springframework.integration.message;
 
 /**
- * A simple Message implementation that encapsulates a single Object payload.
- * 
  * @author Mark Fisher
  */
-public class DocumentMessage extends AbstractMessage {
+public class StringMessage extends GenericMessage<String> {
 
-	public DocumentMessage(Object id, Object payload) {
+	public StringMessage(Object id, String payload) {
 		super(id, payload);
 	}
 

spring-eai-core/src/main/java/org/springframework/integration/samples/Logger.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2002-2007 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.samples;
+
+import org.springframework.integration.endpoint.annotation.Subscriber;
+
+/**
+ * @author Mark Fisher
+ */
+public class Logger {
+
+	//@Subscriber(channel="quotes")
+	public void log(Object o) {
+		System.out.println(o);
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/samples/Quote.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2002-2007 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.samples;
+
+import java.math.BigDecimal;
+
+/**
+ * @author Mark Fisher
+ */
+public class Quote {
+
+	private String ticker;
+
+	private BigDecimal price;
+
+	public Quote(String ticker, BigDecimal price) {
+		this.ticker = ticker;
+		this.price = price;
+	}
+
+	public String toString() {
+		return ticker + ": " + price;
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/samples/QuoteService.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2002-2007 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.samples;
+
+/**
+ * @author Mark Fisher
+ */
+public interface QuoteService {
+
+	Quote lookup(String ticker);
+
+}

spring-eai-core/src/main/java/org/springframework/integration/samples/StockQuoteDemo.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2002-2007 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.samples;
+
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+/**
+ * @author Mark Fisher
+ */
+public class StockQuoteDemo {
+
+	public static void main(String[] args) throws Exception {
+		AbstractApplicationContext context = new ClassPathXmlApplicationContext("stockQuoteDemo.xml", StockQuoteDemo.class);
+		context.start();
+		QuoteService service = (QuoteService) context.getBean("quoteService");
+		service.lookup("SOA");
+		Thread.sleep(1000);
+	}
+}

spring-eai-core/src/main/java/org/springframework/integration/samples/StubQuoteService.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2002-2007 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.samples;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.Random;
+
+import org.springframework.integration.aop.Publisher;
+
+/**
+ * @author Mark Fisher
+ */
+public class StubQuoteService implements QuoteService {
+
+	@Publisher(channel="quotes")
+	public Quote lookup(String ticker) {
+		BigDecimal price = new BigDecimal(new Random().nextDouble() * 100);
+		return new Quote(ticker, price.setScale(2, RoundingMode.HALF_EVEN));
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/samples/stockQuoteDemo.xml

@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
+
+	<message-bus/>
+
+	<annotation-driven/>
+
+	<channel id="quotes"/>
+
+	<endpoint input-channel="quotes" handler="logger" handler-method="log">
+		<consumer period="1000"/>
+	</endpoint>
+
+	<beans:bean id="quoteService" class="org.springframework.integration.samples.StubQuoteService"/>
+
+	<beans:bean id="logger" class="org.springframework.integration.samples.Logger"/>
+
+</beans:beans>

spring-eai-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java

@@ -43,6 +43,9 @@ public class PublisherAnnotationAdvisorTests {
 				}
 				return null;
 			}
+			public MessageChannel getInvalidMessageChannel() {
+				return null;
+			}
 		};
 		PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelMapping);
 		TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor);
@@ -62,6 +65,9 @@ public class PublisherAnnotationAdvisorTests {
 				}
 				return null;
 			}
+			public MessageChannel getInvalidMessageChannel() {
+				return null;
+			}
 		};
 		PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelMapping);
 		TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor);

spring-eai-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml

@@ -10,7 +10,7 @@
 
 	<bean id="testBean" class="org.springframework.integration.aop.PublisherAnnotationTestBean"/>
 
-	<bean class="org.springframework.integration.aop.PublisherAnnotationPostProcessor">
+	<bean class="org.springframework.integration.config.PublisherAnnotationPostProcessor">
 		<property name="channelMapping" ref="messageBus"/>
 	</bean>
 

spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java

@@ -24,14 +24,11 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
-
-import org.springframework.integration.bus.ConsumerPolicy;
-import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.endpoint.GenericMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
-import org.springframework.integration.message.DocumentMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.GenericMessage;
 
 /**
  * @author Mark Fisher
@@ -74,16 +71,16 @@ public class EventDrivenConsumerTests {
 		bus.activateSubscription(subscription);
 		bus.start();
 		for (int i = 0; i < messagesToSend - 110; i++) {
-			channel.send(new DocumentMessage(1, "fast-1." + (i+1)));
+			channel.send(new GenericMessage<String>(1, "fast-1." + (i+1)));
 		}
 		int activeCountAfterFirstBurst = bus.getActiveCountForEndpoint("testEndpoint");
 		for (int i = 0; i < 10; i++) {
-			channel.send(new DocumentMessage(1, "slow-1." + (i+1)));
+			channel.send(new GenericMessage<String>(1, "slow-1." + (i+1)));
 			Thread.sleep(10);
 		}
 		int activeCountAfterSlowDown = bus.getActiveCountForEndpoint("testEndpoint");
 		for (int i = 0; i < 100; i++) {
-			channel.send(new DocumentMessage(1, "fast-2." + (i+1)));
+			channel.send(new GenericMessage<String>(1, "fast-2." + (i+1)));
 		}
 		int activeCountAfterLastBurst = bus.getActiveCountForEndpoint("testEndpoint");
 		latch.await(100, TimeUnit.SECONDS);

spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java

@@ -24,14 +24,11 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
-
-import org.springframework.integration.bus.ConsumerPolicy;
-import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.endpoint.GenericMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
-import org.springframework.integration.message.DocumentMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.GenericMessage;
 
 /**
  * @author Mark Fisher
@@ -66,7 +63,7 @@ public class FixedDelayConsumerTests {
 		bus.activateSubscription(subscription);
 		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
-			channel.send(new DocumentMessage(1, "test " + (i+1)));
+			channel.send(new GenericMessage<String>(1, "test " + (i+1)));
 		}
 		latch.await(250, TimeUnit.MILLISECONDS);
 		assertEquals(messagesToSend, counter.get());
@@ -100,7 +97,7 @@ public class FixedDelayConsumerTests {
 		bus.activateSubscription(subscription);
 		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
-			channel.send(new DocumentMessage(1, "test " + (i+1)));
+			channel.send(new GenericMessage<String>(1, "test " + (i+1)));
 		}
 		latch.await(80, TimeUnit.MILLISECONDS);
 		assertTrue(counter.get() < 10);

spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java

@@ -24,14 +24,11 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
-
-import org.springframework.integration.bus.ConsumerPolicy;
-import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.endpoint.GenericMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
-import org.springframework.integration.message.DocumentMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.GenericMessage;
 
 /**
  * @author Mark Fisher
@@ -63,7 +60,7 @@ public class FixedRateConsumerTests {
 		bus.activateSubscription(subscription);
 		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
-			channel.send(new DocumentMessage(1, "test " + (i+1)));
+			channel.send(new GenericMessage<String>(1, "test " + (i+1)));
 		}
 		latch.await(250, TimeUnit.MILLISECONDS);
 		assertEquals(messagesToSend, counter.get());
@@ -97,7 +94,7 @@ public class FixedRateConsumerTests {
 		bus.activateSubscription(subscription);
 		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
-			channel.send(new DocumentMessage(1, "test " + (i+1)));
+			channel.send(new GenericMessage<String>(1, "test " + (i+1)));
 		}
 		latch.await(80, TimeUnit.MILLISECONDS);
 		assertTrue(counter.get() < 10);

spring-eai-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java

@@ -19,13 +19,10 @@ package org.springframework.integration.bus;
 import static org.junit.Assert.assertEquals;
 
 import org.junit.Test;
-
 import org.springframework.context.support.ClassPathXmlApplicationContext;
-import org.springframework.integration.bus.ConsumerPolicy;
-import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.DocumentMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.GenericMessage;
 
 /**
  * @author Mark Fisher
@@ -65,9 +62,8 @@ public class MessageBusTests {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("messageBusTests.xml", this.getClass());
 		context.start();
 		MessageChannel sourceChannel = (MessageChannel) context.getBean("sourceChannel");
-		sourceChannel.send(new DocumentMessage("123", "test"));		
+		sourceChannel.send(new GenericMessage<String>("123", "test"));		
 		MessageChannel targetChannel = (MessageChannel) context.getBean("targetChannel");
-		// TODO: add metadata for this
 		MessageBus bus = (MessageBus) context.getBean("bus");
 		ConsumerPolicy policy = new ConsumerPolicy();
 		Subscription subscription = new Subscription();
@@ -75,7 +71,7 @@ public class MessageBusTests {
 		subscription.setEndpoint("endpoint");
 		subscription.setPolicy(policy);
 		bus.activateSubscription(subscription);
-		Message result = targetChannel.receive(10);
+		Message<String> result = targetChannel.receive(10);
 		assertEquals("test", result.getPayload());
 	}
 

spring-eai-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java

@@ -29,9 +29,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
-import org.springframework.integration.message.DocumentMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageSelector;
+import org.springframework.integration.message.GenericMessage;
 
 /**
  * @author Mark Fisher
@@ -45,7 +45,7 @@ public class PointToPointChannelTests {
 		final PointToPointChannel channel = new PointToPointChannel();
 		new Thread(new Runnable() {
 			public void run() {
-				Message message = channel.receive();
+				Message<String> message = channel.receive();
 				if (message != null) {
 					messageReceived.set(true);
 					latch.countDown();
@@ -53,7 +53,7 @@ public class PointToPointChannelTests {
 			}
 		}).start();
 		assertFalse(messageReceived.get());
-		channel.send(new DocumentMessage(1, "testing"));
+		channel.send(new GenericMessage<String>(1, "testing"));
 		latch.await(25, TimeUnit.MILLISECONDS);
 		assertTrue(messageReceived.get());
 	}
@@ -67,7 +67,7 @@ public class PointToPointChannelTests {
 		Executor singleThreadExecutor = Executors.newSingleThreadExecutor();
 		Runnable receiveTask1 = new Runnable() {
 			public void run() {
-				Message message = channel.receive(0);
+				Message<String> message = channel.receive(0);
 				if (message != null) {
 					messageReceived.set(true);
 				}
@@ -76,7 +76,7 @@ public class PointToPointChannelTests {
 		};
 		Runnable sendTask = new Runnable() {
 			public void run() {
-				channel.send(new DocumentMessage(1, "testing"));
+				channel.send(new GenericMessage<String>(1, "testing"));
 			}
 		};
 		singleThreadExecutor.execute(receiveTask1);
@@ -85,7 +85,7 @@ public class PointToPointChannelTests {
 		assertFalse(messageReceived.get());
 		Runnable receiveTask2 = new Runnable() {
 			public void run() {
-				Message message = channel.receive(0);
+				Message<String> message = channel.receive(0);
 				if (message != null) {
 					messageReceived.set(true);
 				}
@@ -104,7 +104,7 @@ public class PointToPointChannelTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		Thread t = new Thread(new Runnable() {
 			public void run() {
-				Message message = channel.receive();
+				Message<String> message = channel.receive();
 				receiveInterrupted.set(true);
 				assertTrue(message == null);
 				latch.countDown();
@@ -124,7 +124,7 @@ public class PointToPointChannelTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		Thread t = new Thread(new Runnable() {
 			public void run() {
-				Message message = channel.receive(10000);
+				Message<String> message = channel.receive(10000);
 				receiveInterrupted.set(true);
 				assertTrue(message == null);
 				latch.countDown();
@@ -140,26 +140,26 @@ public class PointToPointChannelTests {
 	@Test
 	public void testImmediateSend() {
 		PointToPointChannel channel = new PointToPointChannel(3);
-		boolean result1 = channel.send(new DocumentMessage(1, "test-1"));
+		boolean result1 = channel.send(new GenericMessage<String>(1, "test-1"));
 		assertTrue(result1);
-		boolean result2 = channel.send(new DocumentMessage(2, "test-2"), 100);
+		boolean result2 = channel.send(new GenericMessage<String>(2, "test-2"), 100);
 		assertTrue(result2);
-		boolean result3 = channel.send(new DocumentMessage(3, "test-3"), 0);
+		boolean result3 = channel.send(new GenericMessage<String>(3, "test-3"), 0);
 		assertTrue(result3);
-		boolean result4 = channel.send(new DocumentMessage(4, "test-4"), 0);
+		boolean result4 = channel.send(new GenericMessage<String>(4, "test-4"), 0);
 		assertFalse(result4);
 	}
 
 	@Test
 	public void testBlockingSendWithNoTimeout() throws Exception{
 		final PointToPointChannel channel = new PointToPointChannel(1);
-		boolean result1 = channel.send(new DocumentMessage(1, "test-1"));
+		boolean result1 = channel.send(new GenericMessage<String>(1, "test-1"));
 		assertTrue(result1);
 		final AtomicBoolean sendInterrupted = new AtomicBoolean(false);
 		final CountDownLatch latch = new CountDownLatch(1);
 		Thread t = new Thread(new Runnable() {
 			public void run() {
-				channel.send(new DocumentMessage(2, "test-2"));
+				channel.send(new GenericMessage<String>(2, "test-2"));
 				sendInterrupted.set(true);
 				latch.countDown();
 			}
@@ -174,13 +174,13 @@ public class PointToPointChannelTests {
 	@Test
 	public void testBlockingSendWithTimeout() throws Exception{
 		final PointToPointChannel channel = new PointToPointChannel(1);
-		boolean result1 = channel.send(new DocumentMessage(1, "test-1"));
+		boolean result1 = channel.send(new GenericMessage<String>(1, "test-1"));
 		assertTrue(result1);
 		final AtomicBoolean sendInterrupted = new AtomicBoolean(false);
 		final CountDownLatch latch = new CountDownLatch(1);
 		Thread t = new Thread(new Runnable() {
 			public void run() {
-				channel.send(new DocumentMessage(2, "test-2"), 10000);
+				channel.send(new GenericMessage<String>(2, "test-2"), 10000);
 				sendInterrupted.set(true);
 				latch.countDown();
 			}
@@ -196,7 +196,7 @@ public class PointToPointChannelTests {
 	public void testSelectorMatchesWithinTimeout() throws Exception {
 		final PointToPointChannel channel = new PointToPointChannel();
 		final CountDownLatch latch = new CountDownLatch(1);
-		final AtomicReference<Message> messageRef = new AtomicReference<Message>();
+		final AtomicReference<Message<String>> messageRef = new AtomicReference<Message<String>>();
 		Thread receiver = new Thread(new Runnable() {
 			public void run() {
 				Message message = channel.receive(new MessageSelector() {
@@ -211,13 +211,13 @@ public class PointToPointChannelTests {
 		receiver.start();
 		Thread sender = new Thread(new Runnable() {
 			public void run() {
-				channel.send(new DocumentMessage(1, "test-1"));
+				channel.send(new GenericMessage<String>(1, "test-1"));
 				try { Thread.sleep(5); } catch (Exception e) {}
-				channel.send(new DocumentMessage(2, "test-2"));
+				channel.send(new GenericMessage<String>(2, "test-2"));
 				try { Thread.sleep(5); } catch (Exception e) {}
-				channel.send(new DocumentMessage(3, "test-3"));
+				channel.send(new GenericMessage<String>(3, "test-3"));
 				try { Thread.sleep(100); } catch (Exception e) {}
-				channel.send(new DocumentMessage(4, "test-4"));
+				channel.send(new GenericMessage<String>(4, "test-4"));
 			}
 		});
 		sender.start();
@@ -229,7 +229,7 @@ public class PointToPointChannelTests {
 	public void testSelectorDoesNotMatchWithinTimeout() throws Exception {
 		final PointToPointChannel channel = new PointToPointChannel();
 		final CountDownLatch latch = new CountDownLatch(1);
-		final AtomicReference<Message> messageRef = new AtomicReference<Message>();
+		final AtomicReference<Message<String>> messageRef = new AtomicReference<Message<String>>();
 		Thread receiver = new Thread(new Runnable() {
 			public void run() {
 				Message message = channel.receive(new MessageSelector() {
@@ -244,11 +244,11 @@ public class PointToPointChannelTests {
 		receiver.start();
 		Thread sender = new Thread(new Runnable() {
 			public void run() {
-				channel.send(new DocumentMessage(1, "test-1"));
+				channel.send(new GenericMessage<String>(1, "test-1"));
 				try { Thread.sleep(5); } catch (Exception e) {}
-				channel.send(new DocumentMessage(2, "test-2"));
+				channel.send(new GenericMessage<String>(2, "test-2"));
 				try { Thread.sleep(5); } catch (Exception e) {}
-				channel.send(new DocumentMessage(3, "test-3"));
+				channel.send(new GenericMessage<String>(3, "test-3"));
 			}
 		});
 		sender.start();

spring-eai-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java

@@ -20,10 +20,9 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 import org.junit.Test;
-
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.DocumentMessage;
+import org.springframework.integration.message.GenericMessage;
 
 /**
  * @author Mark Fisher
@@ -36,9 +35,10 @@ public class ChannelParserTests {
 				"channelParserTests.xml", this.getClass());
 		MessageChannel channel = (MessageChannel) context.getBean("testChannel");
 		for (int i = 0; i < 10; i++) {
-			boolean result = channel.send(new DocumentMessage(1, "test"), 10);
+			boolean result = channel.send(new GenericMessage<String>(1, "test"), 10);
 			assertTrue(result);
 		}
-		assertFalse(channel.send(new DocumentMessage(1, "test"), 3));
+		assertFalse(channel.send(new GenericMessage<String>(1, "test"), 3));
 	}
+
 }

spring-eai-core/src/test/java/org/springframework/integration/config/ComponentConfigurerTests.java

@@ -28,8 +28,8 @@ import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.endpoint.MessageEndpoint;
-import org.springframework.integration.message.DocumentMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.GenericMessage;
 
 /**
  * @author Mark Fisher
@@ -43,10 +43,10 @@ public class ComponentConfigurerTests {
 		context.registerBeanDefinition("tester", new RootBeanDefinition(Tester.class));
 		context.registerBeanDefinition("out", new RootBeanDefinition(PointToPointChannel.class));
 		context.registerBeanDefinition("bus", new RootBeanDefinition(MessageBus.class));
-		String name = cr.serviceActivator("in", "out", "tester", "test");
+		String name = cr.serviceActivator(null, "out", "tester", "test");
 		context.refresh();
 		MessageEndpoint endpoint = (MessageEndpoint) context.getBean(name);
-		endpoint.messageReceived(new DocumentMessage(1, "world"));
+		endpoint.messageReceived(new GenericMessage<String>(1, "world"));
 		MessageChannel channel = (MessageChannel) context.getBean("out");
 		assertEquals("hello world", channel.receive().getPayload());
 	}
@@ -58,7 +58,7 @@ public class ComponentConfigurerTests {
 		context.registerBeanDefinition("tester", new RootBeanDefinition(Tester.class));
 		String adapter = cr.inboundChannelAdapter("tester", "foo");
 		MessageChannel channel = (MessageChannel) context.getBean(adapter);
-		Message message = channel.receive();
+		Message<String> message = channel.receive();
 		assertEquals("bar", message.getPayload());
 	}
 
@@ -71,7 +71,7 @@ public class ComponentConfigurerTests {
 		Tester tester = (Tester) context.getBean("tester");
 		assertNull(tester.getStoredValue());
 		MessageChannel channel = (MessageChannel) context.getBean(adapter);
-		boolean result = channel.send(new DocumentMessage(1, "foo"));
+		boolean result = channel.send(new GenericMessage<String>(1, "foo"));
 		assertTrue(result);
 		assertEquals("foo", tester.getStoredValue());
 	}

spring-eai-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java

@@ -25,7 +25,7 @@ import org.junit.Test;
 
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.DocumentMessage;
+import org.springframework.integration.message.GenericMessage;
 
 /**
  * @author Mark Fisher
@@ -40,7 +40,7 @@ public class EndpointParserTests {
 		MessageChannel channel = (MessageChannel) context.getBean("testChannel");
 		TestHandler handler = (TestHandler) context.getBean("testHandler");
 		assertNull(handler.getMessageString());
-		channel.send(new DocumentMessage(1, "test"));
+		channel.send(new GenericMessage<String>(1, "test"));
 		handler.getLatch().await(50, TimeUnit.MILLISECONDS);
 		assertEquals("test", handler.getMessageString());
 	}
@@ -53,7 +53,7 @@ public class EndpointParserTests {
 		MessageChannel channel = (MessageChannel) context.getBean("testChannel");
 		TestBean bean = (TestBean) context.getBean("testBean");
 		assertNull(bean.getMessage());
-		channel.send(new DocumentMessage(1, "test"));
+		channel.send(new GenericMessage<String>(1, "test"));
 		bean.getLatch().await(50, TimeUnit.MILLISECONDS);
 		assertEquals("test", bean.getMessage());
 	}

spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java

@@ -20,15 +20,14 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 
 import org.junit.Test;
-
 import org.springframework.integration.bus.ConsumerPolicy;
 import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.bus.Subscription;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.handler.MessageHandler;
-import org.springframework.integration.message.DocumentMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.StringMessage;
 
 /**
  * @author Mark Fisher
@@ -40,8 +39,8 @@ public class GenericMessageEndpointTests {
 		MessageChannel channel = new PointToPointChannel();
 		MessageChannel replyChannel = new PointToPointChannel();
 		MessageHandler handler = new MessageHandler() {
-			public Message handle(Message message) {
-				return new DocumentMessage("123", "hello " + message.getPayload());
+			public Message<String> handle(Message message) {
+				return new StringMessage("123", "hello " + message.getPayload());
 			}
 		};
 		GenericMessageEndpoint endpoint = new GenericMessageEndpoint();
@@ -60,9 +59,9 @@ public class GenericMessageEndpointTests {
 		subscription.setPolicy(policy);
 		bus.activateSubscription(subscription);
 		bus.start();
-		DocumentMessage testMessage = new DocumentMessage(1, "test");
+		StringMessage testMessage = new StringMessage(1, "test");
 		channel.send(testMessage);
-		Message reply = replyChannel.receive(10);
+		Message<String> reply = replyChannel.receive(10);
 		assertNotNull(reply);
 		assertEquals("hello test", reply.getPayload());
 	}
@@ -73,7 +72,7 @@ public class GenericMessageEndpointTests {
 		final MessageChannel replyChannel = new PointToPointChannel();
 		MessageHandler handler = new MessageHandler() {
 			public Message handle(Message message) {
-				return new DocumentMessage("123", "hello " + message.getPayload());
+				return new StringMessage("123", "hello " + message.getPayload());
 			}
 		};
 		GenericMessageEndpoint endpoint = new GenericMessageEndpoint();
@@ -91,7 +90,7 @@ public class GenericMessageEndpointTests {
 		subscription.setPolicy(policy);
 		bus.activateSubscription(subscription);
 		bus.start();
-		DocumentMessage testMessage = new DocumentMessage(1, "test");
+		StringMessage testMessage = new StringMessage(1, "test");
 		testMessage.getHeader().setReplyChannelName("replyChannel");
 		channel.send(testMessage);
 		Message reply = replyChannel.receive(10);

spring-eai-core/src/test/java/org/springframework/integration/endpoint/OutboundChannelAdapterTests.java

@@ -22,7 +22,7 @@ import org.junit.Test;
 
 import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter;
-import org.springframework.integration.message.DocumentMessage;
+import org.springframework.integration.message.StringMessage;
 
 /**
  * @author Mark Fisher
@@ -36,7 +36,7 @@ public class OutboundChannelAdapterTests {
 		adapter.setObject(new TestSink());
 		adapter.setMethod("validMethod");
 		adapter.afterPropertiesSet();
-		boolean result = adapter.send(new DocumentMessage(1, "test"));
+		boolean result = adapter.send(new StringMessage(1, "test"));
 		assertTrue(result);
 	}
 
@@ -47,7 +47,7 @@ public class OutboundChannelAdapterTests {
 		adapter.setObject(new TestSink());
 		adapter.setMethod("invalidMethodWithNoArgs");
 		adapter.afterPropertiesSet();
-		adapter.send(new DocumentMessage(1, "test"));
+		adapter.send(new StringMessage(1, "test"));
 	}
 
 	@Test
@@ -57,7 +57,7 @@ public class OutboundChannelAdapterTests {
 		adapter.setObject(new TestSink());
 		adapter.setMethod("validMethodWithIgnoredReturnValue");
 		adapter.afterPropertiesSet();
-		boolean result = adapter.send(new DocumentMessage(1, "test"));
+		boolean result = adapter.send(new StringMessage(1, "test"));
 		assertTrue(result);
 	}
 
@@ -68,7 +68,7 @@ public class OutboundChannelAdapterTests {
 		adapter.setObject(new TestSink());
 		adapter.setMethod("noSuchMethod");
 		adapter.afterPropertiesSet();
-		adapter.send(new DocumentMessage(1, "test"));
+		adapter.send(new StringMessage(1, "test"));
 	}
 
 }

spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/AnnotationAwareMessageEndpointTests.java

@@ -20,7 +20,7 @@ import org.junit.Test;
 
 import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.DocumentMessage;
+import org.springframework.integration.message.GenericMessage;
 
 /**
  * @author Mark Fisher
@@ -37,7 +37,7 @@ public class AnnotationAwareMessageEndpointTests {
 		MessageChannel channel = messageBus.getChannel("testChannel");
 		messageBus.start();
 		endpoint.afterPropertiesSet();
-		channel.send(new DocumentMessage(1, "world"), 10);
+		channel.send(new GenericMessage<String>(1, "world"), 10);
 	}
 
 	@MessageEndpoint(input="testChannel", pollPeriod=10)

spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/EndpointAnnotationPostProcessorTests.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2002-2007 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.endpoint.annotation;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.GenericMessage;
+
+/**
+ * @author Mark Fisher
+ */
+public class EndpointAnnotationPostProcessorTests {
+
+	@Test
+	public void testSimpleHandler() throws InterruptedException {
+		AbstractApplicationContext context = new ClassPathXmlApplicationContext("simpleAnnotatedEndpointTests.xml", this.getClass());
+		context.start();
+		MessageChannel inputChannel = (MessageChannel) context.getBean("inputChannel");
+		MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
+		inputChannel.send(new GenericMessage<String>(1, "world"));
+		Message<String> message = outputChannel.receive();
+		assertEquals("hello world", message.getPayload());
+	}
+
+}

spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2002-2007 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.endpoint.annotation;
+
+/**
+ * @author Mark Fisher
+ */
+@MessageEndpoint(defaultOutput="outputChannel")
+public class InboundChannelAdapterTestBean {
+
+	@Polled(period=100)
+	public String getName() {
+		return "world";
+	}
+
+	@Handler
+	public String sayHello(String name) {
+		return "hello " + name;
+	}
+
+}

spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/OutboundChannelAdapterTestBean.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2002-2007 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.endpoint.annotation;
+
+/**
+ * @author Mark Fisher
+ */
+@MessageEndpoint(input="inputChannel")
+public class OutboundChannelAdapterTestBean {
+
+	@Handler
+	public String sayHello(String name) {
+		return "hello " + name;
+	}
+
+	@DefaultOutput
+	public void sendGreeting(String greeting) {
+		System.out.println(greeting);
+	}
+
+}

spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2002-2007 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.endpoint.annotation;
+
+/**
+ * @author Mark Fisher
+ */
+@MessageEndpoint(input="inputChannel", defaultOutput="outputChannel", pollPeriod=10)
+public class SimpleAnnotatedEndpoint {
+
+	@Handler
+	public String sayHello(String name) {
+		return "hello " + name;
+	}
+
+}

spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml

@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:integration="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
+
+	<bean id="bus" class="org.springframework.integration.bus.MessageBus"/>
+
+	<integration:channel id="inputChannel"/>
+
+	<integration:channel id="outputChannel"/>
+
+	<bean id="endpoint" class="org.springframework.integration.endpoint.annotation.SimpleAnnotatedEndpoint"/>
+
+	<bean class="org.springframework.integration.endpoint.annotation.EndpointAnnotationPostProcessor">
+		<property name="messageBus" ref="bus"/>
+	</bean>
+
+</beans>

spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
+
+	<bean id="bus" class="org.springframework.integration.bus.MessageBus">
+		<property name="autoCreateChannels" value="true"/>
+	</bean>
+
+	<bean id="endpoint" class="org.springframework.integration.endpoint.annotation.SimpleAnnotatedEndpoint"/>
+
+	<bean id="outputChannel" class="org.springframework.integration.channel.PointToPointChannel"/>
+
+	<bean class="org.springframework.integration.endpoint.annotation.EndpointAnnotationPostProcessor">
+		<property name="messageBus" ref="bus"/>
+	</bean>
+
+</beans>

spring-eai-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java

@@ -20,8 +20,8 @@ import static org.junit.Assert.assertEquals;
 
 import org.junit.Test;
 
-import org.springframework.integration.message.DocumentMessage;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.StringMessage;
 
 /**
  * @author Mark Fisher
@@ -35,7 +35,7 @@ public class MessageHandlerChainTests {
 		chain.add(new TestHandler("b"));
 		chain.add(new TestHandler("c"));
 		chain.add(new TestHandler("d"));
-		Message result = chain.handle(new DocumentMessage(1, "!"));
+		Message result = chain.handle(new StringMessage(1, "!"));
 		assertEquals("!abcd", result.getPayload());
 	}
 
@@ -49,7 +49,7 @@ public class MessageHandlerChainTests {
 		chain.add(new TestHandler("a"));
 		chain.add(handler4);
 		chain.add(new TestHandler("b"));		
-		Message result = chain.handle(new DocumentMessage(1, "!"));
+		Message result = chain.handle(new StringMessage(1, "!"));
 		assertEquals("234!a*234b", result.getPayload());
 	}
 
@@ -63,7 +63,7 @@ public class MessageHandlerChainTests {
 		}
 
 		public Message handle(Message message) {
-			return new DocumentMessage(1, message.getPayload() + text);
+			return new StringMessage(1, message.getPayload() + text);
 		}
 	}
 
@@ -78,8 +78,8 @@ public class MessageHandlerChainTests {
 		}
 
 		public Message handle(Message message, MessageHandler target) {
-			message = target.handle(new DocumentMessage(1, text + message.getPayload()));
-			return new DocumentMessage(1, message.getPayload() + text);
+			message = target.handle(new StringMessage(1, text + message.getPayload()));
+			return new StringMessage(1, message.getPayload() + text);
 		}
 	}