From 8a38a5e82958b11fce4e599a1242b16e588c39d8 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 5 Mar 2020 13:08:56 -0500 Subject: [PATCH] Merge spring-integration-kotlin-dsl project * Migrate all the Kotlin tests to use new Kotlin DSL * Upgrade to the latest Kotlin * Generate KDocs --- build.gradle | 50 +- .../integration/dsl/IntegrationFlowDsl.kt | 159 +++ .../dsl/KotlinIntegrationFlowDefinition.kt | 1039 +++++++++++++++++ .../integration/dsl/KotlinDslTests.kt | 335 ++++++ .../integration/dsl/routers/RouterDslTests.kt | 44 +- .../integration/function/FunctionsTests.kt | 19 +- .../integration/jms/dsl/JmsDslKotlinTests.kt | 30 +- 7 files changed, 1628 insertions(+), 48 deletions(-) create mode 100644 spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt create mode 100644 spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt create mode 100644 spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt diff --git a/build.gradle b/build.gradle index 0b03345e20..c9d987ce90 100644 --- a/build.gradle +++ b/build.gradle @@ -1,5 +1,5 @@ buildscript { - ext.kotlinVersion = '1.3.61' + ext.kotlinVersion = '1.3.70' repositories { maven { url 'https://repo.spring.io/plugins-release' } } @@ -17,6 +17,7 @@ plugins { id 'org.ajoberstar.grgit' version '4.0.1' id "io.spring.dependency-management" version '1.0.9.RELEASE' id 'com.jfrog.artifactory' version '4.13.0' apply false + id 'org.jetbrains.dokka' version '0.10.1' } if (System.getenv('TRAVIS') || System.getenv('bamboo_buildKey')) { @@ -188,9 +189,14 @@ configure(javaProjects) { subproject -> targetCompatibility = 1.8 } + compileKotlin { + kotlinOptions { + jvmTarget = '1.8' + allWarningsAsErrors = true + } + } compileTestKotlin { kotlinOptions { - freeCompilerArgs = ['-Xjsr305=strict'] jvmTarget = '1.8' } } @@ -414,10 +420,13 @@ project('spring-integration-core') { optionalApi "com.esotericsoftware:kryo-shaded:$kryoShadedVersion" optionalApi "io.micrometer:micrometer-core:$micrometerVersion" optionalApi "io.github.resilience4j:resilience4j-ratelimiter:$resilience4jVersion" - optionalApi"org.apache.avro:avro:$avroVersion" + optionalApi "org.apache.avro:avro:$avroVersion" + optionalApi 'org.jetbrains.kotlin:kotlin-reflect' + optionalApi 'org.jetbrains.kotlin:kotlin-stdlib-jdk8' testImplementation ("org.aspectj:aspectjweaver:$aspectjVersion") testImplementation ('com.fasterxml.jackson.datatype:jackson-datatype-jsr310') + testRuntime 'com.fasterxml.jackson.module:jackson-module-kotlin' } } @@ -980,6 +989,38 @@ task api(type: Javadoc) { }) } +dokka { + dependsOn api + + outputFormat = 'html' + outputDirectory = "$buildDir/docs/kdoc" + + configuration { + classpath = javaProjects.collect { project -> project.jar.outputs.files.getFiles() }.flatten() + classpath += files(javaProjects.collect { it.sourceSets.main.compileClasspath }) + javaProjects.forEach { project -> + if(project.sourceSets.main.kotlin) { + sourceRoot { + path = project.sourceSets.main.kotlin.srcDirs.first() + } + } + } + moduleName = 'spring-integration' + externalDocumentationLink { + url = new URL("https://docs.spring.io/spring-integration/docs/$version/api/") + } + externalDocumentationLink { + url = new URL('https://docs.spring.io/spring-framework/docs/current/javadoc-api/') + } + externalDocumentationLink { + url = new URL('https://projectreactor.io/docs/core/release/api/') + } + externalDocumentationLink { + url = new URL('https://www.reactive-streams.org/reactive-streams-1.0.1-javadoc/') + } + } +} + task schemaZip(type: Zip) { group = 'Distribution' archiveClassifier = 'schema' @@ -1038,6 +1079,9 @@ task docsZip(type: Zip, dependsOn: reference) { rename 'index-single.pdf', 'spring-integration-reference.pdf' into 'reference/pdf' } + from (dokka) { + into "kdoc-api" + } } task distZip(type: Zip, dependsOn: [docsZip, schemaZip]) { diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt new file mode 100644 index 0000000000..4fb241eeba --- /dev/null +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt @@ -0,0 +1,159 @@ +/* + * Copyright 2020 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.dsl + +import org.reactivestreams.Publisher +import org.springframework.integration.core.MessageSource +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 + +private fun buildIntegrationFlow(flowBuilder: IntegrationFlowBuilder, + flow: (KotlinIntegrationFlowDefinition) -> Unit): IntegrationFlow { + + flow(KotlinIntegrationFlowDefinition(flowBuilder)) + return flowBuilder.get() +} + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow] lambdas. + * + * @author Artem Bilan + */ +fun integrationFlow(flow: KotlinIntegrationFlowDefinition.() -> Unit) = + IntegrationFlow { + flow(KotlinIntegrationFlowDefinition(it)) + } + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(Class, Consumer)` factory method. + * + * @author Artem Bilan + */ +inline fun integrationFlow( + crossinline gateway: GatewayProxySpec.() -> Unit = {}, + flow: KotlinIntegrationFlowDefinition.() -> Unit): IntegrationFlow { + + val flowBuilder = IntegrationFlows.from(T::class.java) { gateway(it) } + flow(KotlinIntegrationFlowDefinition(flowBuilder)) + return flowBuilder.get() +} + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(String, Boolean)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(channelName: String, fixedSubscriber: Boolean = false, + flow: KotlinIntegrationFlowDefinition.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(channelName, fixedSubscriber), flow) + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessageChannel)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(channel: MessageChannel, flow: KotlinIntegrationFlowDefinition.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(channel), flow) + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessageSource<*>, Consumer)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(messageSource: MessageSource<*>, + options: SourcePollingChannelAdapterSpec.() -> Unit = {}, + flow: KotlinIntegrationFlowDefinition.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(messageSource, Consumer { options(it) }), flow) + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessageSourceSpec<*>, Consumer)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(messageSource: MessageSourceSpec<*, out MessageSource<*>>, + options: SourcePollingChannelAdapterSpec.() -> Unit = {}, + flow: KotlinIntegrationFlowDefinition.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(messageSource, options), flow) + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(Supplier<*>, Consumer)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(source: () -> Any, + options: SourcePollingChannelAdapterSpec.() -> Unit = {}, + flow: KotlinIntegrationFlowDefinition.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(source, options), flow) + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(Publisher>)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(publisher: Publisher>, + flow: KotlinIntegrationFlowDefinition.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(publisher), flow) + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessagingGatewaySupport)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(gateway: MessagingGatewaySupport, + flow: KotlinIntegrationFlowDefinition.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(gateway), flow) + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessagingGatewaySpec<*, *>)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(gatewaySpec: MessagingGatewaySpec<*, *>, + flow: KotlinIntegrationFlowDefinition.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(gatewaySpec), flow) + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessageProducerSupport)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(producer: MessageProducerSupport, + flow: KotlinIntegrationFlowDefinition.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(producer), flow) + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessageProducerSpec<*, *>)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(producerSpec: MessageProducerSpec<*, *>, + flow: KotlinIntegrationFlowDefinition.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(producerSpec), flow) diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt new file mode 100644 index 0000000000..afefc24f04 --- /dev/null +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt @@ -0,0 +1,1039 @@ +/* + * Copyright 2020 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.dsl + +import org.reactivestreams.Publisher +import org.springframework.expression.Expression +import org.springframework.integration.aggregator.AggregatingMessageHandler +import org.springframework.integration.channel.FluxMessageChannel +import org.springframework.integration.channel.interceptor.WireTap +import org.springframework.integration.core.MessageSelector +import org.springframework.integration.dsl.support.MessageChannelReference +import org.springframework.integration.filter.MessageFilter +import org.springframework.integration.filter.MethodInvokingSelector +import org.springframework.integration.handler.BridgeHandler +import org.springframework.integration.handler.DelayHandler +import org.springframework.integration.handler.GenericHandler +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.router.RecipientListRouter +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.support.MapBuilder +import org.springframework.integration.transformer.ClaimCheckInTransformer +import org.springframework.integration.transformer.ClaimCheckOutTransformer +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.MessageHeaders +import reactor.core.publisher.Flux +import java.util.concurrent.Executor +import java.util.function.Consumer + +/** + * An [IntegrationFlowDefinition] wrapped for Kotlin DSL. + * + * @property delegate the [IntegrationFlowDefinition] this instance is delegating to. + * + * @author Artem Bilan + * + * @since 5.3 + */ +class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: IntegrationFlowDefinition<*>) { + + /** + * Inline function for [IntegrationFlowDefinition.convert] providing a `convert()` variant + * with reified generic type. + */ + inline fun convert( + crossinline configurer: GenericEndpointSpec.() -> Unit = {}) { + + this.delegate.convert(T::class.java) { configurer(it) } + } + + /** + * Inline function for [IntegrationFlowDefinition.transform] providing a `transform()` variant + * with reified generic type. + */ + inline fun transform(crossinline function: (P) -> Any) { + this.delegate.transform(P::class.java) { function(it) } + } + + /** + * Inline function for [IntegrationFlowDefinition.transform] providing a `transform()` variant + * with reified generic type. + */ + inline fun transform( + crossinline function: (P) -> Any, + crossinline configurer: GenericEndpointSpec.() -> Unit) { + + this.delegate.transform(P::class.java, { function(it) }) { configurer(it) } + } + + /** + * Inline function for [IntegrationFlowDefinition.split] providing a `split()` variant + * with reified generic type. + */ + inline fun split(crossinline function: (P) -> Any) { + this.delegate.split(P::class.java) { function(it) } + } + + + /** + * Inline function for [IntegrationFlowDefinition.split] providing a `split()` variant + * with reified generic type. + */ + inline fun split( + crossinline function: (P) -> Any, + crossinline configurer: SplitterEndpointSpec.() -> Unit) { + + this.delegate.split(P::class.java, { function(it) }) { configurer(it) } + } + + /** + * Inline function for [IntegrationFlowDefinition.filter] providing a `filter()` variant + * with reified generic type. + */ + inline fun filter(crossinline function: (P) -> Boolean) { + this.delegate.filter(P::class.java) { function(it) } + } + + /** + * Inline function for [IntegrationFlowDefinition.filter] providing a `filter()` variant + * with reified generic type. + */ + inline fun filter( + crossinline function: (P) -> Boolean, + crossinline configurer: FilterEndpointSpec.() -> Unit) { + + this.delegate.filter(P::class.java, { function(it) }) { configurer(it) } + } + + + /** + * Inline function for [IntegrationFlowDefinition.filter] providing a `filter()` variant + * with reified generic type. + */ + inline fun route(crossinline function: (P) -> Any?) { + route(function) { } + } + + /** + * Inline function for [IntegrationFlowDefinition.filter] providing a `filter()` variant + * with reified generic type. + */ + inline fun route( + crossinline function: (P) -> T, + crossinline configurer: RouterSpec.() -> Unit) { + + this.delegate.route(P::class.java, { function(it) }) { configurer(it) } + } + + /** + * Populate an [org.springframework.integration.channel.FixedSubscriberChannel] instance + * at the current [IntegrationFlow] chain position. + * The provided `messageChannelName` is used for the bean registration. + */ + fun fixedSubscriberChannel(messageChannelName: String? = null) { + this.delegate.fixedSubscriberChannel(messageChannelName) + } + + /** + * Populate a [MessageChannelReference] instance + * at the current [IntegrationFlow] chain position. + * The provided `messageChannelName` is used for the bean registration + * ([org.springframework.integration.channel.DirectChannel]), if there is no such a bean + * in the application context. Otherwise the existing [MessageChannel] bean is used + * to wire integration endpoints. + */ + fun channel(messageChannelName: String) { + this.delegate.channel(messageChannelName) + } + + /** + * Populate a [MessageChannel] instance + * at the current [IntegrationFlow] chain position using the [MessageChannelSpec] + * fluent API. + */ + fun channel(messageChannelSpec: MessageChannelSpec<*, *>) { + this.delegate.channel(messageChannelSpec) + } + + /** + * Populate the provided [MessageChannel] instance + * at the current [IntegrationFlow] chain position. + * The `messageChannel` can be an existing bean, or fresh instance, in which case + * the [org.springframework.integration.dsl.context.IntegrationFlowBeanPostProcessor] + * will populate it as a bean with a generated name. + */ + fun channel(messageChannel: MessageChannel) { + this.delegate.channel(messageChannel) + } + + /** + * Populate a [MessageChannel] instance + * at the current [IntegrationFlow] chain position using the [Channels] + * factory fluent API. + */ + fun channel(channels: Channels.() -> MessageChannelSpec<*, *>) { + this.delegate.channel(channels) + } + + /** + * The [org.springframework.integration.channel.PublishSubscribeChannel] `channel()` + * method specific implementation to allow the use of the 'subflow' subscriber capability. + */ + fun publishSubscribeChannel(publishSubscribeChannelConfigurer: PublishSubscribeSpec.() -> Unit) { + this.delegate.publishSubscribeChannel(publishSubscribeChannelConfigurer) + } + + /** + * The [org.springframework.integration.channel.PublishSubscribeChannel] `channel()` + * method specific implementation to allow the use of the 'subflow' subscriber capability. + * Use the provided [Executor] for the target subscribers. + */ + fun publishSubscribeChannel(executor: Executor, + publishSubscribeChannelConfigurer: PublishSubscribeSpec.() -> Unit) { + + this.delegate.publishSubscribeChannel(executor, Consumer(publishSubscribeChannelConfigurer)) + } + + /** + * Populate the `Wire Tap` EI Pattern specific + * [org.springframework.messaging.support.ChannelInterceptor] implementation + * to the current channel. + * This method can be used after any `channel()` for explicit [MessageChannel], + * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. + */ + fun wireTap(flow: KotlinIntegrationFlowDefinition.() -> Unit) { + this.delegate.wireTap(IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) }) + } + + /** + * Populate the `Wire Tap` EI Pattern specific + * [org.springframework.messaging.support.ChannelInterceptor] implementation + * to the current channel. + * This method can be used after any `channel()` for explicit [MessageChannel], + * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. + */ + fun wireTap(wireTapConfigurer: WireTapSpec.() -> Unit, flow: KotlinIntegrationFlowDefinition.() -> Unit) { + this.delegate.wireTap( + IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) }, + Consumer(wireTapConfigurer)) + } + + /** + * Populate the `Wire Tap` EI Pattern specific + * [org.springframework.messaging.support.ChannelInterceptor] implementation + * to the current channel. + * This method can be used after any `channel()` for explicit [MessageChannel], + * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. + */ + fun wireTap(wireTapChannel: String, wireTapConfigurer: WireTapSpec.() -> Unit = {}) { + this.delegate.wireTap(wireTapChannel, wireTapConfigurer) + } + + /** + * Populate the `Wire Tap` EI Pattern specific + * [org.springframework.messaging.support.ChannelInterceptor] implementation + * to the current channel. + * This method can be used after any `channel()` for explicit [MessageChannel], + * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. + */ + fun wireTap(wireTapChannel: MessageChannel, wireTapConfigurer: WireTapSpec.() -> Unit = {}) { + this.delegate.wireTap(wireTapChannel, Consumer(wireTapConfigurer)) + } + + /** + * Populate the `Wire Tap` EI Pattern specific + * [org.springframework.messaging.support.ChannelInterceptor] implementation + * to the current channel. + * This method can be used after any `channel()` for explicit [MessageChannel], + * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. + */ + fun wireTap(wireTapSpec: WireTapSpec) { + this.delegate.wireTap(wireTapSpec) + } + + /** + * Populate the `Control Bus` EI Pattern specific [MessageHandler] implementation + * at the current [IntegrationFlow] chain position. + */ + fun controlBus(endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + this.delegate.controlBus(endpointConfigurer) + } + + /** + * Populate the `Transformer` EI Pattern specific [MessageHandler] implementation + * for the SpEL [Expression]. + */ + fun transform(expression: String, + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + + this.delegate.transform(expression, endpointConfigurer) + } + + /** + * Populate the `MessageTransformingHandler` for the [MethodInvokingTransformer] + * to invoke the service method at runtime. + */ + fun transform(service: Any, methodName: String? = null) { + this.delegate.transform(service, methodName) + } + + /** + * Populate the `MessageTransformingHandler` for the [MethodInvokingTransformer] + * to invoke the service method at runtime. + */ + fun transform(service: Any, methodName: String?, + endpointConfigurer: GenericEndpointSpec.() -> Unit) { + + this.delegate.transform(service, methodName, endpointConfigurer) + } + + /** + * Populate the [MessageTransformingHandler] instance for the + * [org.springframework.integration.handler.MessageProcessor] from provided [MessageProcessorSpec]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun transform(messageProcessorSpec: MessageProcessorSpec<*>, + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + + this.delegate.transform(messageProcessorSpec, endpointConfigurer) + } + + /** + * Populate a [MessageFilter] with [MessageSelector] for the provided SpEL expression. + * In addition accept options for the integration endpoint using [FilterEndpointSpec]: + */ + fun filter(expression: String, endpointConfigurer: FilterEndpointSpec.() -> Unit = {}) { + this.delegate.filter(expression, endpointConfigurer) + } + + /** + * Populate a [MessageFilter] with [MethodInvokingSelector] for the + * method of the provided service. + */ + fun filter(service: Any, methodName: String? = null) { + this.delegate.filter(service, methodName) + } + + /** + * Populate a [MessageFilter] with [MethodInvokingSelector] for the + * method of the provided service. + */ + fun filter(service: Any, methodName: String?, endpointConfigurer: FilterEndpointSpec.() -> Unit) { + this.delegate.filter(service, methodName, endpointConfigurer) + } + + /** + * Populate a [MessageFilter] with [MethodInvokingSelector] + * for the [MessageProcessor] from + * the provided [MessageProcessorSpec]. + * In addition accept options for the integration endpoint using [FilterEndpointSpec]. + */ + fun filter(messageProcessorSpec: MessageProcessorSpec<*>, endpointConfigurer: FilterEndpointSpec.() -> Unit = {}) { + this.delegate.filter(messageProcessorSpec, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the selected protocol specific + * [MessageHandler] implementation from `Namespace Factory`: + */ + fun handle(messageHandlerSpec: MessageHandlerSpec<*, H>) { + this.delegate.handle(messageHandlerSpec) + } + + /** + * Populate a [ServiceActivatingHandler] for the provided + * [MessageHandler] implementation. + */ + fun handle(messageHandler: MessageHandler) { + this.delegate.handle(messageHandler) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the `method` for provided `bean` at runtime. + */ + fun handle(beanName: String, methodName: String? = null) { + this.delegate.handle(beanName, methodName) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the `method` for provided `bean` at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(beanName: String, methodName: String?, + endpointConfigurer: GenericEndpointSpec.() -> Unit) { + + this.delegate.handle(beanName, methodName, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the `method` for provided `bean` at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(service: Any, methodName: String? = null) { + this.delegate.handle(service, methodName) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the `method` for provided `bean` at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(service: Any, methodName: String?, + endpointConfigurer: GenericEndpointSpec.() -> Unit) { + + this.delegate.handle(service, methodName, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the provided [GenericHandler] at runtime. + */ + inline fun handle(crossinline handler: (P, MessageHeaders) -> Any) { + this.delegate.handle(P::class.java) { p, h -> handler(p, h) } + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the provided [GenericHandler] at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + inline fun handle( + crossinline handler: (P, MessageHeaders) -> Any, + crossinline endpointConfigurer: GenericEndpointSpec.() -> Unit) { + + this.delegate.handle(P::class.java, { p, h -> handler(p, h) }) { endpointConfigurer(it) } + } + + /** + * Populate a [ServiceActivatingHandler] for the [MessageProcessor] from the provided [MessageProcessorSpec]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(messageProcessorSpec: MessageProcessorSpec<*>, + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + + this.delegate.handle(messageProcessorSpec, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the selected protocol specific + * [MessageHandler] implementation from `Namespace Factory`: + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(messageHandlerSpec: MessageHandlerSpec<*, H>, + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + + this.delegate.handle(messageHandlerSpec, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the provided + * [MessageHandler] lambda. + */ + fun handle(messageHandler: (Message<*>) -> Unit) { + this.delegate.handle(MessageHandler { messageHandler(it) }) + } + + /** + * Populate a [ServiceActivatingHandler] for the provided + * [MessageHandler] lambda. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(messageHandler: (Message<*>) -> Unit, + endpointConfigurer: GenericEndpointSpec.() -> Unit) { + + this.delegate.handle(MessageHandler { messageHandler(it) }, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the provided + * [MessageHandler] implementation. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(messageHandler: H, endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + this.delegate.handle(messageHandler, endpointConfigurer) + } + + /** + * Populate a [BridgeHandler] to the current integration flow position. + */ + fun bridge(endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + this.delegate.bridge(endpointConfigurer) + } + + /** + * Populate a [DelayHandler] to the current integration flow position. + */ + fun delay(groupId: String, endpointConfigurer: DelayerEndpointSpec.() -> Unit = {}) { + this.delegate.delay(groupId, endpointConfigurer) + } + + /** + * Populate a [org.springframework.integration.transformer.ContentEnricher] + * to the current integration flow position + * with provided options. + */ + fun enrich(enricherConfigurer: EnricherSpec.() -> Unit) { + this.delegate.enrich(enricherConfigurer) + } + + /** + * Populate a [MessageTransformingHandler] for + * a [org.springframework.integration.transformer.HeaderEnricher] + * using header values from provided [MapBuilder]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun enrichHeaders(headers: MapBuilder<*, String, Any>, + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + + this.delegate.enrichHeaders(headers, endpointConfigurer) + } + + /** + * Accept a [Map] of values to be used for the + * [Message] header enrichment. + * `values` can apply an [Expression] + * to be evaluated against a request [Message]. + */ + fun enrichHeaders(headers: Map, + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + + this.delegate.enrichHeaders(headers, endpointConfigurer) + } + + /** + * Populate a [MessageTransformingHandler] for + * a [org.springframework.integration.transformer.HeaderEnricher] + * as the result of provided consumer. + */ + fun enrichHeaders(headerEnricherConfigurer: HeaderEnricherSpec.() -> Unit) { + this.delegate.enrichHeaders(headerEnricherConfigurer) + } + + /** + * Populate the [DefaultMessageSplitter] with provided options + * to the current integration flow position. + */ + fun split() { + this.delegate.split() + } + + /** + * Populate the [ExpressionEvaluatingSplitter] with provided + * SpEL expression. + */ + fun split(expression: String, + endpointConfigurer: SplitterEndpointSpec.() -> Unit = {}) { + + this.delegate.split(expression, endpointConfigurer) + } + + /** + * Populate the [MethodInvokingSplitter] to evaluate the provided + * `method` of the `service` at runtime. + */ + fun split(service: Any, methodName: String? = null) { + this.delegate.split(service, methodName) + } + + /** + * Populate the [MethodInvokingSplitter] to evaluate the provided + * `method` of the `bean` at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun split(service: Any, methodName: String?, + endpointConfigurer: SplitterEndpointSpec.() -> Unit) { + + this.delegate.split(service, methodName, endpointConfigurer) + } + + /** + * Populate the [MethodInvokingSplitter] to evaluate the provided + * `method` of the `bean` at runtime. + */ + fun split(beanName: String, methodName: String? = null) { + this.delegate.split(beanName, methodName) + } + + /** + * Populate the [MethodInvokingSplitter] to evaluate the provided + * `method` of the `bean` at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun split(beanName: String, methodName: String?, + endpointConfigurer: SplitterEndpointSpec.() -> Unit) { + + this.delegate.split(beanName, methodName, endpointConfigurer) + } + + /** + * Populate the [MethodInvokingSplitter] to evaluate the + * [MessageProcessor] at runtime + * from provided [MessageProcessorSpec]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun split(messageProcessorSpec: MessageProcessorSpec<*>, + endpointConfigurer: SplitterEndpointSpec.() -> Unit = {}) { + + this.delegate.split(messageProcessorSpec, endpointConfigurer) + } + + /** + * Populate the provided [AbstractMessageSplitter] to the current integration flow position. + */ + fun split(splitterMessageHandlerSpec: MessageHandlerSpec<*, S>, + endpointConfigurer: SplitterEndpointSpec.() -> Unit = {}) { + + this.delegate.split(splitterMessageHandlerSpec, endpointConfigurer) + } + + /** + * Populate the provided [AbstractMessageSplitter] to the current integration + * flow position. + */ + fun split(splitter: S, + endpointConfigurer: SplitterEndpointSpec.() -> Unit = {}) { + + this.delegate.split(splitter, endpointConfigurer) + } + + /** + * Provide the [HeaderFilter] to the current [IntegrationFlow]. + */ + fun headerFilter(headersToRemove: String, patternMatch: Boolean = true) { + this.delegate.headerFilter(headersToRemove, patternMatch) + } + + /** + * Populate the provided [MessageTransformingHandler] for the provided + * [HeaderFilter]. + */ + fun headerFilter(headerFilter: HeaderFilter, + endpointConfigurer: GenericEndpointSpec.() -> Unit) { + + this.delegate.headerFilter(headerFilter, endpointConfigurer) + } + + /** + * Populate the [MessageTransformingHandler] for the [ClaimCheckInTransformer] + * with provided [MessageStore]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun claimCheckIn(messageStore: MessageStore, + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + + this.delegate.claimCheckIn(messageStore, endpointConfigurer) + } + + /** + * Populate the [MessageTransformingHandler] for the [ClaimCheckOutTransformer] + * with provided [MessageStore] and `removeMessage` flag. + */ + fun claimCheckOut(messageStore: MessageStore, removeMessage: Boolean = false) { + this.delegate.claimCheckOut(messageStore, removeMessage) + } + + /** + * Populate the [MessageTransformingHandler] for the [ClaimCheckOutTransformer] + * with provided [MessageStore] and `removeMessage` flag. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun claimCheckOut(messageStore: MessageStore, removeMessage: Boolean, + endpointConfigurer: GenericEndpointSpec.() -> Unit) { + + this.delegate.claimCheckOut(messageStore, removeMessage, endpointConfigurer) + } + + /** + * Populate the + * [org.springframework.integration.aggregator.ResequencingMessageHandler] with + * provided options from [ResequencerSpec]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun resequence(resequencer: ResequencerSpec.() -> Unit = {}) { + this.delegate.resequence(resequencer) + } + + /** + * Populate the [AggregatingMessageHandler] with provided options from [AggregatorSpec]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun aggregate(aggregator: AggregatorSpec.() -> Unit = {}) { + this.delegate.aggregate(aggregator) + } + + /** + * Populate the [MethodInvokingRouter] for provided bean and its method + * with default options. + */ + fun route(beanName: String, method: String? = null) { + this.delegate.route(beanName, method) + } + + /** + * Populate the [MethodInvokingRouter] for provided bean and its method + * with provided options from [RouterSpec]. + */ + fun route(beanName: String, method: String?, routerConfigurer: RouterSpec.() -> Unit) { + this.delegate.route(beanName, method, routerConfigurer) + } + + /** + * Populate the [MethodInvokingRouter] for the method + * of the provided service and its method with default options. + */ + fun route(service: Any, methodName: String? = null) { + this.delegate.route(service, methodName) + } + + /** + * Populate the [MethodInvokingRouter] for the method + * of the provided service and its method with provided options from [RouterSpec]. + */ + fun route(service: Any, methodName: String?, routerConfigurer: RouterSpec.() -> Unit) { + this.delegate.route(service, methodName, routerConfigurer) + } + + /** + * Populate the [ExpressionEvaluatingRouter] for provided SpEL expression + * with provided options from [RouterSpec]. + */ + fun route(expression: String, routerConfigurer: RouterSpec.() -> Unit = {}) { + this.delegate.route(expression, routerConfigurer) + } + + /** + * Populate the [MethodInvokingRouter] for the + * [MessageProcessor] + * from the provided [MessageProcessorSpec] with default options. + */ + fun route(messageProcessorSpec: MessageProcessorSpec<*>, + routerConfigurer: RouterSpec.() -> Unit = {}) { + + this.delegate.route(messageProcessorSpec, routerConfigurer) + } + + /** + * Populate the [RecipientListRouter] with options from the [RecipientListRouterSpec]. + */ + fun routeToRecipients(routerConfigurer: RecipientListRouterSpec.() -> Unit) { + this.delegate.routeToRecipients(routerConfigurer) + } + + /** + * Populate the [ErrorMessageExceptionTypeRouter] with options from the [RouterSpec]. + */ + fun routeByException( + routerConfigurer: RouterSpec, ErrorMessageExceptionTypeRouter>.() -> Unit) { + + this.delegate.routeByException(routerConfigurer) + } + + /** + * Populate the provided [AbstractMessageRouter] implementation to the + * current integration flow position. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun route(router: R, endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + this.delegate.route(router, endpointConfigurer) + } + + /** + * Populate the "artificial" + * [org.springframework.integration.gateway.GatewayMessageHandler] for the + * provided `requestChannel` to send a request with options from + * [GatewayEndpointSpec]. Uses + * [org.springframework.integration.gateway.RequestReplyExchanger] Proxy on the + * background. + */ + fun gateway(requestChannel: String, endpointConfigurer: GatewayEndpointSpec.() -> Unit = {}) { + this.delegate.gateway(requestChannel, endpointConfigurer) + } + + /** + * Populate the "artificial" + * [org.springframework.integration.gateway.GatewayMessageHandler] for the + * provided `requestChannel` to send a request with options from + * [GatewayEndpointSpec]. Uses + * [org.springframework.integration.gateway.RequestReplyExchanger] Proxy on the + * background. + */ + fun gateway(requestChannel: MessageChannel, endpointConfigurer: GatewayEndpointSpec.() -> Unit = {}) { + this.delegate.gateway(requestChannel, Consumer(endpointConfigurer)) + } + + /** + * Populate the "artificial" + * [org.springframework.integration.gateway.GatewayMessageHandler] for the + * provided `subflow` with options from [GatewayEndpointSpec]. + */ + fun gateway(flow: KotlinIntegrationFlowDefinition.() -> Unit) { + this.delegate.gateway(IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) }) + } + + /** + * Populate the "artificial" + * [org.springframework.integration.gateway.GatewayMessageHandler] for the + * provided `subflow` with options from [GatewayEndpointSpec]. + */ + fun gateway(endpointConfigurer: GatewayEndpointSpec.() -> Unit, + flow: KotlinIntegrationFlowDefinition.() -> Unit) { + + this.delegate.gateway( + IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) }, + Consumer(endpointConfigurer)) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the `INFO` + * logging level and `org.springframework.integration.handler.LoggingHandler` + * as a default logging category. + */ + fun log() { + this.delegate.log() + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for provided [LoggingHandler.Level] + * logging level and `org.springframework.integration.handler.LoggingHandler` + * as a default logging category. + */ + fun log(level: LoggingHandler.Level, category: String? = null) { + this.delegate.log(level, category) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided logging category + * and `INFO` logging level. + */ + fun log(category: String) { + this.delegate.log(category) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, logging category + * and SpEL expression for the log message. + */ + fun log(level: LoggingHandler.Level, category: String, logExpression: String) { + this.delegate.log(level, category, logExpression) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the `INFO` logging level, + * the `org.springframework.integration.handler.LoggingHandler` + * as a default logging category and function for the log message. + */ + fun

log(function: (Message

) -> Any) { + this.delegate.log(function) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the `INFO` logging level, + * the `org.springframework.integration.handler.LoggingHandler` + * as a default logging category and SpEL expression to evaluate + * logger message at runtime against the request [Message]. + */ + fun log(logExpression: Expression) { + this.delegate.log(logExpression) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, + * the `org.springframework.integration.handler.LoggingHandler` + * as a default logging category and SpEL expression to evaluate + * logger message at runtime against the request [Message]. + * When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + */ + fun log(level: LoggingHandler.Level, logExpression: Expression) { + this.delegate.log(level, logExpression) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the `INFO` + * [LoggingHandler.Level] logging level, + * the provided logging category and SpEL expression to evaluate + * logger message at runtime against the request [Message]. + */ + fun log(category: String, logExpression: Expression) { + this.delegate.log(category, logExpression) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, + * the `org.springframework.integration.handler.LoggingHandler` + * as a default logging category and function for the log message. + */ + fun

log(level: LoggingHandler.Level, function: (Message

) -> Any) { + this.delegate.log(level, function) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, + * the provided logging category and function for the log message. + */ + fun

log(category: String, function: (Message

) -> Any) { + this.delegate.log(category, function) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, logging category + * and function for the log message. + */ + fun

log(level: LoggingHandler.Level, category: String, function: (Message

) -> Any) { + this.delegate.log(level, category, function) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, logging category + * and SpEL expression for the log message. + */ + fun log(level: LoggingHandler.Level, category: String, logExpression: Expression) { + this.delegate.log(level, category, logExpression) + } + + /** + * Populate a [ScatterGatherHandler] to the current integration flow position + * based on the provided [MessageChannel] for scattering function + * and [AggregatorSpec] for gathering function. + */ + fun scatterGather(scatterChannel: MessageChannel, gatherer: AggregatorSpec.() -> Unit = {}) { + this.delegate.scatterGather(scatterChannel, Consumer(gatherer)) + } + + /** + * Populate a [ScatterGatherHandler] to the current integration flow position + * based on the provided [MessageChannel] for scattering function + * and [AggregatorSpec] for gathering function. + */ + fun scatterGather(scatterChannel: MessageChannel, gatherer: AggregatorSpec.() -> Unit, + scatterGather: ScatterGatherSpec.() -> Unit) { + + this.delegate.scatterGather(scatterChannel, Consumer(gatherer), Consumer(scatterGather)) + } + + /** + * Populate a [ScatterGatherHandler] to the current integration flow position + * based on the provided [RecipientListRouterSpec] for scattering function + * and default [AggregatorSpec] for gathering function. + */ + fun scatterGather(scatterer: RecipientListRouterSpec.() -> Unit) { + this.delegate.scatterGather(scatterer) + } + + /** + * Populate a [ScatterGatherHandler] to the current integration flow position + * based on the provided [RecipientListRouterSpec] for scattering function + * and [AggregatorSpec] for gathering function. + */ + fun scatterGather(scatterer: RecipientListRouterSpec.() -> Unit, gatherer: AggregatorSpec.() -> Unit) { + this.delegate.scatterGather(scatterer, gatherer) + } + + /** + * Populate a [ScatterGatherHandler] to the current integration flow position + * based on the provided [RecipientListRouterSpec] for scattering function + * and [AggregatorSpec] for gathering function. + */ + fun scatterGather(scatterer: RecipientListRouterSpec.() -> Unit, gatherer: AggregatorSpec.() -> Unit, + scatterGather: ScatterGatherSpec.() -> Unit) { + + this.delegate.scatterGather(scatterer, gatherer, scatterGather) + } + + /** + * Populate a [org.springframework.integration.aggregator.BarrierMessageHandler] + * instance for provided timeout and options from [BarrierSpec] and endpoint + * options from [GenericEndpointSpec]. + */ + fun barrier(timeout: Long, barrierConfigurer: BarrierSpec.() -> Unit = {}) { + this.delegate.barrier(timeout, barrierConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] instance to perform [MessageTriggerAction] + * and endpoint options from [GenericEndpointSpec]. + */ + fun trigger(triggerActionId: String, + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + + this.delegate.trigger(triggerActionId, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] instance to perform [MessageTriggerAction] + * and endpoint options from [GenericEndpointSpec]. + */ + fun trigger(triggerAction: MessageTriggerAction, + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { + + this.delegate.trigger(triggerAction, Consumer(endpointConfigurer)) + } + + /** + * Populate a [FluxMessageChannel] to start a reactive processing for upstream data, + * wrap it to a [Flux], apply provided function via [Flux.transform] + * and emit the result to one more [FluxMessageChannel], subscribed in the downstream flow. + */ + fun fluxTransform(fluxFunction: (Flux>) -> Publisher) { + this.delegate.fluxTransform(fluxFunction) + } + +} diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt new file mode 100644 index 0000000000..bab49fbc66 --- /dev/null +++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt @@ -0,0 +1,335 @@ +/* + * Copyright 2020 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.dsl + +import assertk.assertThat +import assertk.assertions.isEqualTo +import assertk.assertions.isGreaterThanOrEqualTo +import assertk.assertions.isInstanceOf +import assertk.assertions.isNotNull +import assertk.assertions.size +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.core.GenericSelector +import org.springframework.integration.core.MessagingTemplate +import org.springframework.integration.dsl.context.IntegrationFlowContext +import org.springframework.integration.endpoint.MessageProcessorMessageSource +import org.springframework.integration.handler.LoggingHandler +import org.springframework.integration.scheduling.PollerMetadata +import org.springframework.integration.support.MessageBuilder +import org.springframework.integration.test.util.OnlyOnceTrigger +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.util.* +import java.util.function.Function + +/** + * @author Artem Bilan + */ +@SpringJUnitConfig +@DirtiesContext +class KotlinDslTests { + + @Autowired + private lateinit var beanFactory: BeanFactory + + @Autowired + private lateinit var integrationFlowContext: IntegrationFlowContext + + @Autowired + private lateinit var convertFlowInput: MessageChannel + + @Test + fun `convert extension`() { + assertThat(this.beanFactory.containsBean("kotlinConverter")) + + val replyChannel = QueueChannel() + val date = Date() + val testMessage = + MessageBuilder.withPayload("{\"name\": \"Test\",\"date\": " + date.time + "}") + .setHeader(MessageHeaders.CONTENT_TYPE, "application/json") + .setReplyChannel(replyChannel) + .build() + this.convertFlowInput.send(testMessage) + + assertThat(replyChannel.receive(10000)?.payload) + .isNotNull() + .isInstanceOf(TestPojo::class.java) + .isEqualTo(TestPojo("Test", date)) + } + + @Autowired + @Qualifier("functionGateway") + private lateinit var upperCaseFunction: Function + + @Test + fun `uppercase function`() { + assertThat(this.upperCaseFunction.apply("test")).isEqualTo("TEST") + } + + @Autowired + private lateinit var fromSupplierQueue: PollableChannel + + @Test + fun `message source flow`() { + assertThat(this.fromSupplierQueue.receive(10_000)?.payload).isNotNull().isEqualTo("testSource") + } + + @Autowired + @Qualifier("functionFlow2.gateway") + private lateinit var lowerCaseFunction: Function + + @Test + fun `lowercase function`() { + assertThat(this.lowerCaseFunction.apply("TEST2")).isEqualTo("test2") + } + + @Autowired + private lateinit var fixedSubscriberInput: MessageChannel + + @Test + fun `fixed subscriber channel`() { + assertThat(MessagingTemplate().convertSendAndReceive(this.fixedSubscriberInput, "test", String::class.java)) + .isEqualTo("test") + } + + @Autowired + private lateinit var fromSupplierQueue2: PollableChannel + + @Test + fun `message source flow2`() { + assertThat(this.fromSupplierQueue2.receive(10_000)?.payload).isNotNull().isEqualTo("testSource2") + } + + @Autowired + private lateinit var testSupplierResult: PollableChannel + + @Test + fun `supplier flow1`() { + assertThat(this.testSupplierResult.receive(10_000)?.payload).isNotNull().isEqualTo("testSupplier") + } + + @Autowired + private lateinit var testSupplierResult2: PollableChannel + + @Test + fun `supplier flow2`() { + assertThat(this.testSupplierResult2.receive(10_000)?.payload).isNotNull().isEqualTo("testSupplier2") + } + + @Test + fun `reactive publisher flow`() { + val fluxChannel = FluxMessageChannel() + + val verifyLater = + StepVerifier + .create(Flux.from(fluxChannel).map { it.payload }.cast(Integer::class.java)) + .expectNext(Integer(4), Integer(6)) + .thenCancel() + .verifyLater() + + val publisher = Flux.just(2, 3).map { GenericMessage(it) } + + val integrationFlow = + integrationFlow(publisher) { + transform>({ it.payload * 2 }) { id("foo") } + channel(fluxChannel) + } + + val registration = this.integrationFlowContext.registration(integrationFlow).register() + + verifyLater.verify() + + registration.destroy() + } + + @Autowired + @Qualifier("flowLambda.input") + private lateinit var flowLambdaInput: MessageChannel + + @Autowired + private lateinit var wireTapChannel: PollableChannel + + @Test + fun `flow from lambda`() { + val replyChannel = QueueChannel() + val message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build() + + this.flowLambdaInput.send(message) + + assertThat(replyChannel.receive(10_000)?.payload).isNotNull().isEqualTo("TEST") + assertThat(this.wireTapChannel.receive(10_000)?.payload).isNotNull().isEqualTo("test") + } + + @Autowired + @Qualifier("scatterGatherFlow.input") + private lateinit var scatterGatherFlowInput: MessageChannel + + @Test + fun `Scatter-Gather`() { + val replyChannel = QueueChannel() + val request = + MessageBuilder.withPayload("foo") + .setReplyChannel(replyChannel) + .build() + this.scatterGatherFlowInput.send(request) + val bestQuoteMessage = replyChannel.receive(10000) + assertThat(bestQuoteMessage).isNotNull() + val payload = bestQuoteMessage!!.payload + assertThat(payload).isInstanceOf(List::class.java).size().isGreaterThanOrEqualTo(1) + } + + @Configuration + @EnableIntegration + class Config { + + @Bean(PollerMetadata.DEFAULT_POLLER) + fun defaultPoller() = + Pollers.fixedDelay(100).maxMessagesPerPoll(1).get() + + @Bean + fun convertFlow() = + integrationFlow("convertFlowInput") { + convert() + convert { id("kotlinConverter") } + handle { m -> (m.headers[MessageHeaders.REPLY_CHANNEL] as MessageChannel).send(m) } + } + + @Bean + fun functionFlow() = + integrationFlow>({ beanName("functionGateway") }) { + transform { it.toUpperCase() } + split> { it.payload } + split({ it }) { id("splitterEndpoint") } + resequence() + aggregate { + id("aggregator") + outputProcessor { it.one } + } + } + + @Bean + fun functionFlow2() = + integrationFlow> { + transform { it.toLowerCase() } + route, Any?>({ null }) { defaultOutputToParentFlow() } + route> { m -> m.headers.replyChannel } + } + + @Bean + fun messageSourceFlow() = + integrationFlow(MessageProcessorMessageSource { "testSource" }, + { poller { it.trigger(OnlyOnceTrigger()) } }) { + channel { queue("fromSupplierQueue") } + } + + @Bean + fun messageSourceFlow2() = + integrationFlow(MessageProcessorMessageSource { "testSource2" }) { + channel { queue("fromSupplierQueue2") } + } + + @Bean + fun fixedSubscriberFlow() = + integrationFlow("fixedSubscriberInput", true) { + log(LoggingHandler.Level.WARN) { it.payload } + transform("payload") { id("spelTransformer") } + } + + @Bean + fun flowFromSupplier() = + integrationFlow({ "testSupplier" }) { + channel { queue("testSupplierResult") } + } + + @Bean + fun flowFromSupplier2() = + integrationFlow({ "testSupplier2" }, + { poller { it.trigger(OnlyOnceTrigger()) } }) { + filter> { m -> m.payload is String } + channel { queue("testSupplierResult2") } + } + + @Bean + fun flowLambda() = + integrationFlow { + filter({ it === "test" }) { id("filterEndpoint") } + wireTap { + channel { queue("wireTapChannel") } + } + delay("delayGroup") { defaultDelay(100) } + transform { it.toUpperCase() } + } + + + /* + A Java variant for the flow below + @Bean + public IntegrationFlow scatterGatherFlow() { + return f -> f + .scatterGather(scatterer -> scatterer + .applySequence(true) + .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10)) + .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10)) + .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10)), + gatherer -> gatherer + .releaseStrategy(group -> + group.size() == 3 || + group.getMessages() + .stream() + .anyMatch(m -> (Double) m.getPayload() > 5)), + scatterGather -> scatterGather + .gatherTimeout(10_000)); + }*/ + @Bean + fun scatterGatherFlow() = + integrationFlow { + scatterGather( + { + applySequence(true) + recipientFlow(GenericSelector { true }, integrationFlow { handle { _, _ -> Math.random() * 10 } }) + recipientFlow(GenericSelector { true }, integrationFlow { handle { _, _ -> Math.random() * 10 } }) + recipientFlow(GenericSelector { true }, integrationFlow { handle { _, _ -> Math.random() * 10 } }) + }, + { + releaseStrategy { + it.size() == 3 || it.messages.stream().anyMatch { it.payload as Double > 5 } + } + }) + { + gatherTimeout(10_000) + } + } + } + + data class TestPojo(val name: String?, val date: Date?) + +} diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt index 5624c9b319..e680aa7681 100644 --- a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt +++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -27,8 +27,8 @@ import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import org.springframework.integration.channel.QueueChannel import org.springframework.integration.config.EnableIntegration -import org.springframework.integration.dsl.IntegrationFlow import org.springframework.integration.dsl.MessageChannels +import org.springframework.integration.dsl.integrationFlow import org.springframework.integration.support.MessageBuilder import org.springframework.messaging.MessageChannel import org.springframework.messaging.PollableChannel @@ -99,39 +99,37 @@ class RouterDslTests { @Bean fun routerTwoSubFlows() = - IntegrationFlow { f -> - f.split() - .route({ p -> p % 2 == 0 }, - { m -> - m.subFlowMapping(true, { sf -> sf.handle { p, _ -> p * 2 } }) - .subFlowMapping(false, { sf -> sf.handle { p, _ -> p * 3 } }) - }) - .aggregate() - .channel { c -> c.queue("routerTwoSubFlowsOutput") } + integrationFlow { + split() + route({ it % 2 == 0 }) { + subFlowMapping(true) { sf -> sf.handle { p, _ -> p * 2 } } + subFlowMapping(false) { sf -> sf.handle { p, _ -> p * 3 } } + } + aggregate() + channel { queue("routerTwoSubFlowsOutput") } } @Bean fun splitRouteAggregate() = - IntegrationFlow { f -> - f.split() - .route({ o -> o % 2 == 0 }, - { m -> - m.subFlowMapping(true) { sf -> sf.gateway(oddFlow()) } - .subFlowMapping(false) { sf -> sf.gateway(evenFlow()) } - }) - .aggregate() + integrationFlow { + split() + route({ it % 2 == 0 }) { + subFlowMapping(true) { sf -> sf.gateway(oddFlow()) } + subFlowMapping(false) { sf -> sf.gateway(evenFlow()) } + } + aggregate() } @Bean fun oddFlow() = - IntegrationFlow { flow -> - flow.handle { _, _ -> "odd" } + integrationFlow { + handle { _, _ -> "odd" } } @Bean fun evenFlow() = - IntegrationFlow { flow -> - flow.handle { _, _ -> "even" } + integrationFlow { + handle { _, _ -> "even" } } @Bean diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt index bc495dd630..bf37592b44 100644 --- a/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt +++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -36,8 +36,10 @@ import org.springframework.integration.channel.DirectChannel import org.springframework.integration.channel.QueueChannel import org.springframework.integration.config.EnableIntegration import org.springframework.integration.dsl.IntegrationFlows +import org.springframework.integration.dsl.integrationFlow import org.springframework.integration.endpoint.SourcePollingChannelAdapter import org.springframework.integration.gateway.GatewayProxyFactoryBean +import org.springframework.integration.test.util.OnlyOnceTrigger import org.springframework.messaging.Message import org.springframework.messaging.MessageChannel import org.springframework.messaging.PollableChannel @@ -179,16 +181,17 @@ class FunctionsTests { @Bean fun flowFromSupplier() = - IntegrationFlows.from({ "" }) { e -> e.poller { p -> p.fixedDelay(10).maxMessagesPerPoll(1) } } - .transform { "blank" } - .channel { c -> c.queue("fromSupplierQueue") } - .get() + integrationFlow({ "" }, { poller { it.fixedDelay(10).maxMessagesPerPoll(1) } }) { + transform { "blank" } + channel { queue("fromSupplierQueue") } + } @Bean fun monoFunctionGateway() = - IntegrationFlows.from(MonoFunction::class.java) { gateway -> gateway.proxyDefaultMethods(true) } - .handle({ p, _ -> Mono.just(p).map(String::toUpperCase) }) { e -> e.async(true) } - .get() + integrationFlow({ proxyDefaultMethods(true) }) { + handle({ p, _ -> Mono.just(p).map(String::toUpperCase) }) { async(true) } + } + } interface MonoFunction : Function>> diff --git a/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt b/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt index 93a546db66..00780afdb1 100644 --- a/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt +++ b/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -30,9 +30,8 @@ import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import org.springframework.integration.IntegrationMessageHeaderAccessor import org.springframework.integration.config.EnableIntegration -import org.springframework.integration.dsl.IntegrationFlow -import org.springframework.integration.dsl.IntegrationFlows import org.springframework.integration.dsl.MessageChannels +import org.springframework.integration.dsl.integrationFlow import org.springframework.integration.jms.DefaultJmsHeaderMapper import org.springframework.integration.support.MessageBuilder import org.springframework.jms.support.JmsHeaders @@ -98,12 +97,15 @@ class JmsDslKotlinTests { @Bean fun jmsOutboundFlow() = - IntegrationFlow { f -> - f.handle(Jms.outboundAdapter(jmsConnectionFactory()) - .destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER) - .deliveryModeFunction { _ -> DeliveryMode.NON_PERSISTENT } - .timeToLiveExpression("10000") - .configureJmsTemplate { t -> t.explicitQosEnabled(true) }) + integrationFlow { + handle(Jms.outboundAdapter(jmsConnectionFactory()) + .apply { + destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER) + deliveryModeFunction { DeliveryMode.NON_PERSISTENT } + timeToLiveExpression("10000") + configureJmsTemplate { it.explicitQosEnabled(true) } + } + ) } @Bean @@ -119,15 +121,15 @@ class JmsDslKotlinTests { @Bean fun jmsMessageDrivenFlowWithContainer() = - IntegrationFlows.from( + integrationFlow( Jms.messageDrivenChannelAdapter( Jms.container(jmsConnectionFactory(), "containerSpecDestination") .pubSubDomain(false) .taskExecutor(Executors.newCachedThreadPool())) - .headerMapper(jmsHeaderMapper())) - .transform({ it: String -> it.trim({ it <= ' ' }) }) - .channel(jmsOutboundInboundReplyChannel()) - .get() + .headerMapper(jmsHeaderMapper())) { + transform { it: String -> it.trim { it <= ' ' } } + channel(jmsOutboundInboundReplyChannel()) + } }