From 36063ec8f2e55e440599b6f1f909eb8b430a0f15 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 9 Jun 2014 13:54:40 +0300 Subject: [PATCH] DSL: Add `Amqp.*channel()` support * Upgrade to the latest Boot snapshot * Polishing and workarounds for `Proxy` beans. Caused by `@EnableJmx` --- spring-integration-java-dsl/build.gradle | 10 +- .../integration/dsl/amqp/Amqp.java | 29 ++++ .../dsl/amqp/AmqpMessageChannelSpec.java | 131 ++++++++++++++++++ .../amqp/AmqpPollableMessageChannelSpec.java | 93 +++++++++++++ ...mqpPublishSubscribeMessageChannelSpec.java | 38 +++++ .../IntegrationFlowBeanPostProcessor.java | 67 +++++---- .../dsl/test/IntegrationFlowTests.java | 104 +++++++++----- .../src/test/resources/application.properties | 1 + 8 files changed, 404 insertions(+), 69 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpMessageChannelSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpPollableMessageChannelSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpPublishSubscribeMessageChannelSpec.java create mode 100644 spring-integration-java-dsl/src/test/resources/application.properties diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index e2da5b1..fc79b3b 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -26,6 +26,7 @@ ext { embedMongoVersion = '1.45' jacocoVersion = '0.7.0.201403182114' log4jVersion = '1.2.17' + slf4jVersion = '1.7.6' springIntegrationVersion = '4.0.2.RELEASE' springBootVersion = '1.1.0.BUILD-SNAPSHOT' @@ -54,10 +55,10 @@ dependencies { , 'spring-integration-ftp' , 'spring-integration-gemfire' , 'spring-integration-http' -// , 'spring-integration-jdbc' + , 'spring-integration-jdbc' , 'spring-integration-jms' , 'spring-integration-jmx' -// , 'spring-integration-jpa' + , 'spring-integration-jpa' , 'spring-integration-mail' , 'spring-integration-mongodb' , 'spring-integration-mqtt' @@ -73,11 +74,14 @@ dependencies { compile("org.springframework.integration:$it:$springIntegrationVersion", optional) } - 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" + testRuntime "log4j:log4j:$log4jVersion" + testRuntime "org.slf4j:jcl-over-slf4j:$slf4jVersion" + testRuntime "org.slf4j:slf4j-log4j12:$slf4jVersion" + jacoco "org.jacoco:org.jacoco.agent:$jacocoVersion:runtime" } 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 index 14998d3..2364c74 100644 --- 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 @@ -77,4 +77,33 @@ public abstract class Amqp { return new AmqpOutboundEndpointSpec(endpoint, expectReply); } + public static > AmqpPollableMessageChannelSpec + pollableChannel(ConnectionFactory connectionFactory) { + return pollableChannel(null, connectionFactory); + } + + public static > AmqpPollableMessageChannelSpec + pollableChannel(String id, ConnectionFactory connectionFactory) { + return new AmqpPollableMessageChannelSpec(connectionFactory).id(id); + } + + public static + > AmqpMessageChannelSpec channel(ConnectionFactory connectionFactory) { + return channel(null, connectionFactory); + } + + public static > AmqpMessageChannelSpec channel(String id, + ConnectionFactory connectionFactory) { + return new AmqpMessageChannelSpec(connectionFactory).id(id); + } + + public static AmqpPublishSubscribeMessageChannelSpec publishSubscribeChannel(ConnectionFactory connectionFactory) { + return publishSubscribeChannel(null, connectionFactory); + } + + public static AmqpPublishSubscribeMessageChannelSpec publishSubscribeChannel(String id, + ConnectionFactory connectionFactory) { + return new AmqpPublishSubscribeMessageChannelSpec(connectionFactory).id(id); + } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpMessageChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpMessageChannelSpec.java new file mode 100644 index 0000000..ca09599 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpMessageChannelSpec.java @@ -0,0 +1,131 @@ +/* + * 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.Arrays; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.Executor; + +import org.aopalliance.aop.Advice; + +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.integration.amqp.channel.AbstractAmqpChannel; +import org.springframework.integration.amqp.config.AmqpChannelFactoryBean; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.interceptor.TransactionAttribute; +import org.springframework.util.ErrorHandler; + +/** + * @author Artem Bilan + */ +public class AmqpMessageChannelSpec> extends AmqpPollableMessageChannelSpec { + + private final List adviceChain = new LinkedList(); + + AmqpMessageChannelSpec(ConnectionFactory connectionFactory) { + super(new AmqpChannelFactoryBean(true), connectionFactory); + } + + public S maxSubscribers(int maxSubscribers) { + this.amqpChannelFactoryBean.setMaxSubscribers(maxSubscribers); + return _this(); + } + + public S acknowledgeMode(AcknowledgeMode acknowledgeMode) { + this.amqpChannelFactoryBean.setAcknowledgeMode(acknowledgeMode); + return _this(); + } + + public S advice(Advice... advice) { + this.adviceChain.addAll(Arrays.asList(advice)); + return _this(); + } + + public S autoStartup(boolean autoStartup) { + this.amqpChannelFactoryBean.setAutoStartup(autoStartup); + return _this(); + } + + public S concurrentConsumers(int concurrentConsumers) { + this.amqpChannelFactoryBean.setConcurrentConsumers(concurrentConsumers); + return _this(); + } + + public S errorHandler(ErrorHandler errorHandler) { + this.amqpChannelFactoryBean.setErrorHandler(errorHandler); + return _this(); + } + + public S exposeListenerChannel(boolean exposeListenerChannel) { + this.amqpChannelFactoryBean.setExposeListenerChannel(exposeListenerChannel); + return _this(); + } + + public S phase(int phase) { + this.amqpChannelFactoryBean.setPhase(phase); + return _this(); + } + + public S prefetchCount(int prefetchCount) { + this.amqpChannelFactoryBean.setPrefetchCount(prefetchCount); + return _this(); + } + + public S receiveTimeout(long receiveTimeout) { + this.amqpChannelFactoryBean.setReceiveTimeout(receiveTimeout); + return _this(); + } + + public S recoveryInterval(long recoveryInterval) { + this.amqpChannelFactoryBean.setRecoveryInterval(recoveryInterval); + return _this(); + } + + public S shutdownTimeout(long shutdownTimeout) { + this.amqpChannelFactoryBean.setShutdownTimeout(shutdownTimeout); + return _this(); + } + + public S taskExecutor(Executor taskExecutor) { + this.amqpChannelFactoryBean.setTaskExecutor(taskExecutor); + return _this(); + } + + public S transactionAttribute(TransactionAttribute transactionAttribute) { + this.amqpChannelFactoryBean.setTransactionAttribute(transactionAttribute); + return _this(); + } + + public S transactionManager(PlatformTransactionManager transactionManager) { + this.amqpChannelFactoryBean.setTransactionManager(transactionManager); + return _this(); + } + + public S txSize(int txSize) { + this.amqpChannelFactoryBean.setTxSize(txSize); + return _this(); + } + + @Override + protected AbstractAmqpChannel doGet() { + this.amqpChannelFactoryBean.setAdviceChain(this.adviceChain.toArray(new Advice[this.adviceChain.size()])); + return super.doGet(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpPollableMessageChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpPollableMessageChannelSpec.java new file mode 100644 index 0000000..7fb2df9 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpPollableMessageChannelSpec.java @@ -0,0 +1,93 @@ +/* + * 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.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.integration.amqp.channel.AbstractAmqpChannel; +import org.springframework.integration.amqp.config.AmqpChannelFactoryBean; +import org.springframework.integration.dsl.channel.MessageChannelSpec; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + */ +public class AmqpPollableMessageChannelSpec> + extends MessageChannelSpec { + + protected final AmqpChannelFactoryBean amqpChannelFactoryBean; + + AmqpPollableMessageChannelSpec(ConnectionFactory connectionFactory) { + this(new AmqpChannelFactoryBean(false), connectionFactory); + } + + AmqpPollableMessageChannelSpec(AmqpChannelFactoryBean amqpChannelFactoryBean, ConnectionFactory connectionFactory) { + this.amqpChannelFactoryBean = amqpChannelFactoryBean; + this.amqpChannelFactoryBean.setConnectionFactory(connectionFactory); + this.amqpChannelFactoryBean.setSingleton(false); + this.amqpChannelFactoryBean.setPubSub(false); + } + + @Override + protected S id(String id) { + this.amqpChannelFactoryBean.setBeanName(id); + return super.id(id); + } + + public S queueName(String queueName) { + if (this.id == null) { + id(queueName + ".channel"); + } + this.amqpChannelFactoryBean.setQueueName(queueName); + return _this(); + } + + public S encoding(String encoding) { + this.amqpChannelFactoryBean.setEncoding(encoding); + return _this(); + } + + public S amqpMessageConverter(MessageConverter messageConverter) { + this.amqpChannelFactoryBean.setMessageConverter(messageConverter); + return _this(); + } + + public S channelTransacted(boolean channelTransacted) { + this.amqpChannelFactoryBean.setChannelTransacted(channelTransacted); + return _this(); + } + + public S messagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) { + this.amqpChannelFactoryBean.setMessagePropertiesConverter(messagePropertiesConverter); + return _this(); + } + + @Override + protected AbstractAmqpChannel doGet() { + Assert.notNull(this.id, "The 'id' or 'queueName' must be specified"); + try { + this.channel = this.amqpChannelFactoryBean.getObject(); + } + catch (Exception e) { + throw new BeanCreationException("Cannot create the AMQP MessageChannel", e); + } + return super.doGet(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpPublishSubscribeMessageChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpPublishSubscribeMessageChannelSpec.java new file mode 100644 index 0000000..b1e4d4b --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpPublishSubscribeMessageChannelSpec.java @@ -0,0 +1,38 @@ +/* + * 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.FanoutExchange; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; + +/** + * @author Artem Bilan + */ +public class AmqpPublishSubscribeMessageChannelSpec + extends AmqpMessageChannelSpec { + + AmqpPublishSubscribeMessageChannelSpec(ConnectionFactory connectionFactory) { + super(connectionFactory); + this.amqpChannelFactoryBean.setPubSub(true); + } + + public AmqpPublishSubscribeMessageChannelSpec exchange(FanoutExchange exchange) { + this.amqpChannelFactoryBean.setExchange(exchange); + return _this(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java index 8aad6aa..f37993a 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java @@ -18,6 +18,7 @@ package org.springframework.integration.dsl.core; import java.util.Collection; +import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; @@ -31,6 +32,7 @@ import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.config.IntegrationConfigUtils; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.support.MessageChannelReference; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; @@ -60,7 +62,7 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { if (bean instanceof IntegrationFlow) { IntegrationFlow flow = (IntegrationFlow) bean; - String flowNamePrefix = beanName + ":"; + String flowNamePrefix = beanName + "."; int channelNameIndex = 0; for (Object component : flow.getIntegrationComponents()) { if (component instanceof ConsumerEndpointSpec) { @@ -69,8 +71,8 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean ConsumerEndpointFactoryBean endpoint = endpointSpec.get().getT1(); String id = endpointSpec.getId(); - Collection messageHandlers = - this.beanFactory.getBeansOfType(messageHandler.getClass(), false, false).values(); + Collection messageHandlers = this.beanFactory.getBeansOfType(MessageHandler.class, false, + false).values(); if (!messageHandlers.contains(messageHandler)) { String handlerBeanName = generateBeanName(messageHandler); @@ -93,35 +95,42 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean registerComponent(endpoint, endpointBeanName); } else { - Collection values = this.beanFactory.getBeansOfType(component.getClass(), false, false).values(); - if (!values.contains(component)) { - if (component instanceof AbstractMessageChannel) { - String channelBeanName = ((AbstractMessageChannel) component).getComponentName(); - if (channelBeanName == null) { - channelBeanName = flowNamePrefix + "channel" + - BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; + //TODO workaround until SF will fix 'TypeDescriptor.forObject' + if (component instanceof MessageChannel) { + Collection messageChannels = + this.beanFactory.getBeansOfType(MessageChannel.class, false, false).values(); + if (!messageChannels.contains(component)) { + if (component instanceof AbstractMessageChannel) { + String channelBeanName = ((AbstractMessageChannel) component).getComponentName(); + if (channelBeanName == null) { + channelBeanName = flowNamePrefix + "channel" + + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; + } + registerComponent(component, channelBeanName); } - registerComponent(component, channelBeanName); - } - else if (component instanceof MessageChannelReference) { - String channelBeanName = ((MessageChannelReference) component).getName(); - if (!this.beanFactory.containsBean(channelBeanName)) { - DirectChannel directChannel = new DirectChannel(); - registerComponent(directChannel, channelBeanName); + else if (component instanceof MessageChannelReference) { + String channelBeanName = ((MessageChannelReference) component).getName(); + if (!this.beanFactory.containsBean(channelBeanName)) { + DirectChannel directChannel = new DirectChannel(); + registerComponent(directChannel, channelBeanName); + } + } + else if (component instanceof FixedSubscriberChannel) { + FixedSubscriberChannel fixedSubscriberChannel = (FixedSubscriberChannel) component; + String channelBeanName = fixedSubscriberChannel.getComponentName(); + if ("Unnamed fixed subscriber channel".equals(channelBeanName)) { + channelBeanName = flowNamePrefix + "channel" + + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; + } + registerComponent(component, channelBeanName); } } - else if (component instanceof FixedSubscriberChannel) { - FixedSubscriberChannel fixedSubscriberChannel = (FixedSubscriberChannel) component; - String channelBeanName = fixedSubscriberChannel.getComponentName(); - if ("Unnamed fixed subscriber channel".equals(channelBeanName)) { - channelBeanName = flowNamePrefix + "channel" + - BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; - } - registerComponent(component, channelBeanName); - } - else { - registerComponent(component, generateBeanName(component)); - } + } + else if (!this.beanFactory + .getBeansOfType(AopUtils.getTargetClass(component), false, false) + .values() + .contains(component)) { + registerComponent(component, generateBeanName(component)); } } } 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 56c3760..7ce93a8 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 @@ -54,11 +54,16 @@ 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.aop.TargetSource; +import org.springframework.aop.framework.Advised; +import org.springframework.aop.support.AopUtils; +import org.springframework.beans.DirectFieldAccessor; 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.boot.test.ConfigFileApplicationContextInitializer; import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.AnnotationConfigApplicationContext; @@ -79,7 +84,6 @@ import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.AbstractPollableChannel; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.FixedSubscriberChannel; -import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.config.GlobalChannelInterceptor; @@ -113,10 +117,12 @@ import org.springframework.integration.xml.transformer.support.XPathExpressionEv import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.ErrorMessage; @@ -130,7 +136,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Artem Bilan */ -@ContextConfiguration +@ContextConfiguration(initializers = ConfigFileApplicationContextInitializer.class) @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext public class IntegrationFlowTests { @@ -158,11 +164,11 @@ public class IntegrationFlowTests { @Autowired @Qualifier("inputChannel") - private DirectChannel inputChannel; + private MessageChannel inputChannel; @Autowired @Qualifier("foo") - private PublishSubscribeChannel foo; + private SubscribableChannel foo; @Autowired @Qualifier("successChannel") @@ -170,7 +176,7 @@ public class IntegrationFlowTests { @Autowired @Qualifier("flow3Input") - private DirectChannel flow3Input; + private MessageChannel flow3Input; @Autowired private AtomicReference eventHolder; @@ -185,7 +191,7 @@ public class IntegrationFlowTests { @Autowired @Qualifier("bridgeFlow2Input") - private DirectChannel bridgeFlow2Input; + private MessageChannel bridgeFlow2Input; @Autowired @Qualifier("bridgeFlow2Output") @@ -193,15 +199,15 @@ public class IntegrationFlowTests { @Autowired @Qualifier("fileFlow1Input") - private DirectChannel fileFlow1Input; + private MessageChannel fileFlow1Input; @Autowired @Qualifier("fileWritingMessageHandler") - private FileWritingMessageHandler fileWritingMessageHandler; + private MessageHandler fileWritingMessageHandler; @Autowired @Qualifier("methodInvokingInput") - private DirectChannel methodInvokingInput; + private MessageChannel methodInvokingInput; @Autowired @Qualifier("delayedAdvice") @@ -213,11 +219,11 @@ public class IntegrationFlowTests { @Autowired @Qualifier("splitInput") - private DirectChannel splitInput; + private MessageChannel splitInput; @Autowired @Qualifier("xpathHeaderEnricherInput") - private DirectChannel xpathHeaderEnricherInput; + private MessageChannel xpathHeaderEnricherInput; @Autowired @Qualifier("splitAggregateInput") @@ -265,7 +271,7 @@ public class IntegrationFlowTests { @Autowired @Qualifier("defaultOutputChannel") - private QueueChannel defaultOutputChannel; + private PollableChannel defaultOutputChannel; @Autowired private MessageStore messageStore; @@ -384,8 +390,9 @@ public class IntegrationFlowTests { assertNotNull(reply); assertEquals("test", reply.getPayload()); - assertTrue(this.beanFactory.containsBean("bridgeFlow2:channel#0")); - assertThat(this.beanFactory.getBean("bridgeFlow2:channel#0"), Matchers.instanceOf(FixedSubscriberChannel.class)); + assertTrue(this.beanFactory.containsBean("bridgeFlow2.channel#0")); + assertThat(this.beanFactory.getBean("bridgeFlow2.channel#0"), Matchers.instanceOf(FixedSubscriberChannel + .class)); try { this.bridgeFlow2Input.send(message); @@ -443,8 +450,7 @@ public class IntegrationFlowTests { @Test - public void testFileHandler() { - assertEquals(1, this.beanFactory.getBeansOfType(FileWritingMessageHandler.class).size()); + public void testFileHandler() throws Exception { Message message = MessageBuilder.withPayload("foo").setHeader(FileHeaders.FILENAME, "foo").build(); try { this.fileFlow1Input.send(message); @@ -456,7 +462,15 @@ public class IntegrationFlowTests { } DefaultFileNameGenerator fileNameGenerator = new DefaultFileNameGenerator(); fileNameGenerator.setBeanFactory(this.beanFactory); - this.fileWritingMessageHandler.setFileNameGenerator(fileNameGenerator); + Object targetFileWritingMessageHandler = this.fileWritingMessageHandler; + if (this.fileWritingMessageHandler instanceof Advised) { + TargetSource targetSource = ((Advised) this.fileWritingMessageHandler).getTargetSource(); + if (targetSource != null) { + targetFileWritingMessageHandler = targetSource.getTarget(); + } + } + DirectFieldAccessor dfa = new DirectFieldAccessor(targetFileWritingMessageHandler); + dfa.setPropertyValue("fileNameGenerator", fileNameGenerator); this.fileFlow1Input.send(message); assertTrue(new File(tmpDir, "foo").exists()); @@ -901,7 +915,7 @@ public class IntegrationFlowTests { private MessageChannel amqpOutboundInput; @Autowired - @Qualifier("amqpReplyChannel") + @Qualifier("amqpReplyChannel.channel") private PollableChannel amqpReplyChannel; @Test @@ -909,13 +923,22 @@ public class IntegrationFlowTests { this.amqpOutboundInput.send(MessageBuilder.withPayload("hello through the amqp") .setHeader("routingKey", "foo") .build()); - Message receive = this.amqpReplyChannel.receive(5000); + Message receive = null; + int i = 0; + do { + receive = this.amqpReplyChannel.receive(); + if (receive != null) { + break; + } + Thread.sleep(100); + i++; + } while (i < 10); + assertNotNull(receive); assertEquals("HELLO THROUGH THE AMQP", receive.getPayload()); } - @MessagingGateway(defaultRequestChannel = "controlBus") private static interface ControlBusGateway { @@ -955,12 +978,12 @@ public class IntegrationFlowTests { } @Bean - public DirectChannel inputChannel() { + public MessageChannel inputChannel() { return MessageChannels.direct().get(); } @Bean - public PublishSubscribeChannel foo() { + public MessageChannel foo() { return MessageChannels.publishSubscribe().get(); } @@ -1106,17 +1129,17 @@ public class IntegrationFlowTests { } @Bean(name = "foo-channel") - public QueueChannel fooChannel() { + public MessageChannel fooChannel() { return new QueueChannel(); } @Bean(name = "bar-channel") - public QueueChannel barChannel() { + public MessageChannel barChannel() { return new QueueChannel(); } @Bean - public QueueChannel defaultOutputChannel() { + public MessageChannel defaultOutputChannel() { return new QueueChannel(); } @@ -1125,7 +1148,7 @@ public class IntegrationFlowTests { return IntegrationFlows.from("recipientListInput") .transform(p -> p.replaceFirst("Payload", "")) .recipientListRoute(r -> - r.defaultOutputChannel(defaultOutputChannel()) + r.defaultOutputChannel(this.defaultOutputChannel()) .recipient("foo-channel", "'foo' == payload") .recipient("bar-channel", m -> m.getHeaders().containsKey("recipient") @@ -1156,28 +1179,28 @@ public class IntegrationFlowTests { public static class ContextConfiguration4 { @Bean - public FileWritingMessageHandler fileWritingMessageHandler() { - return new FileWritingMessageHandler(tmpDir); + public MessageHandler fileWritingMessageHandler() { + FileWritingMessageHandler fileWritingMessageHandler = new FileWritingMessageHandler(tmpDir); + fileWritingMessageHandler.setFileNameGenerator(message -> null); + fileWritingMessageHandler.setExpectReply(false); + return fileWritingMessageHandler; } @Bean public IntegrationFlow fileFlow1() { return IntegrationFlows.from("fileFlow1Input") - .handle(this.fileWritingMessageHandler(), c -> { - FileWritingMessageHandler handler = c.get().getT2(); - handler.setFileNameGenerator(message -> null); - handler.setExpectReply(false); - }) + .handle(this.fileWritingMessageHandler()) .get(); } @Autowired - private GreetingService greetingService; + @Qualifier("integrationFlowTests.GreetingService") + private MessageHandler greetingService; @Bean public IntegrationFlow methodInvokingFlow() { return IntegrationFlows.from("methodInvokingInput") - .handle(Message.class, (p, h) -> this.greetingService.handleRequestMessage(p)) + .handle(this.greetingService) .get(); } @@ -1352,7 +1375,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow amqpOutboundFlow() { - return IntegrationFlows.from("amqpOutboundInput") + return IntegrationFlows.from(Amqp.channel("amqpOutboundInput", this.rabbitConnectionFactory)) .handle(Amqp.outboundAdapter(this.amqpTemplate).routingKeyExpression("headers.routingKey").get()) .get(); } @@ -1362,11 +1385,18 @@ public class IntegrationFlowTests { return new Queue("foo"); } + @Bean + public Queue amqpReplyChannel() { + return new Queue("amqpReplyChannel"); + } + @Bean public IntegrationFlow amqpInboundFlow() { return IntegrationFlows.from(Amqp.inboundAdapter(this.rabbitConnectionFactory, fooQueue())) .transform(String.class, String::toUpperCase) - .channel(MessageChannels.queue("amqpReplyChannel")) + .channel(Amqp.pollableChannel(this.rabbitConnectionFactory) + .queueName("amqpReplyChannel") + .channelTransacted(true)) .get(); } diff --git a/spring-integration-java-dsl/src/test/resources/application.properties b/spring-integration-java-dsl/src/test/resources/application.properties new file mode 100644 index 0000000..eb9d820 --- /dev/null +++ b/spring-integration-java-dsl/src/test/resources/application.properties @@ -0,0 +1 @@ +#spring.jmx.enabled=false