Refactored message dispatchers, executors, and the message bus to support target adapters in addition to message endpoints.

spring-integration-core/src/main/java/org/springframework/integration/adapter/DefaultTargetAdapter.java

@@ -0,0 +1,84 @@
+/*
+ * 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.bus.ConsumerPolicy;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageMapper;
+import org.springframework.integration.message.SimplePayloadMessageMapper;
+import org.springframework.util.Assert;
+
+/**
+ * Target adapter implementation that delegates to a {@link MessageMapper}
+ * and then passes the resulting object to the provided {@link Target}.
+ * 
+ * @author Mark Fisher
+ */
+public class DefaultTargetAdapter implements TargetAdapter {
+
+	private String name;
+
+	private MessageChannel channel;
+
+	private Target target;
+
+	private MessageMapper mapper = new SimplePayloadMessageMapper();
+
+	private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(5);
+
+
+	public DefaultTargetAdapter(Target target) {
+		this.target = target;
+	}
+
+	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 mapper) {
+		Assert.notNull(mapper, "'mapper' must not be null");
+		this.mapper = mapper;
+	}
+
+	public void setConsumerPolicy(ConsumerPolicy policy) {
+		Assert.notNull(policy, "'policy' must not be null");
+		this.policy = policy;
+	}
+
+	public ConsumerPolicy getConsumerPolicy() {
+		return this.policy;
+	}
+
+	public void messageReceived(Message<?> message) {
+		this.target.send(this.mapper.fromMessage(message));
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingTarget.java

@@ -0,0 +1,78 @@
+/*
+ * 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.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.endpoint.ArgumentListPreparer;
+import org.springframework.integration.endpoint.SimpleMethodInvoker;
+import org.springframework.util.Assert;
+
+/**
+ * A messaging target that invokes the specified method on the provided object.
+ * 
+ * @author Mark Fisher
+ */
+public class MethodInvokingTarget<T> implements Target<Object>, InitializingBean {
+
+	private Log logger = LogFactory.getLog(this.getClass());
+
+	private T object;
+
+	private String method;
+
+	private SimpleMethodInvoker<T> invoker;
+
+	private ArgumentListPreparer argumentListPreparer;
+
+
+	public void setObject(T object) {
+		Assert.notNull(object, "'object' must not be null");
+		this.object = object;
+	}
+
+	public void setMethod(String method) {
+		Assert.notNull(method, "'method' must not be null");
+		this.method = method;
+	}
+
+	public void setArgumentListPreparer(ArgumentListPreparer argumentListPreparer) {
+		this.argumentListPreparer = argumentListPreparer;
+	}
+
+	public void afterPropertiesSet() {
+		this.invoker = new SimpleMethodInvoker<T>(this.object, this.method);
+	}
+
+	public boolean send(Object object) {
+		Object args[] = null;
+		if (this.argumentListPreparer != null) {
+			args = this.argumentListPreparer.prepare(object);
+		}
+		else {
+			args = new Object[] { object };
+		}
+		Object result = this.invoker.invokeMethod(args);
+		if (result != null && logger.isWarnEnabled()) {
+			logger.warn("ignoring outbound channel adapter's return value");
+		}
+		return true;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java

@@ -86,7 +86,7 @@ public class PollingSourceAdapter<T> implements SourceAdapter, MessageDispatcher
 		return this.policy;
 	}
 
-	public int receiveAndDispatch() {
+	public int dispatch() {
 		int messagesProcessed = 0;
 		int limit = this.policy.getMaxMessagesPerTask();
 		Collection<T> results = this.source.poll(limit);

spring-integration-core/src/main/java/org/springframework/integration/adapter/Target.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.adapter;
+
+/**
+ * Interface for any external target that may receive data from an outgoing
+ * channel adapter.
+ * 
+ * @author Mark Fisher
+ */
+public interface Target<T> {
+
+	boolean send(T t);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.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.adapter;
+
+import org.springframework.integration.bus.ConsumerPolicy;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.MessageReceiver;
+
+/**
+ * Base interface for target adapters.
+ * 
+ * @author Mark Fisher
+ */
+public interface TargetAdapter extends MessageReceiver {
+
+	String getName();
+
+	void setChannel(MessageChannel channel);
+
+	ConsumerPolicy getConsumerPolicy();
+
+}

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

@@ -37,7 +37,7 @@ public abstract class AbstractMessageDispatcher implements MessageDispatcher {
 
 	private MessageRetriever retriever;
 
-	private List<EndpointExecutor> endpointExecutors = new CopyOnWriteArrayList<EndpointExecutor>();
+	private List<MessageReceivingExecutor> executors = new CopyOnWriteArrayList<MessageReceivingExecutor>();
 
 
 	public AbstractMessageDispatcher(MessageRetriever retriever) {
@@ -45,20 +45,21 @@ public abstract class AbstractMessageDispatcher implements MessageDispatcher {
 	}
 
 
-	public void addEndpointExecutor(EndpointExecutor executor) {
+	public void addExecutor(MessageReceivingExecutor executor) {
 		executor.start();
-		this.endpointExecutors.add(executor);
+		this.executors.add(executor);
 	}
 
-	protected List<EndpointExecutor> getEndpointExecutors() {
-		return this.endpointExecutors;
+	protected List<MessageReceivingExecutor> getExecutors() {
+		return this.executors;
 	}
 
 	/**
-	 * Receives messages and dispatches to the endpoints. Returns the number of
-	 * messages processed.
+	 * Retrieves messages and dispatches to the executors.
+	 * 
+	 * @return the number of messages processed
 	 */
-	public int receiveAndDispatch() {
+	public int dispatch() {
 		int messagesProcessed = 0;
 		Collection<Message<?>> messages = this.retriever.retrieveMessages();
 		if (messages == null) {

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

@@ -30,12 +30,15 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.MessagingException;
+import org.springframework.integration.adapter.DefaultTargetAdapter;
 import org.springframework.integration.adapter.SourceAdapter;
+import org.springframework.integration.adapter.TargetAdapter;
 import org.springframework.integration.channel.ChannelRegistry;
+import org.springframework.integration.channel.DefaultChannelRegistry;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PointToPointChannel;
-import org.springframework.integration.channel.DefaultChannelRegistry;
 import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.message.MessageReceiver;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -53,9 +56,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 
 	private Map<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
 
+	private Map<String, TargetAdapter> targetAdapters = new ConcurrentHashMap<String, TargetAdapter>();
+
 	private List<DispatcherTask> dispatcherTasks = new CopyOnWriteArrayList<DispatcherTask>();
 
-	private Map<MessageEndpoint, EndpointExecutor> endpointExecutors = new ConcurrentHashMap<MessageEndpoint, EndpointExecutor>();
+	private Map<MessageReceiver, MessageReceivingExecutor> receiverExecutors = new ConcurrentHashMap<MessageReceiver, MessageReceivingExecutor>();
 
 	private ScheduledThreadPoolExecutor dispatcherExecutor;
 
@@ -70,6 +75,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		Assert.notNull(applicationContext, "applicationContext must not be null");
 		this.registerChannels(applicationContext);
 		this.registerEndpoints(applicationContext);
+		this.registerSourceAdapters(applicationContext);
+		this.registerTargetAdapters(applicationContext);
 		this.activateSubscriptions(applicationContext);
 	}
 
@@ -95,6 +102,24 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 	}
 
+	@SuppressWarnings("unchecked")
+	private void registerSourceAdapters(ApplicationContext context) {
+		Map<String, SourceAdapter> sourceAdapterBeans =
+				(Map<String, SourceAdapter>) context.getBeansOfType(SourceAdapter.class);
+		for (Map.Entry<String, SourceAdapter> entry : sourceAdapterBeans.entrySet()) {
+			this.registerSourceAdapter(entry.getKey(), entry.getValue());
+		}
+	}
+
+	@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 =
@@ -151,7 +176,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 	}
 
-	public void registerSourceAdapter(SourceAdapter adapter) {
+	public void registerSourceAdapter(String name, SourceAdapter adapter) {
+		// TODO: use the name
 		if (adapter instanceof MessageDispatcher) {
 			ConsumerPolicy policy = adapter.getConsumerPolicy();
 			DispatcherTask dispatcherTask = new DispatcherTask((MessageDispatcher) adapter, policy);
@@ -159,6 +185,20 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 	}
 
+	public void registerTargetAdapter(String name, TargetAdapter targetAdapter) {
+		if (targetAdapter instanceof DefaultTargetAdapter) {
+			DefaultTargetAdapter adapter = (DefaultTargetAdapter) targetAdapter;
+			adapter.setName(name);
+			this.targetAdapters.put(name, targetAdapter);
+			MessageChannel channel = adapter.getChannel();
+			ConsumerPolicy policy = adapter.getConsumerPolicy();
+			MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy);
+			UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy);
+			dispatcher.addExecutor(new MessageReceivingExecutor(adapter, policy.getConcurrency(), policy.getMaxConcurrency()));
+			this.addDispatcherTask(new DispatcherTask(dispatcher, policy));
+		}
+	}
+
 	public void activateSubscription(Subscription subscription) {
 		String channelName = subscription.getChannel();
 		String endpointName = subscription.getEndpoint();
@@ -183,19 +223,19 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 			logger.info("activated subscription to channel '" + channelName + 
 					"' for endpoint '" + endpointName + "'");
 		}
-		EndpointExecutor endpointExecutor = new EndpointExecutor(endpoint, policy.getConcurrency(), policy.getMaxConcurrency());
-		endpointExecutors.put(endpoint, endpointExecutor);
+		MessageReceivingExecutor executor = new MessageReceivingExecutor(endpoint, policy.getConcurrency(), policy.getMaxConcurrency());
+		receiverExecutors.put(endpoint, executor);
 		MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy);
 		UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy);
-		dispatcher.addEndpointExecutor(endpointExecutor);
+		dispatcher.addExecutor(executor);
 		DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy);
 		if (this.isRunning()) {
-			endpointExecutor.start();
+			executor.start();
 		}
 		this.addDispatcherTask(dispatcherTask);
 		if (this.logger.isInfoEnabled()) {
 			logger.info("registered dispatcher task: channel='" +
-					channelName + "' endpoint='" + endpointName + "'");
+					channelName + "' receiver='" + endpointName + "'");
 		}
 	}
 
@@ -209,10 +249,13 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 	}
 
-	public int getActiveCountForEndpoint(String endpointName) {
-		MessageEndpoint endpoint = this.endpoints.get(endpointName);
-		if (endpoint != null) {
-			EndpointExecutor executor = this.endpointExecutors.get(endpoint);
+	public int getActiveCountForReceiver(String receiverName) {
+		MessageReceiver receiver = this.endpoints.get(receiverName);
+		if (receiver == null) {
+			receiver = this.targetAdapters.get(receiverName);
+		}
+		if (receiver != null) {
+			MessageReceivingExecutor executor = this.receiverExecutors.get(receiver);
 			if (executor != null) {
 				return executor.getActiveCount();
 			}
@@ -252,7 +295,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		synchronized (this.lifecycleMonitor) {
 			if (!this.isRunning()) {
 				this.running = true;
-				for (EndpointExecutor executor : endpointExecutors.values()) {
+				for (MessageReceivingExecutor executor : receiverExecutors.values()) {
 					executor.start();
 				}
 				for (DispatcherTask task : this.dispatcherTasks) {
@@ -266,7 +309,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		synchronized (this.lifecycleMonitor) {
 			if (this.isRunning()) {
 				this.running = false;
-				for (EndpointExecutor executor : endpointExecutors.values()) {
+				for (MessageReceivingExecutor executor : receiverExecutors.values()) {
 					executor.stop();
 				}
 				this.dispatcherExecutor.shutdownNow();
@@ -293,7 +336,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 
 		public void run() {
-			dispatcher.receiveAndDispatch();
+			dispatcher.dispatch();
 		}
 
 	}

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

@@ -23,6 +23,6 @@ package org.springframework.integration.bus;
  */
 public interface MessageDispatcher {
 
-	int receiveAndDispatch();
+	int dispatch();
 
 }

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

@@ -25,8 +25,8 @@ import org.apache.commons.logging.LogFactory;
 
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.MessageHandlingException;
-import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageReceiver;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.util.Assert;
 
@@ -35,11 +35,11 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class EndpointExecutor implements Lifecycle {
+public class MessageReceivingExecutor implements Lifecycle {
 
 	private Log logger = LogFactory.getLog(this.getClass());
 
-	private MessageEndpoint endpoint;
+	private MessageReceiver receiver;
 
 	private ThreadPoolExecutor threadPoolExecutor;
 
@@ -60,12 +60,12 @@ public class EndpointExecutor implements Lifecycle {
 	private int totalErrorThreshold = -1;
 
 
-	public EndpointExecutor(MessageEndpoint endpoint, int corePoolSize, int maxPoolSize) {
-		Assert.notNull(endpoint, "'endpoint' must not be null");
+	public MessageReceivingExecutor(MessageReceiver receiver, int corePoolSize, int maxPoolSize) {
+		Assert.notNull(receiver, "'receiver' must not be null");
 		Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1");
 		Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1");
 		Assert.isTrue(maxPoolSize >= corePoolSize, "'corePoolSize' cannot exceed 'maxPoolSize'");
-		this.endpoint = endpoint;
+		this.receiver = receiver;
 		this.corePoolSize = corePoolSize;
 		this.maxPoolSize = maxPoolSize;
 	}
@@ -85,7 +85,7 @@ public class EndpointExecutor implements Lifecycle {
 	public void start() {
 		synchronized (this.lifecycleMonitor) {
 			if (!this.running) {
-				this.threadPoolExecutor = new EndpointThreadPoolExecutor(this.corePoolSize, this.maxPoolSize);
+				this.threadPoolExecutor = new MessageReceivingThreadPoolExecutor(this.corePoolSize, this.maxPoolSize);
 			}
 			this.running = true;
 		}
@@ -105,12 +105,12 @@ public class EndpointExecutor implements Lifecycle {
 		if (threadPoolExecutor == null) {
 			throw new MessageHandlingException("executor is not running");
 		}
-		this.threadPoolExecutor.execute(new EndpointTask(this.endpoint, message));
+		this.threadPoolExecutor.execute(new MessageReceivingTask(this.receiver, message));
 	}
 
 	/**
 	 * Set the maximum number of errors allowed in <em>successive</em>
-	 * endpoint executions. If this threshold is ever exceeded, the executor
+	 * executions. If this threshold is ever exceeded, the executor
 	 * will shutdown.
 	 */
 	public void setSuccessiveErrorThreshold(int successiveErrorThreshold) {
@@ -118,9 +118,8 @@ public class EndpointExecutor implements Lifecycle {
 	}
 
 	/**
-	 * Set the maximum number of <em>total</em> errors allowed in endpoint
-	 * executions. If this threshold is ever exceeded, the executor will
-	 * shutdown.
+	 * Set the maximum number of <em>total</em> errors allowed in executions
+	 * If this threshold is ever exceeded, the executor will shutdown.
 	 */
 	public void setTotalErrorThreshold(int totalErrorThreshold) {
 		this.totalErrorThreshold = totalErrorThreshold;
@@ -138,17 +137,17 @@ public class EndpointExecutor implements Lifecycle {
 	}
 
 
-	private static class EndpointTask implements Runnable {
+	private static class MessageReceivingTask implements Runnable {
 
-		private MessageEndpoint endpoint;
+		private MessageReceiver receiver;
 
 		private Message<?> message;
 
 		private Throwable error;
 
 
-		EndpointTask(MessageEndpoint endpoint, Message<?> message) {
-			this.endpoint = endpoint;
+		MessageReceivingTask(MessageReceiver receiver, Message<?> message) {
+			this.receiver = receiver;
 			this.message = message;
 		}
 
@@ -158,7 +157,7 @@ public class EndpointExecutor implements Lifecycle {
 
 		public void run() {
 			try {
-				this.endpoint.messageReceived(this.message);
+				this.receiver.messageReceived(this.message);
 			}
 			catch (Throwable t) {
 				this.error = t;
@@ -167,9 +166,9 @@ public class EndpointExecutor implements Lifecycle {
 	}
 
 
-	private class EndpointThreadPoolExecutor extends ThreadPoolExecutor {
+	private class MessageReceivingThreadPoolExecutor extends ThreadPoolExecutor {
 
-		public EndpointThreadPoolExecutor(int corePoolSize, int maximumPoolSize) {
+		public MessageReceivingThreadPoolExecutor(int corePoolSize, int maximumPoolSize) {
 			super(corePoolSize, maximumPoolSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>());
 			CustomizableThreadFactory threadFactory = new CustomizableThreadFactory();
 			threadFactory.setThreadNamePrefix("endpoint-executor-");
@@ -178,10 +177,10 @@ public class EndpointExecutor implements Lifecycle {
 
 		@Override
 		protected void afterExecute(Runnable r, Throwable t) {
-			EndpointTask task = (EndpointTask) r;
+			MessageReceivingTask task = (MessageReceivingTask) r;
 			if (task.getError() != null) {
 				if (logger.isWarnEnabled()) {
-					logger.warn("Exception occurred in endpoint execution", task.getError());
+					logger.warn("Exception occurred during task execution", task.getError());
 				}
 				successiveErrorCount++;
 				totalErrorCount++;

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

@@ -23,7 +23,7 @@ import org.springframework.integration.message.Message;
 
 /**
  * A {@link MessageDispatcher} implementation that dispatches each retrieved
- * {@link Message} to a single {@link EndpointExecutor}.
+ * {@link Message} to a single {@link MessageReceivingExecutor}.
  * 
  * @author Mark Fisher
  */
@@ -41,19 +41,19 @@ public class UnicastMessageDispatcher extends AbstractMessageDispatcher {
 	@Override
 	protected boolean dispatchMessage(Message<?> message) {
 		int attempts = 0;
-		Iterator<EndpointExecutor> iter = this.getEndpointExecutors().iterator();
+		Iterator<MessageReceivingExecutor> iter = this.getExecutors().iterator();
 		if (!iter.hasNext()) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("dispatcher has no active endpoint executors");
+				logger.warn("dispatcher has no active executors");
 			}
 			return false;
 		}
 		while (iter.hasNext()) {
-			EndpointExecutor executor = iter.next();
+			MessageReceivingExecutor executor = iter.next();
 			try {
 				if (executor == null || !executor.isRunning()) {
 					if (logger.isInfoEnabled()) {
-						logger.info("removing inactive endpoint executor");
+						logger.info("removing inactive executor");
 					}
 					iter.remove();
 					continue;

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

@@ -25,8 +25,10 @@ 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.endpoint.InboundMethodInvokingChannelAdapter;
-import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter;
+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.util.StringUtils;
 
 /**
@@ -52,21 +54,27 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
 
 
 	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)) {
 			throw new MessagingConfigurationException("'ref' and 'method' are both required");
 		}
-		adapterDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref));
-		adapterDef.getPropertyValues().addPropertyValue("method", method);
+		RootBeanDefinition adapterDef = null;
+		RootBeanDefinition invokerDef = null;
+		if (this.isInbound) {
+			adapterDef = new RootBeanDefinition(PollingSourceAdapter.class);
+			invokerDef = new RootBeanDefinition(MethodInvokingSource.class);
+		}
+		else {
+			adapterDef = new RootBeanDefinition(DefaultTargetAdapter.class);
+			invokerDef = new RootBeanDefinition(MethodInvokingTarget.class);
+		}
+		invokerDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref));
+		invokerDef.getPropertyValues().addPropertyValue("method", method);
+		String invokerBeanName = parserContext.getReaderContext().generateBeanName(invokerDef);
+		parserContext.registerBeanComponent(new BeanComponentDefinition(invokerDef, invokerBeanName));
+		adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName));
+		adapterDef.setSource(parserContext.extractSource(element));
 		String beanName = element.getAttribute(ID_ATTRIBUTE);
 		if (!StringUtils.hasText(beanName)) {
 			beanName = parserContext.getReaderContext().generateBeanName(adapterDef);

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

@@ -1,85 +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.config;
-
-import org.springframework.beans.factory.config.RuntimeBeanReference;
-import org.springframework.beans.factory.support.BeanDefinitionRegistry;
-import org.springframework.beans.factory.support.BeanNameGenerator;
-import org.springframework.beans.factory.support.DefaultBeanNameGenerator;
-import org.springframework.beans.factory.support.RootBeanDefinition;
-import org.springframework.integration.endpoint.GenericMessageEndpoint;
-import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter;
-import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter;
-import org.springframework.integration.handler.DefaultMessageHandlerAdapter;
-import org.springframework.util.Assert;
-
-/**
- * Factory for creating integration component bean definitions.
- * 
- * @author Mark Fisher
- */
-public class ComponentConfigurer {
-
-	private BeanDefinitionRegistry registry;
-
-	private BeanNameGenerator beanNameGenerator;
-
-
-	public ComponentConfigurer(BeanDefinitionRegistry registry, BeanNameGenerator beanNameGenerator) {
-		Assert.notNull(registry, "registry must not be null");
-		this.registry = registry;
-		this.beanNameGenerator = (beanNameGenerator != null ? beanNameGenerator : new DefaultBeanNameGenerator());
-	}
-
-	public String serviceActivator(String inputChannel, String outputChannel, String objectRef, String method) {
-		RootBeanDefinition endpointDef = new RootBeanDefinition(GenericMessageEndpoint.class);
-		RootBeanDefinition adapterDef = new RootBeanDefinition(DefaultMessageHandlerAdapter.class);
-		adapterDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef));
-		adapterDef.getPropertyValues().addPropertyValue("methodName", method);
-		String adapterName = beanNameGenerator.generateBeanName(adapterDef, this.registry);
-		this.registry.registerBeanDefinition(adapterName, adapterDef);
-		endpointDef.getPropertyValues().addPropertyValue("handler", new RuntimeBeanReference(adapterName));
-		if (inputChannel != null) {
-			endpointDef.getPropertyValues().addPropertyValue("inputChannelName", inputChannel);
-		}
-		if (outputChannel != null) {
-			endpointDef.getPropertyValues().addPropertyValue("defaultOutputChannelName", outputChannel);
-		}
-		String endpointName = beanNameGenerator.generateBeanName(endpointDef, this.registry);
-		this.registry.registerBeanDefinition(endpointName, endpointDef);
-		return endpointName;
-	}
-
-	public String inboundChannelAdapter(String objectRef, String method) {
-		RootBeanDefinition bd = new RootBeanDefinition(InboundMethodInvokingChannelAdapter.class);
-		bd.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef));
-		bd.getPropertyValues().addPropertyValue("method", method);
-		String beanName = this.beanNameGenerator.generateBeanName(bd, this.registry);
-		this.registry.registerBeanDefinition(beanName, bd);
-		return beanName;
-	}
-
-	public String outboundChannelAdapter(String objectRef, String method) {
-		RootBeanDefinition bd = new RootBeanDefinition(OutboundMethodInvokingChannelAdapter.class);
-		bd.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef));
-		bd.getPropertyValues().addPropertyValue("method", method);
-		String beanName = this.beanNameGenerator.generateBeanName(bd, this.registry);
-		this.registry.registerBeanDefinition(beanName, bd);
-		return beanName;
-	}
-
-}

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

@@ -33,6 +33,8 @@ import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.core.OrderComparator;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.integration.MessagingConfigurationException;
+import org.springframework.integration.adapter.MethodInvokingSource;
+import org.springframework.integration.adapter.PollingSourceAdapter;
 import org.springframework.integration.annotation.DefaultOutput;
 import org.springframework.integration.annotation.Handler;
 import org.springframework.integration.annotation.MessageEndpoint;
@@ -42,8 +44,9 @@ import org.springframework.integration.annotation.Splitter;
 import org.springframework.integration.bus.ConsumerPolicy;
 import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.ChannelRegistryAware;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.PointToPointChannel;
 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;
@@ -124,14 +127,18 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
 			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);
 					int period = ((Polled) annotation).period();
+					MethodInvokingSource<Object> source = new MethodInvokingSource<Object>();
+					source.setObject(bean);
+					source.setMethod(method.getName());
+					PollingSourceAdapter<Object> adapter = new PollingSourceAdapter<Object>(source);
+					MessageChannel channel = new PointToPointChannel();
+					adapter.setChannel(channel);
+					adapter.setPeriod(period);
+					String channelName = beanName + "-inputChannel";
+					messageBus.registerChannel(channelName, channel);
+					messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter);
+					endpoint.setInputChannelName(channelName);
 					endpoint.getConsumerPolicy().setPeriod(period);
 					if (period > 0) {
 						endpoint.getConsumerPolicy().setConcurrency(1);

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

@@ -1,71 +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.endpoint;
-
-import java.lang.reflect.Method;
-
-import org.springframework.integration.MessagingConfigurationException;
-import org.springframework.util.Assert;
-
-/**
- * An inbound channel adapter for invoking a no-argument method and receiving
- * its return value.
- * 
- * @author Mark Fisher
- */
-public class InboundMethodInvokingChannelAdapter<T> extends AbstractInboundChannelAdapter {
-
-	private T object;
-
-	private String method;
-
-	private SimpleMethodInvoker<T> invoker;
-
-
-	public void setObject(T object) {
-		Assert.notNull(object, "'object' must not be null");
-		this.object = object;
-	}
-
-	public void setMethod(String method) {
-		Assert.notNull(method, "'method' must not be null");
-		this.method = method;
-	}
-
-	@Override
-	public void initialize() {
-		this.invoker = new SimpleMethodInvoker<T>(this.object, this.method);
-		this.invoker.setMethodValidator(new MessageReceivingMethodValidator());
-	}
-
-	@Override
-	protected Object doReceiveObject() {
-		return this.invoker.invokeMethod(new Object[] {});
-	}
-
-
-	public static class MessageReceivingMethodValidator implements MethodValidator {
-
-		public void validate(Method method) {
-			if (method.getReturnType().equals(void.class)) {
-				throw new MessagingConfigurationException(
-						"Inbound channel adapter requires a non-void returning method.");
-			}
-		}
-	}
-
-}

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

@@ -18,14 +18,14 @@ package org.springframework.integration.endpoint;
 
 import org.springframework.integration.bus.ConsumerPolicy;
 import org.springframework.integration.channel.ChannelRegistry;
-import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageReceiver;
 
 /**
  * Base interface for message endpoints.
  * 
  * @author Mark Fisher
  */
-public interface MessageEndpoint {
+public interface MessageEndpoint extends MessageReceiver {
 
 	void setInputChannelName(String inputChannelName);
 
@@ -39,6 +39,4 @@ public interface MessageEndpoint {
 
 	void setChannelRegistry(ChannelRegistry channelRegistry);
 
-	void messageReceived(Message<?> message);
-
 }

spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.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.message;
+
+/**
+ * The primary callback interface for any component capable of receiving
+ * messages. This includes message endpoints as well as target adapters.
+ * 
+ * @author Mark Fisher
+ */
+public interface MessageReceiver {
+
+	void messageReceived(Message<?> message);
+
+}