diff --git a/spring-integration-kotlin-dsl/README.adoc b/spring-integration-kotlin-dsl/README.adoc index c7b6539..f27004e 100644 --- a/spring-integration-kotlin-dsl/README.adoc +++ b/spring-integration-kotlin-dsl/README.adoc @@ -23,6 +23,27 @@ IntegrationFlow { flow -> In this case Kotlin understands that the lambda should be translated into `IntegrationFlow` anonymous instance and 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, this project suggest a Kotlin-specif DSL for declaring integration flows in the *builder* pattern style: + +==== +[source, kotlin] +---- +@Bean +fun flowLambda() = + integrationFlow { + filter { it === "test" } + wireTap( + integrationFlow { + handle { m -> println(m.payload) } + }) + transform { it.toUpperCase() } + } +---- +==== + +Such a global `integrationFlow()` function expects a `@BuilderInference` for an `IntegrationFlowDefinition<*>` and produces a regular `IntegrationFlow` lambda implementation. +See more overloaded `integrationFlow()` variants below. + Many other scenarios require an `IntegrationFlow` to be started from source of data (e.g. `JdbcPollingChannelAdapter`, `JmsInboundGateway` or just an existing `MessageChannel`). For this purpose Spring Integration Java DSL provides an `IntegrationFlows` factory with its bunch of overloaded `from()` methods. This factory can be used in Kotlin as well: @@ -50,14 +71,14 @@ For example: @Bean fun functionFlow() = integrationFlow>({ it.beanName("functionGateway") }) { - it.transform { it.toUpperCase() } + transform { it.toUpperCase() } } @Bean fun messageSourceFlow() = integrationFlow(MessageProcessorMessageSource { "testSource" }, { it.poller { it.fixedDelay(10).maxMessagesPerPoll(1) } }) { - it.channel { c -> c.queue("fromSupplierQueue") } + channel { c -> c.queue("fromSupplierQueue") } } ---- ==== @@ -71,7 +92,7 @@ For example `IntegrationFlowDefinition<*>` requires a reifying for many methods @Bean fun convertFlow() = integrationFlow("convertFlowInput") { - it.convert() + convert() } ---- ==== diff --git a/spring-integration-kotlin-dsl/build.gradle b/spring-integration-kotlin-dsl/build.gradle index aebc7fa..bda2093 100644 --- a/spring-integration-kotlin-dsl/build.gradle +++ b/spring-integration-kotlin-dsl/build.gradle @@ -66,7 +66,7 @@ dependencyManagement { compileKotlin { kotlinOptions { jvmTarget = '1.8' - freeCompilerArgs = ["-Xjsr305=strict"] + freeCompilerArgs = ['-Xjsr305=strict', '-Xuse-experimental=kotlin.Experimental'] allWarningsAsErrors = true } } @@ -188,7 +188,7 @@ task distZip(type: Zip, dependsOn: docsZip) { from project.javadocJar } - from(zipTree(docsZip.archivePath)) { + from(zipTree(docsZip.archiveFile)) { into "${baseDir}/docs" } } diff --git a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowDsl.kt b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowDsl.kt index aaa25b9..3139a20 100644 --- a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowDsl.kt +++ b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowDsl.kt @@ -14,6 +14,8 @@ * limitations under the License. */ +@file:UseExperimental(kotlin.experimental.ExperimentalTypeInference::class) + package org.springframework.integration.dsl.kotlin import org.reactivestreams.Publisher @@ -40,6 +42,16 @@ private fun buildIntegrationFlow(flowBuilder: IntegrationFlowBuilder, return flowBuilder.get() } +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow] lambdas. + * + * @author Artem Bilan + */ +fun integrationFlow(@BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + IntegrationFlow { + flow(it) + } + /** * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - * `IntegrationFlows.from(Class, Consumer)` factory method. @@ -47,7 +59,7 @@ private fun buildIntegrationFlow(flowBuilder: IntegrationFlowBuilder, * @author Artem Bilan */ inline fun integrationFlow(crossinline gateway: (GatewayProxySpec) -> Unit = {}, - crossinline flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { + @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit): IntegrationFlow { val flowBuilder = IntegrationFlows.from(T::class.java) { gateway(it) } flow.invoke(flowBuilder) @@ -61,11 +73,8 @@ inline fun integrationFlow(crossinline gateway: (GatewayProxySpec) - * @author Artem Bilan */ fun integrationFlow(channelName: String, fixedSubscriber: Boolean = false, - flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - - val flowBuilder = IntegrationFlows.from(channelName, fixedSubscriber) - return buildIntegrationFlow(flowBuilder, flow) -} + @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(channelName, fixedSubscriber), flow) /** * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - @@ -73,10 +82,8 @@ fun integrationFlow(channelName: String, fixedSubscriber: Boolean = false, * * @author Artem Bilan */ -fun integrationFlow(channel: MessageChannel, flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - val flowBuilder = IntegrationFlows.from(channel) - return buildIntegrationFlow(flowBuilder, flow) -} +fun integrationFlow(channel: MessageChannel, @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(channel), flow) /** * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - @@ -86,11 +93,8 @@ fun integrationFlow(channel: MessageChannel, flow: (IntegrationFlowDefinition<*> */ fun integrationFlow(messageSource: MessageSource<*>, options: (SourcePollingChannelAdapterSpec) -> Unit = {}, - flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - - val flowBuilder = IntegrationFlows.from(messageSource, Consumer { options(it) }) - return buildIntegrationFlow(flowBuilder, flow) -} + @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(messageSource, Consumer { options(it) }), flow) /** * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - @@ -100,11 +104,8 @@ fun integrationFlow(messageSource: MessageSource<*>, */ fun integrationFlow(messageSource: MessageSourceSpec<*, out MessageSource<*>>, options: (SourcePollingChannelAdapterSpec) -> Unit = {}, - flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - - val flowBuilder = IntegrationFlows.from(messageSource, options) - return buildIntegrationFlow(flowBuilder, flow) -} + @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(messageSource, options), flow) /** * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - @@ -114,11 +115,8 @@ fun integrationFlow(messageSource: MessageSourceSpec<*, out MessageSource<*>>, */ fun integrationFlow(source: () -> Any, options: (SourcePollingChannelAdapterSpec) -> Unit = {}, - flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - - val flowBuilder = IntegrationFlows.from(source, options) - return buildIntegrationFlow(flowBuilder, flow) -} + @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(source, options), flow) /** * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - @@ -127,11 +125,8 @@ fun integrationFlow(source: () -> Any, * @author Artem Bilan */ fun integrationFlow(publisher: Publisher>, - flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - - val flowBuilder = IntegrationFlows.from(publisher) - return buildIntegrationFlow(flowBuilder, flow) -} + @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(publisher), flow) /** * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - @@ -140,11 +135,8 @@ fun integrationFlow(publisher: Publisher>, * @author Artem Bilan */ fun integrationFlow(gateway: MessagingGatewaySupport, - flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - - val flowBuilder = IntegrationFlows.from(gateway) - return buildIntegrationFlow(flowBuilder, flow) -} + @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(gateway), flow) /** * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - @@ -153,11 +145,8 @@ fun integrationFlow(gateway: MessagingGatewaySupport, * @author Artem Bilan */ fun integrationFlow(gatewaySpec: MessagingGatewaySpec<*, *>, - flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - - val flowBuilder = IntegrationFlows.from(gatewaySpec) - return buildIntegrationFlow(flowBuilder, flow) -} + @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(gatewaySpec), flow) /** * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - @@ -166,11 +155,8 @@ fun integrationFlow(gatewaySpec: MessagingGatewaySpec<*, *>, * @author Artem Bilan */ fun integrationFlow(producer: MessageProducerSupport, - flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - - val flowBuilder = IntegrationFlows.from(producer) - return buildIntegrationFlow(flowBuilder, flow) -} + @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(producer), flow) /** * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - @@ -179,8 +165,5 @@ fun integrationFlow(producer: MessageProducerSupport, * @author Artem Bilan */ fun integrationFlow(producerSpec: MessageProducerSpec<*, *>, - flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - - val flowBuilder = IntegrationFlows.from(producerSpec) - return buildIntegrationFlow(flowBuilder, flow) -} + @BuilderInference flow: IntegrationFlowDefinition<*>.() -> Unit) = + buildIntegrationFlow(IntegrationFlows.from(producerSpec), flow) diff --git a/spring-integration-kotlin-dsl/src/test/kotlin/org/springframework/integration/dsl/kotlin/test/KotlinDslTests.kt b/spring-integration-kotlin-dsl/src/test/kotlin/org/springframework/integration/dsl/kotlin/test/KotlinDslTests.kt index 880965d..b9f6edc 100644 --- a/spring-integration-kotlin-dsl/src/test/kotlin/org/springframework/integration/dsl/kotlin/test/KotlinDslTests.kt +++ b/spring-integration-kotlin-dsl/src/test/kotlin/org/springframework/integration/dsl/kotlin/test/KotlinDslTests.kt @@ -164,8 +164,8 @@ class KotlinDslTests { val integrationFlow = integrationFlow(publisher) { - it.transformReified, Int>({ it.payload * 2 }) { it.id("foo") } - .channel(fluxChannel) + transformReified, Int>({ it.payload * 2 }) { it.id("foo") } + channel(fluxChannel) } val registration = this.integrationFlowContext.registration(integrationFlow).register() @@ -175,6 +175,24 @@ class KotlinDslTests { 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") + } + @Configuration @EnableIntegration class Config { @@ -186,56 +204,66 @@ class KotlinDslTests { @Bean fun convertFlow() = integrationFlow("convertFlowInput") { - it.convert() - .convert { it.id("kotlinConverter") } + convert() + convert { it.id("kotlinConverter") } } @Bean fun functionFlow() = integrationFlow>({ it.beanName("functionGateway") }) { - it.transform { it.toUpperCase() } - .split({ p -> p }) + transform { it.toUpperCase() } + split({ p -> p }) } @Bean fun functionFlow2() = integrationFlow> { - it.transform { it.toLowerCase() } - .routeReified, Any?> ({ m -> m.headers.replyChannel }) { it.id("router") } + transform { it.toLowerCase() } + routeReified, Any?>({ m -> m.headers.replyChannel }) { it.id("router") } } @Bean fun messageSourceFlow() = integrationFlow(MessageProcessorMessageSource { "testSource" }, { it.poller { it.trigger(OnlyOnceTrigger()) } }) { - it.channel { c -> c.queue("fromSupplierQueue") } + channel { c -> c.queue("fromSupplierQueue") } } @Bean fun messageSourceFlow2() = integrationFlow(MessageProcessorMessageSource { "testSource2" }) { - it.channel { c -> c.queue("fromSupplierQueue2") } + channel { c -> c.queue("fromSupplierQueue2") } } @Bean fun fixedSubscriberFlow() = integrationFlow("fixedSubscriberInput", true) { - it.logAndReply(LoggingHandler.Level.WARN) + logAndReply(LoggingHandler.Level.WARN) } @Bean fun flowFromSupplier() = integrationFlow({ "testSupplier" }) { - it.channel { c -> c.queue("testSupplierResult") } + channel { c -> c.queue("testSupplierResult") } } @Bean fun flowFromSupplier2() = integrationFlow({ "testSupplier2" }, { it.poller { it.trigger(OnlyOnceTrigger()) } }) { - it - .filterReified>({ m -> m.payload is String }) - .channel { c -> c.queue("testSupplierResult2") } + filterReified>({ m -> m.payload is String }) + channel { c -> c.queue("testSupplierResult2") } + } + + @Bean + fun flowLambda() = + integrationFlow { + filter { it === "test" } + wireTap( + integrationFlow { + channel { c -> c.queue("wireTapChannel") } + }) + transform { it.toUpperCase() } } }