INT-4152: Migrate AMQP DSL

JIRA: https://jira.spring.io/browse/INT-4152

Move the AMQP DSL implementation to spring-integration-amqp.

Polishing

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java

@@ -0,0 +1,245 @@
+/*
+ * Copyright 2014-2016 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.amqp.dsl;
+
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+
+/**
+ * Factory class for AMQP components.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public final class Amqp {
+
+	/**
+	 * Create an initial {@link AmqpInboundGatewaySpec}.
+	 * @param connectionFactory the connectionFactory.
+	 * @param queueNames the queueNames.
+	 * @return the AmqpInboundGatewaySpec.
+	 */
+	public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, String... queueNames) {
+		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
+		listenerContainer.setQueueNames(queueNames);
+		return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer);
+	}
+
+	/**
+	 * Create an initial {@link AmqpInboundGatewaySpec}.
+	 * @param connectionFactory the connectionFactory.
+	 * @param amqpTemplate the {@link AmqpTemplate} to use.
+	 * @param queueNames the queueNames.
+	 * @return the AmqpInboundGatewaySpec.
+	 */
+	public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate,
+			String... queueNames) {
+		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
+		listenerContainer.setQueueNames(queueNames);
+		return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer, amqpTemplate);
+	}
+
+
+	/**
+	 * Create an initial {@link AmqpInboundGatewaySpec}.
+	 * @param connectionFactory the connectionFactory.
+	 * @param queues the queues.
+	 * @return the AmqpInboundGatewaySpec.
+	 */
+	public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, Queue... queues) {
+		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
+		listenerContainer.setQueues(queues);
+		return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer);
+	}
+
+	/**
+	 * Create an initial {@link AmqpInboundGatewaySpec}.
+	 * @param connectionFactory the connectionFactory.
+	 * @param amqpTemplate the {@link AmqpTemplate} to use.
+	 * @param queues the queues.
+	 * @return the AmqpInboundGatewaySpec.
+	 */
+	public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate,
+			Queue... queues) {
+		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
+		listenerContainer.setQueues(queues);
+		return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer, amqpTemplate);
+	}
+
+	/**
+	 * Create an initial {@link AmqpBaseInboundGatewaySpec}
+	 * with provided {@link SimpleMessageListenerContainer}.
+	 * Note: only endpoint options are available from spec.
+	 * The {@code listenerContainer} options should be specified
+	 * on the provided {@link SimpleMessageListenerContainer}.
+	 * @param listenerContainer the listenerContainer
+	 * @return the AmqpBaseInboundGatewaySpec.
+	 */
+	public static AmqpBaseInboundGatewaySpec<?> inboundGateway(SimpleMessageListenerContainer listenerContainer) {
+		return new AmqpInboundGatewaySpec(listenerContainer);
+	}
+
+	/**
+	 * Create an initial {@link AmqpBaseInboundGatewaySpec}
+	 * with provided {@link SimpleMessageListenerContainer}.
+	 * Note: only endpoint options are available from spec.
+	 * The {@code listenerContainer} options should be specified
+	 * on the provided {@link SimpleMessageListenerContainer}.
+	 * @param listenerContainer the listenerContainer
+	 * @param amqpTemplate the {@link AmqpTemplate} to use.
+	 * @return the AmqpBaseInboundGatewaySpec.
+	 */
+	public static AmqpBaseInboundGatewaySpec<?> inboundGateway(SimpleMessageListenerContainer listenerContainer,
+			AmqpTemplate amqpTemplate) {
+		return new AmqpInboundGatewaySpec(listenerContainer, amqpTemplate);
+	}
+
+	/**
+	 * Create an initial AmqpInboundChannelAdapterSpec.
+	 * @param connectionFactory the connectionFactory.
+	 * @param queueNames the queueNames.
+	 * @return the AmqpInboundChannelAdapterSpec.
+	 */
+	public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory,
+			String... queueNames) {
+		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
+		listenerContainer.setQueueNames(queueNames);
+		return (AmqpInboundChannelAdapterSpec) inboundAdapter(listenerContainer);
+	}
+
+	/**
+	 * Create an initial AmqpInboundChannelAdapterSpec.
+	 * @param connectionFactory the connectionFactory.
+	 * @param queues the queues.
+	 * @return the AmqpInboundChannelAdapterSpec.
+	 */
+	public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory, Queue... queues) {
+		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
+		listenerContainer.setQueues(queues);
+		return (AmqpInboundChannelAdapterSpec) inboundAdapter(listenerContainer);
+	}
+
+	/**
+	 * Create an initial AmqpInboundChannelAdapterSpec.
+	 * @param listenerContainer the listenerContainer
+	 * @return the AmqpInboundChannelAdapterSpec.
+	 */
+	public static AmqpBaseInboundChannelAdapterSpec<?> inboundAdapter(
+			SimpleMessageListenerContainer listenerContainer) {
+		return new AmqpInboundChannelAdapterSpec(listenerContainer);
+	}
+
+	/**
+	 * Create an initial AmqpOutboundEndpointSpec (adapter).
+	 * @param amqpTemplate the amqpTemplate.
+	 * @return the AmqpOutboundEndpointSpec.
+	 */
+	public static AmqpOutboundEndpointSpec outboundAdapter(AmqpTemplate amqpTemplate) {
+		return new AmqpOutboundEndpointSpec(amqpTemplate, false);
+	}
+
+	/**
+	 * Create an initial AmqpOutboundEndpointSpec (gateway).
+	 * @param amqpTemplate the amqpTemplate.
+	 * @return the AmqpOutboundEndpointSpec.
+	 */
+	public static AmqpOutboundEndpointSpec outboundGateway(AmqpTemplate amqpTemplate) {
+		return new AmqpOutboundEndpointSpec(amqpTemplate, true);
+	}
+
+	/**
+	 * Create an initial AmqpAsyncOutboundGatewaySpec.
+	 * @param asyncRabbitTemplate the {@link AsyncRabbitTemplate}.
+	 * @return the AmqpOutboundEndpointSpec.
+	 */
+	public static AmqpAsyncOutboundGatewaySpec asyncOutboundGateway(AsyncRabbitTemplate asyncRabbitTemplate) {
+		return new AmqpAsyncOutboundGatewaySpec(asyncRabbitTemplate);
+	}
+
+	/**
+	 * Create an initial AmqpPollableMessageChannelSpec.
+	 * @param connectionFactory the connectionFactory.
+	 * @param <S> the spec type.
+	 * @return the AmqpPollableMessageChannelSpec.
+	 */
+	public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S> pollableChannel(
+			ConnectionFactory connectionFactory) {
+		return pollableChannel(null, connectionFactory);
+	}
+
+	/**
+	 * Create an initial AmqpPollableMessageChannelSpec.
+	 * @param id the id.
+	 * @param connectionFactory the connectionFactory.
+	 * @param <S> the spec type.
+	 * @return the AmqpPollableMessageChannelSpec.
+	 */
+	public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S> pollableChannel(
+			String id, ConnectionFactory connectionFactory) {
+		return new AmqpPollableMessageChannelSpec<S>(connectionFactory).id(id);
+	}
+
+	/**
+	 * Create an initial AmqpMessageChannelSpec.
+	 * @param connectionFactory the connectionFactory.
+	 * @param <S> the spec type.
+	 * @return the AmqpMessageChannelSpec.
+	 */
+	public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(
+			ConnectionFactory connectionFactory) {
+		return channel(null, connectionFactory);
+	}
+
+	/**
+	 * Create an initial AmqpMessageChannelSpec.
+	 * @param id the id.
+	 * @param connectionFactory the connectionFactory.
+	 * @param <S> the spec type.
+	 * @return the AmqpMessageChannelSpec.
+	 */
+	public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(String id,
+			ConnectionFactory connectionFactory) {
+		return new AmqpMessageChannelSpec<S>(connectionFactory).id(id);
+	}
+
+	/**
+	 * Create an initial AmqpPublishSubscribeMessageChannelSpec.
+	 * @param connectionFactory the connectionFactory.
+	 * @return the AmqpPublishSubscribeMessageChannelSpec.
+	 */
+	public static AmqpPublishSubscribeMessageChannelSpec publishSubscribeChannel(ConnectionFactory connectionFactory) {
+		return publishSubscribeChannel(null, connectionFactory);
+	}
+
+	/**
+	 * Create an initial AmqpPublishSubscribeMessageChannelSpec.
+	 * @param id the id.
+	 * @param connectionFactory the connectionFactory.
+	 * @return the AmqpPublishSubscribeMessageChannelSpec.
+	 */
+	public static AmqpPublishSubscribeMessageChannelSpec publishSubscribeChannel(String id,
+			ConnectionFactory connectionFactory) {
+		return new AmqpPublishSubscribeMessageChannelSpec(connectionFactory).id(id);
+	}
+
+	private Amqp() {
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpAsyncOutboundGatewaySpec.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2016 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.amqp.dsl;
+
+import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
+import org.springframework.integration.amqp.outbound.AsyncAmqpOutboundGateway;
+
+/**
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class AmqpAsyncOutboundGatewaySpec
+		extends AmqpBaseOutboundEndpointSpec<AmqpAsyncOutboundGatewaySpec, AsyncAmqpOutboundGateway> {
+
+	AmqpAsyncOutboundGatewaySpec(AsyncRabbitTemplate template) {
+		this.target = new AsyncAmqpOutboundGateway(template);
+		this.target.setRequiresReply(true);
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2014-2016 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.amqp.dsl;
+
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
+import org.springframework.integration.dsl.MessageProducerSpec;
+
+/**
+ * The base {@link MessageProducerSpec} implementation for a {@link AmqpInboundChannelAdapter}.
+ *
+ * @param <S> the target {@link AmqpBaseInboundChannelAdapterSpec} implementation type.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class AmqpBaseInboundChannelAdapterSpec<S extends AmqpBaseInboundChannelAdapterSpec<S>>
+		extends MessageProducerSpec<S, AmqpInboundChannelAdapter> {
+
+	private final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
+
+	AmqpBaseInboundChannelAdapterSpec(AmqpInboundChannelAdapter producer) {
+		super(producer);
+		this.target.setHeaderMapper(this.headerMapper);
+	}
+
+	/**
+	 * Configure the adapter's {@link MessageConverter};
+	 * defaults to {@link org.springframework.amqp.support.converter.SimpleMessageConverter}.
+	 * @param messageConverter the messageConverter.
+	 * @return the spec.
+	 * @see AmqpInboundChannelAdapter#setMessageConverter
+	 */
+	public S messageConverter(MessageConverter messageConverter) {
+		this.target.setMessageConverter(messageConverter);
+		return _this();
+	}
+
+	/**
+	 * Configure the adapter's {@link AmqpHeaderMapper};
+	 * defaults to {@link DefaultAmqpHeaderMapper}.
+	 * @param headerMapper the headerMapper.
+	 * @return the spec.
+	 */
+	public S headerMapper(AmqpHeaderMapper headerMapper) {
+		this.target.setHeaderMapper(headerMapper);
+		return _this();
+	}
+
+	/**
+	 * Only applies if the default header mapper is used.
+	 * @param headers the headers.
+	 * @return the spec.
+	 * @see DefaultAmqpHeaderMapper#setRequestHeaderNames(String[])
+	 */
+	public S mappedRequestHeaders(String... headers) {
+		this.headerMapper.setRequestHeaderNames(headers);
+		return _this();
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2014-2016 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.amqp.dsl;
+
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
+import org.springframework.integration.dsl.MessagingGatewaySpec;
+
+/**
+ * A base {@link MessagingGatewaySpec} implementation for {@link AmqpInboundGateway} endpoint options.
+ * Doesn't allow to specify {@code listenerContainer} options.
+ *
+ * @param <S> the target {@link AmqpBaseInboundGatewaySpec} implementation type.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ *
+ * @see AmqpInboundGateway
+ */
+public class AmqpBaseInboundGatewaySpec<S extends AmqpBaseInboundGatewaySpec<S>>
+		extends MessagingGatewaySpec<S, AmqpInboundGateway> {
+
+	private final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
+
+	AmqpBaseInboundGatewaySpec(AmqpInboundGateway gateway) {
+		super(gateway);
+		this.target.setHeaderMapper(this.headerMapper);
+	}
+
+	/**
+	 * Configure the gateway's {@link MessageConverter};
+	 * defaults to {@link org.springframework.amqp.support.converter.SimpleMessageConverter}.
+	 * @param messageConverter the messageConverter.
+	 * @return the spec.
+	 * @see AmqpInboundGateway#setMessageConverter
+	 */
+	public S messageConverter(MessageConverter messageConverter) {
+		this.target.setMessageConverter(messageConverter);
+		return _this();
+	}
+
+	/**
+	 * Configure the gateway's {@link AmqpHeaderMapper}; defaults to
+	 * {@link org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper}.
+	 * @param headerMapper the headerMapper.
+	 * @return the spec.
+	 */
+	public S headerMapper(AmqpHeaderMapper headerMapper) {
+		this.target.setHeaderMapper(headerMapper);
+		return _this();
+	}
+
+	/**
+	 * Only applies if the default header mapper is used.
+	 * @param headers the headers.
+	 * @return the spec.
+	 * @see org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper#setRequestHeaderNames(String[])
+	 */
+	public S mappedRequestHeaders(String... headers) {
+		this.headerMapper.setRequestHeaderNames(headers);
+		return _this();
+	}
+
+	/**
+	 * Only applies if the default header mapper is used.
+	 * @param headers the headers.
+	 * @return the spec.
+	 * @see org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper#setReplyHeaderNames(String[])
+	 */
+	public S mappedReplyHeaders(String... headers) {
+		this.headerMapper.setReplyHeaderNames(headers);
+		return _this();
+	}
+
+	/**
+	 * The {@code defaultReplyTo} address with the form
+	 * <pre class="code">
+	 * (exchange)/(routingKey)
+	 * </pre>
+	 * or
+	 * <pre class="code">
+	 * (queueName)
+	 * </pre>
+	 * if the request message doesn't have a {@code replyTo} property.
+	 * The second form uses the default exchange ("") and the queue name as
+	 * the routing key.
+	 * @param defaultReplyTo the default {@code replyTo} address to use.
+	 * @return the spec.
+	 * @see AmqpInboundGateway#setDefaultReplyTo
+	 */
+	public S defaultReplyTo(String defaultReplyTo) {
+		this.target.setDefaultReplyTo(defaultReplyTo);
+		return _this();
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseOutboundEndpointSpec.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright 2016 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.amqp.dsl;
+
+import java.util.function.Function;
+
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.expression.Expression;
+import org.springframework.integration.amqp.outbound.AbstractAmqpOutboundEndpoint;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
+import org.springframework.integration.dsl.MessageHandlerSpec;
+import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * The base {@link MessageHandlerSpec} for {@link AbstractAmqpOutboundEndpoint}s.
+ *
+ * @param <S> the target {@link AmqpBaseOutboundEndpointSpec} implementation type.
+ * @param <E> the target {@link AbstractAmqpOutboundEndpoint} implementation type.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public abstract class
+	AmqpBaseOutboundEndpointSpec<S extends AmqpBaseOutboundEndpointSpec<S, E>, E extends AbstractAmqpOutboundEndpoint>
+		extends MessageHandlerSpec<S, E> {
+
+	protected final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
+
+	public S headerMapper(AmqpHeaderMapper headerMapper) {
+		this.target.setHeaderMapper(headerMapper);
+		return _this();
+	}
+
+	public S defaultDeliveryMode(MessageDeliveryMode defaultDeliveryMode) {
+		this.target.setDefaultDeliveryMode(defaultDeliveryMode);
+		return _this();
+	}
+
+	public S routingKey(String routingKey) {
+		this.target.setRoutingKey(routingKey);
+		return _this();
+	}
+
+	public S routingKeyExpression(String routingKeyExpression) {
+		return routingKeyExpression(PARSER.parseExpression(routingKeyExpression));
+	}
+
+	public S routingKeyFunction(Function<Message<?>, String> routingKeyFunction) {
+		return routingKeyExpression(new FunctionExpression<Message<?>>(routingKeyFunction));
+	}
+
+	public S routingKeyExpression(Expression routingKeyExpression) {
+		this.target.setRoutingKeyExpression(routingKeyExpression);
+		return _this();
+	}
+
+	public S returnChannel(MessageChannel returnChannel) {
+		this.target.setReturnChannel(returnChannel);
+		return _this();
+	}
+
+	public S confirmAckChannel(MessageChannel ackChannel) {
+		this.target.setConfirmAckChannel(ackChannel);
+		return _this();
+	}
+
+	public S exchangeName(String exchangeName) {
+		this.target.setExchangeName(exchangeName);
+		return _this();
+	}
+
+	public S exchangeNameExpression(String exchangeNameExpression) {
+		return exchangeNameExpression(PARSER.parseExpression(exchangeNameExpression));
+	}
+
+	public S exchangeNameFunction(Function<Message<?>, String> exchangeNameFunction) {
+		return exchangeNameExpression(new FunctionExpression<Message<?>>(exchangeNameFunction));
+	}
+
+	public S exchangeNameExpression(Expression exchangeNameExpression) {
+		this.target.setExchangeNameExpression(exchangeNameExpression);
+		return _this();
+	}
+
+	public S confirmNackChannel(MessageChannel nackChannel) {
+		this.target.setConfirmNackChannel(nackChannel);
+		return _this();
+	}
+
+	public S confirmCorrelationExpression(String confirmCorrelationExpression) {
+		return confirmCorrelationExpression(PARSER.parseExpression(confirmCorrelationExpression));
+	}
+
+	public S confirmCorrelationFunction(Function<Message<?>, Object> confirmCorrelationFunction) {
+		return confirmCorrelationExpression(new FunctionExpression<Message<?>>(confirmCorrelationFunction));
+	}
+
+
+	public S confirmCorrelationExpression(Expression confirmCorrelationExpression) {
+		this.target.setConfirmCorrelationExpression(confirmCorrelationExpression);
+		return _this();
+	}
+
+	public S mappedRequestHeaders(String... headers) {
+		this.headerMapper.setRequestHeaderNames(headers);
+		return _this();
+	}
+
+	public S mappedReplyHeaders(String... headers) {
+		this.headerMapper.setReplyHeaderNames(headers);
+		return _this();
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java

@@ -0,0 +1,266 @@
+/*
+ * Copyright 2014-2015 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.amqp.dsl;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.concurrent.Executor;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
+import org.springframework.integration.dsl.ComponentsRegistration;
+import org.springframework.integration.dsl.MessageProducerSpec;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.util.ErrorHandler;
+
+/**
+ * A {@link MessageProducerSpec} for {@link AmqpInboundChannelAdapter}s.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class AmqpInboundChannelAdapterSpec extends AmqpBaseInboundChannelAdapterSpec<AmqpInboundChannelAdapterSpec>
+		implements ComponentsRegistration {
+
+	private final SimpleMessageListenerContainer listenerContainer;
+
+	AmqpInboundChannelAdapterSpec(SimpleMessageListenerContainer listenerContainer) {
+		super(new AmqpInboundChannelAdapter(listenerContainer));
+		this.listenerContainer = listenerContainer;
+	}
+
+	/**
+	 * @param acknowledgeMode the acknowledgeMode.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setAcknowledgeMode(AcknowledgeMode)
+	 */
+	public AmqpInboundChannelAdapterSpec acknowledgeMode(AcknowledgeMode acknowledgeMode) {
+		this.listenerContainer.setAcknowledgeMode(acknowledgeMode);
+		return this;
+	}
+
+	/**
+	 * @param queueName a vararg list of queue names to add.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#addQueueNames(String...)
+	 */
+	public AmqpInboundChannelAdapterSpec addQueueNames(String... queueName) {
+		this.listenerContainer.addQueueNames(queueName);
+		return this;
+	}
+
+	/**
+	 * @param queues a vararg list of queues to add.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#addQueueNames(String...)
+	 */
+	public AmqpInboundChannelAdapterSpec addQueues(Queue... queues) {
+		this.listenerContainer.addQueues(queues);
+		return this;
+	}
+
+	/**
+	 * @param errorHandler the errorHandler.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setErrorHandler(ErrorHandler)
+	 */
+	public AmqpInboundChannelAdapterSpec errorHandler(ErrorHandler errorHandler) {
+		this.listenerContainer.setErrorHandler(errorHandler);
+		return this;
+	}
+
+	/**
+	 * @param transactional true for transactional channels.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setChannelTransacted(boolean)
+	 */
+	public AmqpInboundChannelAdapterSpec channelTransacted(boolean transactional) {
+		this.listenerContainer.setChannelTransacted(transactional);
+		return this;
+	}
+
+	/**
+	 * @param adviceChain the adviceChain.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setAdviceChain(Advice[])
+	 */
+	public AmqpInboundChannelAdapterSpec adviceChain(Advice... adviceChain) {
+		this.listenerContainer.setAdviceChain(adviceChain);
+		return this;
+	}
+
+	/**
+	 * @param recoveryInterval the recoveryInterval
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setRecoveryInterval(long)
+	 */
+	public AmqpInboundChannelAdapterSpec recoveryInterval(long recoveryInterval) {
+		this.listenerContainer.setRecoveryInterval(recoveryInterval);
+		return this;
+	}
+
+	/**
+	 * @param concurrentConsumers the concurrentConsumers
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setConcurrentConsumers(int)
+	 */
+	public AmqpInboundChannelAdapterSpec concurrentConsumers(int concurrentConsumers) {
+		this.listenerContainer.setConcurrentConsumers(concurrentConsumers);
+		return this;
+	}
+
+	/**
+	 * @param maxConcurrentConsumers the maxConcurrentConsumers.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setMaxConcurrentConsumers(int)
+	 */
+	public AmqpInboundChannelAdapterSpec maxConcurrentConsumers(int maxConcurrentConsumers) {
+		this.listenerContainer.setMaxConcurrentConsumers(maxConcurrentConsumers);
+		return this;
+	}
+
+	/**
+	 * @param exclusive true for exclusive.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setExclusive(boolean)
+	 */
+	public AmqpInboundChannelAdapterSpec exclusive(boolean exclusive) {
+		this.listenerContainer.setExclusive(exclusive);
+		return this;
+	}
+
+	/**
+	 * @param startConsumerMinInterval the startConsumerMinInterval
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setStartConsumerMinInterval(long)
+	 */
+	public AmqpInboundChannelAdapterSpec startConsumerMinInterval(long startConsumerMinInterval) {
+		this.listenerContainer.setStartConsumerMinInterval(startConsumerMinInterval);
+		return this;
+	}
+
+	/**
+	 * @param stopConsumerMinInterval the stopConsumerMinInterval.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setStopConsumerMinInterval(long)
+	 */
+	public AmqpInboundChannelAdapterSpec stopConsumerMinInterval(long stopConsumerMinInterval) {
+		this.listenerContainer.setStopConsumerMinInterval(stopConsumerMinInterval);
+		return this;
+	}
+
+	/**
+	 * @param consecutiveActiveTrigger the consecutiveActiveTrigger.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setConsecutiveActiveTrigger(int)
+	 */
+	public AmqpInboundChannelAdapterSpec consecutiveActiveTrigger(int consecutiveActiveTrigger) {
+		this.listenerContainer.setConsecutiveActiveTrigger(consecutiveActiveTrigger);
+		return this;
+	}
+
+	/**
+	 * @param consecutiveIdleTrigger the consecutiveIdleTrigger.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setConsecutiveIdleTrigger(int)
+	 */
+	public AmqpInboundChannelAdapterSpec consecutiveIdleTrigger(int consecutiveIdleTrigger) {
+		this.listenerContainer.setConsecutiveIdleTrigger(consecutiveIdleTrigger);
+		return this;
+	}
+
+	/**
+	 * @param receiveTimeout the receiveTimeout
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setReceiveTimeout(long)
+	 */
+	public AmqpInboundChannelAdapterSpec receiveTimeout(long receiveTimeout) {
+		this.listenerContainer.setReceiveTimeout(receiveTimeout);
+		return this;
+	}
+
+	/**
+	 * @param shutdownTimeout the shutdownTimeout.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setShutdownTimeout(long)
+	 */
+	public AmqpInboundChannelAdapterSpec shutdownTimeout(long shutdownTimeout) {
+		this.listenerContainer.setShutdownTimeout(shutdownTimeout);
+		return this;
+	}
+
+	/**
+	 * Configure an {@link Executor} used to invoke the message listener.
+	 * @param taskExecutor the taskExecutor.
+	 * @return the spec.
+	 */
+	public AmqpInboundChannelAdapterSpec taskExecutor(Executor taskExecutor) {
+		this.listenerContainer.setTaskExecutor(taskExecutor);
+		return this;
+	}
+
+	/**
+	 * @param prefetchCount the prefetchCount.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setPrefetchCount(int)
+	 */
+	public AmqpInboundChannelAdapterSpec prefetchCount(int prefetchCount) {
+		this.listenerContainer.setPrefetchCount(prefetchCount);
+		return this;
+	}
+
+	/**
+	 * @param txSize the txSize.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setTxSize(int)
+	 */
+	public AmqpInboundChannelAdapterSpec txSize(int txSize) {
+		this.listenerContainer.setTxSize(txSize);
+		return this;
+	}
+
+	/**
+	 * Configure a {@link PlatformTransactionManager}; used to synchronize the rabbit transaction
+	 * with some other transaction(s).
+	 * @param transactionManager the transactionManager.
+	 * @return the spec.
+	 */
+	public AmqpInboundChannelAdapterSpec transactionManager(PlatformTransactionManager transactionManager) {
+		this.listenerContainer.setTransactionManager(transactionManager);
+		return this;
+	}
+
+	/**
+	 * @param defaultRequeueRejected the defaultRequeueRejected.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setDefaultRequeueRejected(boolean)
+	 */
+	public AmqpInboundChannelAdapterSpec defaultRequeueRejected(boolean defaultRequeueRejected) {
+		this.listenerContainer.setDefaultRequeueRejected(defaultRequeueRejected);
+		return this;
+	}
+
+	@Override
+	public Collection<Object> getComponentsToRegister() {
+		return Collections.<Object>singleton(this.listenerContainer);
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java

@@ -0,0 +1,279 @@
+/*
+ * Copyright 2014-2015 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.amqp.dsl;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.concurrent.Executor;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
+import org.springframework.integration.dsl.ComponentsRegistration;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.util.ErrorHandler;
+
+/**
+ * An {@link AmqpBaseInboundGatewaySpec} implementation for a {@link AmqpInboundGateway}.
+ * Allows to provide {@link SimpleMessageListenerContainer} options.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class AmqpInboundGatewaySpec extends AmqpBaseInboundGatewaySpec<AmqpInboundGatewaySpec>
+		implements ComponentsRegistration {
+
+	private final SimpleMessageListenerContainer listenerContainer;
+
+	AmqpInboundGatewaySpec(SimpleMessageListenerContainer listenerContainer) {
+		super(new AmqpInboundGateway(listenerContainer));
+		this.listenerContainer = listenerContainer;
+	}
+
+	/**
+	 * Instantiate {@link AmqpInboundGateway} based on the provided {@link SimpleMessageListenerContainer}
+	 * and {@link AmqpTemplate}.
+	 * @param listenerContainer the {@link SimpleMessageListenerContainer} to use.
+	 * @param amqpTemplate the {@link AmqpTemplate} to use.
+	 */
+	AmqpInboundGatewaySpec(SimpleMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) {
+		super(new AmqpInboundGateway(listenerContainer, amqpTemplate));
+		this.listenerContainer = listenerContainer;
+	}
+
+	/**
+	 * @param acknowledgeMode the acknowledgeMode.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setAcknowledgeMode(AcknowledgeMode)
+	 */
+	public AmqpInboundGatewaySpec acknowledgeMode(AcknowledgeMode acknowledgeMode) {
+		this.listenerContainer.setAcknowledgeMode(acknowledgeMode);
+		return this;
+	}
+
+	/**
+	 * @param queueName a vararg list of queue names to add.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#addQueueNames(String...)
+	 */
+	public AmqpInboundGatewaySpec addQueueNames(String... queueName) {
+		this.listenerContainer.addQueueNames(queueName);
+		return this;
+	}
+
+	/**
+	 * @param queues a vararg list of queues to add.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#addQueueNames(String...)
+	 */
+	public AmqpInboundGatewaySpec addQueues(Queue... queues) {
+		this.listenerContainer.addQueues(queues);
+		return this;
+	}
+
+	/**
+	 * @param errorHandler the errorHandler.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setErrorHandler(ErrorHandler)
+	 */
+	public AmqpInboundGatewaySpec errorHandler(ErrorHandler errorHandler) {
+		this.listenerContainer.setErrorHandler(errorHandler);
+		return this;
+	}
+
+	/**
+	 * @param transactional true for transactional channels.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setChannelTransacted(boolean)
+	 */
+	public AmqpInboundGatewaySpec channelTransacted(boolean transactional) {
+		this.listenerContainer.setChannelTransacted(transactional);
+		return this;
+	}
+
+	/**
+	 * @param adviceChain the adviceChain.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setAdviceChain(Advice[])
+	 */
+	public AmqpInboundGatewaySpec adviceChain(Advice... adviceChain) {
+		this.listenerContainer.setAdviceChain(adviceChain);
+		return this;
+	}
+
+	/**
+	 * @param recoveryInterval the recoveryInterval
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setRecoveryInterval(long)
+	 */
+	public AmqpInboundGatewaySpec recoveryInterval(long recoveryInterval) {
+		this.listenerContainer.setRecoveryInterval(recoveryInterval);
+		return this;
+	}
+
+	/**
+	 * @param concurrentConsumers the concurrentConsumers
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setConcurrentConsumers(int)
+	 */
+	public AmqpInboundGatewaySpec concurrentConsumers(int concurrentConsumers) {
+		this.listenerContainer.setConcurrentConsumers(concurrentConsumers);
+		return this;
+	}
+
+	/**
+	 * @param maxConcurrentConsumers the maxConcurrentConsumers.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setMaxConcurrentConsumers(int)
+	 */
+	public AmqpInboundGatewaySpec maxConcurrentConsumers(int maxConcurrentConsumers) {
+		this.listenerContainer.setMaxConcurrentConsumers(maxConcurrentConsumers);
+		return this;
+	}
+
+	/**
+	 * @param exclusive true for exclusive.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setExclusive(boolean)
+	 */
+	public AmqpInboundGatewaySpec exclusive(boolean exclusive) {
+		this.listenerContainer.setExclusive(exclusive);
+		return this;
+	}
+
+	/**
+	 * @param startConsumerMinInterval the startConsumerMinInterval
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setStartConsumerMinInterval(long)
+	 */
+	public AmqpInboundGatewaySpec startConsumerMinInterval(long startConsumerMinInterval) {
+		this.listenerContainer.setStartConsumerMinInterval(startConsumerMinInterval);
+		return this;
+	}
+
+	/**
+	 * @param stopConsumerMinInterval the stopConsumerMinInterval.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setStopConsumerMinInterval(long)
+	 */
+	public AmqpInboundGatewaySpec stopConsumerMinInterval(long stopConsumerMinInterval) {
+		this.listenerContainer.setStopConsumerMinInterval(stopConsumerMinInterval);
+		return this;
+	}
+
+	/**
+	 * @param consecutiveActiveTrigger the consecutiveActiveTrigger.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setConsecutiveActiveTrigger(int)
+	 */
+	public AmqpInboundGatewaySpec consecutiveActiveTrigger(int consecutiveActiveTrigger) {
+		this.listenerContainer.setConsecutiveActiveTrigger(consecutiveActiveTrigger);
+		return this;
+	}
+
+	/**
+	 * @param consecutiveIdleTrigger the consecutiveIdleTrigger.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setConsecutiveIdleTrigger(int)
+	 */
+	public AmqpInboundGatewaySpec consecutiveIdleTrigger(int consecutiveIdleTrigger) {
+		this.listenerContainer.setConsecutiveIdleTrigger(consecutiveIdleTrigger);
+		return this;
+	}
+
+	/**
+	 * @param receiveTimeout the receiveTimeout
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setReceiveTimeout(long)
+	 */
+	public AmqpInboundGatewaySpec receiveTimeout(long receiveTimeout) {
+		this.listenerContainer.setReceiveTimeout(receiveTimeout);
+		return this;
+	}
+
+	/**
+	 * @param shutdownTimeout the shutdownTimeout.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setShutdownTimeout(long)
+	 */
+	public AmqpInboundGatewaySpec shutdownTimeout(long shutdownTimeout) {
+		this.listenerContainer.setShutdownTimeout(shutdownTimeout);
+		return this;
+	}
+
+	/**
+	 * Configure an {@link Executor} used to invoke the message listener.
+	 * @param taskExecutor the taskExecutor.
+	 * @return the spec.
+	 */
+	public AmqpInboundGatewaySpec taskExecutor(Executor taskExecutor) {
+		this.listenerContainer.setTaskExecutor(taskExecutor);
+		return this;
+	}
+
+	/**
+	 * @param prefetchCount the prefetchCount.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setPrefetchCount(int)
+	 */
+	public AmqpInboundGatewaySpec prefetchCount(int prefetchCount) {
+		this.listenerContainer.setPrefetchCount(prefetchCount);
+		return this;
+	}
+
+	/**
+	 * @param txSize the txSize.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setTxSize(int)
+	 */
+	public AmqpInboundGatewaySpec txSize(int txSize) {
+		this.listenerContainer.setTxSize(txSize);
+		return this;
+	}
+
+	/**
+	 * Configure a {@link PlatformTransactionManager}; used to synchronize the rabbit transaction
+	 * with some other transaction(s).
+	 * @param transactionManager the transactionManager.
+	 * @return the spec.
+	 */
+	public AmqpInboundGatewaySpec transactionManager(PlatformTransactionManager transactionManager) {
+		this.listenerContainer.setTransactionManager(transactionManager);
+		return this;
+	}
+
+	/**
+	 * @param defaultRequeueRejected the defaultRequeueRejected.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setDefaultRequeueRejected(boolean)
+	 */
+	public AmqpInboundGatewaySpec defaultRequeueRejected(boolean defaultRequeueRejected) {
+		this.listenerContainer.setDefaultRequeueRejected(defaultRequeueRejected);
+		return this;
+	}
+
+	@Override
+	public Collection<Object> getComponentsToRegister() {
+		return Collections.<Object>singleton(this.listenerContainer);
+	}
+
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpMessageChannelSpec.java

@@ -0,0 +1,221 @@
+/*
+ * Copyright 2014-2016 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.amqp.dsl;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.Executor;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
+import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.TransactionAttribute;
+import org.springframework.util.ErrorHandler;
+
+/**
+ * An {@link AmqpPollableMessageChannelSpec} for a message-driven
+ * {@link org.springframework.integration.amqp.channel.PointToPointSubscribableAmqpChannel}.
+ *
+ * @param <S> the target {@link AmqpMessageChannelSpec} implementation type.
+ *
+ * @author Artem Bilan
+ * @author Gary Russell
+ * @since 5.0
+ */
+public class AmqpMessageChannelSpec<S extends AmqpMessageChannelSpec<S>> extends AmqpPollableMessageChannelSpec<S> {
+
+	private final List<Advice> adviceChain = new LinkedList<Advice>();
+
+	AmqpMessageChannelSpec(ConnectionFactory connectionFactory) {
+		super(new AmqpChannelFactoryBean(true), connectionFactory);
+	}
+
+	/**
+	 * @param maxSubscribers the maxSubscribers.
+	 * @return the spec.
+	 * @see org.springframework.integration.amqp.channel.PointToPointSubscribableAmqpChannel#setMaxSubscribers(int)
+	 */
+	public S maxSubscribers(int maxSubscribers) {
+		this.amqpChannelFactoryBean.setMaxSubscribers(maxSubscribers);
+		return _this();
+	}
+
+	/**
+	 * @param acknowledgeMode the acknowledgeMode.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setAcknowledgeMode(AcknowledgeMode)
+	 */
+	public S acknowledgeMode(AcknowledgeMode acknowledgeMode) {
+		this.amqpChannelFactoryBean.setAcknowledgeMode(acknowledgeMode);
+		return _this();
+	}
+
+	/**
+	 * @param advice the advice.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setAdviceChain(Advice[])
+	 */
+	public S advice(Advice... advice) {
+		this.adviceChain.addAll(Arrays.asList(advice));
+		return _this();
+	}
+
+	/**
+	 * @param autoStartup the autoStartup.
+	 * @return the spec.
+	 * @see org.springframework.context.SmartLifecycle
+	 */
+	public S autoStartup(boolean autoStartup) {
+		this.amqpChannelFactoryBean.setAutoStartup(autoStartup);
+		return _this();
+	}
+
+	/**
+	 * @param concurrentConsumers the concurrentConsumers
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setConcurrentConsumers(int)
+	 */
+	public S concurrentConsumers(int concurrentConsumers) {
+		this.amqpChannelFactoryBean.setConcurrentConsumers(concurrentConsumers);
+		return _this();
+	}
+
+	/**
+	 * @param errorHandler the errorHandler.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setErrorHandler(ErrorHandler)
+	 */
+	public S errorHandler(ErrorHandler errorHandler) {
+		this.amqpChannelFactoryBean.setErrorHandler(errorHandler);
+		return _this();
+	}
+
+	/**
+	 * @param exposeListenerChannel the exposeListenerChannel.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setExposeListenerChannel(boolean)
+	 */
+	public S exposeListenerChannel(boolean exposeListenerChannel) {
+		this.amqpChannelFactoryBean.setExposeListenerChannel(exposeListenerChannel);
+		return _this();
+	}
+
+	/**
+	 * @param phase the phase.
+	 * @return the spec.
+	 * @see org.springframework.context.SmartLifecycle
+	 */
+	public S phase(int phase) {
+		this.amqpChannelFactoryBean.setPhase(phase);
+		return _this();
+	}
+
+	/**
+	 * @param prefetchCount the prefetchCount.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setPrefetchCount(int)
+	 */
+	public S prefetchCount(int prefetchCount) {
+		this.amqpChannelFactoryBean.setPrefetchCount(prefetchCount);
+		return _this();
+	}
+
+	/**
+	 * @param receiveTimeout the receiveTimeout
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setReceiveTimeout(long)
+	 */
+	public S receiveTimeout(long receiveTimeout) {
+		this.amqpChannelFactoryBean.setReceiveTimeout(receiveTimeout);
+		return _this();
+	}
+
+	/**
+	 * @param recoveryInterval the recoveryInterval
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setRecoveryInterval(long)
+	 */
+	public S recoveryInterval(long recoveryInterval) {
+		this.amqpChannelFactoryBean.setRecoveryInterval(recoveryInterval);
+		return _this();
+	}
+
+	/**
+	 * @param shutdownTimeout the shutdownTimeout.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setShutdownTimeout(long)
+	 */
+	public S shutdownTimeout(long shutdownTimeout) {
+		this.amqpChannelFactoryBean.setShutdownTimeout(shutdownTimeout);
+		return _this();
+	}
+
+	/**
+	 * Configure an {@link Executor} used to invoke the message listener.
+	 * @param taskExecutor the taskExecutor.
+	 * @return the spec.
+	 */
+	public S taskExecutor(Executor taskExecutor) {
+		this.amqpChannelFactoryBean.setTaskExecutor(taskExecutor);
+		return _this();
+	}
+
+	/**
+	 * Configure a {@link TransactionAttribute} to be used with the
+	 * {@link #transactionManager(PlatformTransactionManager)}.
+	 * @param transactionAttribute the transactionAttribute.
+	 * @return the spec.
+	 */
+	public S transactionAttribute(TransactionAttribute transactionAttribute) {
+		this.amqpChannelFactoryBean.setTransactionAttribute(transactionAttribute);
+		return _this();
+	}
+
+	/**
+	 * Configure a {@link PlatformTransactionManager}; used to synchronize the rabbit transaction
+	 * with some other transaction(s).
+	 * @param transactionManager the transactionManager.
+	 * @return the spec.
+	 */
+	public S transactionManager(PlatformTransactionManager transactionManager) {
+		this.amqpChannelFactoryBean.setTransactionManager(transactionManager);
+		return _this();
+	}
+
+	/**
+	 * Configure the txSize.
+	 * @param txSize the txSize.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setTxSize(int)
+	 */
+	public S txSize(int txSize) {
+		this.amqpChannelFactoryBean.setTxSize(txSize);
+		return _this();
+	}
+
+	@Override
+	protected AbstractAmqpChannel doGet() {
+		this.amqpChannelFactoryBean.setAdviceChain(this.adviceChain.toArray(new Advice[this.adviceChain.size()]));
+		return super.doGet();
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundEndpointSpec.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2014-2016 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.amqp.dsl;
+
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
+import org.springframework.util.Assert;
+
+/**
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class AmqpOutboundEndpointSpec
+		extends AmqpBaseOutboundEndpointSpec<AmqpOutboundEndpointSpec, AmqpOutboundEndpoint> {
+
+	private final boolean expectReply;
+
+	AmqpOutboundEndpointSpec(AmqpTemplate amqpTemplate, boolean expectReply) {
+		this.expectReply = expectReply;
+		this.target = new AmqpOutboundEndpoint(amqpTemplate);
+		this.target.setExpectReply(expectReply);
+		this.target.setHeaderMapper(this.headerMapper);
+		if (expectReply) {
+			this.target.setRequiresReply(true);
+		}
+	}
+
+	@Override
+	public AmqpOutboundEndpointSpec mappedReplyHeaders(String... headers) {
+		Assert.isTrue(this.expectReply, "'mappedReplyHeaders' can be applied only for gateway");
+		return super.mappedReplyHeaders(headers);
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPollableMessageChannelSpec.java

@@ -0,0 +1,208 @@
+/*
+ * Copyright 2014-2016 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.amqp.dsl;
+
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.support.DefaultListableBeanFactory;
+import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
+import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.dsl.channel.MessageChannelSpec;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link MessageChannelSpec} for a {@link AbstractAmqpChannel}s.
+ *
+ * @param <S> the target {@link AmqpPollableMessageChannelSpec} implementation type.
+ *
+ * @author Artem Bilan
+ * @author Gary Russell
+ * @since 5.0
+ */
+public class AmqpPollableMessageChannelSpec<S extends AmqpPollableMessageChannelSpec<S>>
+		extends MessageChannelSpec<S, AbstractAmqpChannel> {
+
+	protected final AmqpChannelFactoryBean amqpChannelFactoryBean;
+
+	AmqpPollableMessageChannelSpec(ConnectionFactory connectionFactory) {
+		this(new AmqpChannelFactoryBean(false), connectionFactory);
+	}
+
+	AmqpPollableMessageChannelSpec(AmqpChannelFactoryBean amqpChannelFactoryBean, ConnectionFactory connectionFactory) {
+		this.amqpChannelFactoryBean = amqpChannelFactoryBean;
+		this.amqpChannelFactoryBean.setConnectionFactory(connectionFactory);
+		this.amqpChannelFactoryBean.setSingleton(false);
+		this.amqpChannelFactoryBean.setPubSub(false);
+
+		/*
+		We need this artificial BeanFactory to overcome AmqpChannelFactoryBean initialization.
+		The real BeanFactory will be applied later for the target AbstractAmqpChannel instance.
+		*/
+		this.amqpChannelFactoryBean.setBeanFactory(new DefaultListableBeanFactory());
+	}
+
+	@Override
+	protected S id(String id) {
+		this.amqpChannelFactoryBean.setBeanName(id);
+		return super.id(id);
+	}
+
+	/**
+	 * Also implicitly sets the {@link #id(String)} (if not explicitly set).
+	 * @param queueName the queueName.
+	 * @return the spec.
+	 * @see AmqpChannelFactoryBean#setQueueName(String)
+	 */
+	public S queueName(String queueName) {
+		if (getId() == null) {
+			id(queueName + ".channel");
+		}
+		this.amqpChannelFactoryBean.setQueueName(queueName);
+		return _this();
+	}
+
+	/**
+	 * @param encoding the encoding.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.core.RabbitTemplate#setEncoding(String)
+	 */
+	public S encoding(String encoding) {
+		this.amqpChannelFactoryBean.setEncoding(encoding);
+		return _this();
+	}
+
+	/**
+	 * @param messageConverter the messageConverter.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.core.RabbitTemplate#setMessageConverter(MessageConverter)
+	 */
+	public S amqpMessageConverter(MessageConverter messageConverter) {
+		this.amqpChannelFactoryBean.setMessageConverter(messageConverter);
+		return _this();
+	}
+
+	/**
+	 * Configure {@code channelTransacted} on both the
+	 * {@link org.springframework.amqp.rabbit.core.RabbitTemplate} (for sends) and
+	 * {@link org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer}
+	 * (for receives) when using Spring Integration 4.0. When using Spring Integration
+	 * 4.1, only the container is configured. See {@link #templateChannelTransacted(boolean)}.
+	 * @param channelTransacted the channelTransacted.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.core.RabbitTemplate#setChannelTransacted(boolean)
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setChannelTransacted(boolean)
+	 */
+	public S channelTransacted(boolean channelTransacted) {
+		this.amqpChannelFactoryBean.setChannelTransacted(channelTransacted);
+		return _this();
+	}
+
+	/**
+	 * Configure {@code channelTransacted} on the
+	 * {@link org.springframework.amqp.rabbit.core.RabbitTemplate} used when sending
+	 * messages to the channel. Only applies when Spring Integration 4.1 or greater is
+	 * being used. Otherwise, see {@link #channelTransacted(boolean)}.
+	 * @param channelTransacted the channelTransacted.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.core.RabbitTemplate#setChannelTransacted(boolean)
+	 */
+	public S templateChannelTransacted(boolean channelTransacted) {
+		this.amqpChannelFactoryBean.setTemplateChannelTransacted(channelTransacted);
+		return _this();
+	}
+
+	/**
+	 * Configure {@code messagePropertiesConverter} on both the
+	 * {@link org.springframework.amqp.rabbit.core.RabbitTemplate} (for sends) and
+	 * {@link org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer}
+	 * (for receives).
+	 * @param messagePropertiesConverter the messagePropertiesConverter.
+	 * @return the spec.
+	 * @see org.springframework.amqp.rabbit.core.RabbitTemplate#setMessagePropertiesConverter
+	 * @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setMessagePropertiesConverter
+	 */
+	public S messagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) {
+		this.amqpChannelFactoryBean.setMessagePropertiesConverter(messagePropertiesConverter);
+		return _this();
+	}
+
+	/**
+	 * Configure the delivery mode for messages that don't have an
+	 * {@link AmqpHeaders#DELIVERY_MODE} header. Default is {@link MessageDeliveryMode#PERSISTENT}.
+	 * @param mode the mode.
+	 * @return the spec.
+	 */
+	public S defaultDeliveryMode(MessageDeliveryMode mode) {
+		this.amqpChannelFactoryBean.setDefaultDeliveryMode(mode);
+		return _this();
+	}
+
+	/**
+	 * Configure whether normal spring-messaging to AMQP message mapping is enabled.
+	 * Default false.
+	 * @param extract true to enable mapping.
+	 * @return the spec.
+	 * @see #outboundHeaderMapper(AmqpHeaderMapper)
+	 * @see #inboundHeaderMapper(AmqpHeaderMapper)
+	 */
+	public S extractPayload(boolean extract) {
+		this.amqpChannelFactoryBean.setExtractPayload(extract);
+		return _this();
+	}
+
+	/**
+	 * Configure the outbound header mapper to use when {@link #extractPayload(boolean)}
+	 * is true. Defaults to a {@code DefaultAmqpHeaderMapper}.
+	 * @param mapper the mapper.
+	 * @return the spec.
+	 * @see #extractPayload(boolean)
+	 */
+	public S outboundHeaderMapper(AmqpHeaderMapper mapper) {
+		this.amqpChannelFactoryBean.setOutboundHeaderMapper(mapper);
+		return _this();
+	}
+
+	/**
+	 * Configure the inbound header mapper to use when {@link #extractPayload(boolean)}
+	 * is true. Defaults to a {@code DefaultAmqpHeaderMapper}.
+	 * @param mapper the mapper.
+	 * @return the spec.
+	 * @see #extractPayload(boolean)
+	 */
+	public S inboundHeaderMapper(AmqpHeaderMapper mapper) {
+		this.amqpChannelFactoryBean.setInboundHeaderMapper(mapper);
+		return _this();
+	}
+
+	@Override
+	protected AbstractAmqpChannel doGet() {
+		Assert.notNull(getId(), "The 'id' or 'queueName' must be specified");
+		try {
+			this.channel = this.amqpChannelFactoryBean.getObject();
+		}
+		catch (Exception e) {
+			throw new BeanCreationException("Cannot create the AMQP MessageChannel", e);
+		}
+		return super.doGet();
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPublishSubscribeMessageChannelSpec.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2014-2016 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.amqp.dsl;
+
+import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.integration.amqp.channel.PublishSubscribeAmqpChannel;
+import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
+
+/**
+ * A {@link AmqpMessageChannelSpec} for {@link PublishSubscribeAmqpChannel}s.
+ *
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class AmqpPublishSubscribeMessageChannelSpec
+		extends AmqpMessageChannelSpec<AmqpPublishSubscribeMessageChannelSpec> {
+
+	AmqpPublishSubscribeMessageChannelSpec(ConnectionFactory connectionFactory) {
+		super(connectionFactory);
+		this.amqpChannelFactoryBean.setPubSub(true);
+	}
+
+	/**
+	 * @param exchange the exchange.
+	 * @return the spec.
+	 * @see AmqpChannelFactoryBean#setExchange(FanoutExchange)
+	 */
+	public AmqpPublishSubscribeMessageChannelSpec exchange(FanoutExchange exchange) {
+		this.amqpChannelFactoryBean.setExchange(exchange);
+		return _this();
+	}
+
+}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides AMQP Component support for the Java DSL.
+ */
+package org.springframework.integration.amqp.dsl;

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java

@@ -0,0 +1,281 @@
+/*
+ * Copyright 2014-2016 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.amqp.dsl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.AnonymousQueue;
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
+import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
+import org.springframework.integration.amqp.rule.BrokerRunning;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlowBuilder;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Artem Bilan
+ * @author Gary Russell
+ * @since 5.0
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class AmqpTests {
+
+	@Rule
+	public BrokerRunning brokerRunning = BrokerRunning.isRunning();
+
+	@Autowired
+	private ConnectionFactory rabbitConnectionFactory;
+
+	@Autowired
+	private AmqpTemplate amqpTemplate;
+
+	@Autowired
+	@Qualifier("queue")
+	private Queue amqpQueue;
+
+	@Autowired
+	private AmqpInboundGateway amqpInboundGateway;
+
+	@Test
+	public void testAmqpInboundGatewayFlow() throws Exception {
+		Object result = this.amqpTemplate.convertSendAndReceive(this.amqpQueue.getName(), "world");
+		assertEquals("HELLO WORLD", result);
+
+		this.amqpInboundGateway.stop();
+		//INTEXT-209
+		this.amqpInboundGateway.start();
+
+		this.amqpTemplate.convertAndSend(this.amqpQueue.getName(), "world");
+		((RabbitTemplate) this.amqpTemplate).setReceiveTimeout(10000);
+		result = this.amqpTemplate.receiveAndConvert("defaultReplyTo");
+		assertEquals("HELLO WORLD", result);
+		assertSame(this.amqpTemplate, TestUtils.getPropertyValue(this.amqpInboundGateway, "amqpTemplate"));
+	}
+
+	@Autowired
+	@Qualifier("amqpOutboundInput")
+	private MessageChannel amqpOutboundInput;
+
+	@Autowired
+	@Qualifier("amqpReplyChannel.channel")
+	private PollableChannel amqpReplyChannel;
+
+	@Test
+	public void testAmqpOutboundFlow() throws Exception {
+		this.amqpOutboundInput.send(MessageBuilder.withPayload("hello through the amqp")
+				.setHeader("routingKey", "foo")
+				.build());
+		Message<?> receive = null;
+		int i = 0;
+		do {
+			receive = this.amqpReplyChannel.receive();
+			if (receive != null) {
+				break;
+			}
+			Thread.sleep(100);
+			i++;
+		}
+		while (i < 10);
+
+		assertNotNull(receive);
+		assertEquals("HELLO THROUGH THE AMQP", receive.getPayload());
+	}
+
+	@Test
+	public void testTemplateChannelTransacted() {
+		IntegrationFlowBuilder flow = IntegrationFlows.from(Amqp.channel("testTemplateChannelTransacted",
+				this.rabbitConnectionFactory)
+				.autoStartup(false)
+				.templateChannelTransacted(true));
+		assertTrue(TestUtils.getPropertyValue(flow, "currentMessageChannel.amqpTemplate.transactional",
+				Boolean.class));
+	}
+
+	@Autowired
+	@Qualifier("amqpAsyncOutboundFlow.input")
+	private MessageChannel amqpAsyncOutboundFlowInput;
+
+	@Test
+	public void testAmqpAsyncOutboundGatewayFlow() throws Exception {
+		QueueChannel replyChannel = new QueueChannel();
+		this.amqpAsyncOutboundFlowInput.send(MessageBuilder.withPayload("async gateway")
+				.setReplyChannel(replyChannel)
+				.build());
+
+		Message<?> receive = replyChannel.receive(10000);
+		assertNotNull(receive);
+		assertEquals("HELLO ASYNC GATEWAY", receive.getPayload());
+	}
+
+	@Autowired
+	private AbstractAmqpChannel unitChannel;
+
+	@Autowired
+	private AmqpHeaderMapper mapperIn;
+
+	@Autowired
+	private AmqpHeaderMapper mapperOut;
+
+	@Test
+	public void unitTestChannel() {
+		assertEquals(MessageDeliveryMode.NON_PERSISTENT,
+				TestUtils.getPropertyValue(this.unitChannel, "defaultDeliveryMode"));
+		assertSame(this.mapperIn, TestUtils.getPropertyValue(this.unitChannel, "inboundHeaderMapper"));
+		assertSame(this.mapperOut, TestUtils.getPropertyValue(this.unitChannel, "outboundHeaderMapper"));
+		assertTrue(TestUtils.getPropertyValue(this.unitChannel, "extractPayload", Boolean.class));
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class ContextConfiguration {
+
+		@Bean
+		public ConnectionFactory rabbitConnectionFactory() {
+			return new CachingConnectionFactory("localhost");
+		}
+
+		@Bean
+		public RabbitTemplate amqpTemplate() {
+			return new RabbitTemplate(rabbitConnectionFactory());
+		}
+
+		@Bean
+		public RabbitAdmin amqpAdmin() {
+			return new RabbitAdmin(rabbitConnectionFactory());
+		}
+
+		@Bean
+		public Queue queue() {
+			return new AnonymousQueue();
+		}
+
+		@Bean
+		public Queue defaultReplyTo() {
+			return new Queue("defaultReplyTo");
+		}
+
+		@Bean
+		public IntegrationFlow amqpFlow(ConnectionFactory rabbitConnectionFactory, AmqpTemplate amqpTemplate) {
+			return IntegrationFlows
+					.from(Amqp.inboundGateway(rabbitConnectionFactory, amqpTemplate, queue())
+							.id("amqpInboundGateway")
+							.defaultReplyTo(defaultReplyTo().getName()))
+					.transform("hello "::concat)
+					.transform(String.class, String::toUpperCase)
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow amqpOutboundFlow(ConnectionFactory rabbitConnectionFactory, AmqpTemplate amqpTemplate) {
+			return IntegrationFlows.from(Amqp.channel("amqpOutboundInput", rabbitConnectionFactory))
+					.handle(Amqp.outboundAdapter(amqpTemplate).routingKeyExpression("headers.routingKey"))
+					.get();
+		}
+
+		@Bean
+		public Queue fooQueue() {
+			return new Queue("foo");
+		}
+
+		@Bean
+		public Queue amqpReplyChannel() {
+			return new Queue("amqpReplyChannel");
+		}
+
+		@Bean
+		public IntegrationFlow amqpInboundFlow(ConnectionFactory rabbitConnectionFactory) {
+			return IntegrationFlows.from(Amqp.inboundAdapter(rabbitConnectionFactory, fooQueue()))
+					.transform(String.class, String::toUpperCase)
+					.channel(Amqp.pollableChannel(rabbitConnectionFactory)
+							.queueName("amqpReplyChannel")
+							.channelTransacted(true))
+					.get();
+		}
+
+		@Bean
+		public Queue asyncReplies() {
+			return new Queue("asyncReplies");
+		}
+
+		@Bean
+		public AsyncRabbitTemplate asyncRabbitTemplate(ConnectionFactory rabbitConnectionFactory) {
+			return new AsyncRabbitTemplate(rabbitConnectionFactory, "", "", "asyncReplies");
+		}
+
+		@Bean
+		public IntegrationFlow amqpAsyncOutboundFlow(AsyncRabbitTemplate asyncRabbitTemplate) {
+			return f -> f
+					.handle(Amqp.asyncOutboundGateway(asyncRabbitTemplate)
+							.routingKeyFunction(m -> queue().getName()));
+		}
+
+		@Bean
+		public AbstractAmqpChannel unitChannel(ConnectionFactory rabbitConnectionFactory) {
+			return Amqp.pollableChannel(rabbitConnectionFactory)
+					.queueName("foo")
+					.channelTransacted(true)
+					.extractPayload(true)
+					.inboundHeaderMapper(mapperIn())
+					.outboundHeaderMapper(mapperOut())
+					.defaultDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
+					.get();
+		}
+
+		@Bean
+		public AmqpHeaderMapper mapperIn() {
+			return DefaultAmqpHeaderMapper.inboundMapper();
+		}
+
+		@Bean
+		public AmqpHeaderMapper mapperOut() {
+			return DefaultAmqpHeaderMapper.outboundMapper();
+		}
+
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java

@@ -2778,7 +2778,6 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		return new PublisherIntegrationFlow<T>(this.integrationComponents, channelForPublisher, executor);
 	}
 
-	@SuppressWarnings("unchecked")
 	private <S extends ConsumerEndpointSpec<S, ? extends MessageHandler>> B register(S endpointSpec,
 			Consumer<S> endpointConfigurer) {
 		if (endpointConfigurer != null) {