Add integrationFlow(source: () -> Any)

* Add more tests to cover all the API

spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowDsl.kt

@@ -19,13 +19,15 @@ package org.springframework.integration.dsl.kotlin
 import org.springframework.integration.core.MessageSource
 import org.springframework.integration.dsl.GatewayProxySpec
 import org.springframework.integration.dsl.IntegrationFlow
+import org.springframework.integration.dsl.IntegrationFlowBuilder
 import org.springframework.integration.dsl.IntegrationFlowDefinition
 import org.springframework.integration.dsl.IntegrationFlows
 import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec
 import java.util.function.Consumer
 
 /**
- * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from(Class)]
+ * Functional [IntegrationFlow] definition in Kotlin DSL for
+ * [IntegrationFlows.from(Class<*>, Consumer<GatewayProxySpec>)]
  *
  * @author Artem Bilan
  */
@@ -38,7 +40,8 @@ inline fun <reified T> integrationFlow(crossinline gateway: (GatewayProxySpec) -
 }
 
 /**
- * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from(String,Boolean)]
+ * Functional [IntegrationFlow] definition in Kotlin DSL for
+ * [IntegrationFlows.from(String, Boolean)]
  *
  * @author Artem Bilan
  */
@@ -46,22 +49,40 @@ fun integrationFlow(channelName: String, fixedSubscriber: Boolean = false,
 					flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow {
 
 	val flowBuilder = IntegrationFlows.from(channelName, fixedSubscriber)
-	flow.invoke(flowBuilder)
+	return buildIntegrationFlow(flowBuilder, flow)
-	return flowBuilder.get()
 }
 
 /**
  * Functional [IntegrationFlow] definition in Kotlin DSL for
- * [IntegrationFlows.from(MessageSource<*>,Consumer<SourcePollingChannelAdapterSpec>)]
+ * [IntegrationFlows.from(MessageSource<*>, Consumer<SourcePollingChannelAdapterSpec>)]
  *
  * @author Artem Bilan
  */
 fun integrationFlow(messageSource: MessageSource<*>,
-						   options: (SourcePollingChannelAdapterSpec) -> Unit = {},
+					options: (SourcePollingChannelAdapterSpec) -> Unit = {},
 					flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow {
 
-	val flowBuilder = IntegrationFlows.from(messageSource, Consumer {options(it)})
+	val flowBuilder = IntegrationFlows.from(messageSource, Consumer { options(it) })
+	return buildIntegrationFlow(flowBuilder, flow)
+}
+
+/**
+ * Functional [IntegrationFlow] definition in Kotlin DSL for
+ * [IntegrationFlows.from(Supplier<*>, Consumer<SourcePollingChannelAdapterSpec>)]
+ *
+ * @author Artem Bilan
+ */
+fun integrationFlow(source: () -> Any,
+						options: (SourcePollingChannelAdapterSpec) -> Unit = {},
+						flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow {
+
+	val flowBuilder = IntegrationFlows.from(source, options)
+	return buildIntegrationFlow(flowBuilder, flow)
+}
+
+private fun buildIntegrationFlow(flowBuilder: IntegrationFlowBuilder,
+								 flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow {
+
 	flow.invoke(flowBuilder)
 	return flowBuilder.get()
 }
-

spring-integration-kotlin-dsl/src/test/kotlin/org/springframework/integration/dsl/kotlin/test/KotlinDslTests.kt

@@ -23,14 +23,20 @@ import assertk.assertions.isNotNull
 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.QueueChannel
 import org.springframework.integration.config.EnableIntegration
+import org.springframework.integration.core.MessagingTemplate
+import org.springframework.integration.dsl.Pollers
 import org.springframework.integration.dsl.kotlin.convert
 import org.springframework.integration.dsl.kotlin.integrationFlow
 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.MessageChannel
 import org.springframework.messaging.MessageHeaders
 import org.springframework.messaging.PollableChannel
@@ -53,7 +59,7 @@ class KotlinDslTests {
 	private lateinit var convertFlowInput: MessageChannel
 
 	@Test
-	fun `test convert extension`() {
+	fun `convert extension`() {
 		assertThat(this.beanFactory.containsBean("kotlinConverter"))
 
 		val replyChannel = QueueChannel()
@@ -72,10 +78,11 @@ class KotlinDslTests {
 	}
 
 	@Autowired
+	@Qualifier("functionGateway")
 	private lateinit var upperCaseFunction: Function<String, String>
 
 	@Test
-	fun `test uppercase function`() {
+	fun `uppercase function`() {
 		assertThat(this.upperCaseFunction.apply("test")).isEqualTo("TEST")
 	}
 
@@ -83,15 +90,60 @@ class KotlinDslTests {
 	private lateinit var fromSupplierQueue: PollableChannel
 
 	@Test
-	fun `verify supplier flow`() {
+	fun `message source flow`() {
-		assertThat(this.fromSupplierQueue.receive(10_000)).isNotNull()
+		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")
+	}
 
 	@Configuration
 	@EnableIntegration
 	class Config {
 
+		@Bean(PollerMetadata.DEFAULT_POLLER)
+		fun defaultPoller() =
+				Pollers.fixedDelay(100).maxMessagesPerPoll(1).get()
+
 		@Bean
 		fun convertFlow() =
 				integrationFlow("convertFlowInput") {
@@ -105,12 +157,44 @@ class KotlinDslTests {
 					it.transform<String, String> { it.toUpperCase() }
 				}
 
+		@Bean
+		fun functionFlow2() =
+				integrationFlow<Function<*, *>> {
+					it.transform<String, String> { it.toLowerCase() }
+				}
+
 		@Bean
 		fun messageSourceFlow() =
 				integrationFlow(MessageProcessorMessageSource { "testSource" },
-						{ it.poller { it.fixedDelay(10).maxMessagesPerPoll(1) } }) {
+						{ it.poller { it.trigger(OnlyOnceTrigger()) } }) {
 					it.channel { c -> c.queue("fromSupplierQueue") }
 				}
+
+		@Bean
+		fun messageSourceFlow2() =
+				integrationFlow(MessageProcessorMessageSource { "testSource2" }) {
+					it.channel { c -> c.queue("fromSupplierQueue2") }
+				}
+
+		@Bean
+		fun fixedSubscriberFlow() =
+				integrationFlow("fixedSubscriberInput", true) {
+					it.logAndReply(LoggingHandler.Level.WARN)
+				}
+
+		@Bean
+		fun flowFromSupplier() =
+				integrationFlow({ "testSupplier" }) {
+					it.channel { c -> c.queue("testSupplierResult") }
+				}
+
+		@Bean
+		fun flowFromSupplier2() =
+				integrationFlow({ "testSupplier2" },
+						{ it.poller { it.trigger(OnlyOnceTrigger()) } }) {
+					it.channel { c -> c.queue("testSupplierResult2") }
+				}
+
 	}
 
 	data class TestPojo(val name: String?, val date: Date?)