From 3d0cec9544dd95e0d84ae907daefc32974d61823 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Mar 2014 16:32:58 +0200 Subject: [PATCH] Add support for `FixedSubscriberChannel` --- .../dsl/IntegrationFlowBuilder.java | 86 +++++++++++-------- .../integration/dsl/IntegrationFlows.java | 18 +++- ...slIntegrationConfigurationInitializer.java | 10 +++ .../FixedSubscriberChannelPrototype.java | 44 ++++++++++ .../MessageChannelReference.java | 2 +- .../dsl/test/IntegrationFlowTests.java | 36 +++++++- .../src/test/resources/log4j.properties | 1 - 7 files changed, 157 insertions(+), 40 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/FixedSubscriberChannelPrototype.java rename spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/{core => support}/MessageChannelReference.java (92%) diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index a62b110..3cf7178 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -19,12 +19,14 @@ package org.springframework.integration.dsl; import org.springframework.beans.factory.BeanCreationException; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.core.GenericSelector; import org.springframework.integration.core.MessageSelector; import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.core.ConsumerEndpointSpec; -import org.springframework.integration.dsl.core.MessageChannelReference; +import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype; +import org.springframework.integration.dsl.support.MessageChannelReference; import org.springframework.integration.dsl.support.BeanNameMethodInvokingMessageHandler; import org.springframework.integration.dsl.support.EndpointConfigurer; import org.springframework.integration.dsl.support.EnricherConfigurer; @@ -71,6 +73,14 @@ public final class IntegrationFlowBuilder { return this; } + public IntegrationFlowBuilder fixedSubscriberChannel() { + return this.fixedSubscriberChannel(null); + } + + public IntegrationFlowBuilder fixedSubscriberChannel(String messageChannelName) { + return this.channel(new FixedSubscriberChannelPrototype(messageChannelName)); + } + public IntegrationFlowBuilder channel(String messageChannelName) { return this.channel(new MessageChannelReference(messageChannelName)); } @@ -78,14 +88,7 @@ public final class IntegrationFlowBuilder { public IntegrationFlowBuilder channel(MessageChannel messageChannel) { Assert.notNull(messageChannel); if (this.currentMessageChannel != null) { - GenericEndpointSpec endpointSpec = new GenericEndpointSpec(new BridgeHandler()); - if (this.currentMessageChannel instanceof MessageChannelReference) { - endpointSpec.get().getT1().setInputChannelName(((MessageChannelReference) this.currentMessageChannel).getName()); - } - else { - endpointSpec.get().getT1().setInputChannel(this.currentMessageChannel); - } - this.addComponent(endpointSpec).currentComponent(endpointSpec.get().getT2()); + this.register(new GenericEndpointSpec(new BridgeHandler()), null); } this.currentMessageChannel = messageChannel; return this.registerOutputChannelIfCan(this.currentMessageChannel); @@ -188,6 +191,14 @@ public final class IntegrationFlowBuilder { endpointSpec.get().getT1().setInputChannelName(((MessageChannelReference) inputChannel).getName()); } else { + if (inputChannel instanceof FixedSubscriberChannelPrototype) { + String beanName = ((FixedSubscriberChannelPrototype) inputChannel).getName(); + inputChannel = new FixedSubscriberChannel(endpointSpec.get().getT2()); + if (beanName != null) { + ((FixedSubscriberChannel) inputChannel).setBeanName(beanName); + } + this.registerOutputChannelIfCan(inputChannel); + } endpointSpec.get().getT1().setInputChannel(inputChannel); } @@ -195,40 +206,47 @@ public final class IntegrationFlowBuilder { } private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) { - this.flow.addComponent(outputChannel); - String channelName = null; - if (outputChannel instanceof MessageChannelReference) { - channelName = ((MessageChannelReference) outputChannel).getName(); - } - if (this.currentComponent != null) { - if (this.currentComponent instanceof AbstractReplyProducingMessageHandler) { - AbstractReplyProducingMessageHandler messageProducer = (AbstractReplyProducingMessageHandler) this.currentComponent; - if (channelName != null) { - messageProducer.setOutputChannelName(channelName); + if (!(outputChannel instanceof FixedSubscriberChannelPrototype)) { + this.flow.addComponent(outputChannel); + if (this.currentComponent != null) { + String channelName = null; + if (outputChannel instanceof MessageChannelReference) { + channelName = ((MessageChannelReference) outputChannel).getName(); + } + if (this.currentComponent instanceof AbstractReplyProducingMessageHandler) { + AbstractReplyProducingMessageHandler messageProducer = (AbstractReplyProducingMessageHandler) this.currentComponent; + if (channelName != null) { + messageProducer.setOutputChannelName(channelName); + } + else { + messageProducer.setOutputChannel(outputChannel); + } + } + else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) { + SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = (SourcePollingChannelAdapterFactoryBean) this.currentComponent; + if (channelName != null) { + pollingChannelAdapterFactoryBean.setOutputChannelName(channelName); + } + else { + pollingChannelAdapterFactoryBean.setOutputChannel(outputChannel); + } } else { - messageProducer.setOutputChannel(outputChannel); + throw new BeanCreationException("The 'currentComponent' (" + this.currentComponent + ") is a one-way 'MessageHandler'" + + " and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow."); } + this.currentComponent = null; } - else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) { - SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = (SourcePollingChannelAdapterFactoryBean) this.currentComponent; - if (channelName != null) { - pollingChannelAdapterFactoryBean.setOutputChannelName(channelName); - } - else { - pollingChannelAdapterFactoryBean.setOutputChannel(outputChannel); - } - } - else { - throw new BeanCreationException("The 'currentComponent' (" + this.currentComponent + ") is a one-way 'MessageHandler'" + - "and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow."); - } - this.currentComponent = null; } return this; } public IntegrationFlow get() { + if (this.currentMessageChannel instanceof FixedSubscriberChannelPrototype) { + throw new BeanCreationException("The 'currentMessageChannel' (" + this.currentMessageChannel + ") is a prototype" + + " for FixedSubscriberChannel which can't be created without MessageHandler constructor argument. " + + "That means that '.fixedSubscriberChannel()' can't be the last EIP-method in the IntegrationFlow definition."); + } return this.flow; } 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 315b4ee..02d415a 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 @@ -19,7 +19,8 @@ package org.springframework.integration.dsl; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.channel.MessageChannelSpec; -import org.springframework.integration.dsl.core.MessageChannelReference; +import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype; +import org.springframework.integration.dsl.support.MessageChannelReference; import org.springframework.integration.dsl.support.EndpointConfigurer; import org.springframework.messaging.MessageChannel; @@ -28,10 +29,25 @@ import org.springframework.messaging.MessageChannel; */ public final class IntegrationFlows { + /** + * @param messageChannelName the name of existing {@link org.springframework.messaging.MessageChannel} bean. + * The new {@link org.springframework.integration.channel.DirectChannel} bean will be + * created on context startup, if there is on bean with this name. + * @return new {@link IntegrationFlowBuilder} + */ public static IntegrationFlowBuilder from(String messageChannelName) { return from(new MessageChannelReference(messageChannelName)); } + /** + * @param messageChannelName the name for {@link org.springframework.integration.channel.FixedSubscriberChannel} + * to be created on context startup, not reference. + * @return new {@link IntegrationFlowBuilder} + */ + public static IntegrationFlowBuilder fromFixedMessageChannel(String messageChannelName) { + return from(new FixedSubscriberChannelPrototype(messageChannelName)); + } + public static IntegrationFlowBuilder from(MessageChannel messageChannel) { return new IntegrationFlowBuilder().channel(messageChannel); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java index 99d30c0..21e6335 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java @@ -34,11 +34,13 @@ import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.GenericBeanDefinition; import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.config.IntegrationConfigUtils; import org.springframework.integration.config.IntegrationConfigurationInitializer; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.config.InstanceBeanDefinition; +import org.springframework.integration.dsl.support.MessageChannelReference; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; @@ -121,6 +123,14 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig IntegrationConfigUtils.autoCreateDirectChannel(channelName, registry); } } + else if (instance instanceof FixedSubscriberChannel) { + FixedSubscriberChannel fixedSubscriberChannel = (FixedSubscriberChannel) instance; + String channelBeanName = fixedSubscriberChannel.getComponentName(); + if ("Unnamed fixed subscriber channel".equals(channelBeanName)) { + channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; + } + registry.registerBeanDefinition(channelBeanName, component); + } else { String beanName = generateInstanceBeanDefinitionName(registry, instance); registry.registerBeanDefinition(beanName, component); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/FixedSubscriberChannelPrototype.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/FixedSubscriberChannelPrototype.java new file mode 100644 index 0000000..c468bfb --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/FixedSubscriberChannelPrototype.java @@ -0,0 +1,44 @@ +package org.springframework.integration.dsl.support; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public class FixedSubscriberChannelPrototype implements MessageChannel { + + private final String name; + + public FixedSubscriberChannelPrototype() { + this(null); + } + + public FixedSubscriberChannelPrototype(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + + @Override + public boolean send(Message message) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean send(Message message, long timeout) { + throw new UnsupportedOperationException(); + } + + @Override + public String toString() { + return "FixedSubscriberChannelPrototype{" + + "name='" + name + '\'' + + '}'; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessageChannelReference.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/MessageChannelReference.java similarity index 92% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessageChannelReference.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/MessageChannelReference.java index 18d9b0e..4483cd9 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessageChannelReference.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/MessageChannelReference.java @@ -1,4 +1,4 @@ -package org.springframework.integration.dsl.core; +package org.springframework.integration.dsl.support; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; 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 d032789..755b1cd 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 @@ -52,6 +52,7 @@ import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; 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; @@ -154,11 +155,12 @@ public class IntegrationFlowTests { @Autowired @Qualifier("enricherInput") - private DirectChannel enricherInput; + private FixedSubscriberChannel enricherInput; @Test public void testPollingFlow() { + assertThat(this.beanFactory.getBean("integerChannel"), Matchers.instanceOf(FixedSubscriberChannel.class)); for (int i = 0; i < 10; i++) { Message message = this.outputChannel.receive(5000); assertNotNull(message); @@ -215,6 +217,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)); + try { this.bridgeFlow2Input.send(message); fail("Expected MessageDispatchingException"); @@ -244,6 +249,18 @@ public class IntegrationFlowTests { } } + @Test + public void testWrongLastMessageChannel() { + try { + new AnnotationConfigApplicationContext(InvalidLastMessageChannelFlowContext.class); + fail("BeanCreationException expected"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(BeanCreationException.class)); + assertThat(e.getMessage(), Matchers.containsString("'.fixedSubscriberChannel()' can't be the last EIP-method in the IntegrationFlow definition")); + } + } + @Test public void testFileHandler() { @@ -318,7 +335,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow flow1() { return IntegrationFlows.from(this.integerMessageSource(), c -> c.poller(Pollers.fixedRate(100))) - .channel("integerChannel") + .fixedSubscriberChannel("integerChannel") .transform("payload.toString()") .channel(MessageChannels.queue("flow1QueueChannel")) .get(); @@ -367,6 +384,7 @@ public class IntegrationFlowTests { return IntegrationFlows.from(this.inputChannel) .filter(p -> p instanceof String, c -> c.id("filter")) .channel("foo") + .fixedSubscriberChannel() .transform(Integer::parseInt) .transform(new PayloadSerializingTransformer(), c -> c.autoStartup(false).id("payloadSerializingTransformer")) @@ -434,6 +452,7 @@ public class IntegrationFlowTests { public IntegrationFlow bridgeFlow2() { return IntegrationFlows.from("bridgeFlow2Input") .bridge(c -> c.autoStartup(false).id("bridge")) + .fixedSubscriberChannel() .delay("delayer", "200", c -> c.advice(this.delayedAdvice)) .channel(MessageChannels.queue("bridgeFlow2Output")) .get(); @@ -486,7 +505,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow enricherFlow() { - return IntegrationFlows.from("enricherInput") + return IntegrationFlows.fromFixedMessageChannel("enricherInput") .enrich(e -> e.requestChannel("enrichChannel") .requestPayloadExpression("payload") .shouldClonePayload(false) @@ -527,6 +546,17 @@ public class IntegrationFlowTests { } + private static class InvalidLastMessageChannelFlowContext { + + @Bean + public IntegrationFlow wrongLastComponent() { + return IntegrationFlows.from(MessageChannels.direct()) + .fixedSubscriberChannel() + .get(); + } + + } + @EnableIntegration public static class InvalidConfigurationWithSpec { diff --git a/spring-integration-java-dsl/src/test/resources/log4j.properties b/spring-integration-java-dsl/src/test/resources/log4j.properties index 9fa9b48..5aa14e4 100644 --- a/spring-integration-java-dsl/src/test/resources/log4j.properties +++ b/spring-integration-java-dsl/src/test/resources/log4j.properties @@ -4,5 +4,4 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n -log4j.category.org.springframework.integration=WARN log4j.category.org.springframework.integration.dsl=INFO