From 77026e8261e9d7e756ec86cbfe54ecac96a64695 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 19 May 2014 20:52:51 +0300 Subject: [PATCH] DSL: Add `IntegrationFlows.from(MGSupport)` * Introduce `Amqp` components factory * Add `AmqpInboundGatewaySpec` builder --- spring-integration-java-dsl/build.gradle | 4 +- .../integration/dsl/IntegrationFlows.java | 13 +- .../integration/dsl/amqp/Amqp.java | 48 +++++ .../dsl/amqp/AmqpInboundGatewaySpec.java | 174 ++++++++++++++++++ .../integration/dsl/amqp/package-info.java | 4 + .../dsl/core/MessagingGatewaySpec.java | 89 +++++++++ .../dsl/test/IntegrationFlowTests.java | 36 +++- 7 files changed, 363 insertions(+), 5 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundGatewaySpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/package-info.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingGatewaySpec.java diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index 64fd173..d7b4a84 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -27,7 +27,8 @@ ext { embedMongoVersion = '1.45' jacocoVersion = '0.7.0.201403182114' log4jVersion = '1.2.17' - springIntegrationVersion = '4.0.0.RELEASE' + springIntegrationVersion = '4.0.1.BUILD-SNAPSHOT' + springBootVersion = '1.1.0.BUILD-SNAPSHOT' linkHomepage = 'https://github.com/spring-projects/spring-integration-extensions' linkCi = 'https://build.spring.io/browse/INTEXT' @@ -76,6 +77,7 @@ dependencies { testCompile "log4j:log4j:$log4jVersion" testCompile "org.springframework.integration:spring-integration-test:$springIntegrationVersion" testCompile "de.flapdoodle.embed:de.flapdoodle.embed.mongo:$embedMongoVersion" + testCompile "org.springframework.boot:spring-boot-autoconfigure:$springBootVersion" jacoco "org.jacoco:org.jacoco.agent:$jacocoVersion:runtime" } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index 92a0543..250621a 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/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() { } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java new file mode 100644 index 0000000..16120b6 --- /dev/null +++ b/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 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); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundGatewaySpec.java new file mode 100644 index 0000000..37faa02 --- /dev/null +++ b/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 { + + 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; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/package-info.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/package-info.java new file mode 100644 index 0000000..ed0d7b6 --- /dev/null +++ b/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; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingGatewaySpec.java new file mode 100644 index 0000000..762ae96 --- /dev/null +++ b/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, G extends MessagingGatewaySupport> + extends IntegrationComponentSpec { + + 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(); + } + +} diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index 4c032ac..e5e4de2 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/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 {