diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index de19646..2a5bb80 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -40,7 +40,7 @@ ext { eclipse.project.natures += 'org.springframework.ide.eclipse.core.springnature' -sourceSets.test.resources.srcDirs = ['src/test/resources', 'src/test/java'] +sourceSets.test.resources.srcDirs = ['src/test/java'] // See http://www.gradle.org/docs/current/userguide/dependency_management.html#sub:configurations // and http://www.gradle.org/docs/current/dsl/org.gradle.api.artifacts.ConfigurationContainer.html @@ -61,8 +61,7 @@ dependencies { } // enable all compiler warnings; individual projects may customize further -ext.xLintArg = '-Xlint:all,-options' -[compileJava, compileTestJava]*.options*.compilerArgs = [xLintArg] +[compileJava, compileTestJava]*.options*.compilerArgs = ['-Xlint:all,-options'] test { // suppress all console output during testing unless running `gradle -i` @@ -190,6 +189,6 @@ task dist(dependsOn: assemble) { task wrapper(type: Wrapper) { description = 'Generates gradlew[.bat] scripts' - gradleVersion = '1.10' + gradleVersion = '1.11' distributionUrl = "http://services.gradle.org/distributions/gradle-${gradleVersion}-all.zip" } diff --git a/spring-integration-java-dsl/gradle/wrapper/gradle-wrapper.jar b/spring-integration-java-dsl/gradle/wrapper/gradle-wrapper.jar index 5838598..3c7abdf 100644 Binary files a/spring-integration-java-dsl/gradle/wrapper/gradle-wrapper.jar and b/spring-integration-java-dsl/gradle/wrapper/gradle-wrapper.jar differ diff --git a/spring-integration-java-dsl/gradle/wrapper/gradle-wrapper.properties b/spring-integration-java-dsl/gradle/wrapper/gradle-wrapper.properties index c8f565d..0c970e7 100644 --- a/spring-integration-java-dsl/gradle/wrapper/gradle-wrapper.properties +++ b/spring-integration-java-dsl/gradle/wrapper/gradle-wrapper.properties @@ -1,6 +1,6 @@ -#Tue Feb 04 18:25:18 EET 2014 +#Wed Feb 12 15:24:07 EET 2014 distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists -distributionUrl=http\://services.gradle.org/distributions/gradle-1.10-all.zip +distributionUrl=http\://services.gradle.org/distributions/gradle-1.11-all.zip diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java index d9973bd..e7700be 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java @@ -16,23 +16,14 @@ package org.springframework.integration.dsl; -import java.util.Collection; import java.util.Map; import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactoryUtils; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; -import org.springframework.beans.factory.parsing.BeanComponentDefinition; -import org.springframework.beans.factory.support.AbstractBeanDefinition; -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.config.ConsumerEndpointFactoryBean; import org.springframework.integration.config.IntegrationConfigurationInitializer; -import org.springframework.integration.config.xml.IntegrationNamespaceUtils; -import org.springframework.integration.dsl.config.InstanceBeanDefinition; -import org.springframework.messaging.MessageHandler; +import org.springframework.integration.dsl.core.Spec; +import org.springframework.util.Assert; /** * The Java DSL Integration infrastructure {@code beanFactory} initializer. @@ -43,71 +34,24 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig @Override public void initialize(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException { - this.initializeIntegrationFlows(configurableListableBeanFactory); + this.populateBeansFromSpecs(configurableListableBeanFactory); + configurableListableBeanFactory.addBeanPostProcessor(new IntegrationFlowBeanPostProcessor(configurableListableBeanFactory)); } - private void initializeIntegrationFlows(ConfigurableListableBeanFactory beanFactory) { - Map integrationFlows = beanFactory.getBeansOfType(IntegrationFlow.class, false, false); + private void populateBeansFromSpecs(ConfigurableListableBeanFactory beanFactory) { + Assert.isInstanceOf(BeanDefinitionRegistry.class, beanFactory, + "To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." + + "Consider using 'GenericApplicationContext' implementation." + ); + Map specs = beanFactory.getBeansOfType(Spec.class, false, false); BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; - for (Map.Entry integrationFlowEntry : integrationFlows.entrySet()) { - String flowName = integrationFlowEntry.getKey(); - String flowNamePrefix = flowName + ":"; - IntegrationFlow flow = integrationFlowEntry.getValue(); - int channelNameIndex = 0; - for (AbstractBeanDefinition beanDefinition : flow.getIntegrationComponents()) { - if (beanDefinition instanceof InstanceBeanDefinition) { - final Object instance = beanDefinition.getSource(); - Collection values = beanFactory.getBeansOfType(instance.getClass(), false, false).values(); - if (!values.contains(instance)) { - if (instance instanceof AbstractMessageChannel) { - String channelBeanName = ((AbstractMessageChannel) instance).getComponentName(); - if (channelBeanName == null) { - channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; - } - registry.registerBeanDefinition(channelBeanName, beanDefinition); - } - else if (instance instanceof EndpointSpec) { - EndpointSpec endpointSpec = (EndpointSpec) instance; - MessageHandler messageHandler = endpointSpec.getHandler(); - ConsumerEndpointFactoryBean endpoint = endpointSpec.getEndpoint(); - String id = endpointSpec.getId(); - - String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler); - String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null; - BeanComponentDefinition definitionHolder = new BeanComponentDefinition(new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias); - BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry); - - String endpointBeanName = id; - if (endpointBeanName == null) { - endpointBeanName = generateInstanceBeanDefinitionName(registry, endpoint); - } - registry.registerBeanDefinition(endpointBeanName, new InstanceBeanDefinition(endpoint)); - } - else { - String beanName = generateInstanceBeanDefinitionName(registry, instance); - registry.registerBeanDefinition(beanName, beanDefinition); - } - } - } - else { - BeanDefinitionReaderUtils.registerWithGeneratedName(beanDefinition, registry); - } - } - registry.removeBeanDefinition(flowName); - beanFactory.destroyBean(flowName); + for (Map.Entry specEntry : specs.entrySet()) { + String id = specEntry.getKey(); + Spec spec = specEntry.getValue(); + registry.removeBeanDefinition(id); + beanFactory.destroyBean(id); + beanFactory.registerSingleton(id, spec.get()); } - - } - - @SuppressWarnings("serial") - private static String generateInstanceBeanDefinitionName(BeanDefinitionRegistry registry, final Object instance) { - return BeanDefinitionReaderUtils.generateBeanName(new GenericBeanDefinition() { - - @Override - public String getBeanClassName() { - return instance.getClass().getName(); - } - }, registry); } } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EndpointSpec.java index 3f05de0..cfce04d 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EndpointSpec.java @@ -23,6 +23,7 @@ import java.util.List; import org.aopalliance.aop.Advice; import org.springframework.integration.config.ConsumerEndpointFactoryBean; +import org.springframework.integration.dsl.support.PollerSpec; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.messaging.MessageHandler; @@ -31,17 +32,17 @@ import org.springframework.messaging.MessageHandler; * @author Artem Bilan * @since 4.0 */ -public abstract class EndpointSpec, C extends MessageHandler> { +public abstract class EndpointSpec, H extends MessageHandler> { private final ConsumerEndpointFactoryBean endpointFactoryBean = new ConsumerEndpointFactoryBean(); - private final C messageHandler; + private final H messageHandler; private final List adviceChain = new LinkedList(); private String id; - EndpointSpec(C messageHandler) { + EndpointSpec(H messageHandler) { this.messageHandler = messageHandler; this.endpointFactoryBean.setHandler(this.messageHandler); if (this.messageHandler instanceof AbstractReplyProducingMessageHandler) { @@ -78,6 +79,11 @@ public abstract class EndpointSpec, C extends Messa return _this(); } + public S poller(PollerSpec pollerMetadataSpec) { + this.endpointFactoryBean.setPollerMetadata(pollerMetadataSpec.get()); + return _this(); + } + String getId() { return id; } @@ -86,7 +92,7 @@ public abstract class EndpointSpec, C extends Messa return this.endpointFactoryBean; } - C getHandler() { + H getHandler() { return this.messageHandler; } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java index 77211a6..8590de3 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java @@ -22,9 +22,9 @@ import org.springframework.messaging.MessageHandler; * @author Artem Bilan * @since 4.0 */ -public final class GenericEndpointSpec extends EndpointSpec, C> { +public final class GenericEndpointSpec extends EndpointSpec, H> { - GenericEndpointSpec(C messageHandler) { + GenericEndpointSpec(H messageHandler) { super(messageHandler); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java new file mode 100644 index 0000000..30d448a --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java @@ -0,0 +1,106 @@ +package org.springframework.integration.dsl; + +import java.util.Collection; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactoryUtils; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.parsing.BeanComponentDefinition; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +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.config.ConsumerEndpointFactoryBean; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.dsl.config.InstanceBeanDefinition; +import org.springframework.messaging.MessageHandler; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor { + + private final ConfigurableListableBeanFactory beanFactory; + + private final BeanDefinitionRegistry registry; + + + public IntegrationFlowBeanPostProcessor(ConfigurableListableBeanFactory beanFactory) { + Assert.isInstanceOf(BeanDefinitionRegistry.class, beanFactory, + "To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." + + "Consider using 'GenericApplicationContext' implementation." + ); + this.beanFactory = beanFactory; + this.registry = (BeanDefinitionRegistry) beanFactory; + } + + @Override + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + if (bean instanceof IntegrationFlow) { + String flowNamePrefix = beanName + ":"; + int channelNameIndex = 0; + for (AbstractBeanDefinition beanDefinition : ((IntegrationFlow) bean).getIntegrationComponents()) { + if (beanDefinition instanceof InstanceBeanDefinition) { + final Object instance = beanDefinition.getSource(); + Collection values = this.beanFactory.getBeansOfType(instance.getClass(), false, false).values(); + if (!values.contains(instance)) { + if (instance instanceof AbstractMessageChannel) { + String channelBeanName = ((AbstractMessageChannel) instance).getComponentName(); + if (channelBeanName == null) { + channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; + } + registry.registerBeanDefinition(channelBeanName, beanDefinition); + } + else if (instance instanceof EndpointSpec) { + EndpointSpec endpointSpec = (EndpointSpec) instance; + MessageHandler messageHandler = endpointSpec.getHandler(); + ConsumerEndpointFactoryBean endpoint = endpointSpec.getEndpoint(); + String id = endpointSpec.getId(); + + String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler); + String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null; + BeanComponentDefinition definitionHolder = new BeanComponentDefinition( + new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias); + BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry); + + String endpointBeanName = id; + if (endpointBeanName == null) { + endpointBeanName = generateInstanceBeanDefinitionName(registry, endpoint); + } + registry.registerBeanDefinition(endpointBeanName, new InstanceBeanDefinition(endpoint)); + } + else { + String name = generateInstanceBeanDefinitionName(registry, instance); + registry.registerBeanDefinition(name, beanDefinition); + } + } + } + else { + BeanDefinitionReaderUtils.registerWithGeneratedName(beanDefinition, registry); + } + } + } + return bean; + } + + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + return bean; + } + + @SuppressWarnings("serial") + private static String generateInstanceBeanDefinitionName(BeanDefinitionRegistry registry, final Object instance) { + return BeanDefinitionReaderUtils.generateBeanName(new GenericBeanDefinition() { + + @Override + public String getBeanClassName() { + return instance.getClass().getName(); + } + }, registry); + } + +} 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 d113b1c..cbd6483 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 @@ -23,6 +23,7 @@ import org.springframework.integration.config.SourcePollingChannelAdapterFactory import org.springframework.integration.core.GenericSelector; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessageSelector; +import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.support.EndpointConfigurer; import org.springframework.integration.filter.ExpressionEvaluatingSelector; import org.springframework.integration.filter.MessageFilter; @@ -34,6 +35,7 @@ import org.springframework.integration.transformer.MessageTransformingHandler; import org.springframework.integration.transformer.MethodInvokingTransformer; import org.springframework.integration.transformer.Transformer; import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; /** * @author Artem Bilan @@ -62,10 +64,53 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder channel(MessageChannel messageChannel) { + Assert.notNull(messageChannel); this.currentMessageChannel = messageChannel; return this.addComponent(this.currentMessageChannel).registerOutputChannelIfCan(this.currentMessageChannel); } + public IntegrationFlowBuilder channel(MessageChannelSpec messageChannelSpec) { + Assert.notNull(messageChannelSpec); + return this.channel(messageChannelSpec.get()); + } + + public IntegrationFlowBuilder transform(String expression) { + return this.transform(PARSER.parseExpression(expression)); + } + + public IntegrationFlowBuilder transform(Expression expression) { + return this.transform(new ExpressionEvaluatingTransformer(expression)); + } + + public IntegrationFlowBuilder transform(GenericTransformer genericTransformer) { + return this.transform(genericTransformer, null); + } + + public IntegrationFlowBuilder transform(GenericTransformer genericTransformer, + EndpointConfigurer> endpointConfigurer) { + Transformer transformer = genericTransformer instanceof Transformer + ? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer); + return this.register(new GenericEndpointSpec(new MessageTransformingHandler(transformer)), endpointConfigurer); + } + + public IntegrationFlowBuilder filter(String expression) { + return this.filter(PARSER.parseExpression(expression)); + } + + public IntegrationFlowBuilder filter(Expression expression) { + return this.filter(new ExpressionEvaluatingSelector(expression)); + } + + public IntegrationFlowBuilder filter(GenericSelector genericSelector) { + return this.filter(genericSelector, null); + } + + public IntegrationFlowBuilder filter(GenericSelector genericSelector, EndpointConfigurer endpointConfigurer) { + MessageSelector selector = genericSelector instanceof MessageSelector + ? (MessageSelector) genericSelector : new MethodInvokingSelector(genericSelector); + return this.register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer); + } + private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) { this.flow.addComponent(outputChannel); if (this.currentComponent != null) { @@ -83,50 +128,10 @@ public final class IntegrationFlowBuilder { return this; } - public IntegrationFlowBuilder transform(String expression) { - return this.transform(PARSER.parseExpression(expression)); - } - - public IntegrationFlowBuilder transform(Expression expression) { - return this.transform(new ExpressionEvaluatingTransformer(expression)); - } - - public IntegrationFlowBuilder transform(GenericTransformer genericTransformer) { - Transformer transformer = genericTransformer instanceof Transformer - ? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer); - return this.transform(genericTransformer, new DefaultEndpointConfigurer>()); - } - - public IntegrationFlowBuilder transform(GenericTransformer genericTransformer, - EndpointConfigurer> endpointConfigurer) { - Transformer transformer = genericTransformer instanceof Transformer - ? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer); - GenericEndpointSpec spec = new GenericEndpointSpec(new MessageTransformingHandler(transformer)); - endpointConfigurer.configure(spec); - return this.register(spec); - } - - public IntegrationFlowBuilder filter(String expression) { - return this.filter(PARSER.parseExpression(expression)); - } - - public IntegrationFlowBuilder filter(Expression expression) { - return this.filter(new ExpressionEvaluatingSelector(expression)); - } - - public IntegrationFlowBuilder filter(GenericSelector genericSelector) { - return this.filter(genericSelector, new DefaultEndpointConfigurer()); - } - - public IntegrationFlowBuilder filter(GenericSelector genericSelector, EndpointConfigurer endpointConfigurer) { - MessageSelector selector = genericSelector instanceof MessageSelector - ? (MessageSelector) genericSelector : new MethodInvokingSelector(genericSelector); - FilterEndpointSpec spec = new FilterEndpointSpec(new MessageFilter(selector)); - endpointConfigurer.configure(spec); - return this.register(spec); - } - - private IntegrationFlowBuilder register(EndpointSpec endpointSpec) { + private > IntegrationFlowBuilder register(S endpointSpec, EndpointConfigurer endpointConfigurer) { + if (endpointConfigurer != null) { + endpointConfigurer.configure(endpointSpec); + } MessageChannel inputChannel = this.currentMessageChannel; this.currentMessageChannel = null; if (inputChannel == null) { @@ -143,12 +148,4 @@ public final class IntegrationFlowBuilder { return this.flow; } - private class DefaultEndpointConfigurer> implements EndpointConfigurer { - - @Override - public void configure(S spec) { - - } - } - } 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 ba60dd3..a85a47f 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 @@ -18,9 +18,11 @@ package org.springframework.integration.dsl; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.core.MessageSource; -import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.dsl.channel.MessageChannelSpec; +import org.springframework.integration.dsl.support.PollerSpec; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; /** * @author Artem Bilan @@ -31,8 +33,17 @@ public final class IntegrationFlows { return new IntegrationFlowBuilder().channel(messageChannel); } + public static IntegrationFlowBuilder from(MessageChannelSpec messageChannelSpec) { + return from(messageChannelSpec.get()); + } + public static IntegrationFlowBuilder from(MessageSource messageSource) { - return from(messageSource, null); + return from(messageSource, (PollerMetadata) null); + } + + public static IntegrationFlowBuilder from(MessageSource messageSource, PollerSpec pollerSpec) { + Assert.notNull(pollerSpec); + return from(messageSource, pollerSpec.get()); } public static IntegrationFlowBuilder from(MessageSource messageSource, PollerMetadata pollerMetadata) { @@ -45,9 +56,9 @@ public final class IntegrationFlows { .currentComponent(factoryBean); } - public static IntegrationFlowBuilder from(AbstractEndpoint endpoint) { + /*public static IntegrationFlowBuilder from(AbstractEndpoint endpoint) { return new IntegrationFlowBuilder(); - } + }*/ private IntegrationFlows() { } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java index 4d06754..15dc281 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java @@ -21,10 +21,10 @@ import org.springframework.integration.channel.DirectChannel; /** * @author Artem Bilan */ -public class DirectChannelSpec extends LoadBalancingChannelSpecSupport { +public class DirectChannelSpec extends LoadBalancingChannelSpec { @Override - public DirectChannel get() { + protected DirectChannel doGet() { this.channel = new DirectChannel(this.loadBalancingStrategy); if (this.failover != null) { this.channel.setFailover(this.failover); @@ -32,7 +32,7 @@ public class DirectChannelSpec extends LoadBalancingChannelSpecSupport { +public class ExecutorChannelSpec extends LoadBalancingChannelSpec { private final Executor executor; @@ -31,7 +31,8 @@ public class ExecutorChannelSpec extends LoadBalancingChannelSpecSupport, C extends AbstractMessageChannel> extends ChannelSpecSupport { +public abstract class LoadBalancingChannelSpec, C extends AbstractMessageChannel> extends MessageChannelSpec { protected LoadBalancingStrategy loadBalancingStrategy = new RoundRobinLoadBalancingStrategy(); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ChannelSpecSupport.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannelSpec.java similarity index 85% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ChannelSpecSupport.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannelSpec.java index 8ea74b6..74951e8 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ChannelSpecSupport.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannelSpec.java @@ -19,12 +19,13 @@ package org.springframework.integration.dsl.channel; import java.util.Arrays; import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.dsl.core.Spec; import org.springframework.messaging.support.ChannelInterceptor; /** * @author Artem Bilan */ -public abstract class ChannelSpecSupport, C extends AbstractMessageChannel> { +public abstract class MessageChannelSpec, C extends AbstractMessageChannel> extends Spec { protected C channel; @@ -34,6 +35,11 @@ public abstract class ChannelSpecSupport, C e private ChannelInterceptor[] interceptors; + S id(String id) { + this.id = id; + return _this(); + } + public S datatypes(Class... datatypes) { this.datatypes = datatypes; return _this(); @@ -44,12 +50,8 @@ public abstract class ChannelSpecSupport, C e return _this(); } - public S id(String id) { - this.id = id; - return _this(); - } - - public C get() { + @Override + protected C doGet() { this.channel.setDatatypes(this.datatypes); this.channel.setBeanName(this.id); if (this.interceptors != null) { @@ -58,9 +60,5 @@ public abstract class ChannelSpecSupport, C e return this.channel; } - @SuppressWarnings("unchecked") - protected S _this() { - return (S) this; - } } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java index 391ade7..a0bbba9 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java @@ -19,7 +19,6 @@ package org.springframework.integration.dsl.channel; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executor; -import org.springframework.integration.dispatcher.LoadBalancingStrategy; import org.springframework.integration.store.MessageGroupStore; import org.springframework.messaging.Message; @@ -32,46 +31,82 @@ public final class MessageChannels { return new DirectChannelSpec(); } - public static DirectChannelSpec direct(LoadBalancingStrategy loadBalancingStrategy) { - return direct().loadBalancer(loadBalancingStrategy); + public static DirectChannelSpec direct(String id) { + return direct().id(id); } public static QueueChannelSpec queue() { return new QueueChannelSpec(); } + public static QueueChannelSpec queue(String id) { + return queue().id(id); + } + public static QueueChannelSpec queue(BlockingQueue> queue) { return new QueueChannelSpec(queue); } + public static QueueChannelSpec queue(String id, BlockingQueue> queue) { + return queue(queue).id(id); + } + public static QueueChannelSpec queue(Integer capacity) { return new QueueChannelSpec(capacity); } + public static QueueChannelSpec queue(String id, Integer capacity) { + return queue(capacity).id(id); + } + public static QueueChannelSpec.MessageStoreSpec queue(MessageGroupStore messageGroupStore, Object groupId) { return new QueueChannelSpec.MessageStoreSpec(messageGroupStore, groupId); } + public static QueueChannelSpec.MessageStoreSpec queue(String id, MessageGroupStore messageGroupStore, Object groupId) { + return queue(messageGroupStore, groupId).id(id); + } + public static ExecutorChannelSpec executor(Executor executor) { return new ExecutorChannelSpec(executor); } + public static ExecutorChannelSpec executor(String id, Executor executor) { + return executor(executor).id(id); + } + public static RendezvousChannelSpec rendezvous() { return new RendezvousChannelSpec(); } + public static RendezvousChannelSpec rendezvous(String id) { + return rendezvous().id(id); + } + public static PriorityChannelSpec priority() { return new PriorityChannelSpec(); } + public static PriorityChannelSpec priority(String id) { + return priority().id(id); + } + public static PublishSubscribeChannelSpec publishSubscribe() { return new PublishSubscribeChannelSpec(); } + public static PublishSubscribeChannelSpec publishSubscribe(String id) { + return publishSubscribe().id(id); + } + public static PublishSubscribeChannelSpec publishSubscribe(Executor executor) { return new PublishSubscribeChannelSpec(executor); } + public static PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor) { + return publishSubscribe(executor).id(id); + } + private MessageChannels() { } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java index 70f8c71..4ea9c17 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java @@ -24,7 +24,7 @@ import org.springframework.messaging.Message; /** * @author Artem Bilan */ -public class PriorityChannelSpec extends ChannelSpecSupport { +public class PriorityChannelSpec extends MessageChannelSpec { private int capacity; @@ -41,9 +41,9 @@ public class PriorityChannelSpec extends ChannelSpecSupport { +public class PublishSubscribeChannelSpec extends MessageChannelSpec { PublishSubscribeChannelSpec() { this.channel = new PublishSubscribeChannel(); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java index c66197b..f3e9ff4 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java @@ -27,7 +27,7 @@ import org.springframework.messaging.Message; /** * @author Artem Bilan */ -public class QueueChannelSpec extends ChannelSpecSupport { +public class QueueChannelSpec extends MessageChannelSpec { protected BlockingQueue> queue; @@ -45,7 +45,7 @@ public class QueueChannelSpec extends ChannelSpecSupport { +public class RendezvousChannelSpec extends MessageChannelSpec { RendezvousChannelSpec() { this.channel = new RendezvousChannel(); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/Spec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/Spec.java new file mode 100644 index 0000000..bd6d5f3 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/Spec.java @@ -0,0 +1,41 @@ +/* + * 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; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public abstract class Spec, T> { + + private volatile T target; + + public final T get() { + if (this.target == null) { + this.target = this.doGet(); + } + return this.target; + } + + protected abstract T doGet(); + + @SuppressWarnings("unchecked") + protected S _this() { + return (S) this; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java new file mode 100644 index 0000000..ef092ed --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java @@ -0,0 +1,94 @@ +/* + * 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.support; + +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.integration.dsl.core.Spec; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.transaction.TransactionSynchronizationFactory; +import org.springframework.scheduling.Trigger; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.util.ErrorHandler; + +/** +* @author Artem Bilan +* @since 4.0 +*/ +public final class PollerSpec extends Spec { + + private final PollerMetadata pollerMetadata = new PollerMetadata(); + + private final List adviceChain = new LinkedList(); + + PollerSpec(Trigger trigger) { + this.pollerMetadata.setTrigger(trigger); + } + + public PollerSpec transactionSynchronizationFactory(TransactionSynchronizationFactory transactionSynchronizationFactory) { + pollerMetadata.setTransactionSynchronizationFactory(transactionSynchronizationFactory); + return this; + } + + public PollerSpec errorHandler(ErrorHandler errorHandler) { + pollerMetadata.setErrorHandler(errorHandler); + return this; + } + + public PollerSpec maxMessagesPerPoll(long maxMessagesPerPoll) { + pollerMetadata.setMaxMessagesPerPoll(maxMessagesPerPoll); + return this; + } + + public PollerSpec receiveTimeout(long receiveTimeout) { + pollerMetadata.setReceiveTimeout(receiveTimeout); + return this; + } + + public PollerSpec advice(Advice... advice) { + this.adviceChain.addAll(Arrays.asList(advice)); + return this; + } + + public PollerSpec transactional(PlatformTransactionManager transactionManager) { + return this.advice(new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource())); + } + + public PollerSpec taskExecutor(Executor taskExecutor) { + pollerMetadata.setTaskExecutor(taskExecutor); + return this; + } + + public PollerSpec sendTimeout(long sendTimeout) { + pollerMetadata.setSendTimeout(sendTimeout); + return this; + } + + @Override + protected PollerMetadata doGet() { + pollerMetadata.setAdviceChain(this.adviceChain); + return this.pollerMetadata; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java index 14a31d6..8fdfe36 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java @@ -16,24 +16,12 @@ package org.springframework.integration.dsl.support; -import java.util.Arrays; -import java.util.LinkedList; -import java.util.List; import java.util.TimeZone; -import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; -import org.aopalliance.aop.Advice; - -import org.springframework.integration.scheduling.PollerMetadata; -import org.springframework.integration.transaction.TransactionSynchronizationFactory; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.support.CronTrigger; import org.springframework.scheduling.support.PeriodicTrigger; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; -import org.springframework.transaction.interceptor.TransactionInterceptor; -import org.springframework.util.ErrorHandler; /** * @author Artem Bilan @@ -77,62 +65,4 @@ public final class Pollers { private Pollers() { } - public static final class PollerSpec { - - private final PollerMetadata pollerMetadata = new PollerMetadata(); - - private final List adviceChain = new LinkedList(); - - private PollerSpec(Trigger trigger) { - this.pollerMetadata.setTrigger(trigger); - } - - public PollerSpec transactionSynchronizationFactory(TransactionSynchronizationFactory transactionSynchronizationFactory) { - pollerMetadata.setTransactionSynchronizationFactory(transactionSynchronizationFactory); - return this; - } - - public PollerSpec errorHandler(ErrorHandler errorHandler) { - pollerMetadata.setErrorHandler(errorHandler); - return this; - } - - public PollerSpec maxMessagesPerPoll(long maxMessagesPerPoll) { - pollerMetadata.setMaxMessagesPerPoll(maxMessagesPerPoll); - return this; - } - - public PollerSpec receiveTimeout(long receiveTimeout) { - pollerMetadata.setReceiveTimeout(receiveTimeout); - return this; - } - - public PollerSpec advice(Advice... advice) { - this.adviceChain.addAll(Arrays.asList(advice)); - return this; - } - - public PollerSpec transactional(PlatformTransactionManager transactionManager) { - return this.advice(new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource())); - } - - public PollerSpec taskExecutor(Executor taskExecutor) { - pollerMetadata.setTaskExecutor(taskExecutor); - return this; - } - - public PollerSpec sendTimeout(long sendTimeout) { - pollerMetadata.setSendTimeout(sendTimeout); - return this; - } - - public PollerMetadata get() { - pollerMetadata.setAdviceChain(this.adviceChain); - return this.pollerMetadata; - } - - - - } - } 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 b9e4ff1..8d25af4 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 @@ -22,6 +22,7 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import org.aopalliance.aop.Advice; @@ -42,7 +43,10 @@ import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.channel.QueueChannelSpec; +import org.springframework.integration.dsl.support.PollerSpec; import org.springframework.integration.dsl.support.Pollers; import org.springframework.integration.endpoint.MethodInvokingMessageSource; import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice; @@ -118,6 +122,16 @@ public class IntegrationFlowTests { @EnableIntegration public static class ContextConfiguration { + @Bean + public DirectChannelSpec inputChannel() { + return MessageChannels.direct(); + } + + @Bean + public QueueChannelSpec successChannel() { + return MessageChannels.queue(); + } + @Bean public MessageSource integerMessageSource() { MethodInvokingMessageSource source = new MethodInvokingMessageSource(); @@ -128,48 +142,51 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow flow1() { - return IntegrationFlows.from(this.integerMessageSource(), Pollers.fixedRate(100).get()) + return IntegrationFlows.from(this.integerMessageSource(), Pollers.fixedRate(100)) .transform("payload.toString()") - .channel(MessageChannels.queue().id("flow1QueueChannel").get()) + .channel(MessageChannels.queue("flow1QueueChannel")) .get(); } - @Bean - public DirectChannel inputChannel() { - return MessageChannels.direct().get(); - } - - @Bean - public QueueChannel successChannel() { - return MessageChannels.queue().get(); - } - @Bean(name = PollerMetadata.DEFAULT_POLLER_METADATA_BEAN_NAME) - public PollerMetadata poller() { - return Pollers.fixedRate(500).get(); + public PollerSpec poller() { + return Pollers.fixedRate(500); } + } + + @Configuration + public static class ContextConfiguration2 { + + @Autowired + @Qualifier("inputChannel") + private DirectChannel inputChannel; + + @Autowired + @Qualifier("successChannel") + private PollableChannel successChannel; + @Bean public Advice expressionAdvice() { ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice(); advice.setOnSuccessExpression("payload"); - advice.setSuccessChannel(this.successChannel()); + advice.setSuccessChannel(this.successChannel); return advice; } @Bean public IntegrationFlow flow2() { - return IntegrationFlows.from(this.inputChannel()) + return IntegrationFlows.from(this.inputChannel) .filter(p -> p instanceof String, c -> c.id("filter")) .transform(Integer::parseInt) .transform(new PayloadSerializingTransformer(), c -> c.autoStartup(false).id("payloadSerializingTransformer")) - .channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue").get()) + .channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue")) .transform(new PayloadDeserializingTransformer()) + .channel(MessageChannels.executor("executor", Executors.newCachedThreadPool())) .transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice())) .get(); } - } }