Added MessageExchangeTemplate, AsyncMessageExchangeTemplate, and AsyncMessage. Removed Poller, SimplePoller, DefaultEndpointPoller, SourceInvoker, and TargetInvoker.

org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java

@@ -112,7 +112,7 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe
 					schedule = this.parseSchedule(childElement);
 				}
 				else if (POLLER_ELEMENT.equals(localName)) {
-					builder.addPropertyReference("poller",
+					builder.addPropertyReference("messageExchangeTemplate",
 							IntegrationNamespaceUtils.parsePoller(childElement, parserContext));
 				}
 				else if (INTERCEPTORS_ELEMENT.equals(localName)) {

org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java

@@ -24,7 +24,8 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
 import org.springframework.beans.factory.xml.ParserContext;
-import org.springframework.integration.endpoint.DefaultEndpointPoller;
+import org.springframework.integration.message.AsyncMessageExchangeTemplate;
+import org.springframework.integration.message.MessageExchangeTemplate;
 import org.springframework.transaction.support.DefaultTransactionDefinition;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
@@ -106,7 +107,12 @@ public abstract class IntegrationNamespaceUtils {
 			}
 			return ref;
 		}
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(DefaultEndpointPoller.class);
+		Class<?> beanClass = (StringUtils.hasText(taskExecutorRef)) ?
+				AsyncMessageExchangeTemplate.class : MessageExchangeTemplate.class;
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(beanClass);
+		if (StringUtils.hasText(taskExecutorRef)) {
+			builder.addConstructorArgReference(taskExecutorRef);
+		}
 		if (txElement != null) {
 			builder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
 			builder.addPropertyValue("propagationBehaviorName", DefaultTransactionDefinition.PREFIX_PROPAGATION + txElement.getAttribute("propagation"));
@@ -114,9 +120,6 @@ public abstract class IntegrationNamespaceUtils {
 			builder.addPropertyValue("transactionTimeout", txElement.getAttribute("timeout"));
 			builder.addPropertyValue("transactionReadOnly", txElement.getAttribute("read-only"));
 		}
-		if (StringUtils.hasText(taskExecutorRef)) {
-			builder.addPropertyReference("taskExecutor", taskExecutorRef);
-		}
 		String receiveTimeout = element.getAttribute("receive-timeout");
 		if (StringUtils.hasText(receiveTimeout)) {
 			builder.addPropertyValue("receiveTimeout", Long.parseLong(receiveTimeout));

org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java

@@ -24,8 +24,8 @@ import org.apache.commons.logging.LogFactory;
 
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageExchangeTemplate;
 import org.springframework.integration.message.MessageTarget;
-import org.springframework.integration.message.TargetInvoker;
 
 /**
  * Base class for {@link MessageDispatcher} implementations.
@@ -38,15 +38,13 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
 
 	protected final List<MessageTarget> targets = new CopyOnWriteArrayList<MessageTarget>();
 
-	private volatile long timeout = 0;
-
 	private volatile TaskExecutor taskExecutor;
 
-	private final TargetInvoker targetInvoker = new TargetInvoker();
+	private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate();
 
 
 	public void setTimeout(long timeout) {
-		this.timeout = timeout;
+		this.messageExchangeTemplate.setSendTimeout(timeout);
 	}
 
 	public boolean addTarget(MessageTarget target) {
@@ -71,7 +69,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
 	}
 
 	protected final boolean sendMessageToTarget(Message<?> message, MessageTarget target) {
-		return this.targetInvoker.invoke(target, message, this.timeout);
+		return this.messageExchangeTemplate.send(message, target);
 	}
 
 }

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

@@ -28,13 +28,11 @@ import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageExchangeTemplate;
 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.Subscribable;
-import org.springframework.integration.message.TargetInvoker;
 import org.springframework.integration.message.selector.MessageSelector;
 import org.springframework.integration.scheduling.Schedule;
 
@@ -57,13 +55,9 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 
 	private volatile MessageTarget target;
 
-	private volatile Poller poller;
-
 	private volatile Schedule schedule;
 
-	private final TargetInvoker targetInvoker = new TargetInvoker();
-
-	private volatile long sendTimeout;
+	private volatile MessageExchangeTemplate messageExchangeTemplate;
 
 	private volatile MessageSelector selector;
 
@@ -92,8 +86,8 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 		return this.schedule;
 	}
 
-	public void setPoller(Poller poller) {
-		this.poller = poller;
+	public void setMessageExchangeTemplate(MessageExchangeTemplate messageExchangeTemplate) {
+		this.messageExchangeTemplate = messageExchangeTemplate;
 	}
 
 	public void setInputChannelName(String inputChannelName) {
@@ -147,7 +141,7 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 	}
 
 	public void setSendTimeout(long sendTimeout) {
-		this.sendTimeout = sendTimeout;
+		this.messageExchangeTemplate.setSendTimeout(sendTimeout);
 	}
 
 	public MessageChannel getOutputChannel() {
@@ -200,9 +194,9 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 	}
 
 	public void afterPropertiesSet() {
-		if (this.poller == null && this.source != null
-				&& !(this.source instanceof Subscribable)) {
-			this.poller = new DefaultEndpointPoller();
+		if (this.messageExchangeTemplate == null) {
+			this.messageExchangeTemplate = new MessageExchangeTemplate();
+			this.messageExchangeTemplate.afterPropertiesSet();
 		}
 		if (this.target == null) {
 			this.target = this.getOutputChannel();
@@ -254,32 +248,31 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 	}
 
 	private boolean doSend(Message<?> message) {
+		if (this.messageExchangeTemplate == null) {
+			this.afterPropertiesSet();
+		}
 		if (!this.supports(message)) {
 			throw new MessageRejectedException(message, "unsupported message");
 		}
 		Message<?> result = this.handleMessage(message);
 		if (result != null) {
-			return this.targetInvoker.invoke(this.target, result, this.sendTimeout);
+			return this.messageExchangeTemplate.send(message, this.target);
 		}
 		return true;
 	}
 
 	public final boolean poll() {
-		if (this.poller == null) {
+		if (this.messageExchangeTemplate == 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() {
+		return this.messageExchangeTemplate.receiveAndForward(this.source, new MessageTarget() {
 			public boolean send(Message<?> message) {
 				return AbstractEndpoint.this.send(message, 0);
 			}
 		});
-		return (result > 0);
 	}
 
 	protected boolean supports(Message<?> message) {

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

@@ -1,119 +0,0 @@
-/*
- * Copyright 2002-2008 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * 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 org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.core.task.TaskExecutor;
-import org.springframework.integration.message.MessageSource;
-import org.springframework.integration.message.MessageTarget;
-import org.springframework.integration.message.SimplePoller;
-import org.springframework.transaction.PlatformTransactionManager;
-import org.springframework.transaction.TransactionStatus;
-import org.springframework.transaction.support.TransactionCallback;
-import org.springframework.transaction.support.TransactionTemplate;
-
-/**
- * An extension of the {@link SimplePoller} that adds concurrency and
- * transactional capabilities.
- * 
- * @author Mark Fisher
- */
-public class DefaultEndpointPoller extends SimplePoller implements InitializingBean {
-
-	private final Log logger = LogFactory.getLog(this.getClass());
-
-	private volatile TaskExecutor taskExecutor;
-
-	private volatile PlatformTransactionManager transactionManager;
-
-	private volatile TransactionTemplate transactionTemplate;
-
-	private volatile String propagationBehaviorName = "PROPAGATION_REQUIRED";
-
-	private volatile String isolationLevelName = "ISOLATION_DEFAULT";
-
-	private volatile int transactionTimeout = -1;
-
-	private volatile boolean readOnly = false;
-
-
-	public void setTaskExecutor(TaskExecutor taskExecutor) {
-		this.taskExecutor = taskExecutor;
-	}
-
-	public void setTransactionManager(PlatformTransactionManager transactionManager) {
-		this.transactionManager = transactionManager;
-	}
-
-	public void setPropagationBehaviorName(String propagationBehaviorName) {
-		this.propagationBehaviorName = propagationBehaviorName;
-	}
-
-	public void setIsolationLevelName(String isolationLevelName) {
-		this.isolationLevelName = isolationLevelName;
-	}
-
-	public void setTransactionTimeout(int transactionTimeout) {
-		this.transactionTimeout = transactionTimeout;
-	}
-
-	public void setTransactionReadOnly(boolean readOnly) {
-		this.readOnly = readOnly;
-	}
-
-	public void afterPropertiesSet() {
-		if (this.transactionManager != null) {
-			TransactionTemplate template = new TransactionTemplate(this.transactionManager);
-			template.setPropagationBehaviorName(this.propagationBehaviorName);
-			template.setIsolationLevelName(this.isolationLevelName);
-			template.setTimeout(this.transactionTimeout);
-			template.setReadOnly(this.readOnly);
-			this.transactionTemplate = template;
-		}
-	}
-
-	public int poll(final MessageSource<?> source, final MessageTarget target) {
-		if (this.taskExecutor != null) {
-			this.taskExecutor.execute(new Runnable() {
-				public void run() {
-					doPoll(source, target);
-				}
-			});
-			return 1;
-		}
-		return doPoll(source, target);
-	}
-
-	private int doPoll(final MessageSource<?> source, final MessageTarget target) {
-		if (this.transactionTemplate != null) {
-			int result = (Integer) this.transactionTemplate.execute(new TransactionCallback() {
-				public Object doInTransaction(TransactionStatus status) {
-					if (logger.isDebugEnabled()) {
-						logger.debug("Polling source '" + source + "' within transaction [" + status + "]");
-					}
-					return DefaultEndpointPoller.super.poll(source, target);
-				}
-            });
-			return result;
-		}
-		return super.poll(source, target);
-	}
-
-}

org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessage.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * 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;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageHeaders;
+import org.springframework.integration.message.MessagingException;
+
+/**
+ * A Message wrapper for asynchronous operations. Implements both
+ * Message and Future so that simple blocking invocations on the
+ * Message (e.g. {@link #getPayload()}) are still allowed, while
+ * timeout-aware methods such as {@link #get(long, TimeUnit)}
+ * are also supported.
+ * 
+ * @author Mark Fisher
+ */
+public class AsyncMessage<T> implements Future<Message<T>>, Message<T> {
+
+	private final Future<Message<T>> future;
+
+
+	public AsyncMessage(Future<Message<T>> future) {
+		this.future = future;
+	}
+
+
+	public boolean cancel(boolean mayInterruptIfRunning) {
+		return this.future.cancel(mayInterruptIfRunning);
+	}
+
+	public Message<T> get() throws InterruptedException, ExecutionException {
+		return this.future.get();
+	}
+
+	public Message<T> get(long timeout, TimeUnit unit)
+			throws InterruptedException, ExecutionException, TimeoutException {
+		return this.future.get(timeout, unit);
+	}
+
+	public boolean isCancelled() {
+		return this.future.isCancelled();
+	}
+
+	public boolean isDone() {
+		return this.future.isDone();
+	}
+
+	public MessageHeaders getHeaders() {
+		try {
+			return this.future.get().getHeaders();
+		} catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			return null;
+		} catch (ExecutionException e) {
+			throw new MessagingException("failure occurred in AsyncMessage", e);
+		}
+	}
+
+	public Object getId() {
+		try {
+			return this.future.get().getId();
+		} catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			return null;
+		} catch (ExecutionException e) {
+			throw new MessagingException("failure occurred in AsyncMessage", e);
+		}
+	}
+
+	public T getPayload() {
+		try {
+			return this.future.get().getPayload();
+		} catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			return null;
+		} catch (ExecutionException e) {
+			throw new MessagingException("failure occurred in AsyncMessage", e);
+		}
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * 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;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.FutureTask;
+
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageExchangeTemplate;
+import org.springframework.integration.message.MessageSource;
+import org.springframework.integration.message.MessageTarget;
+import org.springframework.util.Assert;
+
+/**
+ * An asynchronous version of the {@link MessageExchangeTemplate}.
+ * 
+ * @author Mark Fisher
+ */
+public class AsyncMessageExchangeTemplate extends MessageExchangeTemplate {
+
+	private final TaskExecutor taskExecutor;
+
+
+	public AsyncMessageExchangeTemplate(TaskExecutor taskExecutor) {
+		Assert.notNull(taskExecutor, "TaskExecutor must not be null");
+		this.taskExecutor = taskExecutor;
+	}
+
+
+	/**
+	 * Send the provided message to the given target. Note that the actual
+	 * sending occurs asynchronously, so this method will always return
+	 * <code>true</code> unless an exception is thrown by the executor.
+	 */
+	@Override
+	public boolean send(final Message<?> message, final MessageTarget target) {
+		this.taskExecutor.execute(new Runnable() {
+			public void run() {
+				AsyncMessageExchangeTemplate.super.send(message, target);
+			}
+		});
+		return true;
+	}
+
+	/**
+	 * Send the provided message to the given target and receive the
+	 * result as an {@link AsyncMessage}.
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public Message<?> sendAndReceive(final Message<?> request, final MessageTarget target) {
+		FutureTask<Message<?>> task = new FutureTask<Message<?>>(new Callable<Message<?>>() {
+			public Message<?> call() throws Exception {
+				return AsyncMessageExchangeTemplate.super.sendAndReceive(request, target);
+			}
+		});
+		this.taskExecutor.execute(task);
+		return new AsyncMessage(task);
+	}
+
+	/**
+	 * Receive an {@link AsyncMessage} from the provided source.
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public Message<?> receive(final MessageSource<?> source) {
+		FutureTask<Message<?>> task = new FutureTask<Message<?>>(new Callable<Message<?>>() {
+			public Message<?> call() throws Exception {
+				return AsyncMessageExchangeTemplate.super.receive(source);
+			}
+		});
+		this.taskExecutor.execute(task);
+		return new AsyncMessage(task);
+	}
+
+	/**
+	 * Receive a Message from the provided source and if not <code>null</code>,
+	 * send it to the given target. Note that the receive and send operations
+	 * occur asynchronously, so this method will always return <code>true</code>
+	 * unless an exception is thrown by the executor.
+	 */
+	@Override
+	public boolean receiveAndForward(final MessageSource<?> source, final MessageTarget target) {
+		this.taskExecutor.execute(new Runnable() {
+			public void run() {
+				AsyncMessageExchangeTemplate.super.receiveAndForward(source, target);
+			}
+		});
+		return true;
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java

@@ -0,0 +1,295 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * 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;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.message.BlockingSource;
+import org.springframework.integration.message.BlockingTarget;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.MessageSource;
+import org.springframework.integration.message.MessageTarget;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
+
+/**
+ * This is the central class for invoking message exchange operations
+ * across {@link MessageSource}s and {@link MessageTarget}s. It supports
+ * one-way send and receive calls as well as request/reply. Additionally,
+ * the {@link #receiveAndForward(MessageSource, MessageTarget)} method
+ * plays the role of a polling-consumer while actually sending any
+ * received message to an event-driven consumer.
+ * 
+ * <p>To enable transactions, configure the 'transactionManager' property
+ * with a reference to an instance of Spring's {@link PlatformTransactionManager}
+ * strategy and optionally provide the other transactional attributes
+ * (e.g. 'propagationBehaviorName').
+ * 
+ * @author Mark Fisher
+ */
+public class MessageExchangeTemplate implements InitializingBean {
+
+	protected final Log logger = LogFactory.getLog(this.getClass());
+
+	private volatile long sendTimeout = -1;
+
+	private volatile long receiveTimeout = -1;
+
+	private volatile PlatformTransactionManager transactionManager;
+
+	private volatile TransactionTemplate transactionTemplate;
+
+	private volatile String propagationBehaviorName = "PROPAGATION_REQUIRED";
+
+	private volatile String isolationLevelName = "ISOLATION_DEFAULT";
+
+	private volatile int transactionTimeout = -1;
+
+	private volatile boolean readOnly = false;
+
+	private volatile boolean initialized;
+
+	private final Object initializationMonitor = new Object();
+
+
+	/**
+	 * Specify the timeout value to use for send operations.
+	 * Note that this value will only apply to {@link BlockingTarget}s.
+	 * 
+	 * @param sendTimeout the send timeout in milliseconds
+	 */
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
+	}
+
+	/**
+	 * Specify the timeout value to use for receive operations.
+	 * Note that this value will only apply to {@link BlockingSource}s.
+	 *  
+	 * @param receiveTimeout the receive timeout in milliseconds
+	 */
+	public void setReceiveTimeout(long receiveTimeout) {
+		this.receiveTimeout = receiveTimeout;
+	}
+
+	/**
+	 * Specify a transaction manager to use for all exchange operations.
+	 * If none is provided, then the operations will occur without any
+	 * transactional behavior (i.e. there is no default transaction manager).
+	 */
+	public void setTransactionManager(PlatformTransactionManager transactionManager) {
+		this.transactionManager = transactionManager;
+	}
+
+	public void setPropagationBehaviorName(String propagationBehaviorName) {
+		this.propagationBehaviorName = propagationBehaviorName;
+	}
+
+	public void setIsolationLevelName(String isolationLevelName) {
+		this.isolationLevelName = isolationLevelName;
+	}
+
+	public void setTransactionTimeout(int transactionTimeout) {
+		this.transactionTimeout = transactionTimeout;
+	}
+
+	public void setTransactionReadOnly(boolean readOnly) {
+		this.readOnly = readOnly;
+	}
+
+	private TransactionTemplate getTransactionTemplate() {
+		if (!this.initialized) {
+			this.afterPropertiesSet();
+		}
+		return this.transactionTemplate;
+	}
+
+	public void afterPropertiesSet() {
+		synchronized (this.initializationMonitor) {
+			if (this.initialized) {
+				return;
+			}
+			if (this.transactionManager != null) {
+				TransactionTemplate template = new TransactionTemplate(this.transactionManager);
+				template.setPropagationBehaviorName(this.propagationBehaviorName);
+				template.setIsolationLevelName(this.isolationLevelName);
+				template.setTimeout(this.transactionTimeout);
+				template.setReadOnly(this.readOnly);
+				this.transactionTemplate = template;
+			}
+			this.initialized = true;
+		}
+	}
+
+	public boolean send(final Message<?> message, final MessageTarget target) {
+		TransactionTemplate txTemplate = this.getTransactionTemplate();
+		if (txTemplate != null) {
+			return (Boolean) txTemplate.execute(new TransactionCallback() {
+				public Object doInTransaction(TransactionStatus status) {
+					return doSend(message, target);
+				}
+			});
+		}
+		return this.doSend(message, target);
+	}
+
+	public Message<?> sendAndReceive(final Message<?> request, final MessageTarget target) {
+		TransactionTemplate txTemplate = this.getTransactionTemplate();
+		if (txTemplate != null) {
+			return (Message<?>) txTemplate.execute(new TransactionCallback() {
+				public Object doInTransaction(TransactionStatus status) {
+					return doSendAndReceive(request, target);
+				}
+			});
+		}
+		return this.doSendAndReceive(request, target);
+	}
+
+	public Message<?> receive(final MessageSource<?> source) {
+		TransactionTemplate txTemplate = this.getTransactionTemplate();
+		if (txTemplate != null) {
+			return (Message<?>) txTemplate.execute(new TransactionCallback() {
+				public Object doInTransaction(TransactionStatus status) {
+					return doReceive(source);
+				}
+			});
+		}
+		return this.doReceive(source);
+	}
+
+	public boolean receiveAndForward(final MessageSource<?> source, final MessageTarget target) {
+		TransactionTemplate txTemplate = this.getTransactionTemplate();
+		if (txTemplate != null) {
+			return (Boolean) txTemplate.execute(new TransactionCallback() {
+				public Object doInTransaction(TransactionStatus status) {
+					return doReceiveAndForward(source, target);
+				}
+			});
+		}
+		return this.doReceiveAndForward(source, target);
+	}
+
+
+	private boolean doSend(Message<?> message, MessageTarget target) {
+		long timeout = this.sendTimeout;
+		boolean sent = (timeout >= 0 && target instanceof BlockingTarget)
+				? ((BlockingTarget) target).send(message, timeout)
+				: target.send(message);
+		if (!sent && this.logger.isTraceEnabled()) {
+			this.logger.trace("failed to send message to target '" + target + "' within timeout: " + timeout);
+		}
+		return sent;
+	}
+
+	private Message<?> doReceive(MessageSource<?> source) {
+		long timeout = this.receiveTimeout;
+		Message<?> message = (timeout >= 0 && source instanceof BlockingSource)
+				? ((BlockingSource<?>) source).receive(timeout)
+				: source.receive();
+		if (message == null && this.logger.isTraceEnabled()) {
+			this.logger.trace("failed to receive message from source '" + source + "' within timeout: " + timeout);
+		}
+		return message;
+	}
+
+	private Message<?> doSendAndReceive(Message<?> request, MessageTarget target) {
+		TemporaryReturnAddress returnAddress = new TemporaryReturnAddress(this.receiveTimeout);
+		request = MessageBuilder.fromMessage(request).setReturnAddress(returnAddress).build();
+		if (!this.doSend(request, target)) {
+			return null;
+		}
+		return this.doReceive(returnAddress);
+	}
+
+	private boolean doReceiveAndForward(MessageSource<?> source, MessageTarget target) {
+		Message<?> message = this.doReceive(source);
+		if (message == null) {
+			return false;
+		}
+		try {
+			boolean sent = this.doSend(message, target);
+			if (source instanceof MessageDeliveryAware) {
+				if (sent) {
+					((MessageDeliveryAware) source).onSend(message);
+				}
+				else {
+					((MessageDeliveryAware) source).onFailure(new MessageDeliveryException(message, "failed to send message"));
+				}
+			}
+			return sent;
+		}
+		catch (Exception e) {
+			MessagingException exception = new MessagingException(
+					"exception occurred in receive-and-forward exchange", e);
+			if (source instanceof MessageDeliveryAware) {
+				((MessageDeliveryAware) source).onFailure(exception);
+			}
+			throw exception;
+		}
+	}
+
+
+	@SuppressWarnings("unchecked")
+	private static class TemporaryReturnAddress implements BlockingSource, MessageTarget {
+
+		private volatile Message<?> message;
+
+		private final long receiveTimeout;
+
+		private final CountDownLatch latch = new CountDownLatch(1);
+
+
+		public TemporaryReturnAddress(long receiveTimeout) {
+			this.receiveTimeout = receiveTimeout;
+		}
+
+
+		public Message receive() {
+			return this.receive(-1);
+		}
+
+		public Message receive(long timeout) {
+			try {
+				if (this.receiveTimeout < 0) {
+					this.latch.await();
+				}
+				else {
+					this.latch.await(this.receiveTimeout, TimeUnit.MILLISECONDS);
+				}
+			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+			return this.message;
+		}
+
+		public boolean send(Message<?> message) {
+			this.message = message;
+			this.latch.countDown();
+			return true;
+		}
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/message/Poller.java

@@ -1,29 +0,0 @@
-/*
- * Copyright 2002-2008 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * 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;
-
-/**
- * Strategy interface for polling a {@link MessageSource} and passing
- * any Messages received from that source to a {@link MessageTarget}.
- *
- * @author Mark Fisher
- */
-public interface Poller {
-
-	int poll(MessageSource<?> source, MessageTarget target);
-
-}

org.springframework.integration/src/main/java/org/springframework/integration/message/SimplePoller.java

@@ -1,64 +0,0 @@
-/*
- * Copyright 2002-2008 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * 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;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-/**
- * @author Mark Fisher
- */
-public class SimplePoller implements Poller {
-
-	private final Log logger = LogFactory.getLog(this.getClass());
-
-	private long receiveTimeout = 5000;
-
-	private long sendTimeout = -1;
-
-
-	public void setReceiveTimeout(long receiveTimeout) {
-		this.receiveTimeout = receiveTimeout;
-	}
-
-	public void setSendTimeout(long sendTimeout) {
-		this.sendTimeout = sendTimeout;
-	}
-
-	public int poll(MessageSource<?> source, MessageTarget target) {
-		Message<?> message = (source instanceof BlockingSource && this.receiveTimeout >= 0) ?
-				((BlockingSource<?>) source).receive(this.receiveTimeout) : source.receive();
-		if (message == null) {
-			if (logger.isDebugEnabled()) {
-				logger.debug("received no Message from source '" + source + "'");
-			}
-			return 0;
-		}
-		boolean sent = (target instanceof BlockingTarget && this.sendTimeout >= 0) ?
-				((BlockingTarget) target).send(message, this.sendTimeout) : target.send(message);
-		if (source instanceof MessageDeliveryAware) {
-			if (sent) {
-				((MessageDeliveryAware) source).onSend(message);
-			}
-			else {
-				((MessageDeliveryAware) source).onFailure(new MessageDeliveryException(message, "failed to send message"));
-			}
-		}
-		return (sent ? 1 : 0);
-	}
-
-}

org.springframework.integration/src/main/java/org/springframework/integration/message/SourceInvoker.java

@@ -1,42 +0,0 @@
-/*
- * Copyright 2002-2008 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * 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;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-/**
- * A helper class for receiving {@link Message}s from a {@link MessageSource}.
- * 
- * @author Mark Fisher
- */
-public class SourceInvoker {
-
-	private final Log logger = LogFactory.getLog(this.getClass());
-
-
-	public Message<?> invoke(MessageSource<?> source, long timeout) {
-		Message<?> message = (source instanceof BlockingSource && timeout >= 0)
-				? ((BlockingSource<?>) source).receive(timeout)
-				: source.receive();
-		if (message == null && this.logger.isTraceEnabled()) {
-			this.logger.trace("failed to receive message from source '" + source + "' within timeout: " + timeout);
-		}
-		return message;
-	}
-
-}

org.springframework.integration/src/main/java/org/springframework/integration/message/TargetInvoker.java

@@ -1,42 +0,0 @@
-/*
- * Copyright 2002-2008 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * 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;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-/**
- * A helper class for sending {@link Message}s to a {@link MessageTarget}.
- *  
- * @author Mark Fisher
- */
-public class TargetInvoker {
-
-	private final Log logger = LogFactory.getLog(this.getClass());
-
-
-	public boolean invoke(MessageTarget target, Message<?> message, long timeout) {
-		boolean sent = (target instanceof BlockingTarget && timeout >= 0)
-				? ((BlockingTarget) target).send(message, timeout)
-				: target.send(message);
-		if (!sent && this.logger.isTraceEnabled()) {
-			this.logger.trace("failed to send message to target '" + target + "' within timeout: " + timeout);
-		}
-		return sent;
-	}
-
-}