diff --git a/spring-integration-kotlin-dsl/build.gradle b/spring-integration-kotlin-dsl/build.gradle index 12f0d98..5620dd3 100644 --- a/spring-integration-kotlin-dsl/build.gradle +++ b/spring-integration-kotlin-dsl/build.gradle @@ -5,8 +5,9 @@ plugins { id 'jacoco' id 'org.sonarqube' version '2.7.1' id 'io.spring.dependency-management' version '1.0.8.RELEASE' - id "org.jetbrains.kotlin.jvm" version "$kotlinVersion" + id 'org.jetbrains.kotlin.jvm' version "$kotlinVersion" id "org.jetbrains.kotlin.plugin.spring" version "$kotlinVersion" + id 'org.jetbrains.dokka' version '0.9.18' } apply from: "${rootProject.projectDir}/publish-maven.gradle" @@ -16,6 +17,7 @@ description = 'Spring Integration Kotlin DSL' group = 'org.springframework.integration' repositories { +// mavenLocal() if (version.endsWith('BUILD-SNAPSHOT')) { maven { url 'https://repo.spring.io/libs-snapshot' } } @@ -25,9 +27,10 @@ repositories { ext { assertkVersion = '0.20' - jacksonVersion = '2.10.0.pr3' + jacksonVersion = '2.10.0' junitVersion = '5.5.2' log4jVersion = '2.12.1' + reactorVersion = 'Dysprosium-RELEASE' springIntegrationVersion = '5.2.0.BUILD-SNAPSHOT' idPrefix = 'kotlin-dsl' @@ -55,6 +58,7 @@ dependencyManagement { mavenBom "org.junit:junit-bom:$junitVersion" mavenBom "org.springframework.integration:spring-integration-bom:$springIntegrationVersion" mavenBom "org.apache.logging.log4j:log4j-bom:$log4jVersion" + mavenBom "io.projectreactor:reactor-bom:$reactorVersion" } } @@ -89,6 +93,7 @@ dependencies { testImplementation 'org.springframework.integration:spring-integration-test' testImplementation 'org.junit.jupiter:junit-jupiter-api' testImplementation "com.willowtreeapps.assertk:assertk-jvm:$assertkVersion" + testImplementation 'io.projectreactor:reactor-test' testRuntime 'com.fasterxml.jackson.module:jackson-module-kotlin' testRuntime 'org.junit.jupiter:junit-jupiter-engine' @@ -115,7 +120,21 @@ jacocoTestReport { } } -check.dependsOn javadoc +task api(type: org.jetbrains.dokka.gradle.DokkaTask) { + outputFormat = 'html' + outputDirectory = "$buildDir/javadoc" + externalDocumentationLink { + url = new URL("https://docs.spring.io/spring-integration/docs/$springIntegrationVersion/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/') + } +} + +check.dependsOn api build.dependsOn jacocoTestReport @@ -126,7 +145,7 @@ task sourcesJar(type: Jar) { task javadocJar(type: Jar) { archiveClassifier = 'javadoc' - from javadoc + from api } sonarqube { @@ -141,20 +160,6 @@ sonarqube { } } -task api(type: Javadoc) { - group = 'Documentation' - description = 'Generates the Javadoc API documentation.' - title = "${rootProject.description} ${version} API" - options.memberLevel = org.gradle.external.javadoc.JavadocMemberLevel.PROTECTED - options.author = true - options.header = rootProject.description - options.overview = 'src/api/overview.html' - - source = sourceSets.main.kotlin - classpath = project.sourceSets.main.compileClasspath - destinationDir = new File(buildDir, 'api') -} - task distZip(type: Zip) { group = 'Distribution' archiveClassifier = 'dist' diff --git a/spring-integration-kotlin-dsl/src/api/overview.html b/spring-integration-kotlin-dsl/src/api/overview.html deleted file mode 100644 index dd3e4da..0000000 --- a/spring-integration-kotlin-dsl/src/api/overview.html +++ /dev/null @@ -1,22 +0,0 @@ - - -This document is the API specification for Spring Integration Cassandra Extension -
-
-

- For further API reference and developer documentation, see the - Spring - Integration reference documentation. - That documentation contains more detailed, developer-targeted - descriptions, with conceptual overviews, definitions of terms, - workarounds, and working code examples. -

- -

- If you are interested in commercial training, consultancy, and - support for Spring Integration, please visit - https://spring.io/ -

-
- - diff --git a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowBuilderExtensions.kt b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowBuilderExtensions.kt index e91ee65..a8c277b 100644 --- a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowBuilderExtensions.kt +++ b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowBuilderExtensions.kt @@ -21,10 +21,22 @@ import org.springframework.integration.dsl.IntegrationFlowDefinition import org.springframework.integration.transformer.MessageTransformingHandler /** - * Extension for [IntegrationFlowDefinition.convert()] providing a `convert()` variant. + * Extension for [IntegrationFlowDefinition.convert] providing a `convert()` variant. * * @author Artem Bilan */ inline fun IntegrationFlowDefinition<*>.convert( - crossinline consumer: (GenericEndpointSpec) -> Unit = {}): IntegrationFlowDefinition<*> = - convert(T::class.java) { consumer(it) } + crossinline configurer: (GenericEndpointSpec) -> Unit = {}): + IntegrationFlowDefinition<*> = + convert(T::class.java) { configurer(it) } + +/** + * Extension for [IntegrationFlowDefinition.transform] providing a `transform()` variant. + * + * @author Artem Bilan + */ +inline fun IntegrationFlowDefinition<*>.reifiedTransform( + crossinline function: (P) -> T, + crossinline configurer: (GenericEndpointSpec) -> Unit = {}): + IntegrationFlowDefinition<*> = + transform(P::class.java, { function(it) }) { configurer(it) } 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 0fdfb6d..aaa25b9 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 @@ -16,18 +16,33 @@ package org.springframework.integration.dsl.kotlin +import org.reactivestreams.Publisher 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.MessageProducerSpec +import org.springframework.integration.dsl.MessageSourceSpec +import org.springframework.integration.dsl.MessagingGatewaySpec import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec +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: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { + + flow.invoke(flowBuilder) + return flowBuilder.get() +} + /** - * Functional [IntegrationFlow] definition in Kotlin DSL for - * [IntegrationFlows.from(Class<*>, Consumer)] + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(Class, Consumer)` factory method. * * @author Artem Bilan */ @@ -40,8 +55,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] - + * `IntegrationFlows.from(String, Boolean)` factory method. * * @author Artem Bilan */ @@ -53,8 +68,19 @@ fun integrationFlow(channelName: String, fixedSubscriber: Boolean = false, } /** - * Functional [IntegrationFlow] definition in Kotlin DSL for - * [IntegrationFlows.from(MessageSource<*>, Consumer)] + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessageChannel)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(channel: MessageChannel, flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { + val flowBuilder = IntegrationFlows.from(channel) + return buildIntegrationFlow(flowBuilder, flow) +} + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessageSource<*>, Consumer)` factory method. * * @author Artem Bilan */ @@ -67,22 +93,94 @@ fun integrationFlow(messageSource: MessageSource<*>, } /** - * Functional [IntegrationFlow] definition in Kotlin DSL for - * [IntegrationFlows.from(Supplier<*>, Consumer)] + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessageSourceSpec<*>, Consumer)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(messageSource: MessageSourceSpec<*, out MessageSource<*>>, + options: (SourcePollingChannelAdapterSpec) -> Unit = {}, + flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { + + val flowBuilder = IntegrationFlows.from(messageSource, options) + return buildIntegrationFlow(flowBuilder, flow) +} + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(Supplier<*>, Consumer)` factory method. * * @author Artem Bilan */ fun integrationFlow(source: () -> Any, - options: (SourcePollingChannelAdapterSpec) -> Unit = {}, - flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { + 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 { +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(Publisher>)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(publisher: Publisher>, + flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { - flow.invoke(flowBuilder) - return flowBuilder.get() + val flowBuilder = IntegrationFlows.from(publisher) + return buildIntegrationFlow(flowBuilder, flow) +} + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessagingGatewaySupport)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(gateway: MessagingGatewaySupport, + flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { + + val flowBuilder = IntegrationFlows.from(gateway) + return buildIntegrationFlow(flowBuilder, flow) +} + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessagingGatewaySpec<*, *>)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(gatewaySpec: MessagingGatewaySpec<*, *>, + flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { + + val flowBuilder = IntegrationFlows.from(gatewaySpec) + return buildIntegrationFlow(flowBuilder, flow) +} + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessageProducerSupport)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(producer: MessageProducerSupport, + flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { + + val flowBuilder = IntegrationFlows.from(producer) + return buildIntegrationFlow(flowBuilder, flow) +} + +/** + * Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] - + * `IntegrationFlows.from(MessageProducerSpec<*, *>)` factory method. + * + * @author Artem Bilan + */ +fun integrationFlow(producerSpec: MessageProducerSpec<*, *>, + flow: (IntegrationFlowDefinition<*>) -> Unit): IntegrationFlow { + + val flowBuilder = IntegrationFlows.from(producerSpec) + return buildIntegrationFlow(flowBuilder, 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 bdc6588..9777676 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 @@ -26,22 +26,29 @@ 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.MessagingTemplate import org.springframework.integration.dsl.Pollers +import org.springframework.integration.dsl.context.IntegrationFlowContext import org.springframework.integration.dsl.kotlin.convert +import org.springframework.integration.dsl.kotlin.reifiedTransform 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.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 @@ -55,6 +62,9 @@ class KotlinDslTests { @Autowired private lateinit var beanFactory: BeanFactory + @Autowired + private lateinit var integrationFlowContext: IntegrationFlowContext + @Autowired private lateinit var convertFlowInput: MessageChannel @@ -136,6 +146,32 @@ class KotlinDslTests { 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) { + it.reifiedTransform, Int>({ it.payload * 2 }) { it.id("foo") } + .channel(fluxChannel) + } + + val registration = this.integrationFlowContext.registration(integrationFlow).register() + + verifyLater.verify() + + registration.destroy() + } + @Configuration @EnableIntegration class Config {