diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index a0bf235..de19646 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -1,25 +1,26 @@ description = 'Spring Integration Java DSL' apply plugin: 'java' -apply from: "${rootProject.projectDir}/publish-maven.gradle" +apply from: "${rootProject.projectDir}/publish-maven.gradle" apply plugin: 'eclipse' apply plugin: 'idea' group = 'org.springframework.integration' repositories { +// mavenLocal() if (version.endsWith('BUILD-SNAPSHOT')) { - maven { url 'http://repo.spring.io/libs-snapshot' } - } - maven { url 'http://repo.spring.io/libs-milestone' } + maven { url 'http://repo.spring.io/libs-snapshot' } + } + maven { url 'http://repo.spring.io/libs-milestone' } } compileJava { - sourceCompatibility=1.6 + sourceCompatibility = 1.6 } compileTestJava { - sourceCompatibility=1.8 + sourceCompatibility = 1.8 } ext { @@ -30,10 +31,10 @@ ext { springIntegrationVersion = '4.0.0.BUILD-SNAPSHOT' linkHomepage = 'https://github.com/spring-projects/spring-integration-extensions' - linkCi = 'https://build.springsource.org/browse/INTEXT' - linkIssue = 'https://jira.springsource.org/browse/INTEXT' - linkScmUrl = 'https://github.com/spring-projects/spring-integration-extensions' - linkScmConnection = 'https://github.com/spring-projects/spring-integration-extensions.git' + linkCi = 'https://build.springsource.org/browse/INTEXT' + linkIssue = 'https://jira.springsource.org/browse/INTEXT' + linkScmUrl = 'https://github.com/spring-projects/spring-integration-extensions' + linkScmConnection = 'https://github.com/spring-projects/spring-integration-extensions.git' linkScmDevConnection = 'git@github.com:spring-projects/spring-integration-extensions.git' } @@ -56,10 +57,9 @@ dependencies { testCompile "org.mockito:mockito-all:$mockitoVersion" testCompile "org.springframework:spring-test:$springVersion" - jacoco group: "org.jacoco", name: "org.jacoco.agent", version: "0.6.2.201302030002", classifier: "runtime" + jacoco group: "org.jacoco", name: "org.jacoco.agent", version: "0.5.6.201201232323", classifier: "runtime" } - // enable all compiler warnings; individual projects may customize further ext.xLintArg = '-Xlint:all,-options' [compileJava, compileTestJava]*.options*.compilerArgs = [xLintArg] @@ -119,7 +119,7 @@ task docsZip(type: Zip) { description = "Builds -${classifier} archive containing the api " + "for deployment at static.springframework.org/spring-integration/docs." - from (api) { + from(api) { into 'api' } } @@ -128,7 +128,7 @@ task distZip(type: Zip, dependsOn: docsZip) { group = 'Distribution' classifier = 'dist' description = "Builds -${classifier} archive, containing all jars and docs, " + - "suitable for community download page." + "suitable for community download page." ext.baseDir = "${project.name}-${project.version}"; @@ -142,7 +142,7 @@ task distZip(type: Zip, dependsOn: docsZip) { into "${baseDir}/docs" } - into ("${baseDir}/libs") { + into("${baseDir}/libs") { from project.jar from project.sourcesJar from project.javadocJar @@ -155,7 +155,7 @@ task depsZip(type: Zip, dependsOn: distZip) { zipTask -> group = 'Distribution' classifier = 'dist-with-deps' description = "Builds -${classifier} archive, containing everything " + - "in the -${distZip.classifier} archive plus all dependencies." + "in the -${distZip.classifier} archive plus all dependencies." from zipTree(distZip.archivePath) 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 new file mode 100644 index 0000000..844038d --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java @@ -0,0 +1,97 @@ +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; + +/** + * The Java DSL Integration infrastructure {@code beanFactory} initializer. + * + * @author Artem Bilan + */ +public class DslIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer { + + @Override + public void initialize(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException { + this.initializeIntegrationFlows(configurableListableBeanFactory); + } + + private void initializeIntegrationFlows(ConfigurableListableBeanFactory beanFactory) { + Map integrationFlows = beanFactory.getBeansOfType(IntegrationFlow.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); + } + + } + + @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 new file mode 100644 index 0000000..9e852c5 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EndpointSpec.java @@ -0,0 +1,82 @@ +package org.springframework.integration.dsl; + +import java.util.Arrays; +import java.util.LinkedList; +import java.util.List; + +import org.aopalliance.aop.Advice; + +import org.springframework.integration.config.ConsumerEndpointFactoryBean; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.messaging.MessageHandler; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public abstract class EndpointSpec, C extends MessageHandler> { + + private final ConsumerEndpointFactoryBean endpointFactoryBean = new ConsumerEndpointFactoryBean(); + + private final C messageHandler; + + private final List adviceChain = new LinkedList(); + + private String id; + + EndpointSpec(C messageHandler) { + this.messageHandler = messageHandler; + this.endpointFactoryBean.setHandler(this.messageHandler); + if (this.messageHandler instanceof AbstractReplyProducingMessageHandler) { + ((AbstractReplyProducingMessageHandler) this.messageHandler).setAdviceChain(this.adviceChain); + } + else { + this.endpointFactoryBean.setAdviceChain(this.adviceChain); + } + } + + public S id(String id) { + this.id = id; + this.endpointFactoryBean.setBeanName(id); + return _this(); + } + + public S phase(int phase) { + this.endpointFactoryBean.setPhase(phase); + return _this(); + } + + public S autoStartup(boolean autoStartup) { + this.endpointFactoryBean.setAutoStartup(autoStartup); + return _this(); + } + + public S advice(Advice... advice) { + this.adviceChain.addAll(Arrays.asList(advice)); + return _this(); + } + + public S poller(PollerMetadata pollerMetadata) { + this.endpointFactoryBean.setPollerMetadata(pollerMetadata); + return _this(); + } + + String getId() { + return id; + } + + ConsumerEndpointFactoryBean getEndpoint() { + return this.endpointFactoryBean; + } + + C getHandler() { + return this.messageHandler; + } + + @SuppressWarnings("unchecked") + protected S _this() { + return (S) this; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java new file mode 100644 index 0000000..e63d33c --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java @@ -0,0 +1,31 @@ +package org.springframework.integration.dsl; + +import org.springframework.integration.filter.MessageFilter; +import org.springframework.messaging.MessageChannel; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public final class FilterEndpointSpec extends EndpointSpec { + + FilterEndpointSpec(MessageFilter messageFilter) { + super(messageFilter); + } + + public FilterEndpointSpec throwExceptionOnRejection(boolean throwExceptionOnRejection) { + this.getHandler().setThrowExceptionOnRejection(throwExceptionOnRejection); + return _this(); + } + + public FilterEndpointSpec discardChannel(MessageChannel discardChannel) { + this.getHandler().setDiscardChannel(discardChannel); + return _this(); + } + + public FilterEndpointSpec discardWithinAdvice(boolean discardWithinAdvice) { + this.getHandler().setDiscardWithinAdvice(discardWithinAdvice); + return _this(); + } + +} 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 new file mode 100644 index 0000000..2534e48 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java @@ -0,0 +1,15 @@ +package org.springframework.integration.dsl; + +import org.springframework.messaging.MessageHandler; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public final class GenericEndpointSpec extends EndpointSpec, C> { + + GenericEndpointSpec(C messageHandler) { + super(messageHandler); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java new file mode 100644 index 0000000..7d534ea --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java @@ -0,0 +1,46 @@ +/* + * 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; + +import java.util.LinkedHashSet; +import java.util.Set; + +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.integration.dsl.config.InstanceBeanDefinition; + +/** + * @author Artem Bilan + */ +public final class IntegrationFlow { + + private final Set integrationComponents = new LinkedHashSet(); + + IntegrationFlow() { + } + + public Set getIntegrationComponents() { + return integrationComponents; + } + + IntegrationFlow addComponent(Object component) { + AbstractBeanDefinition beanDefinition = component instanceof AbstractBeanDefinition + ? (AbstractBeanDefinition) component : new InstanceBeanDefinition(component); + this.integrationComponents.add(beanDefinition); + return this; + } + +} 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 new file mode 100644 index 0000000..abed170 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -0,0 +1,154 @@ +/* + * 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; + +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; +import org.springframework.integration.core.GenericSelector; +import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.core.MessageSelector; +import org.springframework.integration.dsl.support.EndpointConfigurer; +import org.springframework.integration.filter.ExpressionEvaluatingSelector; +import org.springframework.integration.filter.MessageFilter; +import org.springframework.integration.filter.MethodInvokingSelector; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.transformer.ExpressionEvaluatingTransformer; +import org.springframework.integration.transformer.GenericTransformer; +import org.springframework.integration.transformer.MessageTransformingHandler; +import org.springframework.integration.transformer.MethodInvokingTransformer; +import org.springframework.integration.transformer.Transformer; +import org.springframework.messaging.MessageChannel; + +/** + * @author Artem Bilan + */ +public final class IntegrationFlowBuilder { + + private final static SpelExpressionParser PARSER = new SpelExpressionParser(); + + private final IntegrationFlow flow = new IntegrationFlow(); + + private MessageChannel currentMessageChannel; + + private Object currentComponent; + + IntegrationFlowBuilder() { + } + + IntegrationFlowBuilder addComponent(Object component) { + this.flow.addComponent(component); + return this; + } + + IntegrationFlowBuilder currentComponent(Object component) { + this.currentComponent = component; + return this; + } + + public IntegrationFlowBuilder channel(MessageChannel messageChannel) { + this.currentMessageChannel = messageChannel; + return this.addComponent(this.currentMessageChannel).registerOutputChannelIfCan(this.currentMessageChannel); + } + + private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) { + this.flow.addComponent(outputChannel); + if (this.currentComponent != null) { + if (this.currentComponent instanceof MessageProducer) { + ((MessageProducer) this.currentComponent).setOutputChannel(outputChannel); + } + if (this.currentComponent instanceof AbstractReplyProducingMessageHandler) { + ((AbstractReplyProducingMessageHandler) this.currentComponent).setOutputChannel(outputChannel); + } + else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) { + ((SourcePollingChannelAdapterFactoryBean) this.currentComponent).setOutputChannel(outputChannel); + } + this.currentComponent = null; + } + 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) { + MessageChannel inputChannel = this.currentMessageChannel; + this.currentMessageChannel = null; + if (inputChannel == null) { + inputChannel = new DirectChannel(); + this.registerOutputChannelIfCan(inputChannel); + } + + endpointSpec.getEndpoint().setInputChannel(inputChannel); + + return this.addComponent(endpointSpec).currentComponent(endpointSpec.getHandler()); + } + + public IntegrationFlow get() { + 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 new file mode 100644 index 0000000..948bb8b --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -0,0 +1,55 @@ +/* + * 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; + +import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.messaging.MessageChannel; + +/** + * @author Artem Bilan + */ +public final class IntegrationFlows { + + public static IntegrationFlowBuilder from(MessageChannel messageChannel) { + return new IntegrationFlowBuilder().channel(messageChannel); + } + + public static IntegrationFlowBuilder from(MessageSource messageSource) { + return from(messageSource, null); + } + + public static IntegrationFlowBuilder from(MessageSource messageSource, PollerMetadata pollerMetadata) { + SourcePollingChannelAdapterFactoryBean factoryBean = new SourcePollingChannelAdapterFactoryBean(); + factoryBean.setSource(messageSource); + factoryBean.setPollerMetadata(pollerMetadata); + return new IntegrationFlowBuilder() + .addComponent(messageSource) + .addComponent(factoryBean) + .currentComponent(factoryBean); + } + + 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/ChannelSpecSupport.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ChannelSpecSupport.java new file mode 100644 index 0000000..8e4fad4 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ChannelSpecSupport.java @@ -0,0 +1,66 @@ +/* + * 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.channel; + +import java.util.Arrays; + +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.messaging.support.ChannelInterceptor; + +/** + * @author Artem Bilan + */ +public abstract class ChannelSpecSupport, C extends AbstractMessageChannel> { + + protected C channel; + + private String id; + + private Class[] datatypes; + + private ChannelInterceptor[] interceptors; + + public S datatypes(Class... datatypes) { + this.datatypes = datatypes; + return _this(); + } + + public S interceptors(ChannelInterceptor... interceptors) { + this.interceptors = interceptors; + return _this(); + } + + public S id(String id) { + this.id = id; + return _this(); + } + + public C get() { + this.channel.setDatatypes(this.datatypes); + this.channel.setBeanName(this.id); + if (this.interceptors != null) { + this.channel.setInterceptors(Arrays.asList(this.interceptors)); + } + 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/DirectChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java new file mode 100644 index 0000000..3c8d5dd --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java @@ -0,0 +1,40 @@ +/* + * 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.channel; + +import org.springframework.integration.channel.DirectChannel; + +/** + * @author Artem Bilan + */ +public class DirectChannelSpec extends LoadBalancingChannelSpecSupport { + + @Override + public DirectChannel get() { + this.channel = new DirectChannel(this.loadBalancingStrategy); + if (this.failover != null) { + this.channel.setFailover(this.failover); + } + if (this.maxSubscribers != null) { + this.channel.setMaxSubscribers(this.maxSubscribers); + } + return super.get(); + } + + DirectChannelSpec() { + } +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ExecutorChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ExecutorChannelSpec.java new file mode 100644 index 0000000..82aa9d8 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ExecutorChannelSpec.java @@ -0,0 +1,45 @@ +/* + * 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.channel; + +import java.util.concurrent.Executor; + +import org.springframework.integration.channel.ExecutorChannel; + +/** + * @author Artem Bilan + */ +public class ExecutorChannelSpec extends LoadBalancingChannelSpecSupport { + + private final Executor executor; + + ExecutorChannelSpec(Executor executor) { + this.executor = executor; + } + + public ExecutorChannel get() { + this.channel = new ExecutorChannel(this.executor, this.loadBalancingStrategy); + if (this.failover != null) { + this.channel.setFailover(this.failover); + } + if (this.maxSubscribers != null) { + this.channel.setMaxSubscribers(this.maxSubscribers); + } + return super.get(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/LoadBalancingChannelSpecSupport.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/LoadBalancingChannelSpecSupport.java new file mode 100644 index 0000000..1d2e086 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/LoadBalancingChannelSpecSupport.java @@ -0,0 +1,49 @@ +/* + * 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.channel; + +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.dispatcher.LoadBalancingStrategy; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; + +/** + * @author Artem Bilan + */ +public class LoadBalancingChannelSpecSupport, C extends AbstractMessageChannel> extends ChannelSpecSupport { + + protected LoadBalancingStrategy loadBalancingStrategy = new RoundRobinLoadBalancingStrategy(); + + protected Boolean failover; + + protected Integer maxSubscribers; + + public S loadBalancer(LoadBalancingStrategy loadBalancingStrategy) { + this.loadBalancingStrategy = loadBalancingStrategy; + return _this(); + } + + public S failover(Boolean failover) { + this.failover = failover; + return _this(); + } + + public S maxSubscribers(Integer maxSubscribers) { + this.maxSubscribers = maxSubscribers; + return _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 new file mode 100644 index 0000000..3e17f71 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java @@ -0,0 +1,78 @@ +/* + * 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.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; + +/** + * @author Artem Bilan + */ +public final class MessageChannels { + + public static DirectChannelSpec direct() { + return new DirectChannelSpec(); + } + + public static DirectChannelSpec direct(LoadBalancingStrategy loadBalancingStrategy) { + return direct().loadBalancer(loadBalancingStrategy); + } + + public static QueueChannelSpec queue() { + return new QueueChannelSpec(); + } + + public static QueueChannelSpec queue(BlockingQueue> queue) { + return new QueueChannelSpec(queue); + } + + public static QueueChannelSpec queue(Integer capacity) { + return new QueueChannelSpec(capacity); + } + + public static QueueChannelSpec.MessageStoreSpec queue(MessageGroupStore messageGroupStore, Object groupId) { + return new QueueChannelSpec.MessageStoreSpec(messageGroupStore, groupId); + } + + public static ExecutorChannelSpec executor(Executor executor) { + return new ExecutorChannelSpec(executor); + } + + public static RendezvousChannelSpec rendezvous() { + return new RendezvousChannelSpec(); + } + + public static PriorityChannelSpec priority() { + return new PriorityChannelSpec(); + } + + public static PublishSubscribeChannelSpec publishSubscribe() { + return new PublishSubscribeChannelSpec(); + } + + public static PublishSubscribeChannelSpec publishSubscribe(Executor executor) { + return new PublishSubscribeChannelSpec(executor); + } + + 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 new file mode 100644 index 0000000..37ff486 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java @@ -0,0 +1,52 @@ +/* + * 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.channel; + +import java.util.Comparator; + +import org.springframework.integration.channel.PriorityChannel; +import org.springframework.messaging.Message; + +/** + * @author Artem Bilan + */ +public class PriorityChannelSpec extends ChannelSpecSupport { + + private int capacity; + + Comparator> comparator; + + public PriorityChannelSpec setCapacity(int capacity) { + this.capacity = capacity; + return this; + } + + public PriorityChannelSpec setComparator(Comparator> comparator) { + this.comparator = comparator; + return this; + } + + @Override + public PriorityChannel get() { + this.channel = new PriorityChannel(this.capacity, this.comparator); + return super.get(); + } + + + PriorityChannelSpec() { + } +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PublishSubscribeChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PublishSubscribeChannelSpec.java new file mode 100644 index 0000000..282d0ce --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PublishSubscribeChannelSpec.java @@ -0,0 +1,62 @@ +/* + * 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.channel; + +import java.util.concurrent.Executor; + +import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.util.ErrorHandler; + +/** + * @author Artem Bilan + */ +public class PublishSubscribeChannelSpec extends ChannelSpecSupport { + + PublishSubscribeChannelSpec() { + this.channel = new PublishSubscribeChannel(); + } + + PublishSubscribeChannelSpec(Executor executor) { + this.channel = new PublishSubscribeChannel(executor); + } + + PublishSubscribeChannelSpec errorHandler(ErrorHandler errorHandler) { + this.channel.setErrorHandler(errorHandler); + return this; + } + + public PublishSubscribeChannelSpec ignoreFailures(boolean ignoreFailures) { + this.channel.setIgnoreFailures(ignoreFailures); + return this; + } + + public PublishSubscribeChannelSpec applySequence(boolean applySequence) { + this.channel.setApplySequence(applySequence); + return this; + } + + public PublishSubscribeChannelSpec maxSubscribers(Integer maxSubscribers) { + this.channel.setMaxSubscribers(maxSubscribers); + return this; + } + + public PublishSubscribeChannelSpec minSubscribers(int minSubscribers) { + this.channel.setMinSubscribers(minSubscribers); + return this; + } + +} 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 new file mode 100644 index 0000000..df3097b --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java @@ -0,0 +1,106 @@ +/* + * 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.channel; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.locks.Lock; + +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.store.MessageGroupQueue; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.messaging.Message; + +/** + * @author Artem Bilan + */ +public class QueueChannelSpec extends ChannelSpecSupport { + + protected BlockingQueue> queue; + + protected Integer capacity; + + QueueChannelSpec() { + } + + QueueChannelSpec(BlockingQueue> queue) { + this.queue = queue; + } + + QueueChannelSpec(Integer capacity) { + this.capacity = capacity; + } + + @Override + public QueueChannel get() { + if (this.queue != null) { + this.channel = new QueueChannel(this.queue); + } + else if (this.capacity != null) { + this.channel = new QueueChannel(this.capacity); + } + else { + this.channel = new QueueChannel(); + } + return super.get(); + } + + public static class MessageStoreSpec extends QueueChannelSpec { + + private final MessageGroupStore messageGroupStore; + + private final Object groupId; + + private Lock storeLock; + + MessageStoreSpec(MessageGroupStore messageGroupStore, Object groupId) { + super(); + this.messageGroupStore = messageGroupStore; + this.groupId = groupId; + } + + public MessageStoreSpec capacity(Integer capacity) { + this.capacity = capacity; + return this; + } + + public MessageStoreSpec storeLock(Lock storeLock) { + this.storeLock = storeLock; + return this; + } + + @Override + public QueueChannel get() { + if (this.capacity != null) { + if (this.storeLock != null) { + this.queue = new MessageGroupQueue(messageGroupStore, groupId, this.capacity, this.storeLock); + } + else { + this.queue = new MessageGroupQueue(messageGroupStore, groupId, this.capacity); + } + } + else if (this.storeLock != null) { + this.queue = new MessageGroupQueue(messageGroupStore, groupId, this.storeLock); + } + else { + this.queue = new MessageGroupQueue(messageGroupStore, groupId); + } + return super.get(); + } + + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/RendezvousChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/RendezvousChannelSpec.java new file mode 100644 index 0000000..e0395f8 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/RendezvousChannelSpec.java @@ -0,0 +1,30 @@ +/* + * 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.channel; + +import org.springframework.integration.channel.RendezvousChannel; + +/** + * @author Artem Bilan + */ +public class RendezvousChannelSpec extends ChannelSpecSupport { + + RendezvousChannelSpec() { + this.channel = new RendezvousChannel(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/InstanceBeanDefinition.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/InstanceBeanDefinition.java new file mode 100644 index 0000000..e665202 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/InstanceBeanDefinition.java @@ -0,0 +1,53 @@ +/* + * 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.config; + +import org.springframework.beans.factory.config.ConstructorArgumentValues; +import org.springframework.beans.factory.support.GenericBeanDefinition; + +/** + * @author Artem Bilan + */ +@SuppressWarnings("serial") +public class InstanceBeanDefinition extends GenericBeanDefinition { + + private final Object instance; + + public InstanceBeanDefinition(Object instance) { + this.instance = instance; + ConstructorArgumentValues args = new ConstructorArgumentValues(); + args.addGenericArgumentValue(this.instance); + this.setConstructorArgumentValues(args); + this.setBeanClass(SimpleFactoryBean.class); + } + + public InstanceBeanDefinition(InstanceBeanDefinition original) { + super(original); + this.instance = original.instance; + } + + @Override + public InstanceBeanDefinition cloneBeanDefinition() { + return new InstanceBeanDefinition(this); + } + + @Override + public Object getSource() { + return this.instance; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/SimpleFactoryBean.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/SimpleFactoryBean.java new file mode 100644 index 0000000..f2ae5f2 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/SimpleFactoryBean.java @@ -0,0 +1,95 @@ +/* + * 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.config; + +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.config.AbstractFactoryBean; +import org.springframework.beans.factory.config.AutowireCapableBeanFactory; +import org.springframework.context.Lifecycle; +import org.springframework.context.SmartLifecycle; + +/** + * @author Artem Bilan + */ +class SimpleFactoryBean extends AbstractFactoryBean implements BeanNameAware, SmartLifecycle { + + private final T target; + + private String name; + + public SimpleFactoryBean(T target) { + this.target = target; + } + + @Override + public void setBeanName(String name) { + this.name = name; + } + + @Override + public Class getObjectType() { + return this.target.getClass(); + } + + @Override + protected T createInstance() throws Exception { + ((AutowireCapableBeanFactory) this.getBeanFactory()).initializeBean(this.target, this.name); + return this.target; + } + + @Override + public boolean isAutoStartup() { + return this.target instanceof SmartLifecycle && ((SmartLifecycle) this.target).isAutoStartup(); + } + + @Override + public void stop(Runnable callback) { + if (this.target instanceof SmartLifecycle) { + ((SmartLifecycle) this.target).stop(callback); + } + } + + @Override + public void start() { + if (this.target instanceof Lifecycle) { + ((Lifecycle) this.target).start(); + } + } + + @Override + public void stop() { + if (this.target instanceof Lifecycle) { + ((Lifecycle) this.target).start(); + } + } + + @Override + public boolean isRunning() { + return this.target instanceof SmartLifecycle && ((SmartLifecycle) this.target).isRunning(); + } + + @Override + public int getPhase() { + if (this.target instanceof SmartLifecycle) { + return ((SmartLifecycle) this.target).getPhase(); + } + else { + return 0; + } + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/package-info.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/package-info.java new file mode 100644 index 0000000..1651fd5 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides configuration classes of the Spring Integration Java DSL. + */ +package org.springframework.integration.dsl.config; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java new file mode 100644 index 0000000..b11d3b6 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java @@ -0,0 +1,13 @@ +package org.springframework.integration.dsl.support; + +import org.springframework.integration.dsl.EndpointSpec; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public interface EndpointConfigurer> { + + void configure(S spec); + +} 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 new file mode 100644 index 0000000..06bcb07 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java @@ -0,0 +1,122 @@ +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 + */ +public final class Pollers { + + public static PollerSpec trigger(Trigger trigger) { + return new PollerSpec(trigger); + } + + public static PollerSpec fixedRate(long period) { + return fixedRate(period, null); + } + + public static PollerSpec fixedRate(long period, TimeUnit timeUnit) { + return periodicTrigger(period, timeUnit, true); + } + + public static PollerSpec fixedDelay(long period) { + return fixedDelay(period, null); + } + + public static PollerSpec fixedDelay(long period, TimeUnit timeUnit) { + return periodicTrigger(period, timeUnit, false); + } + + private static PollerSpec periodicTrigger(long period, TimeUnit timeUnit, boolean fixedRate) { + PeriodicTrigger periodicTrigger = new PeriodicTrigger(period, timeUnit); + periodicTrigger.setFixedRate(fixedRate); + return new PollerSpec(periodicTrigger); + } + + public static PollerSpec cron(String cronExpression) { + return cron(cronExpression, TimeZone.getDefault()); + } + + public static PollerSpec cron(String cronExpression, TimeZone timeZone) { + return new PollerSpec(new CronTrigger(cronExpression, timeZone)); + } + + 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/main/resources/META-INF/spring.factories b/spring-integration-java-dsl/src/main/resources/META-INF/spring.factories index e69de29..52182ce 100644 --- a/spring-integration-java-dsl/src/main/resources/META-INF/spring.factories +++ b/spring-integration-java-dsl/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.integration.config.IntegrationConfigurationInitializer=\ +org.springframework.integration.dsl.DslIntegrationConfigurationInitializer 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 new file mode 100644 index 0000000..b9e4ff1 --- /dev/null +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java @@ -0,0 +1,175 @@ +/* + * 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.test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.concurrent.atomic.AtomicInteger; + +import org.aopalliance.aop.Advice; +import org.hamcrest.Matchers; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.Lifecycle; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.MessageDispatchingException; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +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.MessageChannels; +import org.springframework.integration.dsl.support.Pollers; +import org.springframework.integration.endpoint.MethodInvokingMessageSource; +import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.store.SimpleMessageStore; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.transformer.PayloadDeserializingTransformer; +import org.springframework.integration.transformer.PayloadSerializingTransformer; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.support.AnnotationConfigContextLoader; + +/** + * @author Artem Bilan + */ +@ContextConfiguration(loader = AnnotationConfigContextLoader.class) +@RunWith(SpringJUnit4ClassRunner.class) +public class IntegrationFlowTests { + + @Autowired + @Qualifier("flow1QueueChannel") + private PollableChannel outputChannel; + + @Autowired + private DirectChannel inputChannel; + + @Autowired + private PollableChannel successChannel; + + @Autowired + private BeanFactory beanFactory; + + @Test + public void testPollingFlow() { + for (int i = 0; i < 10; i++) { + Message message = this.outputChannel.receive(5000); + assertNotNull(message); + assertEquals("" + i, message.getPayload()); + } + } + + @Test + public void testDirectFlow() { + assertTrue(this.beanFactory.containsBean("filter")); + assertTrue(this.beanFactory.containsBean("filter.handler")); + QueueChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("100").setReplyChannel(replyChannel).build(); + try { + this.inputChannel.send(message); + fail("Expected MessageDispatchingException"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(MessageDeliveryException.class)); + assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class)); + assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers")); + } + this.beanFactory.getBean("payloadSerializingTransformer", Lifecycle.class).start(); + this.inputChannel.send(message); + Message reply = replyChannel.receive(5000); + assertNotNull(reply); + assertEquals(200, reply.getPayload()); + + Message successMessage = this.successChannel.receive(5000); + assertNotNull(successMessage); + assertEquals(100, successMessage.getPayload()); + } + + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + public MessageSource integerMessageSource() { + MethodInvokingMessageSource source = new MethodInvokingMessageSource(); + source.setObject(new AtomicInteger()); + source.setMethodName("getAndIncrement"); + return source; + } + + @Bean + public IntegrationFlow flow1() { + return IntegrationFlows.from(this.integerMessageSource(), Pollers.fixedRate(100).get()) + .transform("payload.toString()") + .channel(MessageChannels.queue().id("flow1QueueChannel").get()) + .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(); + } + + @Bean + public Advice expressionAdvice() { + ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice(); + advice.setOnSuccessExpression("payload"); + advice.setSuccessChannel(this.successChannel()); + return advice; + } + + @Bean + public IntegrationFlow flow2() { + 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()) + .transform(new PayloadDeserializingTransformer()) + .transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice())) + .get(); + } + + } + +}