INT-4278: DSL: AMQP DirectMessageListenerContainer

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

Support configuration of inbound endpoints using `DirectMessageListenerContainer`.

**Needs migration guide update**

Polishing according PR discussion

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

@@ -0,0 +1,385 @@
+/*
+ * Copyright 2017 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.Map;
+import java.util.concurrent.Executor;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.MessageListener;
+import org.springframework.amqp.core.MessagePostProcessor;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.ListenerContainerIdleEvent;
+import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
+import org.springframework.amqp.support.ConditionalExceptionLogger;
+import org.springframework.amqp.support.ConsumerTagStrategy;
+import org.springframework.integration.dsl.IntegrationComponentSpec;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.TransactionAttribute;
+import org.springframework.util.ErrorHandler;
+import org.springframework.util.backoff.BackOff;
+
+/**
+ * Base class for container specs.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ *
+ */
+public abstract class AbstractMessageListenerContainerSpec<S extends AbstractMessageListenerContainerSpec<S, C>,
+		C extends AbstractMessageListenerContainer>
+		extends IntegrationComponentSpec<S, C> {
+
+	public AbstractMessageListenerContainerSpec(C listenerContainer) {
+		this.target = listenerContainer;
+	}
+
+	@Override
+	public S id(String id) {
+		return super.id(id);
+	}
+
+	/**
+	 * @param acknowledgeMode the acknowledgeMode.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setAcknowledgeMode(AcknowledgeMode)
+	 */
+	public S acknowledgeMode(AcknowledgeMode acknowledgeMode) {
+		this.target.setAcknowledgeMode(acknowledgeMode);
+		return _this();
+	}
+
+	/**
+	 * @param queueName a vararg list of queue names to add.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#addQueueNames(String...)
+	 */
+	public S addQueueNames(String... queueName) {
+		this.target.addQueueNames(queueName);
+		return _this();
+	}
+
+	/**
+	 * @param queues a vararg list of queues to add.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#addQueueNames(String...)
+	 */
+	public S addQueues(Queue... queues) {
+		this.target.addQueues(queues);
+		return _this();
+	}
+
+	/**
+	 * @param errorHandler the errorHandler.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setErrorHandler(ErrorHandler)
+	 */
+	public S errorHandler(ErrorHandler errorHandler) {
+		this.target.setErrorHandler(errorHandler);
+		return _this();
+	}
+
+	/**
+	 * @param transactional true for transactional channels.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setChannelTransacted(boolean)
+	 */
+	public S channelTransacted(boolean transactional) {
+		this.target.setChannelTransacted(transactional);
+		return _this();
+	}
+
+	/**
+	 * @param adviceChain the adviceChain.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setAdviceChain(Advice[])
+	 */
+	public S adviceChain(Advice... adviceChain) {
+		this.target.setAdviceChain(adviceChain);
+		return _this();
+	}
+
+	/**
+	 * @param recoveryInterval the recoveryInterval
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setRecoveryInterval(long)
+	 */
+	public S recoveryInterval(long recoveryInterval) {
+		this.target.setRecoveryInterval(recoveryInterval);
+		return _this();
+	}
+
+	/**
+	 * @param exclusive true for exclusive.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setExclusive(boolean)
+	 */
+	public S exclusive(boolean exclusive) {
+		this.target.setExclusive(exclusive);
+		return _this();
+	}
+
+	/**
+	 * @param shutdownTimeout the shutdownTimeout.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setShutdownTimeout(long)
+	 */
+	public S shutdownTimeout(long shutdownTimeout) {
+		this.target.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.target.setTaskExecutor(taskExecutor);
+		return _this();
+	}
+
+	/**
+	 * @param prefetchCount the prefetchCount.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setPrefetchCount(int)
+	 */
+	public S prefetchCount(int prefetchCount) {
+		this.target.setPrefetchCount(prefetchCount);
+		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.target.setTransactionManager(transactionManager);
+		return _this();
+	}
+
+	/**
+	 * @param defaultRequeueRejected the defaultRequeueRejected.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setDefaultRequeueRejected(boolean)
+	 */
+	public S defaultRequeueRejected(boolean defaultRequeueRejected) {
+		this.target.setDefaultRequeueRejected(defaultRequeueRejected);
+		return _this();
+	}
+
+	/**
+	 * Determine whether or not the container should de-batch batched
+	 * messages (true) or call the listener with the batch (false). Default: true.
+	 * @param deBatchingEnabled the deBatchingEnabled to set.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setDeBatchingEnabled(boolean)
+	 */
+	public S deBatchingEnabled(boolean deBatchingEnabled) {
+		this.target.setDeBatchingEnabled(deBatchingEnabled);
+		return _this();
+	}
+
+	/**
+	 * Set {@link MessagePostProcessor}s that will be applied after message reception, before
+	 * invoking the {@link MessageListener}. Often used to decompress data.  Processors are invoked in order,
+	 * depending on {@code PriorityOrder}, {@code Order} and finally unordered.
+	 * @param afterReceivePostProcessors the post processor.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setAfterReceivePostProcessors(MessagePostProcessor...)
+	 */
+	public S afterReceivePostProcessors(MessagePostProcessor... afterReceivePostProcessors) {
+		this.target.setAfterReceivePostProcessors(afterReceivePostProcessors);
+		return _this();
+	}
+
+	/**
+	 * Set a qualifier that will prefix the connection factory lookup key; default none.
+	 * @param lookupKeyQualifier the qualifier
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setLookupKeyQualifier(String)
+	 */
+	public S lookupKeyQualifier(String lookupKeyQualifier) {
+		this.target.setLookupKeyQualifier(lookupKeyQualifier);
+		return _this();
+	}
+
+	/**
+	 * Set the implementation of {@link ConsumerTagStrategy} to generate consumer tags.
+	 * By default, the RabbitMQ server generates consumer tags.
+	 * @param consumerTagStrategy the consumerTagStrategy to set.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setConsumerTagStrategy(ConsumerTagStrategy)
+	 */
+	public S consumerTagStrategy(ConsumerTagStrategy consumerTagStrategy) {
+		this.target.setConsumerTagStrategy(consumerTagStrategy);
+		return _this();
+	}
+
+	/**
+	 * Set consumer arguments.
+	 * @param args the arguments.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setConsumerArguments(Map)
+	 */
+	public S consumerArguments(Map<String, Object> args) {
+		this.target.setConsumerArguments(args);
+		return _this();
+	}
+
+	/**
+	 * How often to emit {@link ListenerContainerIdleEvent}s in milliseconds.
+	 * @param idleEventInterval the interval.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setIdleEventInterval(long)
+	 */
+	public S idleEventInterval(long idleEventInterval) {
+		this.target.setIdleEventInterval(idleEventInterval);
+		return _this();
+	}
+
+	/**
+	 * Set the transaction attribute to use when using an external transaction manager.
+	 * @param transactionAttribute the transaction attribute to set
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setTransactionAttribute(TransactionAttribute)
+	 */
+	public S transactionAttribute(TransactionAttribute transactionAttribute) {
+		this.target.setTransactionAttribute(transactionAttribute);
+		return _this();
+	}
+
+	/**
+	 * Specify the {@link BackOff} for interval between recovery attempts.
+	 * The default is 5000 ms, that is, 5 seconds.
+	 * With the {@link BackOff} you can supply the {@code maxAttempts} for recovery before
+	 * the {@code stop()} will be performed.
+	 * @param recoveryBackOff The BackOff to recover.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setRecoveryBackOff(BackOff)
+	 */
+	public S recoveryBackOff(BackOff recoveryBackOff) {
+		this.target.setRecoveryBackOff(recoveryBackOff);
+		return _this();
+	}
+
+	/**
+	 * Set the {@link MessagePropertiesConverter} for this listener container.
+	 * @param messagePropertiesConverter The properties converter.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setMessagePropertiesConverter(MessagePropertiesConverter)
+	 */
+	public S messagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) {
+		this.target.setMessagePropertiesConverter(messagePropertiesConverter);
+		return _this();
+	}
+
+	/**
+	 * If all of the configured queue(s) are not available on the broker, this setting
+	 * determines whether the condition is fatal. When true, and
+	 * the queues are missing during startup, the context refresh() will fail.
+	 * <p> When false, the condition is not considered fatal and the container will
+	 * continue to attempt to start the consumers.
+	 * @param missingQueuesFatal the missingQueuesFatal to set.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setMissingQueuesFatal(boolean)
+	 */
+	public S missingQueuesFatal(boolean missingQueuesFatal) {
+		this.target.setMissingQueuesFatal(missingQueuesFatal);
+		return _this();
+	}
+
+	/**
+	 * Prevent the container from starting if any of the queues defined in the context have
+	 * mismatched arguments (TTL etc). Default false.
+	 * @param mismatchedQueuesFatal true to fail initialization when this condition occurs.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setMismatchedQueuesFatal(boolean)
+	 */
+	public S mismatchedQueuesFatal(boolean mismatchedQueuesFatal) {
+		this.target.setMismatchedQueuesFatal(mismatchedQueuesFatal);
+		return _this();
+	}
+
+	/**
+	 * Set to true to automatically declare elements (queues, exchanges, bindings)
+	 * in the application context during container start().
+	 * @param autoDeclare the boolean flag to indicate an declaration operation.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setAutoDeclare(boolean)
+	 */
+	public S autoDeclare(boolean autoDeclare) {
+		this.target.setAutoDeclare(autoDeclare);
+		return _this();
+	}
+
+	/**
+	 * Set the interval between passive queue declaration attempts in milliseconds.
+	 * @param failedDeclarationRetryInterval the interval, default 5000.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setFailedDeclarationRetryInterval(long)
+	 */
+	public S failedDeclarationRetryInterval(long failedDeclarationRetryInterval) {
+		this.target.setFailedDeclarationRetryInterval(failedDeclarationRetryInterval);
+		return _this();
+	}
+
+	/**
+	 * Set whether a message with a null messageId is fatal for the consumer
+	 * when using stateful retry. When false, instead of stopping the consumer,
+	 * the message is rejected and not requeued - it will be discarded or routed
+	 * to the dead letter queue, if so configured. Default true.
+	 * @param statefulRetryFatalWithNullMessageId true for fatal.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setStatefulRetryFatalWithNullMessageId(boolean)
+	 */
+	public S statefulRetryFatalWithNullMessageId(boolean statefulRetryFatalWithNullMessageId) {
+		this.target.setStatefulRetryFatalWithNullMessageId(statefulRetryFatalWithNullMessageId);
+		return _this();
+	}
+
+	/**
+	 * Set a {@link ConditionalExceptionLogger} for logging exclusive consumer failures. The
+	 * default is to log such failures at WARN level.
+	 * @param exclusiveConsumerExceptionLogger the conditional exception logger.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setExclusiveConsumerExceptionLogger(ConditionalExceptionLogger)
+	 */
+	public S exclusiveConsumerExceptionLogger(ConditionalExceptionLogger exclusiveConsumerExceptionLogger) {
+		this.target.setExclusiveConsumerExceptionLogger(exclusiveConsumerExceptionLogger);
+		return _this();
+	}
+
+	/**
+	 * Set to true to always requeue on transaction rollback with an external TransactionManager.
+	 * @param alwaysRequeueWithTxManagerRollback true to always requeue on rollback.
+	 * @return the spec.
+	 * @see AbstractMessageListenerContainer#setAlwaysRequeueWithTxManagerRollback(boolean)
+	 */
+	public S alwaysRequeueWithTxManagerRollback(boolean alwaysRequeueWithTxManagerRollback) {
+		this.target.setAlwaysRequeueWithTxManagerRollback(alwaysRequeueWithTxManagerRollback);
+		return _this();
+	}
+
+}

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

@@ -20,12 +20,14 @@ 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.DirectMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 
 /**
  * Factory class for AMQP components.
  *
  * @author Artem Bilan
+ * @author Gary Russell
  * @since 5.0
  */
 public final class Amqp {
@@ -36,10 +38,10 @@ public final class Amqp {
 	 * @param queueNames the queueNames.
 	 * @return the AmqpInboundGatewaySpec.
 	 */
-	public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, String... queueNames) {
+	public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, String... queueNames) {
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueueNames(queueNames);
-		return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer);
+		return inboundGateway(listenerContainer);
 	}
 
 	/**
@@ -49,11 +51,11 @@ public final class Amqp {
 	 * @param queueNames the queueNames.
 	 * @return the AmqpInboundGatewaySpec.
 	 */
-	public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate,
+	public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate,
 			String... queueNames) {
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueueNames(queueNames);
-		return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer, amqpTemplate);
+		return inboundGateway(listenerContainer, amqpTemplate);
 	}
 
 
@@ -63,10 +65,10 @@ public final class Amqp {
 	 * @param queues the queues.
 	 * @return the AmqpInboundGatewaySpec.
 	 */
-	public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, Queue... queues) {
+	public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, Queue... queues) {
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueues(queues);
-		return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer);
+		return inboundGateway(listenerContainer);
 	}
 
 	/**
@@ -76,74 +78,126 @@ public final class Amqp {
 	 * @param queues the queues.
 	 * @return the AmqpInboundGatewaySpec.
 	 */
-	public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate,
+	public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate,
 			Queue... queues) {
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueues(queues);
-		return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer, amqpTemplate);
+		return inboundGateway(listenerContainer, amqpTemplate);
 	}
 
 	/**
-	 * Create an initial {@link AmqpBaseInboundGatewaySpec}
+	 * Create an initial {@link AmqpInboundGatewaySMLCSpec}
 	 * with provided {@link SimpleMessageListenerContainer}.
 	 * Note: only endpoint options are available from spec.
 	 * The {@code listenerContainer} options should be specified
-	 * on the provided {@link SimpleMessageListenerContainer}.
+	 * on the provided {@link SimpleMessageListenerContainer} using
+	 * {@link AmqpInboundGatewaySMLCSpec#configureContainer(java.util.function.Consumer)}.
 	 * @param listenerContainer the listenerContainer
-	 * @return the AmqpBaseInboundGatewaySpec.
+	 * @return the AmqpInboundGatewaySMLCSpec.
 	 */
-	public static AmqpBaseInboundGatewaySpec<?> inboundGateway(SimpleMessageListenerContainer listenerContainer) {
-		return new AmqpInboundGatewaySpec(listenerContainer);
+	public static AmqpInboundGatewaySMLCSpec inboundGateway(SimpleMessageListenerContainer listenerContainer) {
+		return new AmqpInboundGatewaySMLCSpec(listenerContainer);
 	}
 
 	/**
-	 * Create an initial {@link AmqpBaseInboundGatewaySpec}
+	 * Create an initial {@link AmqpInboundGatewaySMLCSpec}
 	 * with provided {@link SimpleMessageListenerContainer}.
 	 * Note: only endpoint options are available from spec.
 	 * The {@code listenerContainer} options should be specified
-	 * on the provided {@link SimpleMessageListenerContainer}.
+	 * on the provided {@link SimpleMessageListenerContainer}using
+	 * {@link AmqpInboundGatewaySMLCSpec#configureContainer(java.util.function.Consumer)}.
 	 * @param listenerContainer the listenerContainer
 	 * @param amqpTemplate the {@link AmqpTemplate} to use.
-	 * @return the AmqpBaseInboundGatewaySpec.
+	 * @return the AmqpInboundGatewaySMLCSpec.
 	 */
-	public static AmqpBaseInboundGatewaySpec<?> inboundGateway(SimpleMessageListenerContainer listenerContainer,
+	public static AmqpInboundGatewaySMLCSpec inboundGateway(SimpleMessageListenerContainer listenerContainer,
 			AmqpTemplate amqpTemplate) {
-		return new AmqpInboundGatewaySpec(listenerContainer, amqpTemplate);
+		return new AmqpInboundGatewaySMLCSpec(listenerContainer, amqpTemplate);
 	}
 
 	/**
-	 * Create an initial AmqpInboundChannelAdapterSpec.
+	 * Create an initial {@link DirectMessageListenerContainerSpec}
+	 * with provided {@link DirectMessageListenerContainer}.
+	 * Note: only endpoint options are available from spec.
+	 * The {@code listenerContainer} options should be specified
+	 * on the provided {@link DirectMessageListenerContainer} using
+	 * {@link AmqpInboundGatewayDMLCSpec#configureContainer(java.util.function.Consumer)}.
+	 * @param listenerContainer the listenerContainer
+	 * @return the AmqpInboundGatewayDMLCSpec.
+	 */
+	public static AmqpInboundGatewayDMLCSpec inboundGateway(DirectMessageListenerContainer listenerContainer) {
+		return new AmqpInboundGatewayDMLCSpec(listenerContainer);
+	}
+
+	/**
+	 * Create an initial {@link AmqpInboundGatewayDMLCSpec}
+	 * with provided {@link DirectMessageListenerContainer}.
+	 * Note: only endpoint options are available from spec.
+	 * The {@code listenerContainer} options should be specified
+	 * on the provided {@link DirectMessageListenerContainer} using
+	 * {@link AmqpInboundGatewayDMLCSpec#configureContainer(java.util.function.Consumer)}.
+	 * @param listenerContainer the listenerContainer
+	 * @param amqpTemplate the {@link AmqpTemplate} to use.
+	 * @return the AmqpInboundGatewayDMLCSpec.
+	 */
+	public static AmqpInboundGatewayDMLCSpec inboundGateway(DirectMessageListenerContainer listenerContainer,
+			AmqpTemplate amqpTemplate) {
+		return new AmqpInboundGatewayDMLCSpec(listenerContainer, amqpTemplate);
+	}
+
+	/**
+	 * Create an initial AmqpInboundChannelAdapterSpec using a
+	 * {@link SimpleMessageListenerContainer}.
 	 * @param connectionFactory the connectionFactory.
 	 * @param queueNames the queueNames.
 	 * @return the AmqpInboundChannelAdapterSpec.
 	 */
-	public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory,
+	public static AmqpInboundChannelAdapterSMLCSpec inboundAdapter(ConnectionFactory connectionFactory,
 			String... queueNames) {
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueueNames(queueNames);
-		return (AmqpInboundChannelAdapterSpec) inboundAdapter(listenerContainer);
+		return new AmqpInboundChannelAdapterSMLCSpec(listenerContainer);
 	}
 
 	/**
-	 * Create an initial AmqpInboundChannelAdapterSpec.
+	 * Create an initial AmqpInboundChannelAdapterSpec using a
+	 * {@link SimpleMessageListenerContainer}.
 	 * @param connectionFactory the connectionFactory.
 	 * @param queues the queues.
 	 * @return the AmqpInboundChannelAdapterSpec.
 	 */
-	public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory, Queue... queues) {
+	public static AmqpInboundChannelAdapterSMLCSpec inboundAdapter(ConnectionFactory connectionFactory, Queue... queues) {
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueues(queues);
-		return (AmqpInboundChannelAdapterSpec) inboundAdapter(listenerContainer);
+		return new AmqpInboundChannelAdapterSMLCSpec(listenerContainer);
 	}
 
 	/**
-	 * Create an initial AmqpInboundChannelAdapterSpec.
+	 * Create an initial {@link AmqpInboundGatewaySMLCSpec}
+	 * with provided {@link SimpleMessageListenerContainer}.
+	 * Note: only endpoint options are available from spec.
+	 * The {@code listenerContainer} options should be specified
+	 * on the provided {@link SimpleMessageListenerContainer} using
+	 * {@link AmqpInboundGatewaySMLCSpec#configureContainer(java.util.function.Consumer)}.
 	 * @param listenerContainer the listenerContainer
-	 * @return the AmqpInboundChannelAdapterSpec.
+	 * @return the AmqpInboundGatewaySMLCSpec.
 	 */
-	public static AmqpBaseInboundChannelAdapterSpec<?> inboundAdapter(
-			SimpleMessageListenerContainer listenerContainer) {
-		return new AmqpInboundChannelAdapterSpec(listenerContainer);
+	public static AmqpInboundChannelAdapterSMLCSpec inboundAdapter(SimpleMessageListenerContainer listenerContainer) {
+		return new AmqpInboundChannelAdapterSMLCSpec(listenerContainer);
+	}
+
+	/**
+	 * Create an initial {@link AmqpInboundGatewayDMLCSpec}
+	 * with provided {@link DirectMessageListenerContainer}.
+	 * Note: only endpoint options are available from spec.
+	 * The {@code listenerContainer} options should be specified
+	 * on the provided {@link DirectMessageListenerContainer} using
+	 * {@link AmqpInboundGatewaySMLCSpec#configureContainer(java.util.function.Consumer)}.
+	 * @param listenerContainer the listenerContainer
+	 * @return the AmqpInboundGatewaySMLCSpec.
+	 */
+	public static AmqpInboundChannelAdapterDMLCSpec inboundAdapter(DirectMessageListenerContainer listenerContainer) {
+		return new AmqpInboundChannelAdapterDMLCSpec(listenerContainer);
 	}
 
 	/**

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2017 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.Consumer;
+
+import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
+
+/**
+ * Spec for an inbound channel adapter with a {@link DirectMessageListenerContainer}.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public class AmqpInboundChannelAdapterDMLCSpec extends AmqpInboundChannelAdapterSpec<AmqpInboundChannelAdapterDMLCSpec,
+		DirectMessageListenerContainer> {
+
+	AmqpInboundChannelAdapterDMLCSpec(DirectMessageListenerContainer listenerContainer) {
+		super(listenerContainer);
+	}
+
+	AmqpInboundChannelAdapterDMLCSpec configureContainer(Consumer<DirectMessageListenerContainerSpec> configurer) {
+		configurer.accept(new DirectMessageListenerContainerSpec(this.listenerContainer));
+		return this;
+	}
+
+}

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2017 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.Consumer;
+
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+
+/**
+ * Spec for an inbound channel adapter with a {@link SimpleMessageListenerContainer}.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public class AmqpInboundChannelAdapterSMLCSpec extends AmqpInboundChannelAdapterSpec<AmqpInboundChannelAdapterSMLCSpec,
+		SimpleMessageListenerContainer> {
+
+	AmqpInboundChannelAdapterSMLCSpec(SimpleMessageListenerContainer listenerContainer) {
+		super(listenerContainer);
+	}
+
+	AmqpInboundChannelAdapterSMLCSpec configureContainer(Consumer<SimpleMessageListenerContainerSpec> configurer) {
+		configurer.accept(new SimpleMessageListenerContainerSpec(this.listenerContainer));
+		return this;
+	}
+
+}

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

@@ -18,246 +18,33 @@ 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.amqp.rabbit.listener.AbstractMessageListenerContainer;
 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.
  *
+ * @param <S> the spec type.
+ * @param <C> the container type.
+ *
  * @author Artem Bilan
  * @since 5.0
  */
-public class AmqpInboundChannelAdapterSpec extends AmqpBaseInboundChannelAdapterSpec<AmqpInboundChannelAdapterSpec>
+public abstract class AmqpInboundChannelAdapterSpec
+			<S extends AmqpInboundChannelAdapterSpec<S, C>, C extends AbstractMessageListenerContainer>
+		extends AmqpBaseInboundChannelAdapterSpec<S>
 		implements ComponentsRegistration {
 
-	private final SimpleMessageListenerContainer listenerContainer;
+	protected final C listenerContainer;
 
-	AmqpInboundChannelAdapterSpec(SimpleMessageListenerContainer listenerContainer) {
+	AmqpInboundChannelAdapterSpec(C 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/AmqpInboundGatewayDMLCSpec.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2017 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.Consumer;
+
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
+
+/**
+ * Spec for a gateway with a {@link DirectMessageListenerContainer}.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public class AmqpInboundGatewayDMLCSpec
+		extends AmqpInboundGatewaySpec<AmqpInboundGatewayDMLCSpec, DirectMessageListenerContainer> {
+
+	AmqpInboundGatewayDMLCSpec(DirectMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) {
+		super(listenerContainer, amqpTemplate);
+	}
+
+	AmqpInboundGatewayDMLCSpec(DirectMessageListenerContainer listenerContainer) {
+		super(listenerContainer);
+	}
+
+	public AmqpInboundGatewayDMLCSpec configureContainer(Consumer<DirectMessageListenerContainerSpec> configurer) {
+		configurer.accept(new DirectMessageListenerContainerSpec(this.listenerContainer));
+		return this;
+	}
+
+}

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

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2017 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.Consumer;
+
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+
+/**
+ * Spec for a gateway with a {@link SimpleMessageListenerContainer}.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public class AmqpInboundGatewaySMLCSpec
+		extends AmqpInboundGatewaySpec<AmqpInboundGatewaySMLCSpec, SimpleMessageListenerContainer> {
+
+	AmqpInboundGatewaySMLCSpec(SimpleMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) {
+		super(listenerContainer, amqpTemplate);
+	}
+
+	AmqpInboundGatewaySMLCSpec(SimpleMessageListenerContainer listenerContainer) {
+		super(listenerContainer);
+	}
+
+	public AmqpInboundGatewaySMLCSpec configureContainer(Consumer<SimpleMessageListenerContainerSpec> configurer) {
+		configurer.accept(new SimpleMessageListenerContainerSpec(this.listenerContainer));
+		return this;
+	}
+
+}

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

@@ -18,262 +18,45 @@ 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.AbstractMessageListenerContainer;
 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.
+ * Allows to provide {@link AbstractMessageListenerContainer} options.
  *
  * @author Artem Bilan
  * @since 5.0
  */
-public class AmqpInboundGatewaySpec extends AmqpBaseInboundGatewaySpec<AmqpInboundGatewaySpec>
-		implements ComponentsRegistration {
+public abstract class AmqpInboundGatewaySpec
+				<S extends AmqpInboundGatewaySpec<S, C>, C extends AbstractMessageListenerContainer>
+		extends AmqpBaseInboundGatewaySpec<S> implements ComponentsRegistration {
 
-	private final SimpleMessageListenerContainer listenerContainer;
+	protected final C listenerContainer;
 
-	AmqpInboundGatewaySpec(SimpleMessageListenerContainer listenerContainer) {
+	AmqpInboundGatewaySpec(C listenerContainer) {
 		super(new AmqpInboundGateway(listenerContainer));
 		this.listenerContainer = listenerContainer;
 	}
 
 	/**
-	 * Instantiate {@link AmqpInboundGateway} based on the provided {@link SimpleMessageListenerContainer}
+	 * Instantiate {@link AmqpInboundGateway} based on the provided {@link AbstractMessageListenerContainer}
 	 * and {@link AmqpTemplate}.
 	 * @param listenerContainer the {@link SimpleMessageListenerContainer} to use.
 	 * @param amqpTemplate the {@link AmqpTemplate} to use.
 	 */
-	AmqpInboundGatewaySpec(SimpleMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) {
+	AmqpInboundGatewaySpec(C 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/DirectMessageListenerContainerSpec.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2017 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.listener.DirectMessageListenerContainer;
+
+/**
+ * Spec for a {@link DirectMessageListenerContainer}.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public class DirectMessageListenerContainerSpec
+		extends AbstractMessageListenerContainerSpec<DirectMessageListenerContainerSpec, DirectMessageListenerContainer> {
+
+	private final DirectMessageListenerContainer listenerContainer;
+
+	public DirectMessageListenerContainerSpec(DirectMessageListenerContainer listenerContainer) {
+		super(listenerContainer);
+		this.listenerContainer = listenerContainer;
+	}
+
+	/**
+	 * @param consumersPerQueue the consumersPerQueue.
+	 * @return the spec.
+	 * @see DirectMessageListenerContainer#setConsumersPerQueue(int)
+	 */
+	public DirectMessageListenerContainerSpec consumersPerQueue(int consumersPerQueue) {
+		this.listenerContainer.setConsumersPerQueue(consumersPerQueue);
+		return this;
+	}
+
+	/**
+	 * @param messagesPerAck the messages per ack.
+	 * @return the spec.
+	 * @see DirectMessageListenerContainer#setMessagesPerAck(int)
+	 */
+	public DirectMessageListenerContainerSpec messagesPerAck(int messagesPerAck) {
+		this.listenerContainer.setMessagesPerAck(messagesPerAck);
+		return this;
+	}
+
+	/**
+	 * @param ackTimeout the ack timeout.
+	 * @return the spec.
+	 * @see DirectMessageListenerContainer#setAckTimeout(long)
+	 */
+	public DirectMessageListenerContainerSpec ackTimeout(long ackTimeout) {
+		this.listenerContainer.setAckTimeout(ackTimeout);
+		return this;
+	}
+
+}

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

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2017 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.listener.SimpleMessageListenerContainer;
+
+/**
+ * Spec for a {@link SimpleMessageListenerContainer}.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public class SimpleMessageListenerContainerSpec extends
+		AbstractMessageListenerContainerSpec<SimpleMessageListenerContainerSpec, SimpleMessageListenerContainer> {
+
+	private final SimpleMessageListenerContainer listenerContainer;
+
+	public SimpleMessageListenerContainerSpec(SimpleMessageListenerContainer listenerContainer) {
+		super(listenerContainer);
+		this.listenerContainer = listenerContainer;
+	}
+
+	/**
+	 * @param concurrentConsumers the concurrentConsumers
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setConcurrentConsumers(int)
+	 */
+	public SimpleMessageListenerContainerSpec concurrentConsumers(int concurrentConsumers) {
+		this.listenerContainer.setConcurrentConsumers(concurrentConsumers);
+		return this;
+	}
+
+	/**
+	 * @param maxConcurrentConsumers the maxConcurrentConsumers.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setMaxConcurrentConsumers(int)
+	 */
+	public SimpleMessageListenerContainerSpec maxConcurrentConsumers(int maxConcurrentConsumers) {
+		this.listenerContainer.setMaxConcurrentConsumers(maxConcurrentConsumers);
+		return this;
+	}
+
+	/**
+	 * @param startConsumerMinInterval the startConsumerMinInterval
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setStartConsumerMinInterval(long)
+	 */
+	public SimpleMessageListenerContainerSpec startConsumerMinInterval(long startConsumerMinInterval) {
+		this.listenerContainer.setStartConsumerMinInterval(startConsumerMinInterval);
+		return this;
+	}
+
+	/**
+	 * @param stopConsumerMinInterval the stopConsumerMinInterval.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setStopConsumerMinInterval(long)
+	 */
+	public SimpleMessageListenerContainerSpec stopConsumerMinInterval(long stopConsumerMinInterval) {
+		this.listenerContainer.setStopConsumerMinInterval(stopConsumerMinInterval);
+		return this;
+	}
+
+	/**
+	 * @param consecutiveActiveTrigger the consecutiveActiveTrigger.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setConsecutiveActiveTrigger(int)
+	 */
+	public SimpleMessageListenerContainerSpec consecutiveActiveTrigger(int consecutiveActiveTrigger) {
+		this.listenerContainer.setConsecutiveActiveTrigger(consecutiveActiveTrigger);
+		return this;
+	}
+
+	/**
+	 * @param consecutiveIdleTrigger the consecutiveIdleTrigger.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setConsecutiveIdleTrigger(int)
+	 */
+	public SimpleMessageListenerContainerSpec consecutiveIdleTrigger(int consecutiveIdleTrigger) {
+		this.listenerContainer.setConsecutiveIdleTrigger(consecutiveIdleTrigger);
+		return this;
+	}
+
+	/**
+	 * @param receiveTimeout the receiveTimeout
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setReceiveTimeout(long)
+	 */
+	public SimpleMessageListenerContainerSpec receiveTimeout(long receiveTimeout) {
+		this.listenerContainer.setReceiveTimeout(receiveTimeout);
+		return this;
+	}
+
+	/**
+	 * @param txSize the txSize.
+	 * @return the spec.
+	 * @see SimpleMessageListenerContainer#setTxSize(int)
+	 */
+	public SimpleMessageListenerContainerSpec txSize(int txSize) {
+		this.listenerContainer.setTxSize(txSize);
+		return this;
+	}
+
+}

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

@@ -36,6 +36,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.junit.BrokerRunning;
+import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
@@ -211,6 +212,23 @@ public class AmqpTests {
 			return IntegrationFlows
 					.from(Amqp.inboundGateway(rabbitConnectionFactory, amqpTemplate, queue())
 							.id("amqpInboundGateway")
+							.configureContainer(c -> c
+								.recoveryInterval(5000)
+								.concurrentConsumers(1))
+							.defaultReplyTo(defaultReplyTo().getName()))
+					.transform("hello "::concat)
+					.transform(String.class, String::toUpperCase)
+					.get();
+		}
+
+		// syntax only
+		public IntegrationFlow amqpDMLCFlow(ConnectionFactory rabbitConnectionFactory, AmqpTemplate amqpTemplate) {
+			return IntegrationFlows
+					.from(Amqp.inboundGateway(new DirectMessageListenerContainer())
+							.id("amqpInboundGateway")
+							.configureContainer(c -> c
+								.recoveryInterval(5000)
+								.consumersPerQueue(1))
 							.defaultReplyTo(defaultReplyTo().getName()))
 					.transform("hello "::concat)
 					.transform(String.class, String::toUpperCase)