diff --git a/build.gradle b/build.gradle index f946e2e2b1..c8c7a78259 100644 --- a/build.gradle +++ b/build.gradle @@ -67,7 +67,7 @@ ext { ftpServerVersion = '1.2.0' graalvmVersion = '22.2.0' greenmailVersion = '2.0.0-alpha-2' - groovyVersion = '3.0.13' + groovyVersion = '4.0.5' hamcrestVersion = '2.2' hazelcastVersion = '5.1.3' hibernateVersion = '6.1.3.Final' @@ -167,6 +167,7 @@ allprojects { mavenBom "io.micrometer:micrometer-tracing-bom:$micrometerTracingVersion" mavenBom "org.apache.camel:camel-bom:$camelVersion" mavenBom "org.testcontainers:testcontainers-bom:$testcontainersVersion" + mavenBom "org.apache.groovy:groovy-bom:$groovyVersion" } } @@ -617,14 +618,17 @@ project('spring-integration-graphql') { project('spring-integration-groovy') { description = 'Spring Integration Groovy Support' + + apply plugin: 'groovy' + dependencies { api project(':spring-integration-scripting') - api "org.codehaus.groovy:groovy:$groovyVersion" + api 'org.apache.groovy:groovy' api 'org.springframework:spring-context-support' testImplementation 'org.springframework:spring-web' - testRuntimeOnly "org.codehaus.groovy:groovy-dateutil:$groovyVersion" + testRuntimeOnly 'org.apache.groovy:groovy-dateutil' } } @@ -852,7 +856,7 @@ project('spring-integration-scripting') { providedImplementation "org.graalvm.sdk:graal-sdk:$graalvmVersion" testImplementation "org.jruby:jruby-complete:$jrubyVersion" - testImplementation "org.codehaus.groovy:groovy-jsr223:$groovyVersion" + testImplementation 'org.apache.groovy:groovy-jsr223' testImplementation "org.python:jython-standalone:$jythonVersion" testRuntimeOnly 'org.jetbrains.kotlin:kotlin-scripting-compiler-embeddable' @@ -1143,7 +1147,7 @@ task schemaZip(type: Zip) { duplicatesStrategy = DuplicatesStrategy.EXCLUDE javaProjects.each { subproject -> - Properties schemas = new Properties(); + Properties schemas = new Properties() def shortName = subproject.name.replaceFirst("${rootProject.name}-", '') if (subproject.name.endsWith('-core')) { shortName = '' @@ -1204,7 +1208,7 @@ task distZip(type: Zip, dependsOn: [docsZip, schemaZip]) { description = "Builds -${archiveClassifier} archive, containing all jars and docs, " + "suitable for community download page." - ext.baseDir = "${project.name}-${project.version}"; + ext.baseDir = "${project.name}-${project.version}" from('src/dist') { include 'readme.txt' diff --git a/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/GroovyIntegrationFlowDefinition.groovy b/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/GroovyIntegrationFlowDefinition.groovy new file mode 100644 index 0000000000..262d466d07 --- /dev/null +++ b/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/GroovyIntegrationFlowDefinition.groovy @@ -0,0 +1,1309 @@ +/* + * Copyright 2022 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 + * + * https://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.groovy.dsl + +import groovy.transform.CompileStatic +import groovy.transform.stc.ClosureParams +import groovy.transform.stc.SimpleType +import org.reactivestreams.Publisher +import org.springframework.integration.channel.BroadcastCapableChannel +import org.springframework.integration.channel.interceptor.WireTap +import org.springframework.integration.core.GenericHandler +import org.springframework.integration.core.GenericSelector +import org.springframework.integration.core.GenericTransformer +import org.springframework.integration.dsl.AggregatorSpec +import org.springframework.integration.dsl.BarrierSpec +import org.springframework.integration.dsl.BroadcastPublishSubscribeSpec +import org.springframework.integration.dsl.Channels +import org.springframework.integration.dsl.DelayerEndpointSpec +import org.springframework.integration.dsl.EnricherSpec +import org.springframework.integration.dsl.FilterEndpointSpec +import org.springframework.integration.dsl.GatewayEndpointSpec +import org.springframework.integration.dsl.GenericEndpointSpec +import org.springframework.integration.dsl.HeaderEnricherSpec +import org.springframework.integration.dsl.IntegrationFlow +import org.springframework.integration.dsl.IntegrationFlowDefinition +import org.springframework.integration.dsl.MessageChannelSpec +import org.springframework.integration.dsl.MessageHandlerSpec +import org.springframework.integration.dsl.MessageProcessorSpec +import org.springframework.integration.dsl.PublishSubscribeSpec +import org.springframework.integration.dsl.RecipientListRouterSpec +import org.springframework.integration.dsl.ResequencerSpec +import org.springframework.integration.dsl.RouterSpec +import org.springframework.integration.dsl.ScatterGatherSpec +import org.springframework.integration.dsl.SplitterEndpointSpec +import org.springframework.integration.dsl.WireTapSpec +import org.springframework.integration.filter.MethodInvokingSelector +import org.springframework.integration.handler.BridgeHandler +import org.springframework.integration.handler.LoggingHandler +import org.springframework.integration.handler.MessageProcessor +import org.springframework.integration.handler.MessageTriggerAction +import org.springframework.integration.handler.ServiceActivatingHandler +import org.springframework.integration.router.AbstractMessageRouter +import org.springframework.integration.router.ErrorMessageExceptionTypeRouter +import org.springframework.integration.router.ExpressionEvaluatingRouter +import org.springframework.integration.router.MethodInvokingRouter +import org.springframework.integration.scattergather.ScatterGatherHandler +import org.springframework.integration.splitter.AbstractMessageSplitter +import org.springframework.integration.splitter.DefaultMessageSplitter +import org.springframework.integration.splitter.ExpressionEvaluatingSplitter +import org.springframework.integration.splitter.MethodInvokingSplitter +import org.springframework.integration.store.MessageStore +import org.springframework.integration.transformer.ExpressionEvaluatingTransformer +import org.springframework.integration.transformer.HeaderFilter +import org.springframework.integration.transformer.MessageTransformingHandler +import org.springframework.integration.transformer.MethodInvokingTransformer +import org.springframework.messaging.Message +import org.springframework.messaging.MessageChannel +import org.springframework.messaging.MessageHandler +import org.springframework.messaging.support.ChannelInterceptor +import org.springframework.util.StringUtils +import reactor.core.publisher.Flux + +import java.util.concurrent.Executor +import java.util.function.Consumer +import java.util.function.Function + +/** + * The Groovy-specific {@link org.springframework.integration.dsl.IntegrationFlowDefinition} wrapper. + * + * @author Artem Bilan + * + * @since 6.0 + * + * @see org.springframework.integration.dsl.IntegrationFlowDefinition + */ +@CompileStatic +class GroovyIntegrationFlowDefinition { + + private final IntegrationFlowDefinition delegate + + protected GroovyIntegrationFlowDefinition(IntegrationFlowDefinition delegate) { + this.delegate = delegate + } + + /** + * Populate an {@link org.springframework.integration.channel.FixedSubscriberChannel} instance + * at the current {@link IntegrationFlow} chain position. + * The provided {@code messageChannelName} is used for the bean registration. + * @param messageChannelName the bean name to use. + */ + GroovyIntegrationFlowDefinition fixedSubscriberChannel(String messageChannelName = null) { + this.delegate.fixedSubscriberChannel messageChannelName + this + } + + /** + * Populate a {@link org.springframework.integration.dsl.support.MessageChannelReference} instance + * at the current {@link IntegrationFlow} chain position. + * The provided {@code messageChannelName} is used for the bean registration + * ({@link org.springframework.integration.channel.DirectChannel}), if there is no such a bean + * in the application context. Otherwise the existing {@link MessageChannel} bean is used + * to wire integration endpoints. + * @param messageChannelName the bean name to use. + */ + GroovyIntegrationFlowDefinition channel(String messageChannelName) { + this.delegate.channel messageChannelName + this + } + + /** + * Populate a {@link MessageChannel} instance + * at the current {@link IntegrationFlow} chain position using the {@link MessageChannelSpec} + * fluent API. + * @param messageChannelSpec the {@link MessageChannelSpec} to use. + * @see MessageChannels + */ + GroovyIntegrationFlowDefinition channel(MessageChannelSpec messageChannelSpec) { + this.delegate.channel messageChannelSpec + this + } + + /** + * Populate the provided {@link MessageChannel} instance + * at the current {@link IntegrationFlow} chain position. + * The {@code messageChannel} can be an existing bean, or fresh instance, in which case + * the {@link org.springframework.integration.dsl.context.IntegrationFlowBeanPostProcessor} + * will populate it as a bean with a generated name. + * @param messageChannel the {@link MessageChannel} to populate. + */ + GroovyIntegrationFlowDefinition channel(MessageChannel messageChannel) { + this.delegate.channel messageChannel + this + } + + /** + * Populate a {@link MessageChannel} instance + * at the current {@link IntegrationFlow} chain position using the {@link org.springframework.integration.dsl.Channels} + * factory fluent API. + * @param channels the {@link Function} to use. + */ + GroovyIntegrationFlowDefinition channel( + @DelegatesTo(value = Channels, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.Channels') + Closure channels) { + + Function function = + { channelFactory -> + channels.delegate = channelFactory + channels.resolveStrategy = Closure.DELEGATE_FIRST + channels(channelFactory) + } + + this.delegate.channel function + this + } + + /** + * The {@link org.springframework.integration.channel.PublishSubscribeChannel} {@link #channel} + * method specific implementation to allow the use of the 'subflow' subscriber capability. + * @param executor the {@link Executor} to use. + * @param publishSubscribeChannelConfigurer the {@link java.util.function.Consumer} to specify + * {@link org.springframework.integration.dsl.PublishSubscribeSpec} options including 'subflow' definition. + */ + GroovyIntegrationFlowDefinition publishSubscribeChannel( + Executor executor = null, + @DelegatesTo(value = PublishSubscribeSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, + options = 'org.springframework.integration.dsl.PublishSubscribeSpec') + Closure publishSubscribeChannelConfigurer) { + + this.delegate.publishSubscribeChannel executor, createConfigurerIfAny(publishSubscribeChannelConfigurer) + this + } + + /** + * The {@link BroadcastCapableChannel} {@link #channel} + * method specific implementation to allow the use of the 'subflow' subscriber capability. + * @param broadcastCapableChannel the {@link BroadcastCapableChannel} to subscriber sub-flows to. + * @param publishSubscribeChannelConfigurer the {@link Consumer} to specify + * {@link org.springframework.integration.dsl.BroadcastPublishSubscribeSpec} 'subflow' definitions. + */ + GroovyIntegrationFlowDefinition publishSubscribeChannel( + BroadcastCapableChannel broadcastCapableChannel, + @DelegatesTo(value = BroadcastPublishSubscribeSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, + options = 'org.springframework.integration.dsl.BroadcastPublishSubscribeSpec') + Closure publishSubscribeChannelConfigurer) { + + this.delegate.publishSubscribeChannel broadcastCapableChannel, + createConfigurerIfAny(publishSubscribeChannelConfigurer) + this + } + + /** + * Populate the {@code Wire Tap} EI Pattern specific + * {@link ChannelInterceptor} implementation + * to the current message channel. + * It is useful when an implicit {@link MessageChannel} is used between endpoints. + * This method can be used after any {@link #channel} for explicit {@link MessageChannel}, + * but with the caution do not impact existing {@link ChannelInterceptor}s. + * @param flow the {@link IntegrationFlow} for wire-tap subflow as an alternative to the {@code wireTapChannel}. + * @param wireTapConfigurer the {@link Consumer} to accept options for the + * {@link org.springframework.integration.channel.interceptor.WireTap}. + */ + GroovyIntegrationFlowDefinition wireTap( + IntegrationFlow flow, + @DelegatesTo(value = WireTapSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.WireTapSpec') + Closure wireTapConfigurer = null) { + + this.delegate.wireTap flow, createConfigurerIfAny(wireTapConfigurer) + this + } + + /** + * Populate the {@code Wire Tap} EI Pattern specific + * {@link ChannelInterceptor} implementation + * to the current message channel. + * It is useful when an implicit {@link MessageChannel} is used between endpoints. + * This method can be used after any {@link #channel} for explicit {@link MessageChannel}, + * but with the caution do not impact existing {@link ChannelInterceptor}s. + * @param wireTapChannel the {@link MessageChannel} bean name to wire-tap. + * @param wireTapConfigurer the {@link Consumer} to accept options for the + * {@link org.springframework.integration.channel.interceptor.WireTap}. + */ + GroovyIntegrationFlowDefinition wireTap( + String wireTapChannel, + @DelegatesTo(value = WireTapSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.WireTapSpec') + Closure wireTapConfigurer = null) { + + this.delegate.wireTap wireTapChannel, createConfigurerIfAny(wireTapConfigurer) + this + } + + /** + * Populate the {@code Wire Tap} EI Pattern specific + * {@link ChannelInterceptor} implementation + * to the current current MessageChannel. + * It is useful when an implicit {@link MessageChannel} is used between endpoints. + * This method can be used after any {@link #channel} for explicit {@link MessageChannel}, + * but with the caution do not impact existing {@link ChannelInterceptor}s. + * @param wireTapChannel the {@link MessageChannel} to wire-tap. + * @param wireTapConfigurer the {@link Consumer} to accept options for the + * {@link org.springframework.integration.channel.interceptor.WireTap}. + */ + GroovyIntegrationFlowDefinition wireTap( + MessageChannel wireTapChannel, + @DelegatesTo(value = WireTapSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.WireTapSpec') + Closure wireTapConfigurer = null) { + + this.delegate.wireTap wireTapChannel, createConfigurerIfAny(wireTapConfigurer) + this + } + + /** + * Populate the {@code Control Bus} EI Pattern specific {@link MessageHandler} implementation + * at the current {@link IntegrationFlow} chain position. + * @param endpointConfigurer the {@link Consumer} to accept integration endpoint options. + * @see org.springframework.integration.handler.ExpressionCommandMessageProcessor* @see GenericEndpointSpec + */ + GroovyIntegrationFlowDefinition controlBus( + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.controlBus createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@code Transformer} EI Pattern specific {@link MessageHandler} implementation + * for the SpEL {@link org.springframework.expression.Expression}. + * @param expression the {@code Transformer} {@link org.springframework.expression.Expression}. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @see org.springframework.integration.transformer.ExpressionEvaluatingTransformer + */ + GroovyIntegrationFlowDefinition transform( + String expression, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.transform expression, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@code MessageTransformingHandler} for the + * {@link org.springframework.integration.transformer.MethodInvokingTransformer} + * to invoke the service method at runtime. + * @param service the service to use. + * @param methodName the method to invoke. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @see ExpressionEvaluatingTransformer + */ + GroovyIntegrationFlowDefinition transform( + Object service, String methodName = null, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.transform service, methodName, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link MessageTransformingHandler} instance for the + * {@link org.springframework.integration.handler.MessageProcessor} from provided {@link MessageProcessorSpec}. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param messageProcessorSpec the {@link MessageProcessorSpec} to use. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @see MethodInvokingTransformer + */ + GroovyIntegrationFlowDefinition transform( + MessageProcessorSpec messageProcessorSpec, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.transform messageProcessorSpec, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link MessageTransformingHandler} instance + * for the provided {@code payloadType} to convert at runtime. + * In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param payloadType the {@link Class} for expected payload type. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param < P > the payload type - 'transform to'. + */ +

GroovyIntegrationFlowDefinition convert( + Class

payloadType, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.convert payloadType, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer} + * for the specific {@code expectedType} to convert at runtime. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param expectedType the {@link Class} for expected payload type. It can also be + * {@code Message.class} if you wish to access the entire message in the transformer. + * Conversion to this type will be attempted, if necessary. + * @param genericTransformer the {@link GenericTransformer} to populate. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param < P > the payload type - 'transform from', or {@code Message.class}. + * @param < T > the target type - 'transform to'. + */ + GroovyIntegrationFlowDefinition transform( + GenericTransformer genericTransformer, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.transform genericTransformer, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer} + * for the specific {@code expectedType} to convert at runtime. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param expectedType the {@link Class} for expected payload type. It can also be + * {@code Message.class} if you wish to access the entire message in the transformer. + * Conversion to this type will be attempted, if necessary. + * @param genericTransformer the {@link GenericTransformer} to populate. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param < P > the payload type - 'transform from', or {@code Message.class}. + * @param < T > the target type - 'transform to'. + */ + GroovyIntegrationFlowDefinition transform( + Class

expectedType, + GenericTransformer genericTransformer, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + GenericTransformer lambdaWrapper = payload -> genericTransformer(payload) + + this.delegate.transform expectedType, lambdaWrapper, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link org.springframework.integration.filter.MessageFilter} with + * {@link org.springframework.integration.core.MessageSelector} for the provided SpEL expression. + * In addition accept options for the integration endpoint using {@link FilterEndpointSpec} + * @param expression the SpEL expression. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @see FilterEndpointSpec + */ + GroovyIntegrationFlowDefinition filter( + String expression, + @DelegatesTo(value = FilterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.FilterEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.filter expression, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link org.springframework.integration.filter.MessageFilter} + * with {@link org.springframework.integration.filter.MethodInvokingSelector} for the + * method of the provided service. + * @param service the service to use. + * @param methodName the method to invoke + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + */ + GroovyIntegrationFlowDefinition filter( + Object service, String methodName = null, + @DelegatesTo(value = FilterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.FilterEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.filter service, methodName, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link org.springframework.integration.filter.MessageFilter} with {@link MethodInvokingSelector} + * for the {@link MessageProcessor} from + * the provided {@link MessageProcessorSpec}. + * @param messageProcessorSpec the {@link MessageProcessorSpec} to use. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + */ + GroovyIntegrationFlowDefinition filter( + MessageProcessorSpec messageProcessorSpec, + @DelegatesTo(value = FilterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.FilterEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.filter messageProcessorSpec, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link org.springframework.integration.filter.MessageFilter} with {@link MethodInvokingSelector} + * for the provided {@link GenericSelector}. + * In addition, accept options for the integration endpoint using {@link FilterEndpointSpec}. + * @param expectedType the {@link Class} for expected payload type. It can also be + * {@code Message.class} if you wish to access the entire message in the selector. + * Conversion to this type will be attempted, if necessary. + * @param genericSelector the {@link GenericSelector} to use. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param < P > the source payload type or {@code Message.class}. + */ +

GroovyIntegrationFlowDefinition filter( + @DelegatesTo.Target Class

expectedType, + @DelegatesTo(genericTypeIndex = 0, strategy = Closure.DELEGATE_FIRST) + Closure genericSelector, + @DelegatesTo(value = FilterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.FilterEndpointSpec') + Closure endpointConfigurer = null) { + + GenericSelector

lambdaWrapper = payload -> genericSelector(payload) + + this.delegate.filter expectedType, lambdaWrapper, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link ServiceActivatingHandler} for the + * {@link org.springframework.integration.handler.MethodInvokingMessageProcessor} + * to invoke the {@code method} for provided {@code bean} at runtime. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param beanName the bean name to use. + * @param methodName the method to invoke. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + */ + GroovyIntegrationFlowDefinition handle( + String beanName, String methodName, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.handle beanName, methodName, createConfigurerIfAny(endpointConfigurer) + this + } + + + /** + * Populate a {@link ServiceActivatingHandler} for the + * {@link org.springframework.integration.handler.MethodInvokingMessageProcessor} + * to invoke the {@code method} for provided {@code bean} at runtime. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param service the service object to use. + * @param methodName the method to invoke. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + */ + GroovyIntegrationFlowDefinition handle( + Object service, String methodName = null, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.handle service, methodName, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link ServiceActivatingHandler} for the + * {@link org.springframework.integration.handler.MethodInvokingMessageProcessor} + * to invoke the provided {@link GenericHandler} at runtime. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param expectedType the {@link Class} for expected payload type. It can also be + * {@code Message.class} if you wish to access the entire message in the handler. + * Conversion to this type will be attempted, if necessary. + * @param handler the handler to invoke. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param < P > the payload type to expect or {@code Message.class}. + */ +

GroovyIntegrationFlowDefinition handle( + Class

expectedType, GenericHandler

handler, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + GenericHandler

lambdaWrapper = (payload, headers) -> handler(payload, headers) + + this.delegate.handle expectedType, lambdaWrapper, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link ServiceActivatingHandler} for the + * {@link MessageProcessor} from the provided {@link MessageProcessorSpec}. + * @param messageProcessorSpec the {@link MessageProcessorSpec} to use. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + */ + GroovyIntegrationFlowDefinition handle( + MessageProcessorSpec messageProcessorSpec, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.handle messageProcessorSpec, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link ServiceActivatingHandler} for the selected protocol specific + * {@link MessageHandler} implementation from {@code Namespace Factory}: + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param messageHandlerSpec the {@link MessageHandlerSpec} to configure protocol specific + * {@link MessageHandler}. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param < H > the {@link MessageHandler} type. + */ + GroovyIntegrationFlowDefinition handle( + MessageHandlerSpec messageHandlerSpec, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.handle messageHandlerSpec, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link ServiceActivatingHandler} for the provided + * {@link MessageHandler} implementation. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param messageHandler the {@link MessageHandler} to use. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param < H > the {@link MessageHandler} type. + */ + GroovyIntegrationFlowDefinition handle( + H messageHandler, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.handle messageHandler, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link BridgeHandler} to the current integration flow position. + * sed with a Closure expression. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @see GenericEndpointSpec + */ + GroovyIntegrationFlowDefinition bridge( + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.bridge createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link org.springframework.integration.handler.DelayHandler} to the current integration flow position. + * @param groupId the {@code groupId} for delayed messages in the + * {@link org.springframework.integration.store.MessageGroupStore}. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @see org.springframework.integration.dsl.DelayerEndpointSpec + */ + GroovyIntegrationFlowDefinition delay( + String groupId, + @DelegatesTo(value = DelayerEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.DelayerEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.delay groupId, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link org.springframework.integration.transformer.ContentEnricher} + * to the current integration flow position + * with provided options. + * Used with a Closure expression. + * @param enricherConfigurer the {@link Closure} to provide + * {@link org.springframework.integration.transformer.ContentEnricher} options. + * @see org.springframework.integration.dsl.EnricherSpec + */ + GroovyIntegrationFlowDefinition enrich( + @DelegatesTo(value = EnricherSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.EnricherSpec') + Closure enricherConfigurer) { + + this.delegate.enrich createConfigurerIfAny(enricherConfigurer) + this + } + + /** + * Populate a {@link MessageTransformingHandler} for + * a {@link org.springframework.integration.transformer.HeaderEnricher} + * as the result of provided {@link Consumer}. + * @param headerEnricherConfigurer the {@link Consumer} to use. + * @see org.springframework.integration.dsl.HeaderEnricherSpec + */ + GroovyIntegrationFlowDefinition enrichHeaders( + @DelegatesTo(value = HeaderEnricherSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.HeaderEnricherSpec') + Closure enricherConfigurer = null) { + + this.delegate.enrichHeaders createConfigurerIfAny(enricherConfigurer) + this + } + + /** + * Populate the {@link DefaultMessageSplitter} with provided options + * to the current integration flow position. + * Used with a Closure expression (optional). + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options + * and for {@link DefaultMessageSplitter}. + * @see SplitterEndpointSpec + */ + GroovyIntegrationFlowDefinition split( + @DelegatesTo(value = SplitterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.SplitterEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.split createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link ExpressionEvaluatingSplitter} with provided + * SpEL expression. + * @param expression the splitter SpEL expression. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options + * and for {@link ExpressionEvaluatingSplitter}. + */ + GroovyIntegrationFlowDefinition split( + String expression, + @DelegatesTo(value = SplitterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.SplitterEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.split expression, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link MethodInvokingSplitter} to evaluate the provided + * {@code method} of the {@code bean} at runtime. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param service the service to use. + * @param methodName the method to invoke. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options + * and for {@link MethodInvokingSplitter}. + */ + GroovyIntegrationFlowDefinition split( + Object service, String methodName = null, + @DelegatesTo(value = SplitterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.SplitterEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.split service, methodName, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link MethodInvokingSplitter} to evaluate the provided + * {@code method} of the {@code bean} at runtime. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param beanName the bean name to use. + * @param methodName the method to invoke at runtime. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options + * and for {@link MethodInvokingSplitter}. + * @see org.springframework.integration.dsl.SplitterEndpointSpec + */ + GroovyIntegrationFlowDefinition split( + String beanName, String methodName, + @DelegatesTo(value = SplitterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.SplitterEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.split beanName, methodName, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link MethodInvokingSplitter} to evaluate the + * {@link MessageProcessor} at runtime + * from provided {@link MessageProcessorSpec}. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param messageProcessorSpec the splitter {@link MessageProcessorSpec}. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options + * and for {@link MethodInvokingSplitter}. + */ + GroovyIntegrationFlowDefinition split( + MessageProcessorSpec messageProcessorSpec, + @DelegatesTo(value = SplitterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.SplitterEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.split messageProcessorSpec, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link MethodInvokingSplitter} to evaluate the provided + * {@link Function} at runtime. + * In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param expectedType the {@link Class} for expected payload type. It can also be + * {@code Message.class} if you wish to access the entire message in the splitter. + * Conversion to this type will be attempted, if necessary. + * @param splitter the splitter {@link Function}. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param < P > the payload type or {@code Message.class}. + */ +

GroovyIntegrationFlowDefinition split( + Class

expectedType, Function splitter, + @DelegatesTo(value = SplitterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.SplitterEndpointSpec') + Closure endpointConfigurer = null) { + + Function lambdaWrapper = payload -> splitter(payload) + + this.delegate.split expectedType, lambdaWrapper, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the provided {@link AbstractMessageSplitter} to the current integration + * flow position. + * @param splitterMessageHandlerSpec the {@link MessageHandlerSpec} to populate. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param < S > the {@link AbstractMessageSplitter} + * @see org.springframework.integration.dsl.SplitterEndpointSpec + */ + GroovyIntegrationFlowDefinition split( + MessageHandlerSpec splitterMessageHandlerSpec, + @DelegatesTo(value = SplitterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.SplitterEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.split splitterMessageHandlerSpec, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the provided {@link AbstractMessageSplitter} to the current integration + * flow position. + * @param splitter the {@link AbstractMessageSplitter} to populate. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param < S > the {@link AbstractMessageSplitter} + * @see org.springframework.integration.dsl.SplitterEndpointSpec + */ + GroovyIntegrationFlowDefinition split( + S splitter, + @DelegatesTo(value = SplitterEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.SplitterEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.split splitter, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the provided {@link MessageTransformingHandler} for the provided + * {@link HeaderFilter}. + * @param headerFilter the {@link HeaderFilter} to use. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @see GenericEndpointSpec + */ + GroovyIntegrationFlowDefinition headerFilter( + String headersToRemove, + boolean patternMatch = true, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + def headerFilter = new HeaderFilter(StringUtils.delimitedListToStringArray(headersToRemove, ',', ' ')) + headerFilter.patternMatch = patternMatch + + this.delegate.headerFilter headerFilter, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link MessageTransformingHandler} for the + * {@link org.springframework.integration.transformer.ClaimCheckInTransformer} with provided {@link MessageStore}. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param messageStore the {@link MessageStore} to use. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @see GenericEndpointSpec + */ + GroovyIntegrationFlowDefinition claimCheckIn( + MessageStore messageStore, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.claimCheckIn messageStore, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the {@link MessageTransformingHandler} for the + * {@link org.springframework.integration.transformer.ClaimCheckOutTransformer} + * with provided {@link MessageStore} and {@code removeMessage} flag. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param messageStore the {@link MessageStore} to use. + * @param removeMessage the removeMessage boolean flag. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @see GenericEndpointSpec* @see org.springframework.integration.transformer.ClaimCheckOutTransformer#setRemoveMessage(boolean) + */ + GroovyIntegrationFlowDefinition claimCheckOut( + MessageStore messageStore, boolean removeMessage = false, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.claimCheckOut messageStore, removeMessage, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the + * {@link org.springframework.integration.aggregator.ResequencingMessageHandler} with + * provided options from {@link org.springframework.integration.dsl.ResequencerSpec}. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * Used with a Closure expression (optional). + * @param resequencer the {@link Consumer} to provide + * {@link org.springframework.integration.aggregator.ResequencingMessageHandler} options. + * @see org.springframework.integration.dsl.ResequencerSpec + */ + GroovyIntegrationFlowDefinition resequence( + @DelegatesTo(value = ResequencerSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.ResequencerSpec') + Closure resequencer = null) { + + this.delegate.resequence createConfigurerIfAny(resequencer) + this + } + + /** + * A short-cut for the {@code aggregate((aggregator) -> aggregator.processor(aggregatorProcessor))}. + * @param aggregatorProcessor the POJO representing aggregation strategies. + * @see org.springframework.integration.dsl.AggregatorSpec + */ + GroovyIntegrationFlowDefinition aggregate(Object aggregatorProcessor) { + this.delegate.aggregate aggregatorProcessor + this + } + + /** + * Populate the {@link org.springframework.integration.aggregator.AggregatingMessageHandler} + * with provided options from {@link org.springframework.integration.dsl.AggregatorSpec}. + * In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}. + * Used with a Closure expression (optional). + * @param aggregator the {@link Consumer} to provide + * {@link org.springframework.integration.aggregator.AggregatingMessageHandler} options. + */ + GroovyIntegrationFlowDefinition aggregate( + @DelegatesTo(value = AggregatorSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.AggregatorSpec') + Closure aggregator = null) { + + this.delegate.aggregate createConfigurerIfAny(aggregator) + this + } + + /** + * Populate the {@link MethodInvokingRouter} for provided bean and its method + * with provided options from {@link org.springframework.integration.dsl.RouterSpec}. + * @param beanName the bean to use. + * @param method the method to invoke at runtime. + * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. + */ + GroovyIntegrationFlowDefinition route( + String beanName, String method, + @DelegatesTo(value = RouterSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.RouterSpec') + Closure routerConfigurer = null) { + + this.delegate.route beanName, method, createConfigurerIfAny(routerConfigurer) + this + } + + /** + * Populate the {@link MethodInvokingRouter} for the method + * of the provided service and its method with provided options from {@link org.springframework.integration.dsl.RouterSpec}. + * @param service the service to use. + * @param methodName the method to invoke. + * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. + */ + GroovyIntegrationFlowDefinition route( + Object service, String methodName = null, + @DelegatesTo(value = RouterSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.RouterSpec') + Closure routerConfigurer = null) { + + this.delegate.route service, methodName, createConfigurerIfAny(routerConfigurer) + this + } + + /** + * Populate the {@link ExpressionEvaluatingRouter} for provided SpEL expression + * with provided options from {@link org.springframework.integration.dsl.RouterSpec}. + * @param expression the expression to use. + * @param routerConfigurer the {@link Consumer} to provide {@link ExpressionEvaluatingRouter} options. + * @param < T > the target result type. + */ + GroovyIntegrationFlowDefinition route( + String expression, + @DelegatesTo(value = RouterSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.RouterSpec') + Closure routerConfigurer = null) { + + this.delegate.route expression, createConfigurerIfAny(routerConfigurer) + this + } + + /** + * Populate the {@link MethodInvokingRouter} for provided {@link Function} + * and payload type and options from {@link org.springframework.integration.dsl.RouterSpec}. + * In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param expectedType the {@link Class} for expected payload type. It can also be + * {@code Message.class} if you wish to access the entire message in the router. + * Conversion to this type will be attempted, if necessary. + * @param router the {@link Function} to use. + * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. + * @param < P > the source payload type or {@code Message.class}. + * @param < T > the target result type. + */ + GroovyIntegrationFlowDefinition route( + Class

expectedType, Function router, + @DelegatesTo(value = RouterSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.RouterSpec') + Closure routerConfigurer = null) { + + Function lambdaWrapper = payload -> router(payload) + + this.delegate.route expectedType, lambdaWrapper, createConfigurerIfAny(routerConfigurer) + this + } + + /** + * Populate the {@link MethodInvokingRouter} for the + * {@link MessageProcessor} + * from the provided {@link MessageProcessorSpec} with default options. + * @param messageProcessorSpec the {@link MessageProcessorSpec} to use. + * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. + */ + GroovyIntegrationFlowDefinition route( + MessageProcessorSpec messageProcessorSpec, + @DelegatesTo(value = RouterSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.RouterSpec') + Closure routerConfigurer = null) { + + this.delegate.route messageProcessorSpec, createConfigurerIfAny(routerConfigurer) + this + } + + /** + * Populate the {@link org.springframework.integration.router.RecipientListRouter} + * with options from the {@link org.springframework.integration.dsl.RecipientListRouterSpec}. + * Typically used with a Closure expression. + * @param routerConfigurer the {@link Consumer} to provide + * {@link org.springframework.integration.router.RecipientListRouter} options. + */ + GroovyIntegrationFlowDefinition routeToRecipients( + @DelegatesTo(value = RecipientListRouterSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, + options = 'org.springframework.integration.dsl.RecipientListRouterSpec') + Closure routerConfigurer = null) { + + this.delegate.routeToRecipients createConfigurerIfAny(routerConfigurer) + this + } + + /** + * Populate the {@link ErrorMessageExceptionTypeRouter} with options from the {@link RouterSpec}. + * Typically used with a Closure expression. + * @param routerConfigurer the {@link Consumer} to provide {@link ErrorMessageExceptionTypeRouter} options. + * @see ErrorMessageExceptionTypeRouter + */ + GroovyIntegrationFlowDefinition routeByException( + @DelegatesTo(value = RouterSpec, ErrorMessageExceptionTypeRouter>, + strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.RouterSpec') + Closure routerConfigurer = null) { + + this.delegate.routeByException createConfigurerIfAny(routerConfigurer) + this + } + + /** + * Populate the provided {@link AbstractMessageRouter} implementation to the + * current integration flow position. + * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. + * @param router the {@link AbstractMessageRouter} to populate. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + * @param the {@link AbstractMessageRouter} type. + */ + GroovyIntegrationFlowDefinition route( + R router, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.route router, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the "artificial" + * {@link org.springframework.integration.gateway.GatewayMessageHandler} for the + * provided {@code requestChannel} to send a request with options from + * {@link org.springframework.integration.dsl.GatewayEndpointSpec}. Uses + * {@link org.springframework.integration.gateway.RequestReplyExchanger} Proxy on the + * background. + * @param requestChannel the {@link MessageChannel} bean name. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint + * options. + */ + GroovyIntegrationFlowDefinition gateway( + String requestChannel, + @DelegatesTo(value = GatewayEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GatewayEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.gateway requestChannel, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the "artificial" + * {@link org.springframework.integration.gateway.GatewayMessageHandler} for the + * provided {@code requestChannel} to send a request with options from + * {@link GatewayEndpointSpec}. Uses + * {@link org.springframework.integration.gateway.RequestReplyExchanger} Proxy on the + * background. + * @param requestChannel the {@link MessageChannel} to use. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint + * options. + */ + GroovyIntegrationFlowDefinition gateway( + MessageChannel requestChannel, + @DelegatesTo(value = GatewayEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GatewayEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.gateway requestChannel, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate the "artificial" + * {@link org.springframework.integration.gateway.GatewayMessageHandler} for the + * provided {@code subflow} with options from {@link GatewayEndpointSpec}. + * Typically used with a Closure expression. + * @param flow the {@link IntegrationFlow} to to send a request message and wait for reply. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + */ + GroovyIntegrationFlowDefinition gateway( + IntegrationFlow flow, + @DelegatesTo(value = GatewayEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.GatewayEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.gateway flow, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link WireTap} for the current message channel + * with the {@link LoggingHandler} subscriber for provided {@link LoggingHandler.Level} + * logging level and {@code org.springframework.integration.handler.LoggingHandler} + * as a default logging category. + *

The full request {@link Message} will be logged. + *

When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * @param level the {@link LoggingHandler.Level}. + */ + GroovyIntegrationFlowDefinition log(LoggingHandler.Level level = LoggingHandler.Level.INFO, + String category, String logExpression = null) { + + if (logExpression) { + this.delegate.log level, category, logExpression + } else { + this.delegate.log level, category + } + this + } + + /** + * Populate a {@link WireTap} for the current message channel + * with the {@link LoggingHandler} subscriber for the provided + * {@link LoggingHandler.Level} logging level, logging category + * and {@link Function} for the log message. + *

When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * @param level the {@link LoggingHandler.Level}. + * @param category the logging category. + * @param function the function to evaluate logger message at runtime + * @param < P > the expected payload type against the request {@link Message}. + */ +

GroovyIntegrationFlowDefinition log( + LoggingHandler.Level level = LoggingHandler.Level.INFO, + String category = null, + Function, Object> function) { + + this.delegate.log level, category, function + this + } + + /** + * Populate a {@link org.springframework.integration.scattergather.ScatterGatherHandler} + * to the current integration flow position + * based on the provided {@link MessageChannel} for scattering function + * and {@link AggregatorSpec} for gathering function. + * @param scatterChannel the {@link MessageChannel} for scatting requests. + * @param gatherer the {@link Consumer} for {@link AggregatorSpec} to configure gatherer. + * Can be {@code null}. + * @param scatterGather the {@link Consumer} for {@link ScatterGatherSpec} to configure + * {@link org.springframework.integration.scattergather.ScatterGatherHandler} and its endpoint. Can be {@code null}. + */ + GroovyIntegrationFlowDefinition scatterGather( + MessageChannel scatterChannel, + @DelegatesTo(value = AggregatorSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.AggregatorSpec') + Closure gatherer = null, + @DelegatesTo(value = ScatterGatherSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.ScatterGatherSpec') + Closure scatterGather = null) { + + this.delegate.scatterGather scatterChannel, createConfigurerIfAny(gatherer), + createConfigurerIfAny(scatterGather) + this + } + + /** + * Populate a {@link ScatterGatherHandler} to the current integration flow position + * based on the provided {@link RecipientListRouterSpec} for scattering function + * and {@link AggregatorSpec} for gathering function. + * @param scatterer the {@link Consumer} for {@link RecipientListRouterSpec} to configure scatterer. + * @param gatherer the {@link Consumer} for {@link AggregatorSpec} to configure gatherer. + * @param scatterGather the {@link Consumer} for {@link ScatterGatherSpec} to configure + * {@link ScatterGatherHandler} and its endpoint. Can be {@code null}. + */ + GroovyIntegrationFlowDefinition scatterGather( + @DelegatesTo(value = RecipientListRouterSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, + options = 'org.springframework.integration.dsl.RecipientListRouterSpec') + Closure scatterer, + @DelegatesTo(value = AggregatorSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.AggregatorSpec') + Closure gatherer = null, + @DelegatesTo(value = ScatterGatherSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.ScatterGatherSpec') + Closure scatterGather = null) { + + this.delegate.scatterGather createConfigurerIfAny(scatterer), createConfigurerIfAny(gatherer), + createConfigurerIfAny(scatterGather) + this + } + + /** + * Populate a {@link org.springframework.integration.aggregator.BarrierMessageHandler} + * instance for provided timeout and options from {@link org.springframework.integration.dsl.BarrierSpec} and endpoint + * options from {@link GenericEndpointSpec}. + * @param timeout the timeout in milliseconds. + * @param barrierConfigurer the {@link Consumer} to provide + * {@link org.springframework.integration.aggregator.BarrierMessageHandler} options. + */ + GroovyIntegrationFlowDefinition barrier(long timeout, + @DelegatesTo(value = BarrierSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, + options = 'org.springframework.integration.dsl.BarrierSpec') + Closure barrierConfigurer = null) { + + this.delegate.barrier timeout, createConfigurerIfAny(barrierConfigurer) + this + } + + /** + * Populate a {@link ServiceActivatingHandler} instance to perform {@link MessageTriggerAction} + * and endpoint options from {@link GenericEndpointSpec}. + * @param triggerActionId the {@link MessageTriggerAction} bean id. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + */ + GroovyIntegrationFlowDefinition trigger( + String triggerActionId, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, + options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.trigger triggerActionId, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Populate a {@link ServiceActivatingHandler} instance to perform {@link MessageTriggerAction} + * and endpoint options from {@link GenericEndpointSpec}. + * @param triggerAction the {@link MessageTriggerAction}. + * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. + */ + GroovyIntegrationFlowDefinition trigger( + MessageTriggerAction triggerAction, + @DelegatesTo(value = GenericEndpointSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType.class, + options = 'org.springframework.integration.dsl.GenericEndpointSpec') + Closure endpointConfigurer = null) { + + this.delegate.trigger triggerAction, createConfigurerIfAny(endpointConfigurer) + this + } + + /** + * Add one or more {@link ChannelInterceptor} implementations + * to the current {@link MessageChannel}, in the given order, after any interceptors already registered. + * @param interceptorArray one or more {@link ChannelInterceptor}s. + */ + GroovyIntegrationFlowDefinition intercept(ChannelInterceptor... interceptorArray) { + this.delegate.intercept interceptorArray + this + } + + /** + * Populate a {@link org.springframework.integration.channel.FluxMessageChannel} + * to start a reactive processing for upstream data, + * wrap it to a {@link Flux}, apply provided {@link Function} via {@link Flux#transform(Function)} + * and emit the result to one more + * {@link org.springframework.integration.channel.FluxMessageChannel}, subscribed in the downstream flow. + * @param fluxFunction the {@link Function} to process data reactive manner. + * @param < I > the input payload type. + * @param < O > the output type. + */ + GroovyIntegrationFlowDefinition fluxTransform( + Function>, ? extends Publisher> fluxFunction) { + + this.delegate. fluxTransform fluxFunction + this + } + + protected static Consumer createConfigurerIfAny( + @DelegatesTo(strategy = Closure.DELEGATE_FIRST) Closure closure) { + + if (closure) { + return { + closure.delegate = it + closure.resolveStrategy = Closure.DELEGATE_FIRST + closure() + } as Consumer + } + null + } + +} diff --git a/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/IntegrationGroovyDsl.groovy b/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/IntegrationGroovyDsl.groovy new file mode 100644 index 0000000000..16f0bcb6c8 --- /dev/null +++ b/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/IntegrationGroovyDsl.groovy @@ -0,0 +1,270 @@ +/* + * Copyright 2022 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 + * + * https://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.groovy.dsl + +import groovy.transform.CompileStatic +import groovy.transform.stc.ClosureParams +import groovy.transform.stc.SimpleType +import org.reactivestreams.Publisher +import org.springframework.integration.core.MessageSource +import org.springframework.integration.dsl.GatewayProxySpec +import org.springframework.integration.dsl.IntegrationFlow +import org.springframework.integration.dsl.IntegrationFlowBuilder +import org.springframework.integration.dsl.IntegrationFlowDefinition +import org.springframework.integration.dsl.MessageProducerSpec +import org.springframework.integration.dsl.MessageSourceSpec +import org.springframework.integration.dsl.MessagingGatewaySpec +import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec +import org.springframework.integration.endpoint.MessageProducerSupport +import org.springframework.integration.gateway.MessagingGatewaySupport +import org.springframework.messaging.Message +import org.springframework.messaging.MessageChannel + +import java.util.function.Consumer +import java.util.function.Supplier + +/** + * The factory class for Spring Integration Groovy DSL closures. + * + * @author Artem Bilan + * + * @since 6.0 + */ +@CompileStatic +class IntegrationGroovyDsl { + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL. + * @param flow the {@link Closure} for {@link IntegrationFlowDefinition} + */ + static IntegrationFlow integrationFlow( + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + { IntegrationFlowDefinition flowDefinition -> + def delegate = new GroovyIntegrationFlowDefinition(flowDefinition) + flow.delegate = delegate + flow.resolveStrategy = Closure.DELEGATE_FIRST + flow(delegate) + } as IntegrationFlow + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(Class, Consumer)} factory method. + */ + static IntegrationFlow integrationFlow( + Class serviceInterface, + @DelegatesTo(value = GatewayProxySpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.dsl.GatewayProxySpec') + Closure gatewaySpec = null, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + Consumer configurer = GroovyIntegrationFlowDefinition.createConfigurerIfAny(gatewaySpec) + buildIntegrationFlow(IntegrationFlow.from(serviceInterface, configurer), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(String, boolean)} factory method. + */ + static IntegrationFlow integrationFlow( + String channelName, + Boolean fixedSubscriber = false, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + buildIntegrationFlow(IntegrationFlow.from(channelName, fixedSubscriber), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(MessageChannel)} factory method. + */ + static IntegrationFlow integrationFlow( + MessageChannel channel, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + buildIntegrationFlow(IntegrationFlow.from(channel), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(MessageSource, Consumer)} factory method. + */ + static IntegrationFlow integrationFlow( + MessageSource messageSource, + @DelegatesTo(value = SourcePollingChannelAdapterSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = "org.springframework.integration.dsl.SourcePollingChannelAdapterSpec") + Closure adapterSpec = null, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + Consumer configurer = + GroovyIntegrationFlowDefinition.createConfigurerIfAny(adapterSpec) + buildIntegrationFlow(IntegrationFlow.from(messageSource, configurer), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(org.springframework.integration.dsl.MessageSourceSpec, Consumer)} factory method. + */ + static IntegrationFlow integrationFlow( + MessageSourceSpec messageSourceSpec, + @DelegatesTo(value = SourcePollingChannelAdapterSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = "org.springframework.integration.dsl.SourcePollingChannelAdapterSpec") + Closure adapterSpec = null, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + Consumer configurer = + GroovyIntegrationFlowDefinition.createConfigurerIfAny(adapterSpec) + buildIntegrationFlow(IntegrationFlow.from(messageSourceSpec, configurer), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#fromSupplier(Supplier, Consumer)} factory method. + */ + static IntegrationFlow integrationFlow( + Closure source, + @DelegatesTo(value = SourcePollingChannelAdapterSpec, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.dsl.SourcePollingChannelAdapterSpec') + Closure adapterSpec = null, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + Consumer configurer = + GroovyIntegrationFlowDefinition.createConfigurerIfAny(adapterSpec) + buildIntegrationFlow(IntegrationFlow.fromSupplier(source, configurer), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(Publisher)} factory method. + */ + static IntegrationFlow integrationFlow( + Publisher> publisher, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + buildIntegrationFlow(IntegrationFlow.from(publisher), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(MessagingGatewaySupport)} factory method. + */ + static IntegrationFlow integrationFlow( + MessagingGatewaySupport gateway, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + buildIntegrationFlow(IntegrationFlow.from(gateway), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(org.springframework.integration.dsl.MessagingGatewaySpec)} factory method. + */ + static IntegrationFlow integrationFlow( + MessagingGatewaySpec gatewaySpec, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + buildIntegrationFlow(IntegrationFlow.from(gatewaySpec), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(MessageProducerSupport)} factory method. + */ + static IntegrationFlow integrationFlow( + MessageProducerSupport producer, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + buildIntegrationFlow(IntegrationFlow.from(producer), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(org.springframework.integration.dsl.MessageProducerSpec)} factory method. + */ + static IntegrationFlow integrationFlow( + MessageProducerSpec producerSpec, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + buildIntegrationFlow(IntegrationFlow.from(producerSpec), flow) + } + + /** + * Functional {@link IntegrationFlow} definition in Groovy DSL for + * {@link IntegrationFlow#from(IntegrationFlow)} factory method. + */ + static IntegrationFlow integrationFlow( + IntegrationFlow sourceFlow, + @DelegatesTo(value = GroovyIntegrationFlowDefinition, strategy = Closure.DELEGATE_FIRST) + @ClosureParams(value = SimpleType, + options = 'org.springframework.integration.groovy.dsl.GroovyIntegrationFlowDefinition') + Closure flow) { + + buildIntegrationFlow(IntegrationFlow.from(sourceFlow), flow) + } + + private static IntegrationFlow buildIntegrationFlow(IntegrationFlowBuilder flowBuilder, Closure flow) { + flow.delegate = new GroovyIntegrationFlowDefinition(flowBuilder) + flow.resolveStrategy = Closure.DELEGATE_FIRST + flow() + flowBuilder.get() + } + + private IntegrationGroovyDsl() { + } + +} diff --git a/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/package-info.groovy b/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/package-info.groovy new file mode 100644 index 0000000000..df20b89dcd --- /dev/null +++ b/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/package-info.groovy @@ -0,0 +1,4 @@ +/** + * Provides Spring Integration Groovy DSL. + */ +package org.springframework.integration.groovy.dsl diff --git a/spring-integration-groovy/src/test/groovy/org/springframework/integration/groovy/dsl/test/GroovyDslTests.groovy b/spring-integration-groovy/src/test/groovy/org/springframework/integration/groovy/dsl/test/GroovyDslTests.groovy new file mode 100644 index 0000000000..b6431c1a75 --- /dev/null +++ b/spring-integration-groovy/src/test/groovy/org/springframework/integration/groovy/dsl/test/GroovyDslTests.groovy @@ -0,0 +1,306 @@ +/* + * Copyright 2022 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 + * + * https://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.groovy.dsl.test + + +import org.junit.jupiter.api.Test +import org.springframework.beans.factory.BeanFactory +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.beans.factory.annotation.Qualifier +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import org.springframework.integration.channel.FluxMessageChannel +import org.springframework.integration.channel.QueueChannel +import org.springframework.integration.config.EnableIntegration +import org.springframework.integration.dsl.IntegrationFlow +import org.springframework.integration.dsl.IntegrationFlowDefinition +import org.springframework.integration.dsl.Pollers +import org.springframework.integration.dsl.Transformers +import org.springframework.integration.dsl.context.IntegrationFlowContext +import org.springframework.integration.handler.LoggingHandler +import org.springframework.integration.scheduling.PollerMetadata +import org.springframework.integration.support.MessageBuilder +import org.springframework.messaging.Message +import org.springframework.messaging.MessageChannel +import org.springframework.messaging.MessageHeaders +import org.springframework.messaging.PollableChannel +import org.springframework.messaging.support.GenericMessage +import org.springframework.test.annotation.DirtiesContext +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig +import reactor.core.publisher.Flux +import reactor.test.StepVerifier + +import java.time.Duration +import java.util.function.Function + +import static org.springframework.integration.groovy.dsl.IntegrationGroovyDsl.integrationFlow + +/** + * @author Artem Bilan + * + * @since 6.0 + */ +@SpringJUnitConfig +@DirtiesContext +class GroovyDslTests { + + @Autowired + private BeanFactory beanFactory + + @Autowired + private IntegrationFlowContext integrationFlowContext + + @Autowired + private PollableChannel pollerResultChannel + + @Autowired + @Qualifier('requestReplyFlow.input') + private MessageChannel requestReplyFlowInput + + @Autowired + private MessageChannel requestReplyFixedFlowInput + + @Autowired + @Qualifier('functionGateway') + private Function upperCaseFunction + + @Test + void 'when application starts, it emits message to pollerResultChannel'() { + assert this.pollerResultChannel.receive(10000) != null + assert this.pollerResultChannel.receive(10000) != null + } + + @Test + void 'requestReplyFlow has to reply'() { + def replyChannel = new QueueChannel() + def testMessage = + MessageBuilder.withPayload('hello') + .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) + .build() + + this.requestReplyFlowInput.send(testMessage) + + assert replyChannel.receive(1000).payload == 'HELLO' + } + + @Test + void 'requestReplyFixedFlow has to reply'() { + def replyChannel = new QueueChannel() + def testMessage = + MessageBuilder.withPayload(4) + .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) + .build() + + this.requestReplyFixedFlowInput.send(testMessage) + + assert replyChannel.receive(1000).payload == 16 + } + + @Test + void 'uppercase function'() { + assert this.upperCaseFunction.apply('test'.bytes) == 'TEST' + } + + @Test + void 'reactive publisher flow'() { + def fluxChannel = new FluxMessageChannel() + + def verifyLater = + StepVerifier + .create(Flux.from(fluxChannel).map { it.payload }) + .expectNext(4, 6) + .thenCancel() + .verifyLater() + + def publisher = Flux.just(2, 3).map { new GenericMessage<>(it) } + + def integrationFlow = + integrationFlow(publisher) + { + transform Message, { it.payload * 2 }, { id 'foo' } + channel fluxChannel + } + + def registration = this.integrationFlowContext.registration(integrationFlow).register() + + verifyLater.verify(Duration.ofSeconds(10)) + + registration.destroy() + } + + @Autowired + @Qualifier('scatterGatherFlow.input') + private MessageChannel scatterGatherFlowInput + + @Test + void 'Scatter-Gather'() { + def replyChannel = new QueueChannel() + def request = + MessageBuilder.withPayload("foo") + .setReplyChannel(replyChannel) + .build() + + this.scatterGatherFlowInput.send(request) + + def bestQuoteMessage = replyChannel.receive(10000) + assert (bestQuoteMessage?.payload as List).size() >= 1 + } + + @Autowired + @Qualifier('oddFlow.input') + private MessageChannel oddFlowInput + + @Test + void 'oddFlow must reply'() { + def replyChannel = new QueueChannel() + def testMessage = + MessageBuilder.withPayload('test') + .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) + .build() + + this.oddFlowInput.send(testMessage) + + assert replyChannel.receive(1000).payload == 'odd' + } + + + @Autowired + @Qualifier('flowLambda.input') + private MessageChannel flowLambdaInput + + @Autowired + private PollableChannel wireTapChannel + + @Test + void 'flow from lambda'() { + def replyChannel = new QueueChannel() + def message = MessageBuilder.withPayload('test').setReplyChannel(replyChannel).build() + + this.flowLambdaInput.send message + + assert replyChannel.receive(10_000)?.payload == 'TEST' + assert this.wireTapChannel.receive(10_000)?.payload == 'test' + } + + @Configuration + @EnableIntegration + static class Config { + + @Bean(PollerMetadata.DEFAULT_POLLER) + poller() { + Pollers.fixedDelay(1000).get() + } + + + @Bean + someFlow() { + integrationFlow { 'test' } + { + log LoggingHandler.Level.WARN, 'test.category' + channel { queue 'pollerResultChannel' } + } + } + + @Bean + requestReplyFlow() { + integrationFlow { + fluxTransform { it.map { it } } + transform String, { it.toUpperCase() } + } + } + + @Bean + requestReplyFixedFlow() { + integrationFlow 'requestReplyFixedFlowInput', true, + { + handle Integer, { p, h -> p**2 } + } + } + + @Bean + functionFlow() { + integrationFlow Function, + { beanName 'functionGateway' }, + { + transform Transformers.objectToString(), { id 'objectToStringTransformer' } + transform String, { it.toUpperCase() } + split Message, { it.payload } + split Object, { it }, { id 'splitterEndpoint' } + resequence() + aggregate { + id 'aggregator' + outputProcessor { it.one } + } + } + } + + @Bean + scatterGatherFlow() { + integrationFlow { + scatterGather( + { + applySequence true + recipientFlow({ true }, recipientSubFlow()) + recipientFlow({ true }, + integrationFlow { handle Void, { p, h -> Math.random() * 10 } }) + recipientFlow({ true }, + integrationFlow { handle Void, { p, h -> Math.random() * 10 } }) + }, + { + releaseStrategy { + it.size() == 3 || it.messages.any { it.payload as Double > 5 } + } + }) + { + gatherTimeout 10_000 + } + } + } + + static recipientSubFlow() { + integrationFlow { handle Void, { p, h -> Math.random() * 10 } } + } + + @Bean + IntegrationFlow oddFlow() { + { IntegrationFlowDefinition flow -> + flow.handle(Object, { p, h -> 'odd' }) + } + } + + @Bean + flowLambda() { + integrationFlow { + filter String, { it == 'test' }, { id 'filterEndpoint' } + wireTap integrationFlow { + channel { queue 'wireTapChannel' } + } + delay 'delayGroup', { defaultDelay 100 } + transform String, { it.toUpperCase() } + } + } + + @Bean + flowFromSupplier() { + integrationFlow({ 'bar' }, { poller { it.fixedDelay(10).maxMessagesPerPoll(1) } }) { + channel { queue 'fromSupplierQueue' } + } + } + + } + +} diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 6a50748ece..744926b1d8 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -4,6 +4,7 @@ The Spring Integration Java configuration and DSL provides a set of convenient builders and a fluent API that lets you configure Spring Integration message flows from Spring `@Configuration` classes. (See also <<./kotlin-dsl.adoc#kotlin-dsl,Kotlin DSL>>.) +(See also <<./groovy-dsl.adoc#groovy-dsl,Groovy DSL>>.) The Java DSL for Spring Integration is essentially a facade for Spring Integration. The DSL provides a simple way to embed Spring Integration Message Flows into your application by using the fluent `Builder` pattern together with existing Java configuration from Spring Framework and Spring Integration. diff --git a/src/reference/asciidoc/groovy-dsl.adoc b/src/reference/asciidoc/groovy-dsl.adoc new file mode 100644 index 0000000000..9c2aa9bf22 --- /dev/null +++ b/src/reference/asciidoc/groovy-dsl.adoc @@ -0,0 +1,103 @@ +[[groovy-dsl]] +== Groovy DSL + +The Groovy DSL is a wrapper and extension to <<./dsl.adoc#java-dsl,Java DSL>>. +The main goal we pursue here is to make Spring Integration development on Groovy as smooth and straightforward as is it possible with interoperability with existing Java DSL and some Groovy extensions or language-specific structures. +The implementation is a part of <<./groovy.adoc#groovy,Groovy Support>> module. + +All you need to get started is just an import for `import static org.springframework.integration.groovy.dsl.IntegrationGroovyDsl.integrationFlow` - a class containing overloaded factory methods for the Groovy DSL. + +For `IntegrationFlow` definitions as lambdas we typically don't need anything else from Groovy and just declare a bean like this: + +==== +[source, groovy] +---- +@Bean +IntegrationFlow oddFlow() { + { IntegrationFlowDefinition flow -> + flow.handle(Object, { p, h -> 'odd' }) + } +} +---- +==== + +In this case Groovy understands that the closure should be translated into an `IntegrationFlow` anonymous instance and the target Java DSL processor parses this construction properly into Java objects. + +As an alternative to the construction above and for consistency with use-cases explained below, the `spring-integration-groovy` module provides a Groovy-specific DSL for declaring integration flows in a *builder* pattern style: + +==== +[source, groovy] +---- +@Bean +flowLambda() { + integrationFlow { + filter String, { it == 'test' }, { id 'filterEndpoint' } + wireTap integrationFlow { + channel { queue 'wireTapChannel' } + } + delay 'delayGroup', { defaultDelay 100 } + transform String, { it.toUpperCase() } + } +} +---- +==== + +Such a global `integrationFlow()` function expects a closure in the builder style for a `GroovyIntegrationFlowDefinition` (a Groovy wrapper for the `IntegrationFlowDefinition`) and produces a regular `IntegrationFlow` lambda implementation. +See more overloaded `integrationFlow()` variants below. + +Many other scenarios require an `IntegrationFlow` to be started from the source of data (e.g. `JdbcPollingChannelAdapter`, `JmsInboundGateway` or just an existing `MessageChannel`). +For this purpose, Spring Integration Java DSL provides an `IntegrationFlow` factory with a number of overloaded `from()` methods. +This factory can be used in groovy as well: + +==== +[source, groovy] +---- +@Bean +flowFromSupplier() { + IntegrationFlow.fromSupplier({ 'bar' }) { e -> e.poller { p -> p.fixedDelay(10).maxMessagesPerPoll(1) } } + .channel({ c -> c.queue('fromSupplierQueue') } as Function) + .get() +} +---- +==== + +But unfortunately not all `from()` methods are compatible with Groovy structures. +To solve this, Spring Integration provides a Groovy DSL factory around the `IntegrationFlows` factory. +It is implemented as a set of overloaded `integrationFlow()` functions. +With a consumer for a `GroovyIntegrationFlowDefinition` to declare the remainder of the flow as an `IntegrationFlow` closure to reuse the mentioned above experience and also avoid the need for a `get()` call in the end. +For example: + +==== +[source, groovy] +---- +@Bean +functionFlow() { + integrationFlow Function, + { beanName 'functionGateway' }, + { + transform Transformers.objectToString(), { id 'objectToStringTransformer' } + transform String, { it.toUpperCase() } + split Message, { it.payload } + split Object, { it }, { id 'splitterEndpoint' } + resequence() + aggregate { + id 'aggregator' + outputProcessor { it.one } + } + } +} + +@Bean +someFlow() { + integrationFlow ({ 'test' }, + { + poller { it.trigger new OnlyOnceTrigger() } + id 'pollingSource' + }) + { + log LoggingHandler.Level.WARN, 'test.category' + channel { queue 'pollerResultChannel' } + } +} +---- +==== diff --git a/src/reference/asciidoc/groovy.adoc b/src/reference/asciidoc/groovy.adoc index 62ddb1ce75..cf2852e514 100644 --- a/src/reference/asciidoc/groovy.adoc +++ b/src/reference/asciidoc/groovy.adoc @@ -1,5 +1,5 @@ [[groovy]] -=== Groovy support +=== Groovy Support In Spring Integration 2.0, we added Groovy support, letting you use the Groovy scripting language to provide the logic for various integration components -- similar to the way the Spring Expression Language (SpEL) is supported for routing, transformation, and other integration concerns. For more information about Groovy, see the Groovy documentation, which you can find on the https://groovy-lang.org/[project website]. @@ -23,6 +23,8 @@ compile "org.springframework.integration:spring-integration-groovy:{project-vers ---- ==== +In addition, starting with version 6.0, a <<./groovy-dsl.adoc#groovy-dsl,Groovy DSL>> for integration flow configurations is provided. + [[groovy-config]] ==== Groovy Configuration diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 3a4f5c53a2..55ff548920 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -17,6 +17,9 @@ In general the project has been moved to Java 17 baseline and migrated from Java [[x6.0-new-components]] === New Components +A Groovy DSL implementation for integration flow definitions has been added. +See <<./groovy-dsl.adoc#groovy-dsl,Groovy DSL>> for more information. + [[x6.0-mqtt]] ==== MQTT ClientManager