DSL: Add IntegrationFlows.from(MGSupport)

* Introduce `Amqp` components factory
* Add `AmqpInboundGatewaySpec` builder

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java

@@ -25,6 +25,7 @@ import org.springframework.integration.dsl.support.EndpointConfigurer;
 import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
 import org.springframework.integration.dsl.support.MessageChannelReference;
 import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.gateway.MessagingGatewaySupport;
 import org.springframework.messaging.MessageChannel;
 
 /**
@@ -87,9 +88,15 @@ public final class IntegrationFlows {
 		return from(outputChannel).addComponent(messageProducer);
 	}
 
-	/*public static IntegrationFlowBuilder from(AbstractEndpoint endpoint) {
-		return new IntegrationFlowBuilder();
-	}*/
+	public static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway) {
+		DirectFieldAccessor dfa = new DirectFieldAccessor(inboundGateway);
+		MessageChannel outputChannel = (MessageChannel) dfa.getPropertyValue("requestChannel");
+		if (outputChannel == null) {
+			outputChannel = new DirectChannel();
+			inboundGateway.setRequestChannel(outputChannel);
+		}
+		return from(outputChannel).addComponent(inboundGateway);
+	}
 
 	private IntegrationFlows() {
 	}

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

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2014 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.dsl.amqp;
+
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
+import org.springframework.integration.dsl.core.MessagingGatewaySpec;
+
+/**
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public abstract class Amqp {
+
+	public static MessagingGatewaySpec<AmqpInboundGatewaySpec, AmqpInboundGateway> inboundGateway(
+			SimpleMessageListenerContainer listenerContainer) {
+		return new AmqpInboundGatewaySpec(listenerContainer);
+	}
+
+	public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, String... queueNames) {
+		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
+		listenerContainer.setQueueNames(queueNames);
+		return new AmqpInboundGatewaySpec(listenerContainer);
+	}
+
+	public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, Queue... queues) {
+		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
+		listenerContainer.setQueues(queues);
+		return new AmqpInboundGatewaySpec(listenerContainer);
+	}
+
+}

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

@@ -0,0 +1,174 @@
+/*
+ * Copyright 2014 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.dsl.amqp;
+
+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.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.core.MessagingGatewaySpec;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.util.ErrorHandler;
+
+/**
+ * @author Artem Bilan
+ */
+public class AmqpInboundGatewaySpec extends MessagingGatewaySpec<AmqpInboundGatewaySpec, AmqpInboundGateway> {
+
+	private final SimpleMessageListenerContainer listenerContainer;
+
+	private final DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
+
+	AmqpInboundGatewaySpec(SimpleMessageListenerContainer listenerContainer) {
+		super(new AmqpInboundGateway(listenerContainer));
+		this.listenerContainer = listenerContainer;
+		this.target.setHeaderMapper(headerMapper);
+	}
+
+	public AmqpInboundGatewaySpec acknowledgeMode(AcknowledgeMode acknowledgeMode) {
+		this.listenerContainer.setAcknowledgeMode(acknowledgeMode);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec addQueueNames(String... queueName) {
+		this.listenerContainer.addQueueNames(queueName);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec addQueues(Queue... queues) {
+		this.listenerContainer.addQueues(queues);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec errorHandler(ErrorHandler errorHandler) {
+		this.listenerContainer.setErrorHandler(errorHandler);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec channelTransacted(boolean transactional) {
+		this.listenerContainer.setChannelTransacted(transactional);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec adviceChain(Advice... adviceChain) {
+		this.listenerContainer.setAdviceChain(adviceChain);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec recoveryInterval(long recoveryInterval) {
+		this.listenerContainer.setRecoveryInterval(recoveryInterval);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec concurrentConsumers(int concurrentConsumers) {
+		this.listenerContainer.setConcurrentConsumers(concurrentConsumers);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec maxConcurrentConsumers(int maxConcurrentConsumers) {
+		this.listenerContainer.setMaxConcurrentConsumers(maxConcurrentConsumers);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec exclusive(boolean exclusive) {
+		this.listenerContainer.setExclusive(exclusive);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec startConsumerMinInterval(long startConsumerMinInterval) {
+		this.listenerContainer.setStartConsumerMinInterval(startConsumerMinInterval);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec stopConsumerMinInterval(long stopConsumerMinInterval) {
+		this.listenerContainer.setStopConsumerMinInterval(stopConsumerMinInterval);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec consecutiveActiveTrigger(int consecutiveActiveTrigger) {
+		this.listenerContainer.setConsecutiveActiveTrigger(consecutiveActiveTrigger);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec consecutiveIdleTrigger(int consecutiveIdleTrigger) {
+		this.listenerContainer.setConsecutiveIdleTrigger(consecutiveIdleTrigger);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec receiveTimeout(long receiveTimeout) {
+		this.listenerContainer.setReceiveTimeout(receiveTimeout);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec shutdownTimeout(long shutdownTimeout) {
+		this.listenerContainer.setShutdownTimeout(shutdownTimeout);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec taskExecutor(Executor taskExecutor) {
+		this.listenerContainer.setTaskExecutor(taskExecutor);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec prefetchCount(int prefetchCount) {
+		this.listenerContainer.setPrefetchCount(prefetchCount);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec txSize(int txSize) {
+		this.listenerContainer.setTxSize(txSize);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec transactionManager(PlatformTransactionManager transactionManager) {
+		this.listenerContainer.setTransactionManager(transactionManager);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec defaultRequeueRejected(boolean defaultRequeueRejected) {
+		this.listenerContainer.setDefaultRequeueRejected(defaultRequeueRejected);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec messageConverter(MessageConverter messageConverter) {
+		this.target.setMessageConverter(messageConverter);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec headerMapper(AmqpHeaderMapper headerMapper) {
+		this.target.setHeaderMapper(headerMapper);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec mappedRequestHeaders(String... headers) {
+		this.headerMapper.setRequestHeaderNames(headers);
+		return this;
+	}
+
+	public AmqpInboundGatewaySpec mappedReplyHeaders(String... headers) {
+		this.headerMapper.setReplyHeaderNames(headers);
+		return this;
+	}
+
+}

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

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

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

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2014 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.dsl.core;
+
+import org.springframework.integration.gateway.MessagingGatewaySupport;
+import org.springframework.integration.mapping.InboundMessageMapper;
+import org.springframework.integration.mapping.OutboundMessageMapper;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Artem Bilan
+ */
+public abstract class MessagingGatewaySpec<S extends MessagingGatewaySpec<S, G>, G extends MessagingGatewaySupport>
+		extends IntegrationComponentSpec<S, G> {
+
+	public MessagingGatewaySpec(G gateway) {
+		this.target = gateway;
+	}
+
+	public S id(String id) {
+		this.target.setBeanName(id);
+		return super.id(id);
+	}
+
+	public S phase(int phase) {
+		this.target.setPhase(phase);
+		return _this();
+	}
+
+	public S autoStartup(boolean autoStartup) {
+		this.target.setAutoStartup(autoStartup);
+		return _this();
+	}
+
+	public S replyChannel(MessageChannel replyChannel) {
+		this.target.setReplyChannel(replyChannel);
+		return _this();
+	}
+
+	public S requestChannel(MessageChannel requestChannel) {
+		target.setRequestChannel(requestChannel);
+		return _this();
+	}
+
+	public S errorChannel(MessageChannel errorChannel) {
+		target.setErrorChannel(errorChannel);
+		return _this();
+	}
+
+	public S requestTimeout(long requestTimeout) {
+		target.setRequestTimeout(requestTimeout);
+		return _this();
+	}
+
+	public S replyTimeout(long replyTimeout) {
+		target.setReplyTimeout(replyTimeout);
+		return _this();
+	}
+
+	public S requestMapper(InboundMessageMapper<?> requestMapper) {
+		target.setRequestMapper(requestMapper);
+		return _this();
+	}
+
+	public S replyMapper(OutboundMessageMapper<?> replyMapper) {
+		target.setReplyMapper(replyMapper);
+		return _this();
+	}
+
+	@Override
+	protected final G doGet() {
+		throw new UnsupportedOperationException();
+	}
+
+}

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java

@@ -51,10 +51,15 @@ import org.junit.BeforeClass;
 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.Queue;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -82,6 +87,7 @@ import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.ResequencerSpec;
 import org.springframework.integration.dsl.SplitterEndpointSpec;
+import org.springframework.integration.dsl.amqp.Amqp;
 import org.springframework.integration.dsl.channel.DirectChannelSpec;
 import org.springframework.integration.dsl.channel.MessageChannels;
 import org.springframework.integration.dsl.support.GenericSplitter;
@@ -284,6 +290,13 @@ public class IntegrationFlowTests {
 	@Qualifier("tailChannel")
 	private PollableChannel tailChannel;
 
+	@Autowired
+	private AmqpTemplate amqpTemplate;
+
+	@Autowired
+	private Queue amqpQueue;
+
+
 	@BeforeClass
 	public static void setup() throws IOException {
 		mongoPort = Network.getFreeServerPort();
@@ -843,6 +856,12 @@ public class IntegrationFlowTests {
 	}
 
 
+	@Test
+	public void testAmqpInboundGatewayFlow() throws Exception {
+		Object result = this.amqpTemplate.convertSendAndReceive(this.amqpQueue.getName(), "world");
+		assertEquals("HELLO WORLD", result);
+	}
+
 	@MessagingGateway(defaultRequestChannel = "controlBus")
 	private static interface ControlBusGateway {
 
@@ -850,7 +869,7 @@ public class IntegrationFlowTests {
 	}
 
 	@Configuration
-	@EnableIntegration
+	@EnableAutoConfiguration
 	@IntegrationComponentScan
 	public static class ContextConfiguration {
 
@@ -1258,6 +1277,21 @@ public class IntegrationFlowTests {
 					.get();
 		}
 
+		@Autowired
+		private ConnectionFactory rabbitConnectionFactory;
+
+		@Bean
+		public Queue queue() {
+			return new AnonymousQueue();
+		}
+
+		@Bean
+		public IntegrationFlow amqpFlow() {
+			return IntegrationFlows.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue()).get())
+					.transform("hello "::concat)
+					.transform((String p) -> p.toUpperCase())
+					.get();
+		}
 	}
 
 	private static class RoutingTestBean {