From 800b398fc133b1d74eaccbba032c75e4684029ce Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 25 Sep 2019 18:25:35 -0400 Subject: [PATCH] Add `integrationFlow(source: () -> Any)` * Add more tests to cover all the API --- .../dsl/kotlin/IntegrationFlowDsl.kt | 37 ++++++-- .../dsl/kotlin/test/KotlinDslTests.kt | 94 ++++++++++++++++++- 2 files changed, 118 insertions(+), 13 deletions(-) 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 f937eca..0fdfb6d 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 @@ -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)] * * @author Artem Bilan */ @@ -38,7 +40,8 @@ inline fun 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 flowBuilder.get() + return buildIntegrationFlow(flowBuilder, flow) } /** * Functional [IntegrationFlow] definition in Kotlin DSL for - * [IntegrationFlows.from(MessageSource<*>,Consumer)] + * [IntegrationFlows.from(MessageSource<*>, Consumer)] * * @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)] + * + * @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() } - 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 ed735ca..bdc6588 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 @@ -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 @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`() { - assertThat(this.fromSupplierQueue.receive(10_000)).isNotNull() + 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") + } @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 { it.toUpperCase() } } + @Bean + fun functionFlow2() = + integrationFlow> { + it.transform { 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?)