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

@@ -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]) {

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<GatewayProxySpec>)` factory method.
+ *
+ * @author Artem Bilan
+ */
+inline fun <reified T> 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<SourcePollingChannelAdapterSpec>)` 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<SourcePollingChannelAdapterSpec>)` 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<SourcePollingChannelAdapterSpec>)` 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<out Message<*>>)` factory method.
+ *
+ * @author Artem Bilan
+ */
+fun integrationFlow(publisher: Publisher<out Message<*>>,
+					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)

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<String, String>
+
+	@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<String, String>
+
+	@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<Message<Int>>({ 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<TestPojo>()
+					convert<TestPojo> { id("kotlinConverter") }
+					handle { m -> (m.headers[MessageHeaders.REPLY_CHANNEL] as MessageChannel).send(m) }
+				}
+
+		@Bean
+		fun functionFlow() =
+				integrationFlow<Function<String, String>>({ beanName("functionGateway") }) {
+					transform<String> { it.toUpperCase() }
+					split<Message<*>> { it.payload }
+					split<String>({ it }) { id("splitterEndpoint") }
+					resequence()
+					aggregate {
+						id("aggregator")
+						outputProcessor { it.one }
+					}
+				}
+
+		@Bean
+		fun functionFlow2() =
+				integrationFlow<Function<*, *>> {
+					transform<String> { it.toLowerCase() }
+					route<Message<*>, Any?>({ null }) { defaultOutputToParentFlow() }
+					route<Message<*>> { 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<Any>(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<Message<*>> { m -> m.payload is String }
+					channel { queue("testSupplierResult2") }
+				}
+
+		@Bean
+		fun flowLambda() =
+				integrationFlow {
+					filter<String>({ it === "test" }) { id("filterEndpoint") }
+					wireTap {
+						channel { queue("wireTapChannel") }
+					}
+					delay("delayGroup") { defaultDelay(100) }
+					transform<String> { 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<Any> { true }, integrationFlow { handle<Any> { _, _ -> Math.random() * 10 } })
+								recipientFlow(GenericSelector<Any> { true }, integrationFlow { handle<Any> { _, _ -> Math.random() * 10 } })
+								recipientFlow(GenericSelector<Any> { true }, integrationFlow { handle<Any> { _, _ -> 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?)
+
+}

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<Int, Boolean>({ p -> p % 2 == 0 },
-									{ m ->
-										m.subFlowMapping(true, { sf -> sf.handle<Int> { p, _ -> p * 2 } })
-												.subFlowMapping(false, { sf -> sf.handle<Int> { p, _ -> p * 3 } })
-									})
-							.aggregate()
-							.channel { c -> c.queue("routerTwoSubFlowsOutput") }
+				integrationFlow {
+					split()
+					route<Int, Boolean>({ it % 2 == 0 }) {
+						subFlowMapping(true) { sf -> sf.handle<Int> { p, _ -> p * 2 } }
+						subFlowMapping(false) { sf -> sf.handle<Int> { p, _ -> p * 3 } }
+					}
+					aggregate()
+					channel { queue("routerTwoSubFlowsOutput") }
 				}
 
 		@Bean
 		fun splitRouteAggregate() =
-				IntegrationFlow { f ->
-					f.split()
-							.route<Int, Boolean>({ o -> o % 2 == 0 },
-									{ m ->
-										m.subFlowMapping(true) { sf -> sf.gateway(oddFlow()) }
-												.subFlowMapping(false) { sf -> sf.gateway(evenFlow()) }
-									})
-							.aggregate()
+				integrationFlow {
+					split()
+					route<Int, Boolean>({ it % 2 == 0 }) {
+						subFlowMapping(true) { sf -> sf.gateway(oddFlow()) }
+						subFlowMapping(false) { sf -> sf.gateway(evenFlow()) }
+					}
+					aggregate()
 				}
 
 		@Bean
 		fun oddFlow() =
-				IntegrationFlow { flow ->
-					flow.handle<Any> { _, _ -> "odd" }
+				integrationFlow {
+					handle<Any> { _, _ -> "odd" }
 				}
 
 		@Bean
 		fun evenFlow() =
-				IntegrationFlow { flow ->
-					flow.handle<Any> { _, _ -> "even" }
+				integrationFlow {
+					handle<Any> { _, _ -> "even" }
 				}
 
 		@Bean

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<String>({ "" }) { e -> e.poller { p -> p.fixedDelay(10).maxMessagesPerPoll(1) } }
-						.transform<String, String> { "blank" }
-						.channel { c -> c.queue("fromSupplierQueue") }
-						.get()
+				integrationFlow({ "" }, { poller { it.fixedDelay(10).maxMessagesPerPoll(1) } }) {
+					transform<String> { "blank" }
+					channel { queue("fromSupplierQueue") }
+				}
 
 		@Bean
 		fun monoFunctionGateway() =
-				IntegrationFlows.from(MonoFunction::class.java) { gateway -> gateway.proxyDefaultMethods(true) }
-						.handle<String>({ p, _ -> Mono.just(p).map(String::toUpperCase) }) { e -> e.async(true) }
-						.get()
+				integrationFlow<MonoFunction>({ proxyDefaultMethods(true) }) {
+					handle<String>({ p, _ -> Mono.just(p).map(String::toUpperCase) }) { async(true) }
+				}
+
 	}
 
 	interface MonoFunction : Function<String, Mono<Message<*>>>

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<Any> { _ -> DeliveryMode.NON_PERSISTENT }
-							.timeToLiveExpression("10000")
-							.configureJmsTemplate { t -> t.explicitQosEnabled(true) })
+				integrationFlow {
+					handle(Jms.outboundAdapter(jmsConnectionFactory())
+							.apply {
+								destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)
+								deliveryModeFunction<Any> { 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())
+				}
 
 	}