Endpoints now expose a setter for a Poller strategy and no longer implement Lifecycle.

org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java

@@ -263,7 +263,7 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 	private void configureEndpoint(AbstractEndpoint endpoint, String name, Object input, Schedule schedule) {
 		endpoint.setName(name);
 		if (input instanceof MessageChannel) {
-			endpoint.setInputChannel((MessageChannel) input);
+			endpoint.setSource((MessageChannel) input);
 		}
 		else if (input instanceof String) {
 			endpoint.setInputChannelName((String) input);
@@ -322,15 +322,12 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 		if (endpoint.getOutputChannel() == null) {
 			this.lookupOrCreateChannel(endpoint.getOutputChannelName());
 		}
-		try {
-			endpoint.afterPropertiesSet();
-		}
-		catch (Exception e) {
-			throw new ConfigurationException("failed to initialize endpoint", e);
-		}
 		MessageChannel channel = endpoint.getInputChannel();
 		if (channel == null) {
 			channel = this.lookupOrCreateChannel(endpoint.getInputChannelName());
+			if (channel != null) {
+				endpoint.setSource(channel);
+			}
 		}
 		if (channel != null && channel instanceof Subscribable) {
 			((Subscribable) channel).subscribe(endpoint);
@@ -343,6 +340,12 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 		Schedule schedule = endpoint.getSchedule();
 		EndpointTrigger trigger = new EndpointTrigger(schedule != null ? schedule : this.defaultPollerSchedule);
 		trigger.addTarget(endpoint);
+		try {
+			endpoint.afterPropertiesSet();
+		}
+		catch (Exception e) {
+			throw new ConfigurationException("failed to initialize endpoint '" + endpoint + "'", e);
+		}
 		if (this.endpointTriggers.add(trigger)) {
 			this.taskScheduler.schedule(trigger);
 		}

org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java

@@ -74,7 +74,7 @@ public class SourceAnnotationPostProcessor extends AbstractAnnotationMethodPostP
 		if (!StringUtils.hasText(outputChannelName)) {
 			MessageChannel outputChannel = new DirectChannel();
 			this.getMessageBus().registerChannel(beanName + ".output", outputChannel);
-			endpoint.setOutputChannel(outputChannel);
+			endpoint.setTarget(outputChannel);
 		}
 		else {
 			endpoint.setOutputChannelName(outputChannelName);

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java

@@ -23,13 +23,17 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.factory.BeanNameAware;
-import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.ChannelRegistry;
+import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.handler.MessageHandlerNotRunningException;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageRejectedException;
+import org.springframework.integration.message.MessageSource;
 import org.springframework.integration.message.MessageTarget;
+import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.Poller;
+import org.springframework.integration.message.TargetInvoker;
+import org.springframework.integration.message.selector.MessageSelector;
 import org.springframework.integration.scheduling.Schedule;
 
 /**
@@ -37,7 +41,7 @@ import org.springframework.integration.scheduling.Schedule;
  * 
  * @author Mark Fisher
  */
-public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware, Lifecycle {
+public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware {
 
 	protected final Log logger = LogFactory.getLog(this.getClass());
 
@@ -45,26 +49,26 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
 
 	private volatile String inputChannelName;
 
-	private MessageChannel inputChannel;
-
 	private volatile String outputChannelName;
 
-	private MessageChannel outputChannel;
+	private volatile MessageSource<?> source;
 
-	private final List<EndpointInterceptor> interceptors = new ArrayList<EndpointInterceptor>();
+	private volatile MessageTarget target;
+
+	private volatile Poller poller;
 
 	private volatile Schedule schedule;
 
-	private volatile EndpointTrigger trigger;
+	private final TargetInvoker targetInvoker = new TargetInvoker();
+
+	private volatile long sendTimeout;
+
+	private volatile MessageSelector selector;
+
+	private final List<EndpointInterceptor> interceptors = new ArrayList<EndpointInterceptor>();
 
 	private volatile ChannelRegistry channelRegistry;
 
-	private volatile boolean autoStartup = true;
-
-	private volatile boolean running;
-
-	private final Object lifecycleMonitor = new Object();
-
 
 	public String getName() {
 		return this.name;
@@ -86,12 +90,8 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
 		return this.schedule;
 	}
 
-	public void setTrigger(EndpointTrigger trigger) {
-		this.trigger = trigger;
-	}
-
-	public EndpointTrigger getTrigger() {
-		return this.trigger;
+	public void setPoller(Poller poller) {
+		this.poller = poller;
 	}
 
 	public void setInputChannelName(String inputChannelName) {
@@ -102,17 +102,27 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
 		return this.inputChannelName;
 	}
 
-	public void setInputChannel(MessageChannel channel) {
-		this.inputChannel = channel; 
-		this.inputChannelName = channel.getName();
+	public void setSource(MessageSource<?> source) {
+		if (source instanceof MessageChannel) {
+			this.inputChannelName = ((MessageChannel) source).getName();
+		}
+		this.source = source;
 	}
 
 	public MessageChannel getInputChannel() {
-		if (this.inputChannel == null &&
-				(this.inputChannelName != null && this.channelRegistry != null)) {
-			this.inputChannel = this.channelRegistry.lookupChannel(this.inputChannelName);
+		if (this.source != null) {
+			if (this.source instanceof MessageChannel) {
+				return (MessageChannel) this.source;
+			}
 		}
-		return this.inputChannel;
+		else if (this.inputChannelName != null && this.channelRegistry != null) {
+			MessageChannel inputChannel = this.channelRegistry.lookupChannel(this.inputChannelName);
+			if (inputChannel != null) {
+				this.source = inputChannel;
+			}
+			return inputChannel;
+		}
+		return null;
 	}
 
 	/**
@@ -127,17 +137,31 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
 		return this.outputChannelName;
 	}
 
-	public void setOutputChannel(MessageChannel outputChannel) {
-		this.outputChannel = outputChannel;
-		this.outputChannelName = outputChannel.getName();
+	public void setTarget(MessageTarget target) {
+		if (target instanceof MessageChannel) {
+			this.outputChannelName = ((MessageChannel) target).getName();
+		}
+		this.target = target;
+	}
+
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
 	}
 
 	public MessageChannel getOutputChannel() {
-		if (this.outputChannel == null &&
-				(this.outputChannelName != null && this.channelRegistry != null)) {
-			this.outputChannel = this.channelRegistry.lookupChannel(this.outputChannelName);
+		if (this.target != null) {
+			if (this.target instanceof MessageChannel) {
+				return (MessageChannel) this.target;
+			}
 		}
-		return this.outputChannel;
+		else if (this.outputChannelName != null && this.channelRegistry != null) {
+			MessageChannel outputChannel = this.channelRegistry.lookupChannel(this.outputChannelName);
+			if (outputChannel != null) {
+				this.target = outputChannel;
+			}
+			return outputChannel;
+		}
+		return null;
 	}
 
 	/**
@@ -151,8 +175,8 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
 		return this.channelRegistry;
 	}
 
-	public void setAutoStartup(boolean autoStartup) {
-		this.autoStartup = autoStartup;
+	public void setMessageSelector(MessageSelector selector) {
+		this.selector = selector;
 	}
 
 	public void addInterceptor(EndpointInterceptor interceptor) {
@@ -166,10 +190,6 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
 		}
 	}
 
-	public List<EndpointInterceptor> getInterceptors() {
-		return this.interceptors;
-	}
-
 	public String toString() {
 		return (this.name != null) ? this.name : super.toString();
 	}
@@ -177,39 +197,31 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
 	protected void initialize() {
 	}
 
-	public boolean isRunning() {
-		return this.running;
-	}
-
 	public void afterPropertiesSet() {
-		if (this.autoStartup) {
-			this.start();
+		if (this.source != null && this.poller == null) {
+			this.poller = new DefaultEndpointPoller();
 		}
-		else {
-			this.initialize();
+		if (this.target == null) {
+			this.target = this.getOutputChannel();
+		}
+		if (this.target != null && this.target instanceof ChannelRegistryAware
+				&& this.channelRegistry != null) {
+			((ChannelRegistryAware) this.target).setChannelRegistry(this.channelRegistry);
 		}
-	}
-
-	public void start() {
 		this.initialize();
-		synchronized (this.lifecycleMonitor) {
-			if (this.running) {
-				return;
-			}
-			this.running = true;
-		}
-	}
-
-	public void stop() {
-		synchronized (this.lifecycleMonitor) {
-			if (!this.running) {
-				return;
-			}
-			this.running = false;
-		}
 	}
 
 	public final boolean send(Message<?> message) {
+		if (message == null || message.getPayload() == null) {
+			throw new IllegalArgumentException("Message and its payload must not be null.");
+		}
+		if (logger.isDebugEnabled()) {
+			logger.debug("endpoint '" + this + "' handling message: " + message);
+		}
+		if (message.getPayload() instanceof EndpointVisitor) {
+			((EndpointVisitor) message.getPayload()).visitEndpoint(this);
+			return true;
+		}
 		return this.send(message, 0);
 	}
 
@@ -239,27 +251,46 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
 	}
 
 	private boolean doSend(Message<?> message) {
-		if (message == null || message.getPayload() == null) {
-			throw new IllegalArgumentException("Message and its payload must not be null.");
-		}
-		if (logger.isDebugEnabled()) {
-			logger.debug("endpoint '" + this + "' handling message: " + message);
-		}
-		if (!this.isRunning()) {
-			throw new MessageHandlerNotRunningException(message);
-		}
-		if (message.getPayload() instanceof EndpointVisitor) {
-			((EndpointVisitor) message.getPayload()).visitEndpoint(this);
-			return true;
-		}
 		if (!this.supports(message)) {
 			throw new MessageRejectedException(message, "unsupported message");
 		}
-		return this.handleMessage(message);
+		Message<?> result = this.handleMessage(message);
+		if (result != null) {
+			return this.targetInvoker.invoke(this.target, result, this.sendTimeout);
+		}
+		return true;
 	}
 
-	protected abstract boolean supports(Message<?> message);
+	public final boolean poll() {
+		if (this.poller == null) {
+			this.afterPropertiesSet();
+			if (this.poller == null) {
+				throw new MessagingException("endpoint '" + this + "' has no poller");
+			}
+		}
+		if (this.source == null) {
+			throw new MessagingException("endpoint '" + this + "' has no source");
+		}
+		int result = this.poller.poll(this.source, new MessageTarget() {
+			public boolean send(Message<?> message) {
+				return AbstractEndpoint.this.send(message, 0);
+			}
+		});
+		return (result > 0);
+	}
 
-	protected abstract boolean handleMessage(Message<?> message);
+	protected boolean supports(Message<?> message) {
+		if (this.selector != null && !this.selector.accept(message)) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("selector for endpoint '" + this + "' rejected message: " + message);
+			}
+			return false;
+		}
+		return true;
+	}
+
+	protected Message<?> handleMessage(Message<?> message) {
+		return message;
+	}
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java

@@ -25,7 +25,6 @@ import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageDeliveryException;
 import org.springframework.integration.message.MessageHandlingException;
 import org.springframework.integration.message.MessageHeader;
-import org.springframework.integration.message.MessageTarget;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -35,7 +34,7 @@ import org.springframework.util.StringUtils;
  * 
  * @author Mark Fisher
  */
-public class HandlerEndpoint extends TargetEndpoint {
+public class HandlerEndpoint extends AbstractEndpoint {
 
 	private volatile MessageHandler handler;
 
@@ -80,7 +79,6 @@ public class HandlerEndpoint extends TargetEndpoint {
 		if (this.handler instanceof ChannelRegistryAware) {
 			((ChannelRegistryAware) this.handler).setChannelRegistry(this.getChannelRegistry());
 		}
-		super.setTarget(new HandlerInvokingTarget(this.handler, this.replyHandler));
 		super.initialize();
 	}
 
@@ -118,31 +116,16 @@ public class HandlerEndpoint extends TargetEndpoint {
 		return null;
 	}
 
-
-	private static class HandlerInvokingTarget implements MessageTarget {
-
-		private final MessageHandler handler;
-
-		private final ReplyHandler replyHandler;
-
-
-		public HandlerInvokingTarget(MessageHandler handler, ReplyHandler replyHandler) {
-			this.handler = handler;
-			this.replyHandler = replyHandler;
-		}
-
-
-		public boolean send(Message<?> message) {
-			Message<?> replyMessage = this.handler.handle(message);
-			if (replyMessage != null) {
-				if (replyMessage.getHeader().getCorrelationId() == null) {
-					replyMessage.getHeader().setCorrelationId(message.getId());
-				}
-				this.replyHandler.handle(replyMessage, message.getHeader());
+	@Override
+	protected Message<?> handleMessage(Message<?> message) {
+		Message<?> replyMessage = this.handler.handle(message);
+		if (replyMessage != null) {
+			if (replyMessage.getHeader().getCorrelationId() == null) {
+				replyMessage.getHeader().setCorrelationId(message.getId());
 			}
-			return true;
+			this.replyHandler.handle(replyMessage, message.getHeader());
 		}
-
+		return null;
 	}
 
 

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java

@@ -19,6 +19,7 @@ package org.springframework.integration.endpoint;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.MessageSource;
 import org.springframework.integration.message.MessageTarget;
 import org.springframework.integration.scheduling.Schedule;
 
@@ -35,6 +36,10 @@ public interface MessageEndpoint extends MessageTarget, ChannelRegistryAware, In
 
 	Schedule getSchedule();
 
+	void setSource(MessageSource<?> source);
+
+	void setTarget(MessageTarget target);
+
 	String getInputChannelName();
 
 	MessageChannel getInputChannel();

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java

@@ -18,10 +18,6 @@ package org.springframework.integration.endpoint;
 
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.BlockingSource;
-import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageDeliveryAware;
-import org.springframework.integration.message.MessageDeliveryException;
 import org.springframework.integration.message.MessageSource;
 import org.springframework.util.Assert;
 
@@ -33,27 +29,12 @@ import org.springframework.util.Assert;
  */
 public class SourceEndpoint extends AbstractEndpoint {
 
-	private final MessageSource<?> source;
-
-	private volatile long receiveTimeout = 5000;
-
-	private volatile long sendTimeout = -1;
-
-
 	public SourceEndpoint(MessageSource<?> source) {
 		Assert.notNull(source, "source must not be null");
-		this.source = source;
+		this.setSource(source);
 	}
 
 
-	public void setReceiveTimeout(long receiveTimeout) {
-		this.receiveTimeout = receiveTimeout;
-	}
-
-	public void setSendTimeout(long sendTimeout) {
-		this.sendTimeout = sendTimeout;
-	}
-
 	@Override
 	public void initialize() {
 		if (this.getOutputChannelName() == null && this.getOutputChannel() == null) {
@@ -62,32 +43,4 @@ public class SourceEndpoint extends AbstractEndpoint {
 		}
 	}
 
-	@Override
-	protected final boolean supports(Message<?> message) {
-		return false;
-	}
-
-	@Override
-	protected final boolean handleMessage(Message<?> message) {
-		return false;
-	}
-
-	public boolean poll() {
-		Message<?> message = (this.source instanceof BlockingSource && this.receiveTimeout >= 0) ?
-				((BlockingSource<?>) this.source).receive(this.receiveTimeout) : this.source.receive();
-		if (message == null) {
-			return false;
-		}
-		boolean sent = this.getOutputChannel().send(message, this.sendTimeout);
-		if (this.source instanceof MessageDeliveryAware) {
-			if (sent) {
-				((MessageDeliveryAware) this.source).onSend(message);
-			}
-			else {
-				((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message"));
-			}
-		}
-		return sent;
-	}
-
 }

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java

@@ -16,12 +16,7 @@
 
 package org.springframework.integration.endpoint;
 
-import org.springframework.integration.channel.ChannelRegistryAware;
-import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.BlockingTarget;
-import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageTarget;
-import org.springframework.integration.message.selector.MessageSelector;
 import org.springframework.util.Assert;
 
 /**
@@ -31,90 +26,12 @@ import org.springframework.util.Assert;
  */
 public class TargetEndpoint extends AbstractEndpoint {
 
-	private volatile MessageTarget target;
-
-	private volatile MessageSelector selector;
-
-	private volatile long receiveTimeout = 5000;
-
-	private volatile long sendTimeout = 0;
-
-	private volatile boolean initialized;
-
-	private final Object initializationMonitor = new Object();
-
-
 	public TargetEndpoint() {
 	}
 
 	public TargetEndpoint(MessageTarget target) {
 		Assert.notNull(target, "target must not be null");
-		this.target = target;
-	}
-
-
-	public MessageTarget getTarget() {
-		return this.target;
-	}
-
-	public void setTarget(MessageTarget target) {
-		Assert.notNull(target, "target must not be null");
-		this.target = target;
-	}
-
-	public void setMessageSelector(MessageSelector selector) {
-		this.selector = selector;
-	}
-
-	public void setReceiveTimeout(long receiveTimeout) {
-		this.receiveTimeout = receiveTimeout;
-	}
-
-	public void setSendTimeout(long sendTimeout) {
-		this.sendTimeout = sendTimeout;
-	}
-
-	protected void initialize() {
-		synchronized (this.initializationMonitor) {
-	        if (this.initialized) {
-	        	return;
-	        }
-	        if (this.target instanceof ChannelRegistryAware && this.getChannelRegistry() != null) {
-	        	((ChannelRegistryAware) this.target).setChannelRegistry(this.getChannelRegistry());
-	        }
-	        this.initialized = true;
-		}
-	}
-
-	@Override
-	protected boolean supports(Message<?> message) {
-		if (this.selector != null && !this.selector.accept(message)) {
-			if (logger.isDebugEnabled()) {
-				logger.debug("selector for endpoint '" + this + "' rejected message: " + message);
-			}
-			return false;
-		}
-		return true;
-	}
-
-	@Override
-	protected final boolean handleMessage(Message<?> message) {
-		return (this.sendTimeout >= 0 && this.target instanceof BlockingTarget) ?
-				((BlockingTarget) this.target).send(message) : this.target.send(message);
-	}
-
-	public final boolean poll() {
-		MessageChannel channel = this.getInputChannel();
-		if (channel != null) {
-			Message<?> receivedMessage = channel.receive(this.receiveTimeout);
-			if (receivedMessage != null) {
-				return this.handleMessage(receivedMessage);
-			}
-		}
-		else if (logger.isDebugEnabled()) {
-			logger.debug("TargetEndpoint unable to resolve channel '" + this.getInputChannelName() + "'");
-		}
-		return false;
+		this.setTarget(target);
 	}
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java

@@ -213,7 +213,7 @@ public class RequestReplyTemplate implements MessageBusAware {
 			}
 			ReplyMessageCorrelator correlator = new ReplyMessageCorrelator(10);
 			HandlerEndpoint endpoint = new HandlerEndpoint(correlator);
-			endpoint.setInputChannel(this.replyChannel);
+			endpoint.setSource(this.replyChannel);
 			endpoint.setName("internal.correlator." + this);
 			this.endpointRegistry.registerEndpoint(endpoint);
 			this.replyMessageCorrelator = correlator;
@@ -249,11 +249,11 @@ public class RequestReplyTemplate implements MessageBusAware {
         public void setName(String name) {
         }
 
-        public Message receive() {
+        public Message<?> receive() {
 	        return null;
         }
 
-        public Message receive(long timeout) {
+        public Message<?> receive(long timeout) {
 	        return null;
         }