From d4c0cf01700868a0315a02abdf07d170e93c4499 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 3 Jan 2020 14:14:40 -0500 Subject: [PATCH] Rework more lambdas into Kotlin builder pattern * Move classes in this project to their own `kotlin` package to avoid conflict with other Spring Integration artifacts * Add lambda-based `handle()` instead of `GenericHandler` one * Swap arguments in some EIP-method for better code flow when build with Kotlin DSL * Add a `scatterGatherFlow` configuration into a test suite to demonstrate a complex configuration with Java API interoperability. The Java sample is present in the comment for this flow definition --- spring-integration-kotlin-dsl/build.gradle | 2 +- .../dsl/{ => kotlin}/IntegrationFlowDsl.kt | 18 +- .../KotlinIntegrationFlowDefinition.kt | 188 ++++++++++-------- .../dsl/kotlin/test/KotlinDslTests.kt | 94 +++++++-- 4 files changed, 193 insertions(+), 109 deletions(-) rename spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/{ => kotlin}/IntegrationFlowDsl.kt (87%) rename spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/{ => kotlin}/KotlinIntegrationFlowDefinition.kt (82%) diff --git a/spring-integration-kotlin-dsl/build.gradle b/spring-integration-kotlin-dsl/build.gradle index bbe9115..a65e5d2 100644 --- a/spring-integration-kotlin-dsl/build.gradle +++ b/spring-integration-kotlin-dsl/build.gradle @@ -29,7 +29,7 @@ ext { assertkVersion = '0.20' jacksonVersion = '2.10.1' junitVersion = '5.5.2' - log4jVersion = '2.12.1' + log4jVersion = '2.13.0' reactorVersion = 'Dysprosium-SR1' springIntegrationVersion = '5.2.2.RELEASE' diff --git a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowDsl.kt similarity index 87% rename from spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt rename to spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowDsl.kt index 0f7302d..33ea1c1 100644 --- a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt +++ b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/IntegrationFlowDsl.kt @@ -14,10 +14,18 @@ * limitations under the License. */ -package org.springframework.integration.dsl +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.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 @@ -48,7 +56,7 @@ fun integrationFlow(flow: KotlinIntegrationFlowDefinition.() -> Unit) = * @author Artem Bilan */ inline fun integrationFlow( - crossinline gateway: (GatewayProxySpec) -> Unit = {}, + crossinline gateway: GatewayProxySpec.() -> Unit = {}, flow: KotlinIntegrationFlowDefinition.() -> Unit): IntegrationFlow { val flowBuilder = IntegrationFlows.from(T::class.java) { gateway(it) } @@ -82,7 +90,7 @@ fun integrationFlow(channel: MessageChannel, flow: KotlinIntegrationFlowDefiniti * @author Artem Bilan */ fun integrationFlow(messageSource: MessageSource<*>, - options: (SourcePollingChannelAdapterSpec) -> Unit = {}, + options: SourcePollingChannelAdapterSpec.() -> Unit = {}, flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(messageSource, Consumer { options(it) }), flow) @@ -93,7 +101,7 @@ fun integrationFlow(messageSource: MessageSource<*>, * @author Artem Bilan */ fun integrationFlow(messageSource: MessageSourceSpec<*, out MessageSource<*>>, - options: (SourcePollingChannelAdapterSpec) -> Unit = {}, + options: SourcePollingChannelAdapterSpec.() -> Unit = {}, flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(messageSource, options), flow) @@ -104,7 +112,7 @@ fun integrationFlow(messageSource: MessageSourceSpec<*, out MessageSource<*>>, * @author Artem Bilan */ fun integrationFlow(source: () -> Any, - options: (SourcePollingChannelAdapterSpec) -> Unit = {}, + options: SourcePollingChannelAdapterSpec.() -> Unit = {}, flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(source, options), flow) diff --git a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/KotlinIntegrationFlowDefinition.kt similarity index 82% rename from spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt rename to spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/KotlinIntegrationFlowDefinition.kt index f5cff38..39ee511 100644 --- a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt +++ b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/kotlin/KotlinIntegrationFlowDefinition.kt @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dsl +package org.springframework.integration.dsl.kotlin import org.reactivestreams.Publisher import org.springframework.expression.Expression @@ -22,6 +22,27 @@ import org.springframework.integration.aggregator.AggregatingMessageHandler import org.springframework.integration.channel.FluxMessageChannel import org.springframework.integration.channel.interceptor.WireTap import org.springframework.integration.core.MessageSelector +import org.springframework.integration.dsl.AggregatorSpec +import org.springframework.integration.dsl.BarrierSpec +import org.springframework.integration.dsl.Channels +import org.springframework.integration.dsl.DelayerEndpointSpec +import org.springframework.integration.dsl.EnricherSpec +import org.springframework.integration.dsl.FilterEndpointSpec +import org.springframework.integration.dsl.GatewayEndpointSpec +import org.springframework.integration.dsl.GenericEndpointSpec +import org.springframework.integration.dsl.HeaderEnricherSpec +import org.springframework.integration.dsl.IntegrationFlow +import org.springframework.integration.dsl.IntegrationFlowDefinition +import org.springframework.integration.dsl.MessageChannelSpec +import org.springframework.integration.dsl.MessageHandlerSpec +import org.springframework.integration.dsl.MessageProcessorSpec +import org.springframework.integration.dsl.PublishSubscribeSpec +import org.springframework.integration.dsl.RecipientListRouterSpec +import org.springframework.integration.dsl.ResequencerSpec +import org.springframework.integration.dsl.RouterSpec +import org.springframework.integration.dsl.ScatterGatherSpec +import org.springframework.integration.dsl.SplitterEndpointSpec +import org.springframework.integration.dsl.WireTapSpec import org.springframework.integration.dsl.support.MessageChannelReference import org.springframework.integration.filter.MessageFilter import org.springframework.integration.filter.MethodInvokingSelector @@ -53,10 +74,10 @@ import org.springframework.integration.transformer.MethodInvokingTransformer import org.springframework.messaging.Message import org.springframework.messaging.MessageChannel import org.springframework.messaging.MessageHandler +import org.springframework.messaging.MessageHeaders import reactor.core.publisher.Flux import java.util.concurrent.Executor import java.util.function.Consumer -import java.util.function.Function /** * An [IntegrationFlowDefinition] wrapped for Kotlin DSL. @@ -72,7 +93,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * with reified generic type. */ inline fun convert( - crossinline configurer: (GenericEndpointSpec) -> Unit = {}) { + crossinline configurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.convert(T::class.java) { configurer(it) } } @@ -91,7 +112,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ */ inline fun transform( crossinline function: (P) -> Any, - crossinline configurer: (GenericEndpointSpec) -> Unit) { + crossinline configurer: GenericEndpointSpec.() -> Unit) { this.delegate.transform(P::class.java, { function(it) }) { configurer(it) } } @@ -100,7 +121,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Inline function for [IntegrationFlowDefinition.split] providing a `split()` variant * with reified generic type. */ - inline fun split(noinline function: (P) -> Any) { + inline fun split(crossinline function: (P) -> Any) { this.delegate.split(P::class.java) { function(it) } } @@ -111,7 +132,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ */ inline fun split( crossinline function: (P) -> Any, - crossinline configurer: (SplitterEndpointSpec) -> Unit) { + crossinline configurer: SplitterEndpointSpec.() -> Unit) { this.delegate.split(P::class.java, { function(it) }) { configurer(it) } } @@ -130,7 +151,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ */ inline fun filter( crossinline function: (P) -> Boolean, - crossinline configurer: (FilterEndpointSpec) -> Unit) { + crossinline configurer: FilterEndpointSpec.() -> Unit) { this.delegate.filter(P::class.java, { function(it) }) { configurer(it) } } @@ -141,7 +162,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * with reified generic type. */ inline fun route(crossinline function: (P) -> Any?) { - this.delegate.route(P::class.java, Function { function(it) }) + route(function) { } } /** @@ -150,7 +171,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ */ inline fun route( crossinline function: (P) -> T, - crossinline configurer: (RouterSpec) -> Unit) { + crossinline configurer: RouterSpec.() -> Unit) { this.delegate.route(P::class.java, { function(it) }) { configurer(it) } } @@ -201,7 +222,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * at the current [IntegrationFlow] chain position using the [Channels] * factory fluent API. */ - fun channel(channels: (Channels) -> MessageChannelSpec<*, *>) { + fun channel(channels: Channels.() -> MessageChannelSpec<*, *>) { this.delegate.channel(channels) } @@ -209,7 +230,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * The [org.springframework.integration.channel.PublishSubscribeChannel] `channel()` * method specific implementation to allow the use of the 'subflow' subscriber capability. */ - fun publishSubscribeChannel(publishSubscribeChannelConfigurer: (PublishSubscribeSpec) -> Unit) { + fun publishSubscribeChannel(publishSubscribeChannelConfigurer: PublishSubscribeSpec.() -> Unit) { this.delegate.publishSubscribeChannel(publishSubscribeChannelConfigurer) } @@ -219,7 +240,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Use the provided [Executor] for the target subscribers. */ fun publishSubscribeChannel(executor: Executor, - publishSubscribeChannelConfigurer: (PublishSubscribeSpec) -> Unit) { + publishSubscribeChannelConfigurer: PublishSubscribeSpec.() -> Unit) { this.delegate.publishSubscribeChannel(executor, Consumer(publishSubscribeChannelConfigurer)) } @@ -242,9 +263,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * This method can be used after any `channel()` for explicit [MessageChannel], * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. */ - fun wireTap(flow: KotlinIntegrationFlowDefinition.() -> Unit, - wireTapConfigurer: (WireTapSpec) -> Unit) { - + fun wireTap(wireTapConfigurer: WireTapSpec.() -> Unit, flow: KotlinIntegrationFlowDefinition.() -> Unit) { this.delegate.wireTap( IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) }, Consumer(wireTapConfigurer)) @@ -257,7 +276,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * This method can be used after any `channel()` for explicit [MessageChannel], * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. */ - fun wireTap(wireTapChannel: String, wireTapConfigurer: (WireTapSpec) -> Unit = {}) { + fun wireTap(wireTapChannel: String, wireTapConfigurer: WireTapSpec.() -> Unit = {}) { this.delegate.wireTap(wireTapChannel, wireTapConfigurer) } @@ -268,7 +287,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * This method can be used after any `channel()` for explicit [MessageChannel], * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. */ - fun wireTap(wireTapChannel: MessageChannel, wireTapConfigurer: (WireTapSpec) -> Unit = {}) { + fun wireTap(wireTapChannel: MessageChannel, wireTapConfigurer: WireTapSpec.() -> Unit = {}) { this.delegate.wireTap(wireTapChannel, Consumer(wireTapConfigurer)) } @@ -287,7 +306,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate the `Control Bus` EI Pattern specific [MessageHandler] implementation * at the current [IntegrationFlow] chain position. */ - fun controlBus(endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + fun controlBus(endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.controlBus(endpointConfigurer) } @@ -296,7 +315,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * for the SpEL [Expression]. */ fun transform(expression: String, - endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.transform(expression, endpointConfigurer) } @@ -312,14 +331,9 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ /** * Populate the `MessageTransformingHandler` for the [MethodInvokingTransformer] * to invoke the service method at runtime. - * @param service the service to use. - * @param methodName the method to invoke. - * @param endpointConfigurer the consumer to provide integration endpoint options. - * @return the current [BaseIntegrationFlowDefinition]. - * @see ExpressionEvaluatingTransformer */ fun transform(service: Any, methodName: String?, - endpointConfigurer: (GenericEndpointSpec) -> Unit) { + endpointConfigurer: GenericEndpointSpec.() -> Unit) { this.delegate.transform(service, methodName, endpointConfigurer) } @@ -330,7 +344,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun transform(messageProcessorSpec: MessageProcessorSpec<*>, - endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.transform(messageProcessorSpec, endpointConfigurer) } @@ -339,7 +353,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate a [MessageFilter] with [MessageSelector] for the provided SpEL expression. * In addition accept options for the integration endpoint using [FilterEndpointSpec]: */ - fun filter(expression: String, endpointConfigurer: (FilterEndpointSpec) -> Unit = {}) { + fun filter(expression: String, endpointConfigurer: FilterEndpointSpec.() -> Unit = {}) { this.delegate.filter(expression, endpointConfigurer) } @@ -355,7 +369,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate a [MessageFilter] with [MethodInvokingSelector] for the * method of the provided service. */ - fun filter(service: Any, methodName: String?, endpointConfigurer: (FilterEndpointSpec) -> Unit) { + fun filter(service: Any, methodName: String?, endpointConfigurer: FilterEndpointSpec.() -> Unit) { this.delegate.filter(service, methodName, endpointConfigurer) } @@ -365,7 +379,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * the provided [MessageProcessorSpec]. * In addition accept options for the integration endpoint using [FilterEndpointSpec]. */ - fun filter(messageProcessorSpec: MessageProcessorSpec<*>, endpointConfigurer: (FilterEndpointSpec) -> Unit = {}) { + fun filter(messageProcessorSpec: MessageProcessorSpec<*>, endpointConfigurer: FilterEndpointSpec.() -> Unit = {}) { this.delegate.filter(messageProcessorSpec, endpointConfigurer) } @@ -401,7 +415,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun handle(beanName: String, methodName: String?, - endpointConfigurer: (GenericEndpointSpec) -> Unit) { + endpointConfigurer: GenericEndpointSpec.() -> Unit) { this.delegate.handle(beanName, methodName, endpointConfigurer) } @@ -423,11 +437,20 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun handle(service: Any, methodName: String?, - endpointConfigurer: (GenericEndpointSpec) -> Unit) { + endpointConfigurer: GenericEndpointSpec.() -> Unit) { this.delegate.handle(service, methodName, endpointConfigurer) } + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the provided [GenericHandler] at runtime. + */ + inline fun handle(crossinline handler: (P, MessageHeaders) -> Any) { + this.delegate.handle(P::class.java) { p, h -> handler(p, h) } + } + /** * Populate a [ServiceActivatingHandler] for the * [org.springframework.integration.handler.MethodInvokingMessageProcessor] @@ -435,20 +458,18 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ inline fun handle( - handler: GenericHandler

, - crossinline endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + crossinline handler: (P, MessageHeaders) -> Any, + crossinline endpointConfigurer: GenericEndpointSpec.() -> Unit) { - this.delegate.handle(P::class.java, handler, Consumer { endpointConfigurer(it) }) + this.delegate.handle(P::class.java, { p, h -> handler(p, h) }) { endpointConfigurer(it) } } /** - * Populate a [ServiceActivatingHandler] for the - * [MessageProcessor] from the provided - * [MessageProcessorSpec]. + * Populate a [ServiceActivatingHandler] for the [MessageProcessor] from the provided [MessageProcessorSpec]. * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun handle(messageProcessorSpec: MessageProcessorSpec<*>, - endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.handle(messageProcessorSpec, endpointConfigurer) } @@ -459,7 +480,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun handle(messageHandlerSpec: MessageHandlerSpec<*, H>, - endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.handle(messageHandlerSpec, endpointConfigurer) } @@ -469,21 +490,21 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * [MessageHandler] implementation. * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ - fun handle(messageHandler: H, endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + fun handle(messageHandler: H, endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.handle(messageHandler, endpointConfigurer) } /** * Populate a [BridgeHandler] to the current integration flow position. */ - fun bridge(endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + fun bridge(endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.bridge(endpointConfigurer) } /** * Populate a [DelayHandler] to the current integration flow position. */ - fun delay(groupId: String, endpointConfigurer: (DelayerEndpointSpec) -> Unit = {}) { + fun delay(groupId: String, endpointConfigurer: DelayerEndpointSpec.() -> Unit = {}) { this.delegate.delay(groupId, endpointConfigurer) } @@ -492,7 +513,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * to the current integration flow position * with provided options. */ - fun enrich(enricherConfigurer: (EnricherSpec) -> Unit) { + fun enrich(enricherConfigurer: EnricherSpec.() -> Unit) { this.delegate.enrich(enricherConfigurer) } @@ -503,7 +524,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun enrichHeaders(headers: MapBuilder<*, String, Any>, - endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.enrichHeaders(headers, endpointConfigurer) } @@ -515,7 +536,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * to be evaluated against a request [Message]. */ fun enrichHeaders(headers: Map, - endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.enrichHeaders(headers, endpointConfigurer) } @@ -525,7 +546,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * a [org.springframework.integration.transformer.HeaderEnricher] * as the result of provided consumer. */ - fun enrichHeaders(headerEnricherConfigurer: (HeaderEnricherSpec) -> Unit) { + fun enrichHeaders(headerEnricherConfigurer: HeaderEnricherSpec.() -> Unit) { this.delegate.enrichHeaders(headerEnricherConfigurer) } @@ -542,7 +563,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * SpEL expression. */ fun split(expression: String, - endpointConfigurer: (SplitterEndpointSpec) -> Unit = {}) { + endpointConfigurer: SplitterEndpointSpec.() -> Unit = {}) { this.delegate.split(expression, endpointConfigurer) } @@ -561,7 +582,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun split(service: Any, methodName: String?, - endpointConfigurer: (SplitterEndpointSpec) -> Unit) { + endpointConfigurer: SplitterEndpointSpec.() -> Unit) { this.delegate.split(service, methodName, endpointConfigurer) } @@ -580,7 +601,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun split(beanName: String, methodName: String?, - endpointConfigurer: (SplitterEndpointSpec) -> Unit) { + endpointConfigurer: SplitterEndpointSpec.() -> Unit) { this.delegate.split(beanName, methodName, endpointConfigurer) } @@ -592,7 +613,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun split(messageProcessorSpec: MessageProcessorSpec<*>, - endpointConfigurer: (SplitterEndpointSpec) -> Unit = {}) { + endpointConfigurer: SplitterEndpointSpec.() -> Unit = {}) { this.delegate.split(messageProcessorSpec, endpointConfigurer) } @@ -600,8 +621,8 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ /** * Populate the provided [AbstractMessageSplitter] to the current integration flow position. */ - fun split(splitterMessageHandlerSpec: MessageHandlerSpec<*, S>, - endpointConfigurer: (SplitterEndpointSpec) -> Unit = {}) { + fun split(splitterMessageHandlerSpec: MessageHandlerSpec<*, S>, + endpointConfigurer: SplitterEndpointSpec.() -> Unit = {}) { this.delegate.split(splitterMessageHandlerSpec, endpointConfigurer) } @@ -610,14 +631,14 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate the provided [AbstractMessageSplitter] to the current integration * flow position. */ - fun split(splitter: S, - endpointConfigurer: (SplitterEndpointSpec) -> Unit = {}) { + fun split(splitter: S, + endpointConfigurer: SplitterEndpointSpec.() -> Unit = {}) { this.delegate.split(splitter, endpointConfigurer) } /** - * Provide the [HeaderFilter] to the current [StandardIntegrationFlow]. + * Provide the [HeaderFilter] to the current [IntegrationFlow]. */ fun headerFilter(headersToRemove: String, patternMatch: Boolean = true) { this.delegate.headerFilter(headersToRemove, patternMatch) @@ -628,7 +649,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * [HeaderFilter]. */ fun headerFilter(headerFilter: HeaderFilter, - endpointConfigurer: (GenericEndpointSpec) -> Unit) { + endpointConfigurer: GenericEndpointSpec.() -> Unit) { this.delegate.headerFilter(headerFilter, endpointConfigurer) } @@ -639,7 +660,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun claimCheckIn(messageStore: MessageStore, - endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.claimCheckIn(messageStore, endpointConfigurer) } @@ -658,7 +679,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ fun claimCheckOut(messageStore: MessageStore, removeMessage: Boolean, - endpointConfigurer: (GenericEndpointSpec) -> Unit) { + endpointConfigurer: GenericEndpointSpec.() -> Unit) { this.delegate.claimCheckOut(messageStore, removeMessage, endpointConfigurer) } @@ -669,7 +690,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * provided options from [ResequencerSpec]. * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ - fun resequence(resequencer: (ResequencerSpec) -> Unit = {}) { + fun resequence(resequencer: ResequencerSpec.() -> Unit = {}) { this.delegate.resequence(resequencer) } @@ -677,7 +698,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate the [AggregatingMessageHandler] with provided options from [AggregatorSpec]. * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ - fun aggregate(aggregator: (AggregatorSpec) -> Unit = {}) { + fun aggregate(aggregator: AggregatorSpec.() -> Unit = {}) { this.delegate.aggregate(aggregator) } @@ -693,7 +714,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate the [MethodInvokingRouter] for provided bean and its method * with provided options from [RouterSpec]. */ - fun route(beanName: String, method: String?, routerConfigurer: (RouterSpec) -> Unit) { + fun route(beanName: String, method: String?, routerConfigurer: RouterSpec.() -> Unit) { this.delegate.route(beanName, method, routerConfigurer) } @@ -709,7 +730,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate the [MethodInvokingRouter] for the method * of the provided service and its method with provided options from [RouterSpec]. */ - fun route(service: Any, methodName: String?, routerConfigurer: (RouterSpec) -> Unit) { + fun route(service: Any, methodName: String?, routerConfigurer: RouterSpec.() -> Unit) { this.delegate.route(service, methodName, routerConfigurer) } @@ -717,7 +738,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate the [ExpressionEvaluatingRouter] for provided SpEL expression * with provided options from [RouterSpec]. */ - fun route(expression: String, routerConfigurer: (RouterSpec) -> Unit = {}) { + fun route(expression: String, routerConfigurer: RouterSpec.() -> Unit = {}) { this.delegate.route(expression, routerConfigurer) } @@ -727,7 +748,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * from the provided [MessageProcessorSpec] with default options. */ fun route(messageProcessorSpec: MessageProcessorSpec<*>, - routerConfigurer: (RouterSpec) -> Unit = {}) { + routerConfigurer: RouterSpec.() -> Unit = {}) { this.delegate.route(messageProcessorSpec, routerConfigurer) } @@ -735,7 +756,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ /** * Populate the [RecipientListRouter] with options from the [RecipientListRouterSpec]. */ - fun routeToRecipients(routerConfigurer: (RecipientListRouterSpec) -> Unit) { + fun routeToRecipients(routerConfigurer: RecipientListRouterSpec.() -> Unit) { this.delegate.routeToRecipients(routerConfigurer) } @@ -743,7 +764,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate the [ErrorMessageExceptionTypeRouter] with options from the [RouterSpec]. */ fun routeByException( - routerConfigurer: (RouterSpec, ErrorMessageExceptionTypeRouter>) -> Unit) { + routerConfigurer: RouterSpec, ErrorMessageExceptionTypeRouter>.() -> Unit) { this.delegate.routeByException(routerConfigurer) } @@ -753,7 +774,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * current integration flow position. * In addition accept options for the integration endpoint using [GenericEndpointSpec]. */ - fun route(router: R, endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + fun route(router: R, endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.route(router, endpointConfigurer) } @@ -765,7 +786,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * [org.springframework.integration.gateway.RequestReplyExchanger] Proxy on the * background. */ - fun gateway(requestChannel: String, endpointConfigurer: (GatewayEndpointSpec) -> Unit = {}) { + fun gateway(requestChannel: String, endpointConfigurer: GatewayEndpointSpec.() -> Unit = {}) { this.delegate.gateway(requestChannel, endpointConfigurer) } @@ -777,7 +798,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * [org.springframework.integration.gateway.RequestReplyExchanger] Proxy on the * background. */ - fun gateway(requestChannel: MessageChannel, endpointConfigurer: (GatewayEndpointSpec) -> Unit = {}) { + fun gateway(requestChannel: MessageChannel, endpointConfigurer: GatewayEndpointSpec.() -> Unit = {}) { this.delegate.gateway(requestChannel, Consumer(endpointConfigurer)) } @@ -795,8 +816,8 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * [org.springframework.integration.gateway.GatewayMessageHandler] for the * provided `subflow` with options from [GatewayEndpointSpec]. */ - fun gateway(flow: KotlinIntegrationFlowDefinition.() -> Unit, - endpointConfigurer: (GatewayEndpointSpec) -> Unit) { + fun gateway(endpointConfigurer: GatewayEndpointSpec.() -> Unit, + flow: KotlinIntegrationFlowDefinition.() -> Unit) { this.delegate.gateway( IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) }, @@ -934,7 +955,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * based on the provided [MessageChannel] for scattering function * and [AggregatorSpec] for gathering function. */ - fun scatterGather(scatterChannel: MessageChannel, gatherer: (AggregatorSpec) -> Unit = {}) { + fun scatterGather(scatterChannel: MessageChannel, gatherer: AggregatorSpec.() -> Unit = {}) { this.delegate.scatterGather(scatterChannel, Consumer(gatherer)) } @@ -943,8 +964,8 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * based on the provided [MessageChannel] for scattering function * and [AggregatorSpec] for gathering function. */ - fun scatterGather(scatterChannel: MessageChannel, gatherer: (AggregatorSpec) -> Unit, - scatterGather: (ScatterGatherSpec) -> Unit) { + fun scatterGather(scatterChannel: MessageChannel, gatherer: AggregatorSpec.() -> Unit, + scatterGather: ScatterGatherSpec.() -> Unit) { this.delegate.scatterGather(scatterChannel, Consumer(gatherer), Consumer(scatterGather)) } @@ -954,7 +975,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * based on the provided [RecipientListRouterSpec] for scattering function * and default [AggregatorSpec] for gathering function. */ - fun scatterGather(scatterer: (RecipientListRouterSpec) -> Unit) { + fun scatterGather(scatterer: RecipientListRouterSpec.() -> Unit) { this.delegate.scatterGather(scatterer) } @@ -963,7 +984,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * based on the provided [RecipientListRouterSpec] for scattering function * and [AggregatorSpec] for gathering function. */ - fun scatterGather(scatterer: (RecipientListRouterSpec) -> Unit, gatherer: (AggregatorSpec) -> Unit) { + fun scatterGather(scatterer: RecipientListRouterSpec.() -> Unit, gatherer: AggregatorSpec.() -> Unit) { this.delegate.scatterGather(scatterer, gatherer) } @@ -971,14 +992,9 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate a [ScatterGatherHandler] to the current integration flow position * based on the provided [RecipientListRouterSpec] for scattering function * and [AggregatorSpec] for gathering function. - * @param scatterer the consumer for [RecipientListRouterSpec] to configure scatterer. - * @param gatherer the consumer for [AggregatorSpec] to configure gatherer. - * @param scatterGather the consumer for [ScatterGatherSpec] to configure - * [ScatterGatherHandler] and its endpoint. Can be `null`. - * @return the current [BaseIntegrationFlowDefinition]. */ - fun scatterGather(scatterer: (RecipientListRouterSpec) -> Unit, gatherer: (AggregatorSpec) -> Unit, - scatterGather: (ScatterGatherSpec) -> Unit) { + fun scatterGather(scatterer: RecipientListRouterSpec.() -> Unit, gatherer: AggregatorSpec.() -> Unit, + scatterGather: ScatterGatherSpec.() -> Unit) { this.delegate.scatterGather(scatterer, gatherer, scatterGather) } @@ -988,7 +1004,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * instance for provided timeout and options from [BarrierSpec] and endpoint * options from [GenericEndpointSpec]. */ - fun barrier(timeout: Long, barrierConfigurer: (BarrierSpec) -> Unit = {}) { + fun barrier(timeout: Long, barrierConfigurer: BarrierSpec.() -> Unit = {}) { this.delegate.barrier(timeout, barrierConfigurer) } @@ -997,7 +1013,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * and endpoint options from [GenericEndpointSpec]. */ fun trigger(triggerActionId: String, - endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.trigger(triggerActionId, endpointConfigurer) } @@ -1007,7 +1023,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * and endpoint options from [GenericEndpointSpec]. */ fun trigger(triggerAction: MessageTriggerAction, - endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + endpointConfigurer: GenericEndpointSpec.() -> Unit = {}) { this.delegate.trigger(triggerAction, Consumer(endpointConfigurer)) } 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 5b24774..e0cfe11 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 @@ -18,8 +18,10 @@ package org.springframework.integration.dsl.kotlin.test 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 @@ -29,10 +31,11 @@ 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.Pollers import org.springframework.integration.dsl.context.IntegrationFlowContext -import org.springframework.integration.dsl.integrationFlow +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 @@ -159,7 +162,7 @@ class KotlinDslTests { val integrationFlow = integrationFlow(publisher) { - transform>({ it.payload * 2 }) { it.id("foo") } + transform>({ it.payload * 2 }) { id("foo") } channel(fluxChannel) } @@ -188,6 +191,24 @@ class KotlinDslTests { 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 { @@ -200,72 +221,111 @@ class KotlinDslTests { fun convertFlow() = integrationFlow("convertFlowInput") { convert() - convert { it.id("kotlinConverter") } + convert { id("kotlinConverter") } } @Bean fun functionFlow() = - integrationFlow>({ it.beanName("functionGateway") }) { + integrationFlow>({ beanName("functionGateway") }) { transform { it.toUpperCase() } split> { it.payload } - split({ it }) { it.id("splitterEndpoint") } + split({ it }) { id("splitterEndpoint") } resequence() - aggregate { it.id("aggregator").outputProcessor { it.one } } + aggregate { id("aggregator").outputProcessor { it.one } } } @Bean fun functionFlow2() = integrationFlow> { transform { it.toLowerCase() } - route, Any?>({ null }) { it.defaultOutputToParentFlow() } + route, Any?>({ null }) { defaultOutputToParentFlow() } route> { m -> m.headers.replyChannel } } @Bean fun messageSourceFlow() = integrationFlow(MessageProcessorMessageSource { "testSource" }, - { it.poller { it.trigger(OnlyOnceTrigger()) } }) { - channel { it.queue("fromSupplierQueue") } + { poller { it.trigger(OnlyOnceTrigger()) } }) { + channel { queue("fromSupplierQueue") } } @Bean fun messageSourceFlow2() = integrationFlow(MessageProcessorMessageSource { "testSource2" }) { - channel { it.queue("fromSupplierQueue2") } + channel { queue("fromSupplierQueue2") } } @Bean fun fixedSubscriberFlow() = integrationFlow("fixedSubscriberInput", true) { log(LoggingHandler.Level.WARN) { it.payload } - transform("payload") { it.id("spelTransformer") } + transform("payload") { id("spelTransformer") } } @Bean fun flowFromSupplier() = integrationFlow({ "testSupplier" }) { - channel { it.queue("testSupplierResult") } + channel { queue("testSupplierResult") } } @Bean fun flowFromSupplier2() = integrationFlow({ "testSupplier2" }, - { it.poller { it.trigger(OnlyOnceTrigger()) } }) { + { poller { it.trigger(OnlyOnceTrigger()) } }) { filter> { m -> m.payload is String } - channel { it.queue("testSupplierResult2") } + channel { queue("testSupplierResult2") } } @Bean fun flowLambda() = integrationFlow { - filter({ it === "test" }) { it.id("filterEndpoint") } + filter({ it === "test" }) { id("filterEndpoint") } wireTap { - channel { it.queue("wireTapChannel") } + channel { queue("wireTapChannel") } } - delay("delayGroup") { it.defaultDelay(100) } + delay("delayGroup") { defaultDelay(100) } transform { 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 { true }, integrationFlow { handle { _, _ -> Math.random() * 10 } }) + recipientFlow(GenericSelector { true }, integrationFlow { handle { _, _ -> Math.random() * 10 } }) + recipientFlow(GenericSelector { true }, integrationFlow { handle { _, _ -> 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?)