Added channel reference to Subscription and refactored/simplified TargetAdapter (implements MessageHandler).

2008-01-15 21:57:12 +00:00
parent 0a273b016c
commit 6cd1e7be7c
21 changed files with 276 additions and 362 deletions
--- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java
@@ -19,12 +19,10 @@ package org.springframework.integration.adapter;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;

-import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageMapper;
 import org.springframework.integration.message.SimplePayloadMessageMapper;
-import org.springframework.integration.scheduling.PollingSchedule;
-import org.springframework.integration.scheduling.Schedule;
 import org.springframework.util.Assert;

 /**
@@ -32,35 +30,12 @@ import org.springframework.util.Assert;
 * 
 * @author Mark Fisher
 */
-public abstract class AbstractTargetAdapter<T> implements TargetAdapter {
+public abstract class AbstractTargetAdapter<T> implements MessageHandler {

 	protected Log logger = LogFactory.getLog(this.getClass());

-	private String name;
-
-	private MessageChannel channel;
-
 	private MessageMapper<?,T> mapper = new SimplePayloadMessageMapper<T>();

-	private Schedule schedule = new PollingSchedule(5);
-
-
-	public void setName(String name) {
-		this.name = name;
-	}
-
-	public String getName() {
-		return this.name;
-	}
-
-	public void setChannel(MessageChannel channel) {
-		Assert.notNull(channel, "'channel' must not be null");
-		this.channel = channel;
-	}
-
-	public MessageChannel getChannel() {
-		return this.channel;
-	}

 	public void setMessageMapper(MessageMapper<?,T> mapper) {
 		Assert.notNull(mapper, "'mapper' must not be null");
@@ -71,15 +46,6 @@ public abstract class AbstractTargetAdapter<T> implements TargetAdapter {
 		return this.mapper;
 	}

-	public void setSchedule(Schedule schedule) {
-		Assert.notNull(schedule, "'schedule' must not be null");
-		this.schedule = schedule;
-	}
-
-	public Schedule getSchedule() {
-		return this.schedule;
-	}
-
 	public final Message handle(Message message) {
 		this.sendToTarget(this.mapper.fromMessage(message));
 		return null;
--- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java
@@ -1,33 +0,0 @@
-/*
- * 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.adapter;
-
-import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.handler.MessageHandler;
-
-/**
- * Base interface for target adapters.
- * 
- * @author Mark Fisher
- */
-public interface TargetAdapter extends MessageHandler {
-
-	String getName();
-
-	void setChannel(MessageChannel channel);
-
-}
--- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
@@ -28,11 +28,11 @@ import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.Lifecycle;
+import org.springframework.integration.MessagingConfigurationException;
 import org.springframework.integration.MessagingException;
-import org.springframework.integration.adapter.AbstractTargetAdapter;
 import org.springframework.integration.adapter.SourceAdapter;
-import org.springframework.integration.adapter.TargetAdapter;
 import org.springframework.integration.channel.ChannelRegistry;
+import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.DefaultChannelRegistry;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
@@ -40,6 +40,7 @@ import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
 import org.springframework.integration.dispatcher.DispatcherPolicy;
 import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
+import org.springframework.integration.endpoint.DefaultMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.handler.PooledMessageHandler;
@@ -50,7 +51,6 @@ import org.springframework.integration.scheduling.Schedule;
 import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;

 /**
 * The messaging bus. Serves as a registry for channels and endpoints, manages their lifecycle,
@@ -64,7 +64,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif

 	private ChannelRegistry channelRegistry = new DefaultChannelRegistry();

-	private Map<String, MessageHandler> handlers = new ConcurrentHashMap<String, MessageHandler>();
+	private Map<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();

 	private Map<MessageChannel, MessageDispatcher> dispatchers = new ConcurrentHashMap<MessageChannel, MessageDispatcher>();

@@ -90,8 +90,6 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		this.registerChannels(applicationContext);
 		this.registerEndpoints(applicationContext);
 		this.registerSourceAdapters(applicationContext);
-		this.registerTargetAdapters(applicationContext);
-		this.activateSubscriptions(applicationContext);
 	}

 	/**
@@ -140,28 +138,6 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 	}

-	@SuppressWarnings("unchecked")
-	private void registerTargetAdapters(ApplicationContext context) {
-		Map<String, TargetAdapter> targetAdapterBeans =
-				(Map<String, TargetAdapter>) context.getBeansOfType(TargetAdapter.class);
-		for (Map.Entry<String, TargetAdapter> entry : targetAdapterBeans.entrySet()) {
-			this.registerTargetAdapter(entry.getKey(), entry.getValue());
-		}
-	}
-
-	@SuppressWarnings("unchecked")
-	private void activateSubscriptions(ApplicationContext context) {
-		Map<String, Subscription> subscriptionBeans =
-				(Map<String, Subscription>) context.getBeansOfType(Subscription.class);
-		for (Subscription subscription : subscriptionBeans.values()) {
-			this.activateSubscription(subscription);
-			if (logger.isInfoEnabled()) {
-				logger.info("activated subscription to channel '" + subscription.getChannel() + 
-						"' for handler '" + subscription.getHandler() + "'");
-			}
-		}
-	}
-
 	public void initialize() {
 		if (this.getInvalidMessageChannel() == null) {
 			this.setInvalidMessageChannel(new SimpleChannel(Integer.MAX_VALUE));
@@ -212,32 +188,72 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 		this.dispatchers.put(channel, dispatcher);
 		this.channelRegistry.registerChannel(name, channel);
+		if (logger.isInfoEnabled()) {
+			logger.info("registered channel '" + name + "'");
+		}
 	}

-	public void registerEndpoint(String name, MessageEndpoint endpoint) {
+	public void registerHandler(String name, MessageHandler handler, Subscription subscription) {
+		this.registerHandler(name, handler, subscription, null);
+	}
+
+	public void registerHandler(String name, MessageHandler handler, Subscription subscription, ConcurrencyPolicy concurrencyPolicy) {
 		if (!this.initialized) {
 			this.initialize();
 		}
 		Assert.notNull(name, "'name' must not be null");
-		Assert.notNull(endpoint, "'endpoint' must not be null");
+		Assert.notNull(handler, "'handler' must not be null");
+		Assert.notNull(subscription, "'subscription' must not be null");
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
 		endpoint.setName(name);
-		this.handlers.put(name, endpoint);
-		endpoint.setChannelRegistry(this);
-		Schedule schedule = endpoint.getSchedule();
-		if (endpoint.getInputChannelName() != null) {
-			Subscription subscription = new Subscription();
-			subscription.setHandler(name);
-			subscription.setChannel(endpoint.getInputChannelName());
-			if (schedule != null) {
-				subscription.setSchedule(schedule);
-			}
-			this.activateSubscription(subscription);
+		endpoint.setHandler(handler);
+		endpoint.setSubscription(subscription);
+		endpoint.setConcurrencyPolicy(concurrencyPolicy);
+		this.registerEndpoint(name, endpoint);
+	}
+
+	public void registerEndpoint(String name, MessageEndpoint endpoint) {
+		if (endpoint instanceof ChannelRegistryAware) {
+			((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry);
 		}
-		if (this.autoCreateChannels) {
-			String defaultOutputChannelName = endpoint.getDefaultOutputChannelName();
-			if (StringUtils.hasText(defaultOutputChannelName) && this.lookupChannel(defaultOutputChannelName) == null) {
-				this.registerChannel(defaultOutputChannelName, new SimpleChannel());
+		this.endpoints.put(name, endpoint);
+		if (logger.isInfoEnabled()) {
+			logger.info("registered endpoint '" + name + "'");
+		}
+	}
+
+	private void activateEndpoints() {
+		for (MessageEndpoint endpoint : this.endpoints.values()) {
+			this.activateEndpoint(endpoint);
+		}
+	}
+
+	private void activateEndpoint(MessageEndpoint endpoint) {
+		Subscription subscription = endpoint.getSubscription();
+		MessageChannel channel = subscription.getChannel();
+		if (channel == null) {
+			String channelName = subscription.getChannelName();
+			if (channelName == null) {
+				throw new MessagingConfigurationException("endpoint '" + endpoint.getName() +
+						"' must provide either 'channel' or 'channelName' in its subscription metadata");
 			}
+			channel = this.lookupChannel(channelName);
+			if (channel == null) {
+				if (this.autoCreateChannels == false) {
+					throw new MessagingException("Cannot activate subscription, unknown channel '" + channelName +
+							"'. Consider enabling the 'autoCreateChannels' option for the message bus.");
+				}
+				if (this.logger.isInfoEnabled()) {
+					logger.info("auto-creating channel '" + channel.getName() + "'");
+				}
+				channel = new SimpleChannel(); 
+				this.registerChannel(channelName, channel);
+			}
+		}
+		this.registerWithDispatcher(channel, endpoint, subscription.getSchedule(), endpoint.getConcurrencyPolicy());
+		if (logger.isInfoEnabled()) {
+			logger.info("activated subscription to channel '" + channel.getName() + 
+					"' for endpoint '" + endpoint.getName() + "'");
 		}
 	}

@@ -259,56 +275,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 	}

-	public void registerTargetAdapter(String name, TargetAdapter targetAdapter) {
-		if (targetAdapter instanceof AbstractTargetAdapter) {
-			AbstractTargetAdapter<?> adapter = (AbstractTargetAdapter<?>) targetAdapter;
-			adapter.setName(name);
-			this.handlers.put(name, adapter);
-			MessageChannel channel = adapter.getChannel();
-			Schedule schedule = adapter.getSchedule();
-			ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy();
-			concurrencyPolicy.setCoreConcurrency(1);
-			concurrencyPolicy.setMaxConcurrency(1);
-			this.doActivate(channel, adapter, schedule, concurrencyPolicy);
-		}
-		if (logger.isInfoEnabled()) {
-			logger.info("registered target adapter '" + name + "'");
-		}
-	}
-
-	public void activateSubscription(Subscription subscription) {
-		String channelName = subscription.getChannel();
-		String handlerName = subscription.getHandler();
-		Schedule schedule = subscription.getSchedule();
-		ConcurrencyPolicy concurrencyPolicy = subscription.getConcurrencyPolicy();
-		MessageHandler handler = this.handlers.get(handlerName);
-		if (handler == null) {
-			throw new MessagingException("Cannot activate subscription, unknown handler '" + handlerName + "'");
-		}
-		MessageChannel channel = this.lookupChannel(channelName);
-		if (channel == null) {
-			if (this.autoCreateChannels == false) {
-				throw new MessagingException("Cannot activate subscription, unknown channel '" + channelName +
-						"'. Consider enabling the 'autoCreateChannels' option for the message bus.");
-			}
-			if (this.logger.isInfoEnabled()) {
-				logger.info("auto-creating channel '" + channelName + "'");
-			}
-			channel = new SimpleChannel(); 
-			this.registerChannel(channelName, channel);
-		}
-		this.doActivate(channel, handler, schedule, concurrencyPolicy);
-		if (logger.isInfoEnabled()) {
-			logger.info("activated subscription to channel '" + channelName + 
-					"' for handler '" + handlerName + "'");
-		}
-	}
-
-	private void doActivate(MessageChannel channel, MessageHandler handler, Schedule schedule, ConcurrencyPolicy concurrencyPolicy) {
+	private void registerWithDispatcher(MessageChannel channel, MessageHandler handler, Schedule schedule, ConcurrencyPolicy concurrencyPolicy) {
 		MessageDispatcher dispatcher = dispatchers.get(channel);
 		if (dispatcher == null) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("no dispatcher available for channel '" + channel + "', be sure to register the channel");
+				logger.warn("no dispatcher available for channel '" + channel.getName() + "', be sure to register the channel");
 			}
 		}
 		if (concurrencyPolicy != null) {
@@ -335,8 +306,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 		this.starting = true;
 		synchronized (this.lifecycleMonitor) {
+			this.activateEndpoints();
 			this.taskScheduler.start();
-			this.running = true;
 			for (MessageDispatcher dispatcher : this.dispatchers.values()) {
 				dispatcher.start();
 				if (logger.isInfoEnabled()) {
--- a/spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java
@@ -16,7 +16,7 @@

 package org.springframework.integration.bus;

-import org.springframework.integration.endpoint.ConcurrencyPolicy;
+import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.scheduling.Schedule;

 /**
@@ -26,29 +26,39 @@ import org.springframework.integration.scheduling.Schedule;
 */
 public class Subscription {

-	private String channel;
+	private MessageChannel channel;

-	private String handler;
+	private String channelName;

 	private Schedule schedule;

-	private ConcurrencyPolicy concurrencyPolicy;

-
-	public String getChannel() {
-		return this.channel;
+	public Subscription() {
 	}

-	public void setChannel(String channel) {
+	public Subscription(MessageChannel channel) {
 		this.channel = channel;
 	}

-	public String getHandler() {
-		return this.handler;
+	public Subscription(String channelName) {
+		this.channelName = channelName;
 	}

-	public void setHandler(String handler) {
-		this.handler = handler;
+
+	public MessageChannel getChannel() {
+		return this.channel;
+	}
+
+	public void setChannel(MessageChannel channel) {
+		this.channel = channel;
+	}
+
+	public String getChannelName() {
+		return (this.channel != null) ? this.channel.getName() : this.channelName;
+	}
+
+	public void setChannelName(String channelName) {
+		this.channelName = channelName;
 	}

 	public Schedule getSchedule() {
@@ -59,12 +69,4 @@ public class Subscription {
 		this.schedule = schedule;
 	}

-	public ConcurrencyPolicy getConcurrencyPolicy() {
-		return this.concurrencyPolicy;
-	}
-
-	public void setConcurrencyPolicy(ConcurrencyPolicy concurrencyPolicy) {
-		this.concurrencyPolicy = concurrencyPolicy;
-	}
-
 }
--- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java
@@ -19,9 +19,6 @@ package org.springframework.integration.channel;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;

-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
 import org.springframework.util.Assert;

 /**
@@ -31,8 +28,6 @@ import org.springframework.util.Assert;
 */
 public class DefaultChannelRegistry implements ChannelRegistry {

-	private Log logger = LogFactory.getLog(this.getClass());
-
 	private Map<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();

 	private MessageChannel invalidMessageChannel;
@@ -55,9 +50,6 @@ public class DefaultChannelRegistry implements ChannelRegistry {
 		Assert.notNull(channel, "'channel' must not be null");
 		channel.setName(name);
 		this.channels.put(name, channel);
-		if (logger.isInfoEnabled()) {
-			logger.info("registered channel '" + name + "'");
-		}
 	}

 }
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java
@@ -29,6 +29,8 @@ import org.springframework.integration.adapter.DefaultTargetAdapter;
 import org.springframework.integration.adapter.MethodInvokingSource;
 import org.springframework.integration.adapter.MethodInvokingTarget;
 import org.springframework.integration.adapter.PollingSourceAdapter;
+import org.springframework.integration.bus.Subscription;
+import org.springframework.integration.endpoint.DefaultMessageEndpoint;
 import org.springframework.util.StringUtils;

 /**
@@ -79,6 +81,7 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
 			if (StringUtils.hasText(period)) {
 				adapterDef.getPropertyValues().addPropertyValue("period", period);
 			}
+			adapterDef.getPropertyValues().addPropertyValue("channel", new RuntimeBeanReference(channel));
 		}
 		else {
 			adapterDef = new RootBeanDefinition(DefaultTargetAdapter.class);
@@ -89,12 +92,22 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
 		String invokerBeanName = parserContext.getReaderContext().generateBeanName(invokerDef);
 		parserContext.registerBeanComponent(new BeanComponentDefinition(invokerDef, invokerBeanName));
 		adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName));
-		adapterDef.getPropertyValues().addPropertyValue("channel", new RuntimeBeanReference(channel));
 		adapterDef.setSource(parserContext.extractSource(element));
 		String beanName = element.getAttribute(ID_ATTRIBUTE);
 		if (!StringUtils.hasText(beanName)) {
 			beanName = parserContext.getReaderContext().generateBeanName(adapterDef);
 		}
+		if (!this.isInbound) {
+			RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class);
+			RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class);
+			subscriptionDef.getPropertyValues().addPropertyValue("channel", new RuntimeBeanReference(channel));
+			String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef);
+			parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName));
+			endpointDef.getPropertyValues().addPropertyValue("subscription", new RuntimeBeanReference(subscriptionBeanName));
+			endpointDef.getPropertyValues().addPropertyValue("handler", new RuntimeBeanReference(beanName));
+			String endpointBeanName = parserContext.getReaderContext().generateBeanName(endpointDef);
+			parserContext.registerBeanComponent(new BeanComponentDefinition(endpointDef, endpointBeanName));
+		}
 		parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDef, beanName));
 		return adapterDef;
 	}
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java
@@ -31,6 +31,7 @@ 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.MessagingConfigurationException;
+import org.springframework.integration.bus.Subscription;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.DefaultMessageEndpoint;
 import org.springframework.integration.handler.DefaultMessageHandlerAdapter;
@@ -49,7 +50,9 @@ public class EndpointParser implements BeanDefinitionParser {

 	private static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";

-	private static final String INPUT_CHANNEL_PROPERTY = "inputChannelName";
+	private static final String SUBSCRIPTION_PROPERTY = "subscription";
+
+	private static final String CHANNEL_PROPERTY = "channel";

 	private static final String DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE = "default-output-channel";

@@ -90,8 +93,9 @@ public class EndpointParser implements BeanDefinitionParser {
 		RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class);
 		endpointDef.setSource(parserContext.extractSource(element));
 		String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
+		RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class);
 		if (StringUtils.hasText(inputChannel)) {
-			endpointDef.getPropertyValues().addPropertyValue(INPUT_CHANNEL_PROPERTY, inputChannel);
+			subscriptionDef.getPropertyValues().addPropertyValue(CHANNEL_PROPERTY, new RuntimeBeanReference(inputChannel));
 		}
 		String defaultOutputChannel = element.getAttribute(DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE);
 		if (StringUtils.hasText(defaultOutputChannel)) {
@@ -104,7 +108,7 @@ public class EndpointParser implements BeanDefinitionParser {
 			if (child.getNodeType() == Node.ELEMENT_NODE) {
 				String localName = child.getLocalName();
 				if (CONCURRENCY_ELEMENT.equals(localName)) {
-					parseConcurrencyPolicy((Element) child, endpointDef);
+					parseConcurrencyPolicy((Element) child, subscriptionDef);
 				}
 				else if (HANDLER_ELEMENT.equals(localName)) {
 					String ref = ((Element) child).getAttribute(REF_ATTRIBUTE);
@@ -113,6 +117,9 @@ public class EndpointParser implements BeanDefinitionParser {
 				}
 			}
 		}
+		String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef);
+		parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName));
+		endpointDef.getPropertyValues().addPropertyValue(SUBSCRIPTION_PROPERTY, new RuntimeBeanReference(subscriptionBeanName));
 		if (childHandlerRefs.size() > 0) {
 			if (childHandlerRefs.size() == 1) {
 				endpointDef.getPropertyValues().addPropertyValue(
@@ -153,7 +160,7 @@ public class EndpointParser implements BeanDefinitionParser {
 		return endpointDef;
 	}

-	private void parseConcurrencyPolicy(Element concurrencyElement, RootBeanDefinition endpointDefinition) {
+	private void parseConcurrencyPolicy(Element concurrencyElement, RootBeanDefinition subscriptionDefinition) {
 		ConcurrencyPolicy policy = new ConcurrencyPolicy();
 		String coreConcurrency = concurrencyElement.getAttribute(CORE_CONCURRENCY_ATTRIBUTE);
 		String maxConcurrency = concurrencyElement.getAttribute(MAX_CONCURRENCY_ATTRIBUTE);
@@ -163,7 +170,7 @@ public class EndpointParser implements BeanDefinitionParser {
 		if (StringUtils.hasText(maxConcurrency)) {
 			policy.setMaxConcurrency(Integer.parseInt(maxConcurrency));
 		}
-		endpointDefinition.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy);
+		subscriptionDefinition.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy);
 	}

 	private void parseSchedule(Element scheduleElement, RootBeanDefinition endpointDefinition) {
--- 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
@@ -44,10 +44,10 @@ import org.springframework.integration.annotation.Polled;
 import org.springframework.integration.annotation.Router;
 import org.springframework.integration.annotation.Splitter;
 import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.bus.Subscription;
 import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
-import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.DefaultMessageEndpoint;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.handler.MessageHandlerChain;
@@ -106,24 +106,23 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
 			return bean;
 		}
 		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
-		this.configureInputChannel(bean, beanName, endpointAnnotation, endpoint);
-		this.configureDefaultOutputChannel(bean, beanName, endpointAnnotation, endpoint);
+		this.configureInput(bean, beanName, endpointAnnotation, endpoint);
 		MessageHandlerChain handlerChain = this.createHandlerChain(bean);
-		if (handlerChain != null) {
-			endpoint.setHandler(handlerChain);
-		}
-		this.messageBus.registerEndpoint(beanName, endpoint);
-		return endpoint;
+		endpoint.setHandler(handlerChain);
+		this.configureDefaultOutput(bean, beanName, endpointAnnotation, endpoint);
+		this.messageBus.registerEndpoint(bean + "-endpoint", endpoint);
+		return bean;
 	}

-	private void configureInputChannel(final Object bean, final String beanName,
-			MessageEndpoint annotation, final DefaultMessageEndpoint endpoint) {
+	private void configureInput(final Object bean, final String beanName, MessageEndpoint annotation,
+			final DefaultMessageEndpoint endpoint) {
 		String channelName = annotation.input();
 		if (StringUtils.hasText(channelName)) {
-			endpoint.setInputChannelName(channelName);
+			Subscription subscription = new Subscription();
+			subscription.setChannelName(channelName);
 			Schedule schedule = new PollingSchedule(annotation.pollPeriod());
-			endpoint.setSchedule(schedule);
-			return;
+			subscription.setSchedule(schedule);
+			endpoint.setSubscription(subscription);
 		}
 		ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
 			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
@@ -140,22 +139,16 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
 					String channelName = beanName + "-inputChannel";
 					messageBus.registerChannel(channelName, channel);
 					messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter);
-					endpoint.setInputChannelName(channelName);
+					Subscription subscription = new Subscription(channel);
 					Schedule schedule = new PollingSchedule(period);
-					endpoint.setSchedule(schedule);
-					if (period > 0) {
-						ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy();
-						concurrencyPolicy.setCoreConcurrency(1);
-						concurrencyPolicy.setMaxConcurrency(1);
-						endpoint.setConcurrencyPolicy(concurrencyPolicy);
-					}
-					return;
+					subscription.setSchedule(schedule);
+					endpoint.setSubscription(subscription);
 				}
 			}
 		});
 	}

-	private void configureDefaultOutputChannel(final Object bean, final String beanName,
+	private void configureDefaultOutput(final Object bean, final String beanName,
 			final MessageEndpoint annotation, final DefaultMessageEndpoint endpoint) {
 		String channelName = annotation.defaultOutput();
 		if (StringUtils.hasText(channelName)) {
@@ -177,8 +170,9 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
 					DefaultTargetAdapter adapter = new DefaultTargetAdapter(target);
 					SimpleChannel channel = new SimpleChannel();
 					String channelName = beanName + "-defaultOutputChannel";
+					Subscription subscription = new Subscription(channel);
 					messageBus.registerChannel(channelName, channel);
-					messageBus.registerTargetAdapter(beanName + "-targetAdapter", adapter);
+					messageBus.registerHandler(beanName + "-targetAdapter", adapter, subscription);
 					endpoint.setDefaultOutputChannelName(channelName);
 					foundDefaultOutput = true;
 					return;
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java
@@ -28,7 +28,7 @@ import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.integration.annotation.Subscriber;
 import org.springframework.integration.bus.MessageBus;
-import org.springframework.integration.endpoint.DefaultMessageEndpoint;
+import org.springframework.integration.bus.Subscription;
 import org.springframework.integration.handler.DefaultMessageHandlerAdapter;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -85,13 +85,10 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor {
 					adapter.setMethodName(method.getName());
 					adapter.setObject(bean);
 					adapter.afterPropertiesSet();
-					DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
-					endpoint.setInputChannelName(channelName);
-					endpoint.setChannelRegistry(messageBus);
-					endpoint.setHandler(adapter);
-					String endpointName = ClassUtils.getShortNameAsProperty(targetClass) + 
+					String adapterName = ClassUtils.getShortNameAsProperty(targetClass) + 
 							"-" + method.getName() + "-endpoint";
-					messageBus.registerEndpoint(endpointName, endpoint);
+					Subscription subscription = new Subscription(channelName);
+					messageBus.registerHandler(adapterName, adapter, subscription);
 				}
 			}
 		});
--- 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
@@ -19,31 +19,30 @@ package org.springframework.integration.endpoint;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.MessagingConfigurationException;
+import org.springframework.integration.bus.Subscription;
 import org.springframework.integration.channel.ChannelRegistry;
+import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.scheduling.Schedule;

 /**
 * Default implementation of the {@link MessageEndpoint} interface.
 * 
 * @author Mark Fisher
 */
-public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware {
+public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryAware, BeanNameAware {

 	private String name;

-	private String inputChannelName;
-
-	private String defaultOutputChannelName;
-
 	private MessageHandler handler;

-	private Schedule schedule;
+	private Subscription subscription;

 	private ConcurrencyPolicy concurrencyPolicy;

+	private String defaultOutputChannelName;
+
 	private ChannelRegistry channelRegistry;


@@ -59,18 +58,8 @@ public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware {
 		this.setName(beanName);
 	}

-	/**
-	 * Set the name of the channel from which this endpoint receives messages.
-	 */
-	public void setInputChannelName(String inputChannelName) {
-		this.inputChannelName = inputChannelName;
-	}
-
-	/**
-	 * Return the name of the channel from which this endpoint receives messages.
-	 */
-	public String getInputChannelName() {
-		return this.inputChannelName;
+	public String getDefaultOutputChannelName() {
+		return this.defaultOutputChannelName;
 	}

 	/**
@@ -80,8 +69,8 @@ public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware {
 		this.defaultOutputChannelName = defaultOutputChannelName;
 	}

-	public String getDefaultOutputChannelName() {
-		return this.defaultOutputChannelName;
+	public MessageHandler getHandler() {
+		return this.handler;
 	}

 	/**
@@ -91,12 +80,12 @@ public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware {
 		this.handler = handler;
 	}

-	public Schedule getSchedule() {
-		return this.schedule;
+	public Subscription getSubscription() {
+		return this.subscription;
 	}

-	public void setSchedule(Schedule schedule) {
-		this.schedule = schedule;
+	public void setSubscription(Subscription subscription) {
+		this.subscription = subscription;
 	}

 	public ConcurrencyPolicy getConcurrencyPolicy() {
@@ -114,7 +103,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware {
 		this.channelRegistry = channelRegistry;
 	}

-	public Message handle(Message<?> message) {
+	public Message<?> handle(Message<?> message) {
 		if (this.handler == null) {
 			if (this.defaultOutputChannelName == null) {
 				throw new MessagingConfigurationException(
--- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
@@ -16,9 +16,8 @@

 package org.springframework.integration.endpoint;

-import org.springframework.integration.channel.ChannelRegistry;
+import org.springframework.integration.bus.Subscription;
 import org.springframework.integration.handler.MessageHandler;
-import org.springframework.integration.scheduling.Schedule;

 /**
 * Base interface for message endpoints.
@@ -27,19 +26,11 @@ import org.springframework.integration.scheduling.Schedule;
 */
 public interface MessageEndpoint extends MessageHandler {

-	void setName(String name);
+	String getName();

-	void setInputChannelName(String inputChannelName);
+	MessageHandler getHandler();

-	String getInputChannelName();
-
-	void setDefaultOutputChannelName(String defaultOutputChannelName);
-
-	String getDefaultOutputChannelName();
-
-	void setChannelRegistry(ChannelRegistry channelRegistry);
-
-	Schedule getSchedule();
+	Subscription getSubscription();

 	ConcurrencyPolicy getConcurrencyPolicy();

--- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java
@@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit;
 import org.junit.Test;

 import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.bus.Subscription;
 import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
@@ -46,15 +47,16 @@ public class DefaultTargetAdapterTests {
 		target.afterPropertiesSet();
 		DefaultTargetAdapter adapter = new DefaultTargetAdapter(target);
 		SimpleChannel channel = new SimpleChannel();
-		adapter.setChannel(channel);
+		Subscription subscription = new Subscription();
+		subscription.setChannel(channel);
 		Message<String> message = new GenericMessage<String>("123", "testing");
 		channel.send(message);
 		assertNull(queue.poll());
 		MessageBus bus = new MessageBus();
 		bus.registerChannel("channel", channel);
-		bus.registerTargetAdapter("targetAdapter", adapter);
+		bus.registerHandler("targetAdapter", adapter, subscription);
 		bus.start();
-		String result = queue.poll(100, TimeUnit.MILLISECONDS);
+		String result = queue.poll(500, TimeUnit.MILLISECONDS);
 		assertNotNull(result);
 		assertEquals("testing", result);
 		bus.stop();
--- 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
@@ -29,13 +29,25 @@
 				<property name="method" value="store"/>
 			</bean>
 		</constructor-arg>
-		<property name="channel" ref="outputChannel"/>
 	</bean>

+	<bean id="targetEndpoint" class="org.springframework.integration.endpoint.DefaultMessageEndpoint">
+		<property name="handler" ref="targetAdapter"/>
+		<property name="subscription">
+			<bean class="org.springframework.integration.bus.Subscription">
+				<constructor-arg ref="outputChannel"/>
+			</bean>
+		</property>
+	</bean>
+
 	<bean id="sink" class="org.springframework.integration.adapter.TestSink"/>

-	<bean id="endpoint" class="org.springframework.integration.endpoint.DefaultMessageEndpoint">
-		<property name="inputChannelName" value="inputChannel"/>
+	<bean class="org.springframework.integration.endpoint.DefaultMessageEndpoint">
+		<property name="subscription">
+			<bean class="org.springframework.integration.bus.Subscription">
+				<constructor-arg ref="inputChannel"/>
+			</bean>
+		</property>
 		<property name="defaultOutputChannelName" value="outputChannel"/>
 	</bean>

--- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java
@@ -26,6 +26,7 @@ import org.junit.Test;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.bus.Subscription;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.message.StringMessage;
@@ -46,10 +47,9 @@ public class ApplicationEventTargetAdapterTests {
 		MessageChannel channel = new SimpleChannel();
 		ApplicationEventTargetAdapter adapter = new ApplicationEventTargetAdapter();
 		adapter.setApplicationEventPublisher(publisher);
-		adapter.setChannel(channel);
 		MessageBus bus = new MessageBus();
 		bus.registerChannel("channel", channel);
-		bus.registerTargetAdapter("adapter", adapter);
+		bus.registerHandler("adapter", adapter, new Subscription(channel));
 		bus.start();
 		assertEquals(1, latch.getCount());
 		channel.send(new StringMessage("123", "testing"));
--- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java
@@ -39,7 +39,6 @@ public class CharacterStreamTargetAdapterTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		adapter.setChannel(channel);
 		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new StringMessage("foo"));
@@ -54,7 +53,6 @@ public class CharacterStreamTargetAdapterTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		adapter.setChannel(channel);
 		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new StringMessage("foo"));
@@ -72,7 +70,6 @@ public class CharacterStreamTargetAdapterTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		adapter.setChannel(channel);
 		adapter.setShouldAppendNewLine(true);
 		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		dispatcherTask.addHandler(adapter);
@@ -92,7 +89,6 @@ public class CharacterStreamTargetAdapterTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		adapter.setChannel(channel);
 		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
 		retriever.setMaxMessagesPerTask(2);
 		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
@@ -109,7 +105,6 @@ public class CharacterStreamTargetAdapterTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		adapter.setChannel(channel);
 		adapter.setShouldAppendNewLine(true);
 		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
 		retriever.setReceiveTimeout(0);
@@ -129,7 +124,6 @@ public class CharacterStreamTargetAdapterTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		adapter.setChannel(channel);
 		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		dispatcherTask.addHandler(adapter);
 		TestObject testObject = new TestObject("foo");
@@ -145,7 +139,6 @@ public class CharacterStreamTargetAdapterTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		adapter.setChannel(channel);
 		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
 		retriever.setReceiveTimeout(0);
 		retriever.setMaxMessagesPerTask(2);
@@ -165,7 +158,6 @@ public class CharacterStreamTargetAdapterTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		adapter.setChannel(channel);
 		adapter.setShouldAppendNewLine(true);
 		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
 		retriever.setReceiveTimeout(0);
--- a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java
@@ -30,7 +30,7 @@ import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
 import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException;
-import org.springframework.integration.endpoint.DefaultMessageEndpoint;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.handler.PooledMessageHandler;
 import org.springframework.integration.message.ErrorMessage;
 import org.springframework.integration.message.Message;
@@ -467,19 +467,18 @@ public class DefaultMessageDispatcherTests {
 		assertEquals("endpoint2 should have accepted the message", 1, counter2.get());
 	}

-	private static class TestEndpoint extends DefaultMessageEndpoint {
+
+	private static class TestEndpoint implements MessageHandler {

 		private AtomicInteger counter;

 		private CountDownLatch latch;

-
 		public TestEndpoint(AtomicInteger counter, CountDownLatch latch) {
 			this.counter = counter;
 			this.latch = latch;
 		}

-		@Override
 		public Message<?> handle(Message<?> message) {
 			counter.incrementAndGet();
 			latch.countDown();
--- a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java
@@ -27,8 +27,7 @@ import org.junit.Test;

 import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
-import org.springframework.integration.endpoint.DefaultMessageEndpoint;
-import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.scheduling.PollingSchedule;
@@ -44,8 +43,7 @@ public class FixedDelayConsumerTests {
 		final AtomicInteger counter = new AtomicInteger(0);
 		final CountDownLatch latch = new CountDownLatch(messagesToSend);
 		SimpleChannel channel = new SimpleChannel();
-		MessageEndpoint endpoint = new DefaultMessageEndpoint() {
-			@Override
+		MessageHandler handler = new MessageHandler() {
 			public Message<?> handle(Message<?> message) {
 				counter.incrementAndGet();
 				latch.countDown();
@@ -55,7 +53,6 @@ public class FixedDelayConsumerTests {
 		MessageBus bus = new MessageBus();
 		bus.initialize();
 		bus.registerChannel("testChannel", channel);
-		bus.registerEndpoint("testEndpoint", endpoint);
 		PollingSchedule schedule = new PollingSchedule(10);
 		schedule.setFixedRate(false);
 		ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy();
@@ -63,10 +60,8 @@ public class FixedDelayConsumerTests {
 		concurrencyPolicy.setMaxConcurrency(1);
 		Subscription subscription = new Subscription();
 		subscription.setSchedule(schedule);
-		subscription.setConcurrencyPolicy(concurrencyPolicy);
-		subscription.setChannel("testChannel");
-		subscription.setHandler("testEndpoint");
-		bus.activateSubscription(subscription);
+		subscription.setChannelName("testChannel");
+		bus.registerHandler("testHandler", handler, subscription, concurrencyPolicy);
 		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
 			channel.send(new GenericMessage<String>(1, "test " + (i+1)));
@@ -81,8 +76,7 @@ public class FixedDelayConsumerTests {
 		final AtomicInteger counter = new AtomicInteger(0);
 		final CountDownLatch latch = new CountDownLatch(messagesToSend);
 		SimpleChannel channel = new SimpleChannel();
-		MessageEndpoint endpoint = new DefaultMessageEndpoint() {
-			@Override
+		MessageHandler handler = new MessageHandler() {
 			public Message<?> handle(Message<?> message) {
 				counter.incrementAndGet();
 				latch.countDown();
@@ -92,14 +86,12 @@ public class FixedDelayConsumerTests {
 		MessageBus bus = new MessageBus();
 		bus.initialize();
 		bus.registerChannel("testChannel", channel);
-		bus.registerEndpoint("testEndpoint", endpoint);
 		PollingSchedule schedule = new PollingSchedule(10);
 		schedule.setFixedRate(false);
-		Subscription subscription = new Subscription();
-		subscription.setChannel("testChannel");
-		subscription.setHandler("testEndpoint");
+		Subscription subscription = new Subscription(channel);
+		subscription.setChannelName("testChannel");
 		subscription.setSchedule(schedule);
-		bus.activateSubscription(subscription);
+		bus.registerHandler("testHandler", handler, subscription);
 		for (int i = 0; i < messagesToSend; i++) {
 			channel.send(new GenericMessage<String>(1, "test " + (i+1)));
 		}
--- a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java
@@ -27,8 +27,7 @@ import org.junit.Test;

 import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
-import org.springframework.integration.endpoint.DefaultMessageEndpoint;
-import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.scheduling.PollingSchedule;
@@ -44,8 +43,7 @@ public class FixedRateConsumerTests {
 		final AtomicInteger counter = new AtomicInteger(0);
 		final CountDownLatch latch = new CountDownLatch(messagesToSend);
 		SimpleChannel channel = new SimpleChannel();
-		MessageEndpoint endpoint = new DefaultMessageEndpoint() {
-			@Override
+		MessageHandler handler = new MessageHandler() {
 			public Message<?> handle(Message<?> message) {
 				counter.incrementAndGet();
 				latch.countDown();
@@ -55,14 +53,12 @@ public class FixedRateConsumerTests {
 		MessageBus bus = new MessageBus();
 		bus.initialize();
 		bus.registerChannel("testChannel", channel);
-		bus.registerEndpoint("testEndpoint", endpoint);
 		PollingSchedule schedule = new PollingSchedule(10);
 		schedule.setFixedRate(true);
 		Subscription subscription = new Subscription();
-		subscription.setChannel("testChannel");
-		subscription.setHandler("testEndpoint");
+		subscription.setChannelName("testChannel");
 		subscription.setSchedule(schedule);
-		bus.activateSubscription(subscription);
+		bus.registerHandler("testHandler", handler, subscription);
 		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
 			channel.send(new GenericMessage<String>(1, "test " + (i+1)));
@@ -77,8 +73,7 @@ public class FixedRateConsumerTests {
 		final AtomicInteger counter = new AtomicInteger(0);
 		final CountDownLatch latch = new CountDownLatch(messagesToSend);
 		SimpleChannel channel = new SimpleChannel();
-		MessageEndpoint endpoint = new DefaultMessageEndpoint() {
-			@Override
+		MessageHandler handler = new MessageHandler() {
 			public Message<?> handle(Message<?> message) {
 				counter.incrementAndGet();
 				latch.countDown();
@@ -86,20 +81,16 @@ public class FixedRateConsumerTests {
 			}
 		};
 		MessageBus bus = new MessageBus();
-		bus.initialize();
-		bus.registerChannel("testChannel", channel);
-		bus.registerEndpoint("testEndpoint", endpoint);
 		PollingSchedule schedule = new PollingSchedule(5);
 		schedule.setFixedRate(true);
 		ConcurrencyPolicy  concurrencyPolicy = new ConcurrencyPolicy();
 		concurrencyPolicy.setCoreConcurrency(1);
 		concurrencyPolicy.setMaxConcurrency(1);
 		Subscription subscription = new Subscription();
-		subscription.setChannel("testChannel");
-		subscription.setHandler("testEndpoint");
+		subscription.setChannelName("testChannel");
 		subscription.setSchedule(schedule);
-		subscription.setConcurrencyPolicy(concurrencyPolicy);
-		bus.activateSubscription(subscription);
+		bus.registerChannel("testChannel", channel);
+		bus.registerHandler("testHandler", handler, subscription, concurrencyPolicy);
 		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
 			channel.send(new GenericMessage<String>(1, "test " + (i+1)));
--- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
@@ -33,7 +33,7 @@ import org.springframework.integration.adapter.PollingSourceAdapter;
 import org.springframework.integration.adapter.SourceAdapter;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
-import org.springframework.integration.endpoint.DefaultMessageEndpoint;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.ErrorMessage;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
@@ -45,25 +45,30 @@ import org.springframework.integration.message.StringMessage;
 public class MessageBusTests {

 	@Test
-	public void testChannelsConnectedWithEndpoint() {
+	public void testOutputChannel() {
 		MessageBus bus = new MessageBus();
 		MessageChannel sourceChannel = new SimpleChannel();
 		MessageChannel targetChannel = new SimpleChannel();
 		bus.registerChannel("sourceChannel", sourceChannel);
-		sourceChannel.send(new StringMessage("123", "test"));
+		StringMessage message = new StringMessage("test");
+		message.getHeader().setReplyChannelName("targetChannel");
+		sourceChannel.send(message);
 		bus.registerChannel("targetChannel", targetChannel);
-		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
-		endpoint.setInputChannelName("sourceChannel");
-		endpoint.setDefaultOutputChannelName("targetChannel");
-		bus.registerEndpoint("endpoint", endpoint);
+		MessageHandler handler = new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				return message;
+			}
+		};
+		Subscription subscription = new Subscription(sourceChannel);
+		bus.registerHandler("handler", handler, subscription);
 		bus.start();
-		Message<?> result = targetChannel.receive(100);
+		Message<?> result = targetChannel.receive(3000);
 		assertEquals("test", result.getPayload());
 		bus.stop();
 	}

 	@Test
-	public void testChannelsWithoutEndpoint() {
+	public void testChannelsWithoutHandlers() {
 		MessageBus bus = new MessageBus();
 		MessageChannel sourceChannel = new SimpleChannel();
 		sourceChannel.send(new StringMessage("123", "test"));
@@ -84,31 +89,34 @@ public class MessageBusTests {
 		sourceChannel.send(new GenericMessage<String>("123", "test"));		
 		MessageChannel targetChannel = (MessageChannel) context.getBean("targetChannel");
 		MessageBus bus = (MessageBus) context.getBean("bus");
-		Subscription subscription = new Subscription();
-		subscription.setChannel("sourceChannel");
-		subscription.setHandler("endpoint");
-		bus.activateSubscription(subscription);
-		Message<?> result = targetChannel.receive(100);
+		bus.start();
+		Message<?> result = targetChannel.receive(1000);
 		assertEquals("test", result.getPayload());
 	}

 	@Test
-	public void testExactlyOneEndpointReceivesUnicastMessage() {
+	public void testExactlyOneHandlerReceivesPointToPointMessage() {
 		SimpleChannel inputChannel = new SimpleChannel();
 		SimpleChannel outputChannel1 = new SimpleChannel();
 		SimpleChannel outputChannel2 = new SimpleChannel();
-		DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint();
-		endpoint1.setDefaultOutputChannelName("output1");
-		endpoint1.setInputChannelName("input");
-		DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint();
-		endpoint2.setDefaultOutputChannelName("output2");
-		endpoint2.setInputChannelName("input");
+		MessageHandler handler1 = new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				message.getHeader().setReplyChannelName("output1");
+				return message;
+			}
+		};
+		MessageHandler handler2 = new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				message.getHeader().setReplyChannelName("output2");
+				return message;
+			}
+		};
 		MessageBus bus = new MessageBus();
 		bus.registerChannel("input", inputChannel);
 		bus.registerChannel("output1", outputChannel1);
 		bus.registerChannel("output2", outputChannel2);
-		bus.registerEndpoint("endpoint1", endpoint1);
-		bus.registerEndpoint("endpoint2", endpoint2);
+		bus.registerHandler("handler1", handler1, new Subscription(inputChannel));
+		bus.registerHandler("handler2", handler2, new Subscription(inputChannel));
 		bus.start();
 		inputChannel.send(new StringMessage(1, "testing"));
 		Message<?> message1 = outputChannel1.receive(100);
@@ -117,6 +125,39 @@ public class MessageBusTests {
 		assertTrue("exactly one message should be null", message1 == null ^ message2 == null);
 	}

+	@Test
+	public void testBothHandlersReceivePublishSubscribeMessage() {
+		SimpleChannel inputChannel = new SimpleChannel();
+		inputChannel.setBroadcaster(true);
+		SimpleChannel outputChannel1 = new SimpleChannel();
+		SimpleChannel outputChannel2 = new SimpleChannel();
+		MessageHandler handler1 = new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				message.getHeader().setReplyChannelName("output1");
+				return message;
+			}
+		};
+		MessageHandler handler2 = new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				message.getHeader().setReplyChannelName("output2");
+				return message;
+			}
+		};
+		MessageBus bus = new MessageBus();
+		bus.registerChannel("input", inputChannel);
+		bus.registerChannel("output1", outputChannel1);
+		bus.registerChannel("output2", outputChannel2);
+		bus.registerHandler("handler1", handler1, new Subscription(inputChannel));
+		bus.registerHandler("handler2", handler2, new Subscription(inputChannel));
+		bus.start();
+		inputChannel.send(new StringMessage(1, "testing"));
+		Message<?> message1 = outputChannel1.receive(100);
+		Message<?> message2 = outputChannel2.receive(0);
+		bus.stop();
+		assertTrue("both handlers should have received and replied to the message",
+				(message1 != null && message2 != null));
+	}
+
 	@Test
 	public void testInvalidMessageChannelWithFailedDispatch() throws InterruptedException {
 		MessageBus bus = new MessageBus();
--- 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,7 +11,11 @@
 	<bean id="targetChannel" class="org.springframework.integration.channel.SimpleChannel"/>

 	<bean id="endpoint" class="org.springframework.integration.endpoint.DefaultMessageEndpoint">
-		<property name="inputChannelName" value="sourceChannel"/>
+		<property name="subscription">
+			<bean class="org.springframework.integration.bus.Subscription">
+				<constructor-arg ref="sourceChannel"/>
+			</bean>
+		</property>
 		<property name="defaultOutputChannelName" value="targetChannel"/>
 	</bean>

--- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java
@@ -21,7 +21,8 @@ import static org.junit.Assert.assertNotNull;

 import org.junit.Test;

-import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.channel.ChannelRegistry;
+import org.springframework.integration.channel.DefaultChannelRegistry;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.handler.MessageHandler;
@@ -35,49 +36,40 @@ public class DefaultMessageEndpointTests {

 	@Test
 	public void testDefaultReplyChannel() throws Exception {
-		MessageChannel channel = new SimpleChannel();
 		MessageChannel replyChannel = new SimpleChannel();
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		channelRegistry.registerChannel("replyChannel", replyChannel);
 		MessageHandler handler = new MessageHandler() {
 			public Message<String> handle(Message<?> message) {
 				return new StringMessage("123", "hello " + message.getPayload());
 			}
 		};
 		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
-		endpoint.setInputChannelName("testChannel");
+		endpoint.setChannelRegistry(channelRegistry);
 		endpoint.setHandler(handler);
 		endpoint.setDefaultOutputChannelName("replyChannel");
-		MessageBus bus = new MessageBus();
-		bus.registerChannel("testChannel", channel);
-		bus.registerEndpoint("testEndpoint", endpoint);
-		bus.registerChannel("replyChannel", replyChannel);
-		bus.start();
-		StringMessage testMessage = new StringMessage(1, "test");
-		channel.send(testMessage);
-		Message<String> reply = replyChannel.receive(50);
+		endpoint.handle(new StringMessage(1, "test"));
+		Message<?> reply = replyChannel.receive(50);
 		assertNotNull(reply);
 		assertEquals("hello test", reply.getPayload());
 	}

 	@Test
 	public void testExplicitReplyChannel() throws Exception {
-		MessageChannel channel = new SimpleChannel();
 		final MessageChannel replyChannel = new SimpleChannel();
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		channelRegistry.registerChannel("replyChannel", replyChannel);
 		MessageHandler handler = new MessageHandler() {
 			public Message<?> handle(Message<?> message) {
 				return new StringMessage("123", "hello " + message.getPayload());
 			}
 		};
 		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
-		endpoint.setInputChannelName("testChannel");
+		endpoint.setChannelRegistry(channelRegistry);
 		endpoint.setHandler(handler);
-		MessageBus bus = new MessageBus();
-		bus.registerChannel("testChannel", channel);
-		bus.registerEndpoint("testEndpoint", endpoint);
-		bus.registerChannel("replyChannel", replyChannel);
-		bus.start();
 		StringMessage testMessage = new StringMessage(1, "test");
 		testMessage.getHeader().setReplyChannelName("replyChannel");
-		channel.send(testMessage);
+		endpoint.handle(testMessage);
 		Message<?> reply = replyChannel.receive(50);
 		assertNotNull(reply);
 		assertEquals("hello test", reply.getPayload());