Apply @BuilderInference for integrationFlow() functions

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<String> { it === "test" }
+        wireTap(
+                integrationFlow {
+                    handle { m -> println(m.payload) }
+                })
+        transform<String, String> { 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<Function<String, String>>({ it.beanName("functionGateway") }) {
-            it.transform<String, String> { it.toUpperCase() }
+            transform<String, String> { 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<TestPojo>()
+        convert<TestPojo>()
     }
 ----
 ====

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"
     }
 }

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<GatewayProxySpec>)` factory method.
@@ -47,7 +59,7 @@ private fun buildIntegrationFlow(flowBuilder: IntegrationFlowBuilder,
  * @author Artem Bilan
  */
 inline fun <reified T> 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 <reified T> 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<out Message<*>>,
-					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<out Message<*>>,
  * @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)

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<Message<Int>, Int>({ it.payload * 2 }) { it.id("foo") }
-							.channel(fluxChannel)
+					transformReified<Message<Int>, 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<TestPojo>()
-							.convert<TestPojo> { it.id("kotlinConverter") }
+					convert<TestPojo>()
+					convert<TestPojo> { it.id("kotlinConverter") }
 				}
 
 		@Bean
 		fun functionFlow() =
 				integrationFlow<Function<String, String>>({ it.beanName("functionGateway") }) {
-					it.transform<String, String> { it.toUpperCase() }
-							.split<String>({ p -> p })
+					transform<String, String> { it.toUpperCase() }
+					split<String>({ p -> p })
 				}
 
 		@Bean
 		fun functionFlow2() =
 				integrationFlow<Function<*, *>> {
-					it.transform<String, String> { it.toLowerCase() }
-							.routeReified<Message<*>, Any?> ({ m -> m.headers.replyChannel }) { it.id("router") }
+					transform<String, String> { it.toLowerCase() }
+					routeReified<Message<*>, 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<Message<Any>>({ m -> m.payload is String })
-							.channel { c -> c.queue("testSupplierResult2") }
+					filterReified<Message<Any>>({ m -> m.payload is String })
+					channel { c -> c.queue("testSupplierResult2") }
+				}
+
+		@Bean
+		fun flowLambda() =
+				integrationFlow {
+					filter<String> { it === "test" }
+					wireTap(
+							integrationFlow {
+								channel { c -> c.queue("wireTapChannel") }
+							})
+					transform<String, String> { it.toUpperCase() }
 				}
 
 	}