The <jms-gateway/> element is now available for message-driven JMS adapters. The <jms-source/> is strictly for polling adapters. Provided basic foundation for converting other request-reply adapters into "gateways". This includes the following namespace changes: 'send-timeout' is now 'request-timeout', 'receive-timeout' is now 'reply-timeout', and the 'channel' element is now 'request-channel'. The Source interface now includes the receive() method itself. PollableSource and SubscribableSource interfaces have both been removed. The Subscribable interface was added (e.g. for SynchronousChannel), but "gateway" adapter types will be configured directly with a request channel (and optionally a response channel).

spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers

@@ -7,6 +7,7 @@ httpinvoker-source=org.springframework.integration.adapter.httpinvoker.config.Ht
 httpinvoker-target=org.springframework.integration.adapter.httpinvoker.config.HttpInvokerTargetAdapterParser
 jms-source=org.springframework.integration.adapter.jms.config.JmsSourceAdapterParser
 jms-target=org.springframework.integration.adapter.jms.config.JmsTargetParser
+jms-gateway=org.springframework.integration.adapter.jms.config.JmsGatewayParser
 mail-target=org.springframework.integration.adapter.mail.config.MailTargetParser
 rmi-source=org.springframework.integration.adapter.rmi.config.RmiSourceAdapterParser
 rmi-target=org.springframework.integration.adapter.rmi.config.RmiTargetAdapterParser

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java

@@ -37,16 +37,12 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
-	private final MessageChannel channel;
+	private final MessageChannel requestChannel;
 
-	private volatile RequestReplyTemplate requestReplyTemplate;
+	private final RequestReplyTemplate requestReplyTemplate = new RequestReplyTemplate();
 
 	private volatile boolean expectReply = true;
 
-	private volatile long sendTimeout = -1;
-
-	private volatile long receiveTimeout = -1;
-
 	protected final Object lifecycleMonitor = new Object();
 
 	private volatile boolean initialized;
@@ -55,12 +51,13 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin
 	/**
 	 * Create an adapter that sends to the provided channel.
 	 * 
-	 * @param channel the channel where messages will be sent, must not be
+	 * @param requestChannel the channel where messages will be sent, must not be
 	 * <code>null</code>.
 	 */
-	public MessageHandlingSourceAdapter(MessageChannel channel) {
-		Assert.notNull(channel, "channel must not be null");
-		this.channel = channel;
+	public MessageHandlingSourceAdapter(MessageChannel requestChannel) {
+		Assert.notNull(requestChannel, "request channel must not be null");
+		this.requestChannel = requestChannel;
+		this.requestReplyTemplate.setRequestChannel(requestChannel);
 	}
 
 
@@ -72,16 +69,16 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin
 		this.expectReply = expectReply;
 	}
 
-	public void setSendTimeout(long sendTimeout) {
-		this.sendTimeout = sendTimeout;
+	public void setRequestTimeout(long requestTimeout) {
+		this.requestReplyTemplate.setRequestTimeout(requestTimeout);
 	}
 
-	public void setReceiveTimeout(long receiveTimeout) {
-		this.receiveTimeout = receiveTimeout;
+	public void setReplyTimeout(long replyTimeout) {
+		this.requestReplyTemplate.setReplyTimeout(replyTimeout);
 	}
 
 	protected MessageChannel getChannel() {
-		return this.channel;
+		return this.requestChannel;
 	}
 
 	public final void afterPropertiesSet() throws Exception {
@@ -89,9 +86,6 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin
 			if (this.initialized) {
 				return;
 			}
-			if (this.requestReplyTemplate == null) {
-				this.requestReplyTemplate = this.createRequestReplyTemplate();
-			}
 		}
 		this.initialize();
 		this.initialized = true;
@@ -103,13 +97,6 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin
 	protected void initialize() throws Exception {
 	}
 
-	private RequestReplyTemplate createRequestReplyTemplate() {
-		RequestReplyTemplate template = new RequestReplyTemplate(this.channel);
-		template.setRequestTimeout(this.sendTimeout);
-		template.setReplyTimeout(this.receiveTimeout);
-		return template;
-	}
-
 	public final Message<?> handle(Message<?> message) {
 		if (!this.initialized) {
 			try {
@@ -120,9 +107,9 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin
 			}
 		}
 		if (!this.expectReply) {
-			boolean sent = (this.sendTimeout < 0) ? this.channel.send(message) : this.channel.send(message, this.sendTimeout);
+			boolean sent = this.requestReplyTemplate.send(message);
 			if (!sent && logger.isWarnEnabled()) {
-				logger.warn("failed to send message to channel within timeout of " + this.sendTimeout + " milliseconds");
+				logger.warn("failed to send message to channel within timeout");
 			}
 			return null;
 		}

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java

@@ -50,24 +50,24 @@ public abstract class AbstractRequestReplySourceAdapterParser extends AbstractSi
 
 	@Override
 	protected boolean isEligibleAttribute(String attributeName) {
-		return !attributeName.equals("name") && !attributeName.equals("channel") && super.isEligibleAttribute(attributeName);
+		return !attributeName.equals("name") && !attributeName.equals("request-channel") && super.isEligibleAttribute(attributeName);
 	}
 
 	@Override
 	protected void postProcess(BeanDefinitionBuilder builder, Element element) {
-		String channelRef = element.getAttribute("channel");
+		String channelRef = element.getAttribute("request-channel");
 		if (!StringUtils.hasText(channelRef)) {
-			throw new ConfigurationException("a 'channel' reference is required");
+			throw new ConfigurationException("a 'request-channel' reference is required");
 		}
 		builder.addConstructorArgReference(channelRef);
 		builder.addPropertyValue("expectReply", element.getAttribute("expect-reply").equals("true"));
-		String sendTimeout = element.getAttribute("send-timeout");
-		if (StringUtils.hasText(sendTimeout)) {
-			builder.addPropertyValue("sendTimeout", Long.parseLong(sendTimeout));
+		String requestTimeout = element.getAttribute("request-timeout");
+		if (StringUtils.hasText(requestTimeout)) {
+			builder.addPropertyValue("requestTimeout", Long.parseLong(requestTimeout));
 		}
-		String receiveTimeout = element.getAttribute("receive-timeout");
-		if (StringUtils.hasText(receiveTimeout)) {
-			builder.addPropertyValue("receiveTimeout", Long.parseLong(receiveTimeout));
+		String replyTimeout = element.getAttribute("reply-timeout");
+		if (StringUtils.hasText(replyTimeout)) {
+			builder.addPropertyValue("replyTimeout", Long.parseLong(replyTimeout));
 		}
 		this.doPostProcess(builder, element);
 	}

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd

@@ -58,54 +58,83 @@
 		</xsd:complexType>
 	</xsd:element>
 
-	<xsd:element name="jms-source">
+	<xsd:element name="jms-source" type="jmsInboundAdapterType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Defines a JMS-based source channel adapter.
+			</xsd:documentation>
+		</xsd:annotation>
+	</xsd:element>
+
+	<xsd:element name="jms-gateway">
+		<xsd:annotation>
+			<xsd:documentation>
+	Defines a JMS-based gateway adapter.
+			</xsd:documentation>
+		</xsd:annotation>
 		<xsd:complexType>
-			<xsd:annotation>
-				<xsd:documentation>
-	Defines a jms-based source channel adapter.
-				</xsd:documentation>
-			</xsd:annotation>
-			<xsd:attribute name="id" type="xsd:string"/>
-			<xsd:attribute name="message-driven" type="xsd:boolean" default="true"/>
-			<xsd:attribute name="jms-template" type="xsd:string"/>
-			<xsd:attribute name="connection-factory" type="xsd:string"/>
-			<xsd:attribute name="destination" type="xsd:string"/>
-			<xsd:attribute name="destination-name" type="xsd:string"/>
-			<xsd:attribute name="message-converter" type="xsd:string"/>
-			<xsd:attribute name="acknowledge" default="auto">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-	The native JMS acknowledge mode: "auto", "client", "dups-ok" or "transacted".
-	The latter effectively activates a locally transacted Session.
-					]]></xsd:documentation>
-				</xsd:annotation>
-				<xsd:simpleType>
-					<xsd:restriction base="xsd:NMTOKEN">
-						<xsd:enumeration value="auto"/>
-						<xsd:enumeration value="client"/>
-						<xsd:enumeration value="dups-ok"/>
-						<xsd:enumeration value="transacted"/>
-					</xsd:restriction>
-				</xsd:simpleType>
-			</xsd:attribute>
+			<xsd:complexContent>
+				<xsd:extension base="jmsInboundAdapterType">
+					<xsd:attribute name="message-converter" type="xsd:string"/>
+					<xsd:attribute name="expect-reply" type="xsd:boolean" default="true"/>
+					<xsd:attribute name="request-channel" type="xsd:string" use="required"/>
+					<xsd:attribute name="reply-channel" type="xsd:string"/>
+					<xsd:attribute name="request-timeout" type="xsd:long"/>
+					<xsd:attribute name="reply-timeout" type="xsd:long"/>
+				</xsd:extension>
+			</xsd:complexContent>
 		</xsd:complexType>
 	</xsd:element>
 
-	<xsd:element name="jms-target">
-		<xsd:complexType>
-			<xsd:annotation>
-				<xsd:documentation>
+	<xsd:element name="jms-target" type="jmsAdapterType">
+		<xsd:annotation>
+			<xsd:documentation>
 	Defines a target that sends JMS Messages.
-				</xsd:documentation>
-			</xsd:annotation>
-			<xsd:attribute name="id" type="xsd:string"/>
-			<xsd:attribute name="jms-template" type="xsd:string"/>
-			<xsd:attribute name="connection-factory" type="xsd:string"/>
-			<xsd:attribute name="destination" type="xsd:string"/>
-			<xsd:attribute name="destination-name" type="xsd:string"/>
-		</xsd:complexType>
+			</xsd:documentation>
+		</xsd:annotation>
 	</xsd:element>
 
+	<xsd:complexType name="jmsInboundAdapterType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Common configuration for inbound JMS-based adapters.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexContent>
+			<xsd:extension base="jmsAdapterType">
+				<xsd:attribute name="acknowledge" default="auto">
+					<xsd:annotation>
+						<xsd:documentation><![CDATA[
+	The native JMS acknowledge mode: "auto", "client", "dups-ok" or "transacted".
+	The latter effectively activates a locally transacted Session.
+						]]></xsd:documentation>
+					</xsd:annotation>
+					<xsd:simpleType>
+						<xsd:restriction base="xsd:NMTOKEN">
+							<xsd:enumeration value="auto"/>
+							<xsd:enumeration value="client"/>
+							<xsd:enumeration value="dups-ok"/>
+							<xsd:enumeration value="transacted"/>
+						</xsd:restriction>
+					</xsd:simpleType>
+				</xsd:attribute>
+			</xsd:extension>
+		</xsd:complexContent>
+	</xsd:complexType>
+
+	<xsd:complexType name="jmsAdapterType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Common configuration for JMS-based adapters.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:attribute name="id" type="xsd:string"/>
+		<xsd:attribute name="jms-template" type="xsd:string"/>
+		<xsd:attribute name="connection-factory" type="xsd:string"/>
+		<xsd:attribute name="destination" type="xsd:string"/>
+		<xsd:attribute name="destination-name" type="xsd:string"/>
+	</xsd:complexType>
+
 	<xsd:element name="rmi-source">
 		<xsd:complexType>
 			<xsd:annotation>
@@ -114,7 +143,7 @@
 				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:complexContent>
-				<xsd:extension base="requestReplySource">
+				<xsd:extension base="gatewayType">
 					<xsd:attribute name="registry-host" type="xsd:string"/>
 					<xsd:attribute name="registry-port" type="xsd:integer"/>
 					<xsd:attribute name="remote-invocation-executor" type="xsd:string"/>
@@ -138,7 +167,7 @@
 		</xsd:complexType>
 	</xsd:element>
 
-	<xsd:element name="httpinvoker-source" type="requestReplySource">
+	<xsd:element name="httpinvoker-source" type="gatewayType">
 		<xsd:annotation>
 			<xsd:documentation>
 	Defines an httpinvoker-based source channel adapter.
@@ -201,18 +230,19 @@
 		</xsd:complexType>
 	</xsd:element>
 
-	<xsd:complexType name="requestReplySource">
+	<xsd:complexType name="gatewayType">
 		<xsd:annotation>
 			<xsd:documentation>
-	Defines common configuration for request-reply source adapters.
+	Defines common configuration for gateway adapters.
 			</xsd:documentation>
 		</xsd:annotation>
 		<xsd:attribute name="id" type="xsd:ID"/>
 		<xsd:attribute name="name" type="xsd:string"/>
 		<xsd:attribute name="expect-reply" type="xsd:boolean" default="true"/>
-		<xsd:attribute name="send-timeout" type="xsd:long"/>
-		<xsd:attribute name="receive-timeout" type="xsd:long"/>
-		<xsd:attribute name="channel" type="xsd:string" use="required"/>
+		<xsd:attribute name="request-channel" type="xsd:string" use="required"/>
+		<xsd:attribute name="reply-channel" type="xsd:string"/>
+		<xsd:attribute name="request-timeout" type="xsd:long"/>
+		<xsd:attribute name="reply-timeout" type="xsd:long"/>
 	</xsd:complexType>
 
 </xsd:schema>

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java

@@ -23,7 +23,7 @@ import java.io.FilenameFilter;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessagingException;
-import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.Source;
 import org.springframework.util.Assert;
 
 /**
@@ -31,7 +31,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class FileSource implements PollableSource<Object>, InitializingBean {
+public class FileSource implements Source<Object>, InitializingBean {
 
 	private final File directory;
 

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java

@@ -35,7 +35,7 @@ import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageCreator;
 import org.springframework.integration.message.MessageDeliveryAware;
 import org.springframework.integration.message.MessagingException;
-import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.Source;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -45,7 +45,7 @@ import org.springframework.util.StringUtils;
  * @author Marius Bogoevici
  * @author Mark Fisher
  */
-public class FtpSource implements PollableSource<Object>, MessageDeliveryAware {
+public class FtpSource implements Source<Object>, MessageDeliveryAware {
 
 	private final static String DEFAULT_HOST = "localhost";
 

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java

@@ -25,10 +25,11 @@ import org.springframework.context.Lifecycle;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.SubscribableSource;
-import org.springframework.integration.message.Target;
+import org.springframework.integration.channel.RequestReplyTemplate;
+import org.springframework.integration.gateway.MessagingGateway;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
 import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.jms.listener.adapter.MessageListenerAdapter;
 import org.springframework.jms.support.converter.MessageConverter;
 import org.springframework.jms.support.converter.SimpleMessageConverter;
 import org.springframework.util.Assert;
@@ -38,9 +39,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecycle, DisposableBean {
-
-	private volatile MessageChannel channel;
+public class JmsMessageDrivenSourceAdapter extends MessagingGateway implements Lifecycle, DisposableBean {
 
 	private volatile AbstractMessageListenerContainer container;
 
@@ -58,8 +57,6 @@ public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecy
 
 	private volatile int sessionAcknowledgeMode = Session.AUTO_ACKNOWLEDGE;
 
-	private volatile long receiveTimeout = 1000;
-
 	private volatile int concurrentConsumers = 1;
 
 	private volatile int maxConcurrentConsumers = 1;
@@ -68,37 +65,8 @@ public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecy
 
 	private volatile int idleTaskExecutionLimit = 1;
 
-	private volatile long sendTimeout = -1;
+	private boolean expectReply = false;
 
-	private volatile boolean initialized;
-
-	private final Object lifecycleMonitor = new Object();
-
-
-	public boolean subscribe(Target target) {
-		if (target instanceof MessageChannel) {
-			this.setChannel((MessageChannel) target);
-			return true;
-		}
-		return false;
-	}
-
-	public boolean unsubscribe(Target target) {
-		if (target.equals(this.channel)) {
-			this.stop();
-			this.channel = null;
-			return true;
-		}
-		return false;
-	}
-
-	public void setChannel(MessageChannel channel) {
-		this.channel = channel;
-	}
-
-	public MessageChannel getChannel() {
-		return this.channel;
-	}
 
 	public void setContainer(AbstractMessageListenerContainer container) {
 		this.container = container;
@@ -121,10 +89,6 @@ public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecy
 		this.messageConverter = messageConverter;
 	}
 
-	public void setSendTimeout(long sendTimeout) {
-		this.sendTimeout = sendTimeout;
-	}
-
 	public void setTaskExecutor(TaskExecutor taskExecutor) {
 		this.taskExecutor = taskExecutor;
 	}
@@ -137,15 +101,18 @@ public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecy
 		this.sessionAcknowledgeMode = sessionAcknowledgeMode;
 	}
 
+	public void setExpectReply(boolean expectReply) {
+		this.expectReply = expectReply;
+	}
+
 	private void initialize() {
-		if (this.channel == null) {
-			throw new ConfigurationException("channel must not be null");
-		}
 		if (this.container == null) {
 			this.container = createDefaultContainer();
 		}
-		ChannelPublishingJmsListener listener = new ChannelPublishingJmsListener(this.getChannel(), this.messageConverter);
-		listener.setTimeout(this.sendTimeout);
+		MessageListenerAdapter listener = new MessageListenerAdapter();
+		listener.setDelegate(this);
+		listener.setDefaultListenerMethod(this.expectReply ? "request" : "send");
+		listener.setMessageConverter(this.messageConverter);
 		this.container.setMessageListener(listener);
 		if (!this.container.isActive()) {
 			this.container.afterPropertiesSet();
@@ -169,7 +136,6 @@ public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecy
 		if (this.destinationName != null) {
 			dmlc.setDestinationName(this.destinationName);
 		}
-		dmlc.setReceiveTimeout(this.receiveTimeout);
 		dmlc.setSessionTransacted(this.sessionTransacted);
 		dmlc.setSessionAcknowledgeMode(this.sessionAcknowledgeMode);
 		dmlc.setAutoStartup(false);

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java

@@ -21,7 +21,7 @@ import javax.jms.Destination;
 
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.Source;
 import org.springframework.jms.core.JmsTemplate;
 
 /**
@@ -32,7 +32,7 @@ import org.springframework.jms.core.JmsTemplate;
  * 
  * @author Mark Fisher
  */
-public class JmsPollableSource extends AbstractJmsTemplateBasedAdapter implements PollableSource<Object> {
+public class JmsPollableSource extends AbstractJmsTemplateBasedAdapter implements Source<Object> {
 
 	public JmsPollableSource(JmsTemplate jmsTemplate) {
 		super(jmsTemplate);

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsAdapterParserUtils.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.adapter.jms.config;
 
+import javax.jms.Session;
+
 import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.BeanCreationException;
@@ -44,6 +46,20 @@ public abstract class JmsAdapterParserUtils {
 
 	public static final String DESTINATION_NAME_PROPERTY = "destinationName";
 
+	public static final String MESSAGE_CONVERTER_ATTRIBUTE = "message-converter";
+
+	public static final String MESSAGE_CONVERTER_PROPERTY = "messageConverter";
+
+	private static final String ACKNOWLEDGE_ATTRIBUTE = "acknowledge";
+
+	private static final String ACKNOWLEDGE_AUTO = "auto";
+
+	private static final String ACKNOWLEDGE_CLIENT = "client";
+
+	private static final String ACKNOWLEDGE_DUPS_OK = "dups-ok";
+
+	private static final String ACKNOWLEDGE_TRANSACTED = "transacted";
+
 
 	public static String determineConnectionFactoryBeanName(Element element) {
 		String connectionFactoryBeanName = "connectionFactory";
@@ -57,4 +73,28 @@ public abstract class JmsAdapterParserUtils {
 		return connectionFactoryBeanName;
 	}
 
+	public static Integer parseAcknowledgeMode(Element element) {
+		String acknowledge = element.getAttribute(ACKNOWLEDGE_ATTRIBUTE);
+		if (StringUtils.hasText(acknowledge)) {
+			int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
+			if (ACKNOWLEDGE_TRANSACTED.equals(acknowledge)) {
+				acknowledgeMode = Session.SESSION_TRANSACTED;
+			}
+			else if (ACKNOWLEDGE_DUPS_OK.equals(acknowledge)) {
+				acknowledgeMode = Session.DUPS_OK_ACKNOWLEDGE;
+			}
+			else if (ACKNOWLEDGE_CLIENT.equals(acknowledge)) {
+				acknowledgeMode = Session.CLIENT_ACKNOWLEDGE;
+			}
+			else if (!ACKNOWLEDGE_AUTO.equals(acknowledge)) {
+				throw new BeanCreationException("Invalid JMS 'acknowledge' setting: " +
+						"only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported.");
+			}
+			return acknowledgeMode;
+		}
+		else {
+			return null;
+		}
+	}
+
 }

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsGatewayParser.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.adapter.jms.config;
+
+import javax.jms.Session;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.adapter.jms.JmsMessageDrivenSourceAdapter;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the <jms-gateway> element.
+ * 
+ * @author Mark Fisher
+ */
+public class JmsGatewayParser extends AbstractSingleBeanDefinitionParser {
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return JmsMessageDrivenSourceAdapter.class;
+	}
+
+	@Override
+	protected boolean shouldGenerateId() {
+		return false;
+	}
+
+	@Override
+	protected boolean shouldGenerateIdAsFallback() {
+		return true;
+	}
+
+	@Override
+	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+		String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE);
+		String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE);
+		String messageConverter = element.getAttribute(JmsAdapterParserUtils.MESSAGE_CONVERTER_ATTRIBUTE);
+		if (StringUtils.hasText(element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE))) {
+			throw new BeanCreationException(JmsMessageDrivenSourceAdapter.class.getSimpleName() +
+					" does not accept a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE +
+					"' reference. One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" +
+					JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided.");
+		}
+		if (StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) {
+			builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY,
+					JmsAdapterParserUtils.determineConnectionFactoryBeanName(element));
+			if (StringUtils.hasText(destination)) {
+				builder.addPropertyReference(JmsAdapterParserUtils.DESTINATION_PROPERTY, destination);
+			}
+			else {
+				builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_NAME_PROPERTY, destinationName);
+			}
+		}
+		else {
+			throw new BeanCreationException("One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE +
+					"' or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided.");
+		}
+		if (StringUtils.hasText(messageConverter)) {
+			builder.addPropertyReference(JmsAdapterParserUtils.MESSAGE_CONVERTER_PROPERTY, messageConverter);
+		}
+		Integer acknowledgeMode = JmsAdapterParserUtils.parseAcknowledgeMode(element);
+		if (acknowledgeMode != null) {
+			if (acknowledgeMode.intValue() == Session.SESSION_TRANSACTED) {
+				builder.addPropertyValue("sessionTransacted", Boolean.TRUE);
+			}
+			else {
+				builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode);
+			}
+		}
+		String requestChannel = element.getAttribute("request-channel");
+		if (StringUtils.hasText(requestChannel)) {
+			builder.addPropertyReference("requestChannel", requestChannel);
+		}
+		String requestTimeout = element.getAttribute("request-timeout");
+		if (StringUtils.hasText(requestTimeout)) {
+			builder.addPropertyValue("requestTimeout", Long.parseLong(requestTimeout));
+		}
+		String replyChannel = element.getAttribute("reply-channel");
+		if (StringUtils.hasText(replyChannel)) {
+			builder.addPropertyReference("replyChannel", replyChannel);
+		}
+		String replyTimeout = element.getAttribute("reply-timeout");
+		if (StringUtils.hasText(replyTimeout)) {
+			builder.addPropertyValue("replyTimeout", Long.parseLong(replyTimeout));
+		}
+	}
+
+}

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java

@@ -36,21 +36,6 @@ import org.springframework.util.StringUtils;
  */
 public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser {
 
-	private static final String MESSAGE_CONVERTER_ATTRIBUTE = "message-converter";
-
-	private static final String MESSAGE_CONVERTER_PROPERTY = "messageConverter";
-
-	private static final String ACKNOWLEDGE_ATTRIBUTE = "acknowledge";
-
-	private static final String ACKNOWLEDGE_AUTO = "auto";
-
-	private static final String ACKNOWLEDGE_CLIENT = "client";
-
-	private static final String ACKNOWLEDGE_DUPS_OK = "dups-ok";
-
-	private static final String ACKNOWLEDGE_TRANSACTED = "transacted";
-
-
 	protected boolean shouldGenerateId() {
 		return false;
 	}
@@ -61,20 +46,7 @@ public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser {
 
 	@Override
 	protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
-		if ("true".equals(element.getAttribute("message-driven"))) {
-			return parseMessageDrivenSource(element, parserContext);
-		}
-		return parsePollableSource(element, parserContext);
-	}
-
-	private AbstractBeanDefinition parsePollableSource(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsPollableSource.class);
-		if (StringUtils.hasText(element.getAttribute(MESSAGE_CONVERTER_ATTRIBUTE))) {
-			throw new BeanCreationException(
-					"The '" + MESSAGE_CONVERTER_ATTRIBUTE + "' attribute is not supported for a polling JMS adapter. " +
-					". Consider providing a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + 
-					"' reference where the template contains a 'messageConverter' property instead.");
-		}
 		String jmsTemplate = element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE);
 		String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE);
 		String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE);
@@ -107,68 +79,4 @@ public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser {
 		return builder.getBeanDefinition();
 	}
 
-	private AbstractBeanDefinition parseMessageDrivenSource(Element element, ParserContext parserContext) {
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsMessageDrivenSourceAdapter.class);
-		String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE);
-		String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE);
-		String messageConverter = element.getAttribute(MESSAGE_CONVERTER_ATTRIBUTE);
-		if (StringUtils.hasText(element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE))) {
-			throw new BeanCreationException(JmsMessageDrivenSourceAdapter.class.getSimpleName() +
-					" does not accept a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE +
-					"' reference. One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" +
-					JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided.");
-		}
-		if (StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) {
-			builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY,
-					JmsAdapterParserUtils.determineConnectionFactoryBeanName(element));
-			if (StringUtils.hasText(destination)) {
-				builder.addPropertyReference(JmsAdapterParserUtils.DESTINATION_PROPERTY, destination);
-			}
-			else {
-				builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_NAME_PROPERTY, destinationName);
-			}
-		}
-		else {
-			throw new BeanCreationException("One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE +
-					"' or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided.");
-		}
-		if (StringUtils.hasText(messageConverter)) {
-			builder.addPropertyReference(MESSAGE_CONVERTER_PROPERTY, messageConverter);
-		}
-		Integer acknowledgeMode = parseAcknowledgeMode(element);
-		if (acknowledgeMode != null) {
-			if (acknowledgeMode.intValue() == Session.SESSION_TRANSACTED) {
-				builder.addPropertyValue("sessionTransacted", Boolean.TRUE);
-			}
-			else {
-				builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode);
-			}
-		}
-		return builder.getBeanDefinition();
-	}
-
-	private Integer parseAcknowledgeMode(Element element) {
-		String acknowledge = element.getAttribute(ACKNOWLEDGE_ATTRIBUTE);
-		if (StringUtils.hasText(acknowledge)) {
-			int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
-			if (ACKNOWLEDGE_TRANSACTED.equals(acknowledge)) {
-				acknowledgeMode = Session.SESSION_TRANSACTED;
-			}
-			else if (ACKNOWLEDGE_DUPS_OK.equals(acknowledge)) {
-				acknowledgeMode = Session.DUPS_OK_ACKNOWLEDGE;
-			}
-			else if (ACKNOWLEDGE_CLIENT.equals(acknowledge)) {
-				acknowledgeMode = Session.CLIENT_ACKNOWLEDGE;
-			}
-			else if (!ACKNOWLEDGE_AUTO.equals(acknowledge)) {
-				throw new BeanCreationException("Invalid jms-source 'acknowledge' setting: " +
-						"only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported.");
-			}
-			return acknowledgeMode;
-		}
-		else {
-			return null;
-		}
-	}
-
 }

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java

@@ -23,14 +23,14 @@ import java.io.InputStream;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessagingException;
-import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.Source;
 
 /**
  * A pollable source for receiving bytes from an {@link InputStream}.
  * 
  * @author Mark Fisher
  */
-public class ByteStreamSource implements PollableSource<byte[]> {
+public class ByteStreamSource implements Source<byte[]> {
 
 	private BufferedInputStream stream;
 

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java

@@ -24,7 +24,7 @@ import java.io.UnsupportedEncodingException;
 
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.message.MessagingException;
-import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.Source;
 import org.springframework.integration.message.StringMessage;
 import org.springframework.util.Assert;
 
@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class CharacterStreamSource implements PollableSource<String> {
+public class CharacterStreamSource implements Source<String> {
 
 	private final BufferedReader reader;
 

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/config/HttpInvokerSourceAdapterParserTests.java

@@ -25,6 +25,7 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.adapter.httpinvoker.HttpInvokerSourceAdapter;
 import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.RequestReplyTemplate;
 
 /**
  * @author Mark Fisher
@@ -38,10 +39,13 @@ public class HttpInvokerSourceAdapterParserTests {
 		MessageChannel channel = (MessageChannel) context.getBean("testChannel");
 		HttpInvokerSourceAdapter adapter = (HttpInvokerSourceAdapter) context.getBean("adapterWithDefaults");
 		DirectFieldAccessor accessor = new DirectFieldAccessor(adapter);
-		assertEquals(channel, accessor.getPropertyValue("channel"));
+		assertEquals(channel, accessor.getPropertyValue("requestChannel"));
 		assertEquals(true, accessor.getPropertyValue("expectReply"));
-		assertEquals(-1L, accessor.getPropertyValue("sendTimeout"));
-		assertEquals(-1L, accessor.getPropertyValue("receiveTimeout"));
+		RequestReplyTemplate template = (RequestReplyTemplate)
+				accessor.getPropertyValue("requestReplyTemplate");
+		DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template);
+		assertEquals(-1L, templateAccessor.getPropertyValue("requestTimeout"));
+		assertEquals(-1L, templateAccessor.getPropertyValue("replyTimeout"));
 	}
 
 	@Test
@@ -51,10 +55,13 @@ public class HttpInvokerSourceAdapterParserTests {
 		MessageChannel channel = (MessageChannel) context.getBean("testChannel");
 		HttpInvokerSourceAdapter adapter = (HttpInvokerSourceAdapter) context.getBean("/adapter/with/name");
 		DirectFieldAccessor accessor = new DirectFieldAccessor(adapter);
-		assertEquals(channel, accessor.getPropertyValue("channel"));
+		assertEquals(channel, accessor.getPropertyValue("requestChannel"));
 		assertEquals(true, accessor.getPropertyValue("expectReply"));
-		assertEquals(-1L, accessor.getPropertyValue("sendTimeout"));
-		assertEquals(-1L, accessor.getPropertyValue("receiveTimeout"));
+		RequestReplyTemplate template = (RequestReplyTemplate)
+				accessor.getPropertyValue("requestReplyTemplate");
+		DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template);
+		assertEquals(-1L, templateAccessor.getPropertyValue("requestTimeout"));
+		assertEquals(-1L, templateAccessor.getPropertyValue("replyTimeout"));
 	}
 
 	@Test
@@ -64,10 +71,13 @@ public class HttpInvokerSourceAdapterParserTests {
 		MessageChannel channel = (MessageChannel) context.getBean("testChannel");
 		HttpInvokerSourceAdapter adapter = (HttpInvokerSourceAdapter) context.getBean("adapterWithCustomProperties");
 		DirectFieldAccessor accessor = new DirectFieldAccessor(adapter);
-		assertEquals(channel, accessor.getPropertyValue("channel"));
+		assertEquals(channel, accessor.getPropertyValue("requestChannel"));
 		assertEquals(false, accessor.getPropertyValue("expectReply"));
-		assertEquals(123L, accessor.getPropertyValue("sendTimeout"));
-		assertEquals(456L, accessor.getPropertyValue("receiveTimeout"));
+		RequestReplyTemplate template = (RequestReplyTemplate)
+				accessor.getPropertyValue("requestReplyTemplate");
+		DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template);
+		assertEquals(123L, templateAccessor.getPropertyValue("requestTimeout"));
+		assertEquals(456L, templateAccessor.getPropertyValue("replyTimeout"));
 	}
 
 }

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/config/httpInvokerSourceAdapterParserTests.xml

@@ -11,11 +11,12 @@
 
 	<channel id="testChannel"/>
 
-	<httpinvoker-source id="adapterWithDefaults" channel="testChannel"/>
+	<httpinvoker-source id="adapterWithDefaults" request-channel="testChannel"/>
 
-	<httpinvoker-source name="/adapter/with/name" channel="testChannel"/>
+	<httpinvoker-source name="/adapter/with/name" request-channel="testChannel"/>
 
-	<httpinvoker-source id="adapterWithCustomProperties" channel="testChannel"
-			expect-reply="false" send-timeout="123" receive-timeout="456"/>
+	<httpinvoker-source id="adapterWithCustomProperties"
+			request-channel="testChannel" request-timeout="123" 
+			expect-reply="false" reply-timeout="456"/>
 
 </beans:beans>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java

@@ -81,7 +81,7 @@ public class JmsSourceAdapterParserTests {
 				"messageDrivenAdapterWithConnectionFactoryAndDestination.xml", this.getClass());
 		MessageChannel channel = new QueueChannel(1);
 		JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource");
-		source.setChannel(channel);
+		source.setRequestChannel(channel);
 		context.start();
 		Message<?> message = channel.receive(3000);
 		assertNotNull("message should not be null", message);
@@ -95,7 +95,7 @@ public class JmsSourceAdapterParserTests {
 				"messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass());
 		MessageChannel channel = new QueueChannel(1);
 		JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource");
-		source.setChannel(channel);
+		source.setRequestChannel(channel);
 		context.start();
 		assertEquals(JmsMessageDrivenSourceAdapter.class, source.getClass());
 		Message<?> message = channel.receive(3000);
@@ -110,7 +110,7 @@ public class JmsSourceAdapterParserTests {
 				"messageDrivenAdapterWithMessageConverter.xml", this.getClass());
 		MessageChannel channel = new QueueChannel(1);
 		JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource");
-		source.setChannel(channel);
+		source.setRequestChannel(channel);
 		context.start();
 		Message<?> message = channel.receive(3000);
 		assertNotNull("message should not be null", message);
@@ -194,7 +194,7 @@ public class JmsSourceAdapterParserTests {
 				"messageDrivenAdapterWithDefaultConnectionFactory.xml", this.getClass());
 		MessageChannel channel = new QueueChannel(1);
 		JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource");
-		source.setChannel(channel);
+		source.setRequestChannel(channel);
 		context.start();
 		Message<?> message = channel.receive(3000);
 		assertNotNull("message should not be null", message);

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestination.xml

@@ -7,9 +7,12 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
 
-	<si:jms-source id="jmsSource"
-	               connection-factory="testConnectionFactory"
-	               destination="testDestination"/>
+	<si:channel id="requestChannel"/>
+
+	<si:jms-gateway id="jmsSource"
+	                connection-factory="testConnectionFactory"
+	                destination="testDestination"
+	                request-channel="requestChannel"/>
 
 	<bean id="testConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 		<constructor-arg>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml

@@ -9,9 +9,12 @@
 
 	<si:message-bus/>
 
-	<si:jms-source id="jmsSource"
-	               connection-factory="testConnectionFactory"
-	               destination-name="testDestinationName"/>
+	<si:channel id="requestChannel"/>
+
+	<si:jms-gateway id="jmsSource"
+	                connection-factory="testConnectionFactory"
+	                destination-name="testDestinationName"
+	                request-channel="requestChannel"/>
 
 	<bean id="testConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 		<constructor-arg>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryOnly.xml

@@ -9,7 +9,11 @@
 
 	<si:message-bus/>
 
-	<si:jms-source id="jmsSource" connection-factory="testConnectionFactory"/>
+	<si:channel id="requestChannel"/>
+
+	<si:jms-gateway id="jmsSource"
+					connection-factory="testConnectionFactory"
+		 			request-channel="requestChannel"/>
 
 	<bean id="testConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 		<constructor-arg>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithDefaultConnectionFactory.xml

@@ -9,7 +9,11 @@
 
 	<si:message-bus/>
 
-	<si:jms-source id="jmsSource" destination-name="testDestinationName"/>
+	<si:channel id="requestChannel"/>
+
+	<si:jms-gateway id="jmsSource"
+					destination-name="testDestinationName"
+					request-channel="requestChannel"/>
 
 	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 		<constructor-arg>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithEmptyConnectionFactory.xml

@@ -7,6 +7,11 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
 
-	<si:jms-source id="jmsSource" connection-factory="" destination-name="testDestinationName"/>
+	<si:channel id="requestChannel"/>
+
+	<si:jms-gateway id="jmsSource"
+					connection-factory=""
+					destination-name="testDestinationName"
+					request-channel="requestChannel"/>
 
 </beans>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithMessageConverter.xml

@@ -9,10 +9,13 @@
 
 	<si:message-bus/>
 
-	<si:jms-source id="jmsSource"
-	               connection-factory="testConnectionFactory"
-	               destination="testDestination"
-	               message-converter="converter"/>
+	<si:channel id="requestChannel"/>
+
+	<si:jms-gateway id="jmsSource"
+	                connection-factory="testConnectionFactory"
+	                destination="testDestination"
+	                message-converter="converter"
+	                request-channel="requestChannel"/>
 
 	<bean id="converter" class="org.springframework.integration.adapter.jms.config.JmsSourceAdapterParserTests$TestMessageConverter"/>
 

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestination.xml

@@ -9,8 +9,7 @@
 
 	<si:jms-source id="jmsSource"
 	               connection-factory="testConnectionFactory"
-	               destination="testDestination"
-	               message-driven="false"/>
+	               destination="testDestination"/>
 
 	<bean id="testConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 		<constructor-arg>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestinationName.xml

@@ -9,8 +9,7 @@
 
 	<si:jms-source id="jmsSource"
 	               connection-factory="testConnectionFactory"
-	               destination-name="testDestinationName"
-	               message-driven="false"/>
+	               destination-name="testDestinationName"/>
 
 	<bean id="testConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 		<constructor-arg>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryOnly.xml

@@ -7,7 +7,7 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
 
-	<si:jms-source id="adapter" connection-factory="testConnectionFactory" message-driven="false"/>
+	<si:jms-source id="adapter" connection-factory="testConnectionFactory"/>
 
 	<bean id="testConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 		<constructor-arg>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationAndDefaultConnectionFactory.xml

@@ -7,7 +7,7 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
 
-	<si:jms-source id="jmsSource" destination="testDestination" message-driven="false"/>
+	<si:jms-source id="jmsSource" destination="testDestination"/>
 
 	<bean id="testDestination" class="org.springframework.integration.adapter.jms.StubDestination"/>
 

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameAndDefaultConnectionFactory.xml

@@ -7,7 +7,7 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
 
-	<si:jms-source id="jmsSource" destination-name="testDestinationName" message-driven="false"/>
+	<si:jms-source id="jmsSource" destination-name="testDestinationName"/>
 
 	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 		<constructor-arg>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameOnly.xml

@@ -7,6 +7,6 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
 
-	<si:jms-source id="jmsSource" destination-name="testDestinationName" message-driven="false"/>
+	<si:jms-source id="jmsSource" destination-name="testDestinationName"/>
 
 </beans>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationOnly.xml

@@ -7,7 +7,7 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
 
-	<si:jms-source id="jmsSource" destination="testDestination" message-driven="false"/>
+	<si:jms-source id="jmsSource" destination="testDestination"/>
 
 	<bean id="testDestination" class="org.springframework.integration.adapter.jms.StubDestination"/>
 

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithJmsTemplate.xml

@@ -7,7 +7,7 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
 
-	<si:jms-source id="jmsSource" jms-template="jmsTemplate" message-driven="false"/>
+	<si:jms-source id="jmsSource" jms-template="jmsTemplate"/>
 
 	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
 		<property name="connectionFactory" ref="connectionFactory"/>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingJmsSourceEndpoint.xml

@@ -15,7 +15,7 @@
 		<si:schedule period="5000"/>
 	</si:source-endpoint>
 
-	<si:jms-source id="jmsSource" jms-template="jmsTemplate" message-driven="false"/>
+	<si:jms-source id="jmsSource" jms-template="jmsTemplate"/>
 
 	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
 		<property name="connectionFactory" ref="connectionFactory"/>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParserTests.java

@@ -26,6 +26,7 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.adapter.rmi.RmiSourceAdapter;
 import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.RequestReplyTemplate;
 
 /**
  * @author Mark Fisher
@@ -39,10 +40,13 @@ public class RmiSourceAdapterParserTests {
 		MessageChannel channel = (MessageChannel) context.getBean("testChannel");
 		RmiSourceAdapter adapter = (RmiSourceAdapter) context.getBean("adapterWithDefaults");
 		DirectFieldAccessor accessor = new DirectFieldAccessor(adapter);
-		assertEquals(channel, accessor.getPropertyValue("channel"));
+		assertEquals(channel, accessor.getPropertyValue("requestChannel"));
 		assertEquals(true, accessor.getPropertyValue("expectReply"));
-		assertEquals(-1L, accessor.getPropertyValue("sendTimeout"));
-		assertEquals(-1L, accessor.getPropertyValue("receiveTimeout"));
+		RequestReplyTemplate template = (RequestReplyTemplate)
+				accessor.getPropertyValue("requestReplyTemplate");
+		DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template);
+		assertEquals(-1L, templateAccessor.getPropertyValue("requestTimeout"));
+		assertEquals(-1L, templateAccessor.getPropertyValue("replyTimeout"));
 	}
 
 	@Test
@@ -52,10 +56,13 @@ public class RmiSourceAdapterParserTests {
 		MessageChannel channel = (MessageChannel) context.getBean("testChannel");
 		RmiSourceAdapter adapter = (RmiSourceAdapter) context.getBean("adapterWithCustomProperties");
 		DirectFieldAccessor accessor = new DirectFieldAccessor(adapter);
-		assertEquals(channel, accessor.getPropertyValue("channel"));
+		assertEquals(channel, accessor.getPropertyValue("requestChannel"));
 		assertEquals(false, accessor.getPropertyValue("expectReply"));
-		assertEquals(123L, accessor.getPropertyValue("sendTimeout"));
-		assertEquals(456L, accessor.getPropertyValue("receiveTimeout"));
+		RequestReplyTemplate template = (RequestReplyTemplate)
+				accessor.getPropertyValue("requestReplyTemplate");
+		DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template);
+		assertEquals(123L, templateAccessor.getPropertyValue("requestTimeout"));
+		assertEquals(456L, templateAccessor.getPropertyValue("replyTimeout"));
 	}
 
 	@Test

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/rmiSourceAdapterParserTests.xml

@@ -11,16 +11,16 @@
 
 	<channel id="testChannel"/>
 
-	<rmi-source id="adapterWithDefaults" channel="testChannel"/>
+	<rmi-source id="adapterWithDefaults" request-channel="testChannel"/>
 
-	<rmi-source id="adapterWithCustomProperties" channel="testChannel"
-			expect-reply="false" send-timeout="123" receive-timeout="456"/>
+	<rmi-source id="adapterWithCustomProperties" request-channel="testChannel"
+			expect-reply="false" request-timeout="123" reply-timeout="456"/>
 
-	<rmi-source id="adapterWithHost" channel="testChannel" registry-host="localhost"/>
+	<rmi-source id="adapterWithHost" request-channel="testChannel" registry-host="localhost"/>
 
-	<rmi-source id="adapterWithPort" channel="testChannel" registry-port="1234"/>
+	<rmi-source id="adapterWithPort" request-channel="testChannel" registry-port="1234"/>
 
-	<rmi-source id="adapterWithExecutorRef" channel="testChannel" remote-invocation-executor="invocationExecutor"/>
+	<rmi-source id="adapterWithExecutorRef" request-channel="testChannel" remote-invocation-executor="invocationExecutor"/>
 
 	<beans:bean id="invocationExecutor" class="org.springframework.integration.adapter.rmi.config.StubRemoteInvocationExecutor"/>
 

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

@@ -23,7 +23,7 @@ import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.handler.HandlerMethodInvoker;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.Source;
 import org.springframework.integration.util.MethodValidator;
 import org.springframework.util.Assert;
 
@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class MethodInvokingSource<T> implements PollableSource<Object>, InitializingBean {
+public class MethodInvokingSource<T> implements Source<Object>, InitializingBean {
 
 	private T object;
 

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

@@ -18,11 +18,9 @@ package org.springframework.integration.channel;
 
 import java.util.List;
 
-import org.springframework.beans.BeansException;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.ApplicationContextAware;
 import org.springframework.integration.ConfigurationException;
-import org.springframework.integration.config.MessageBusParser;
+import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.bus.MessageBusAware;
 import org.springframework.integration.endpoint.EndpointRegistry;
 import org.springframework.integration.endpoint.HandlerEndpoint;
 import org.springframework.integration.handler.ReplyHandler;
@@ -38,7 +36,7 @@ import org.springframework.integration.scheduling.Subscription;
  * 
  * @author Mark Fisher
  */
-public class RequestReplyTemplate implements ApplicationContextAware {
+public class RequestReplyTemplate implements MessageBusAware {
 
 	private MessageChannel requestChannel;
 
@@ -122,9 +120,9 @@ public class RequestReplyTemplate implements ApplicationContextAware {
 		this.endpointRegistry = endpointRegistry;
 	}
 
-	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-		if (applicationContext.containsBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME)) {
-			this.setEndpointRegistry((EndpointRegistry) applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
+	public void setMessageBus(MessageBus messageBus) {
+		if (this.endpointRegistry == null) {
+			this.setEndpointRegistry(messageBus);
 		}
 	}
 

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

@@ -23,7 +23,6 @@ import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.endpoint.PollingSourceEndpoint;
-import org.springframework.integration.endpoint.SimpleSourceEndpoint;
 import org.springframework.integration.scheduling.PollingSchedule;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.util.StringUtils;
@@ -37,10 +36,7 @@ import org.springframework.util.xml.DomUtils;
 public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser {
 
 	protected final Class<?> getBeanClass(Element element) {
-		if (this.getScheduleElement(element) != null) {
-			return PollingSourceEndpoint.class;
-		}
-		return SimpleSourceEndpoint.class;
+		return PollingSourceEndpoint.class;
 	}
 
 	protected boolean shouldGenerateId() {
@@ -68,9 +64,10 @@ public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser {
 		builder.addConstructorArgReference(source);
 		builder.addConstructorArgReference(output);
 		Element scheduleElement = this.getScheduleElement(element);
-		if (scheduleElement != null) {
-			builder.addConstructorArgValue(this.parseSchedule(scheduleElement));
+		if (scheduleElement == null) {
+			throw new ConfigurationException("The <schedule/> sub-element is required for a <source-endpoint/>.");
 		}
+		builder.addConstructorArgValue(this.parseSchedule(scheduleElement));
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java

@@ -25,7 +25,7 @@ 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.PollableSource;
+import org.springframework.integration.message.Source;
 import org.springframework.util.Assert;
 
 /**
@@ -46,14 +46,14 @@ import org.springframework.util.Assert;
  */
 public class DefaultPollingDispatcher extends SimpleDispatcher implements PollingDispatcher {
 
-	private final PollableSource<?> source;
+	private final Source<?> source;
 
 
 	public DefaultPollingDispatcher(MessageChannel channel) {
 		this(channel, channel.getDispatcherPolicy());
 	}
 
-	public DefaultPollingDispatcher(PollableSource<?> source, DispatcherPolicy dispatcherPolicy) {
+	public DefaultPollingDispatcher(Source<?> source, DispatcherPolicy dispatcherPolicy) {
 		super(dispatcherPolicy);
 		Assert.notNull(source, "source must not be null");
 		this.source = source;

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java

@@ -17,11 +17,11 @@
 package org.springframework.integration.dispatcher;
 
 import org.springframework.integration.message.Poller;
-import org.springframework.integration.message.SubscribableSource;
+import org.springframework.integration.message.Subscribable;
 
 /**
  * @author Mark Fisher
  */
-public interface PollingDispatcher extends Poller, MessageDispatcher, SubscribableSource {
+public interface PollingDispatcher extends Poller, MessageDispatcher, Subscribable {
 
 }

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java

@@ -30,7 +30,7 @@ import org.springframework.integration.handler.MessageHandlerRejectedExecutionEx
 import org.springframework.integration.message.BlockingTarget;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageDeliveryException;
-import org.springframework.integration.message.SubscribableSource;
+import org.springframework.integration.message.Subscribable;
 import org.springframework.integration.message.Target;
 
 /**
@@ -38,7 +38,7 @@ import org.springframework.integration.message.Target;
  * 
  * @author Mark Fisher
  */
-public class SimpleDispatcher implements MessageDispatcher, SubscribableSource {
+public class SimpleDispatcher implements MessageDispatcher, Subscribable {
 
 	protected final Log logger = LogFactory.getLog(this.getClass());
 

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java

@@ -26,8 +26,8 @@ import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.DispatcherPolicy;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.PollableSource;
-import org.springframework.integration.message.SubscribableSource;
+import org.springframework.integration.message.Source;
+import org.springframework.integration.message.Subscribable;
 import org.springframework.integration.message.Target;
 import org.springframework.integration.message.selector.MessageSelector;
 
@@ -45,12 +45,12 @@ import org.springframework.integration.message.selector.MessageSelector;
  * @author Dave Syer
  * @author Mark Fisher
  */
-public class SynchronousChannel extends AbstractMessageChannel implements SubscribableSource {
+public class SynchronousChannel extends AbstractMessageChannel implements Subscribable {
 
 	private static final ThreadLocalMessageHolder messageHolder = new ThreadLocalMessageHolder();
 
 
-	private volatile PollableSource<?> source;
+	private volatile Source<?> source;
 
 	private final SimpleDispatcher dispatcher;
 
@@ -61,14 +61,14 @@ public class SynchronousChannel extends AbstractMessageChannel implements Subscr
 		this(null);
 	}
 
-	public SynchronousChannel(PollableSource<?> source) {
+	public SynchronousChannel(Source<?> source) {
 		super(defaultDispatcherPolicy());
 		this.source = source;
 		this.dispatcher = new SimpleDispatcher(this.getDispatcherPolicy());
 	}
 
 
-	public void setSource(PollableSource<?> source) {
+	public void setSource(Source<?> source) {
 		this.source = source;
 	}
 

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

@@ -32,14 +32,14 @@ public abstract class AbstractSourceEndpoint implements SourceEndpoint {
 
 	protected final Log logger = LogFactory.getLog(this.getClass());
 
-	private final Source source;
+	private final Source<?> source;
 
 	private final MessageChannel channel;
 
 	private volatile String name;
 
 
-	public AbstractSourceEndpoint(Source source, MessageChannel channel) {
+	public AbstractSourceEndpoint(Source<?> source, MessageChannel channel) {
 		Assert.notNull(source, "source must not be null");
 		Assert.notNull(channel, "channel must not be null");
 		this.source = source;
@@ -47,7 +47,7 @@ public abstract class AbstractSourceEndpoint implements SourceEndpoint {
 	}
 
 
-	public Source getSource() {
+	public Source<?> getSource() {
 		return this.source;
 	}
 

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

@@ -29,7 +29,7 @@ import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.dispatcher.DefaultPollingDispatcher;
 import org.springframework.integration.dispatcher.PollingDispatcher;
 import org.springframework.integration.dispatcher.PollingDispatcherTask;
-import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.Source;
 import org.springframework.integration.scheduling.MessagingTask;
 import org.springframework.integration.scheduling.PollingSchedule;
 import org.springframework.integration.scheduling.Schedule;
@@ -60,7 +60,7 @@ public class PollingSourceEndpoint extends AbstractSourceEndpoint implements Mes
 	private final Object taskMonitor = new Object();
 
 
-	public PollingSourceEndpoint(PollableSource<?> source, MessageChannel channel, PollingSchedule schedule) {
+	public PollingSourceEndpoint(Source<?> source, MessageChannel channel, PollingSchedule schedule) {
 		super(source, channel);
 		Assert.notNull(schedule, "schedule must not be null");
 		this.dispatcher = new DefaultPollingDispatcher(source, this.dispatcherPolicy);

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

@@ -1,32 +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.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.SubscribableSource;
-
-/**
- * @author Mark Fisher
- */
-public class SimpleSourceEndpoint extends AbstractSourceEndpoint {
-
-	public SimpleSourceEndpoint(SubscribableSource source, MessageChannel channel) {
-		super(source, channel);
-		source.subscribe(channel);
-	}
-
-}

spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java

@@ -23,8 +23,12 @@ import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.beans.SimpleTypeConverter;
 import org.springframework.beans.TypeConverter;
 import org.springframework.beans.factory.BeanClassLoaderAware;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.config.MessageBusParser;
 import org.springframework.integration.message.Message;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -35,7 +39,8 @@ import org.springframework.util.ClassUtils;
  * 
  * @author Mark Fisher
  */
-public class GatewayProxyFactoryBean extends MessagingGateway implements FactoryBean, MethodInterceptor, InitializingBean, BeanClassLoaderAware {
+public class GatewayProxyFactoryBean extends MessagingGateway
+		implements FactoryBean, MethodInterceptor, InitializingBean, BeanClassLoaderAware, BeanFactoryAware {
 
 	private Class<?> serviceInterface;
 
@@ -59,6 +64,11 @@ public class GatewayProxyFactoryBean extends MessagingGateway implements Factory
 		this.beanClassLoader = beanClassLoader;
 	}
 
+	public void setBeanFactory(BeanFactory beanFactory) {
+		this.getRequestReplyTemplate().setMessageBus(
+				(MessageBus) beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
+	}
+
 	public void afterPropertiesSet() {
 		this.serviceProxy = new ProxyFactory(this.serviceInterface, this).getProxy(this.beanClassLoader);
 	}
@@ -84,7 +94,7 @@ public class GatewayProxyFactoryBean extends MessagingGateway implements Factory
 			if (shouldReturnMessage) {
 				return this.receive();
 			}
-			response = this.invoke();
+			response = this.receive();
 		}
 		else {
 			Object payload = (paramCount == 1) ? invocation.getArguments()[0] : invocation.getArguments();
@@ -92,7 +102,7 @@ public class GatewayProxyFactoryBean extends MessagingGateway implements Factory
 				this.send(payload);
 				return null;
 			}
-			response = this.invoke(payload, !shouldReturnMessage);
+			response = shouldReturnMessage ? this.sendAndReceiveMessage(payload) : this.sendAndReceive(payload);
 		}
 		return (response != null) ? this.typeConverter.convertIfNecessary(response, returnType) : null;
 	}

spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGateway.java

@@ -33,19 +33,17 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class MessagingGateway extends RequestReplyTemplate {
+public class MessagingGateway {
+
+	private final RequestReplyTemplate requestReplyTemplate = new RequestReplyTemplate();
 
 	private MessageCreator messageCreator = new DefaultMessageCreator();
 
 	private MessageMapper messageMapper = new DefaultMessageMapper();
 
 
-	public MessagingGateway(MessageChannel requestChannel, MessageChannel replyChannel) {
-		super(requestChannel, replyChannel);
-	}
-
 	public MessagingGateway(MessageChannel requestChannel) {
-		super(requestChannel);
+		this.requestReplyTemplate.setRequestChannel(requestChannel);
 	}
 
 	public MessagingGateway() {
@@ -53,6 +51,22 @@ public class MessagingGateway extends RequestReplyTemplate {
 	}
 
 
+	public void setRequestChannel(MessageChannel requestChannel) {
+		this.requestReplyTemplate.setRequestChannel(requestChannel);
+	}
+
+	public void setReplyChannel(MessageChannel replyChannel) {
+		this.requestReplyTemplate.setReplyChannel(replyChannel);
+	}
+
+	public void setRequestTimeout(long requestTimeout) {
+		this.requestReplyTemplate.setRequestTimeout(requestTimeout);
+	}
+
+	public void setReplyTimeout(long replyTimeout) {
+		this.requestReplyTemplate.setReplyTimeout(replyTimeout);
+	}
+
 	public void setMessageCreator(MessageCreator<?, ?> messageCreator) {
 		Assert.notNull(messageCreator, "messageCreator must not be null");
 		this.messageCreator = messageCreator;
@@ -63,30 +77,38 @@ public class MessagingGateway extends RequestReplyTemplate {
 		this.messageMapper = messageMapper;
 	}
 
+	protected RequestReplyTemplate getRequestReplyTemplate() {
+		return this.requestReplyTemplate;
+	}
+
 	public void send(Object object) {
 		Message<?> message = (object instanceof Message) ? (Message) object :
 				this.messageCreator.createMessage(object);
 		if (message != null) {
-			this.send(message);
+			this.requestReplyTemplate.send(message);
 		}
 	}
 
-	public Object invoke() {
-		Message<?> message = this.receive();
+	public Object receive() {
+		Message<?> message = this.requestReplyTemplate.receive();
 		return (message != null) ? this.messageMapper.mapMessage(message) : null;
 	}
 
-	public Object invoke(Object object) {
-		return this.invoke(object, true);
+	public Object sendAndReceive(Object object) {
+		return this.sendAndReceive(object, true);
 	}
 
-	public Object invoke(Object object, boolean shouldMapMessage) {
+	public Message<?> sendAndReceiveMessage(Object object) {
+		return (Message<?>) this.sendAndReceive(object, false);
+	}
+
+	private Object sendAndReceive(Object object, boolean shouldMapMessage) {
 		Message<?> request = (object instanceof Message) ? (Message) object :
 				this.messageCreator.createMessage(object);
 		if (request == null) {
 			return null;
 		}
-		Message<?> reply = this.request(request);
+		Message<?> reply = this.requestReplyTemplate.request(request);
 		if (!shouldMapMessage) {
 			return reply;
 		}

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

@@ -21,7 +21,7 @@ package org.springframework.integration.message;
  * 
  * @author Mark Fisher
  */
-public interface BlockingSource<T> extends PollableSource<T> {
+public interface BlockingSource<T> extends Source<T> {
 
 	/**
 	 * Receive a message, blocking indefinitely if necessary.

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

@@ -1,31 +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;
-
-/**
- * Interface for any external message source that can be polled.
- * 
- * @author Mark Fisher
- */
-public interface PollableSource<T> extends Source {
-
-	/**
-	 * Retrieve a message from this source or <code>null</code> if no message is available.
-	 */
-	Message<T> receive();
-
-}

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

@@ -21,6 +21,11 @@ package org.springframework.integration.message;
  * 
  * @author Mark Fisher
  */
-public interface Source {
+public interface Source<T> {
+
+	/**
+	 * Retrieve a message from this source or <code>null</code> if no message is available.
+	 */
+	Message<T> receive();
 
 }

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

@@ -17,11 +17,11 @@
 package org.springframework.integration.message;
 
 /**
- * Interface for any message source that accepts subscribers.
+ * Interface for any component that accepts subscribers.
  * 
  * @author Mark Fisher
  */
-public interface SubscribableSource extends Source {
+public interface Subscribable {
 
 	/**
 	 * Register a {@link Target} as a subscriber to this source.

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

@@ -25,6 +25,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
+
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.channel.DispatcherPolicy;
@@ -38,7 +39,7 @@ import org.springframework.integration.message.ErrorMessage;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageDeliveryException;
-import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.Source;
 import org.springframework.integration.message.StringMessage;
 import org.springframework.integration.scheduling.PollingSchedule;
 import org.springframework.integration.scheduling.Subscription;
@@ -219,9 +220,7 @@ public class MessageBusTests {
 
 	@Test(expected = BeanCreationException.class)
 	public void testMultipleMessageBusBeans() {
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("multipleMessageBusBeans.xml", 
-				this.getClass());
-
+		new ClassPathXmlApplicationContext("multipleMessageBusBeans.xml", this.getClass());
 	}
 
 	@Test
@@ -258,7 +257,7 @@ public class MessageBusTests {
 		assertTrue(messageBusAwareBean.getMessageBus() == context.getBean("bus"));
 	}
 
-	private static class FailingSource implements PollableSource<Object> {
+	private static class FailingSource implements Source<Object> {
 
 		private CountDownLatch latch;
 

spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java

@@ -29,7 +29,7 @@ import java.util.concurrent.TimeUnit;
 import org.junit.Test;
 
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.Source;
 import org.springframework.integration.message.StringMessage;
 import org.springframework.integration.message.Target;
 
@@ -96,7 +96,7 @@ public class SynchronousChannelTests {
 
 	@Test
 	public void testReceive() {
-		SynchronousChannel channel = new SynchronousChannel(new PollableSource<String>() {
+		SynchronousChannel channel = new SynchronousChannel(new Source<String>() {
 			public Message<String> receive() {
 				return new StringMessage("foo");
 			}
@@ -175,7 +175,7 @@ public class SynchronousChannelTests {
 	}
 
 
-	private static class MessageReturningTestSource implements PollableSource<String> {
+	private static class MessageReturningTestSource implements Source<String> {
 
 		private final String messageText;
 

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

@@ -35,7 +35,7 @@ import org.springframework.aop.MethodBeforeAdvice;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.Source;
 import org.springframework.integration.scheduling.PollingSchedule;
 
 /**
@@ -429,7 +429,7 @@ public class PollingSourceEndpointTests {
 	}
 
 
-	private static class TestSource implements PollableSource<String> {
+	private static class TestSource implements Source<String> {
 
 		private String message;
 

spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java

@@ -37,8 +37,8 @@ public class GatewayProxyFactoryBeanTests {
 		final MessageChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
 		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
-		proxyFactory.setServiceInterface(TestService.class);
 		proxyFactory.setRequestChannel(requestChannel);
+		proxyFactory.setServiceInterface(TestService.class);
 		proxyFactory.afterPropertiesSet();
 		TestService service = (TestService) proxyFactory.getObject();
 		String result = service.requestReply("foo");