From 92421f75755991c7bf2b1773f53ed168731153df Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 17 Mar 2020 12:56:45 -0400 Subject: [PATCH] Add Kotlin wrappers for router specs (#3218) * Add Kotlin wrappers for router specs To avoid casting and extra logic logic in the end-user code, it is better to provide Kotlin-specific API to let end-users to do whatever is really dictated by API and don't think about specific types to cast * * Fix typos; code clean up --- .../dsl/AbstractKotlinRouterSpec.kt | 58 +++++++++++ .../dsl/KotlinIntegrationFlowDefinition.kt | 64 ++++++------ .../dsl/KotlinRecipientListRouterSpec.kt | 99 +++++++++++++++++++ .../integration/dsl/KotlinRouterSpec.kt | 66 +++++++++++++ .../integration/dsl/KotlinDslTests.kt | 6 +- .../integration/dsl/routers/RouterDslTests.kt | 8 +- 6 files changed, 266 insertions(+), 35 deletions(-) create mode 100644 spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/AbstractKotlinRouterSpec.kt create mode 100644 spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRecipientListRouterSpec.kt create mode 100644 spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRouterSpec.kt diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/AbstractKotlinRouterSpec.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/AbstractKotlinRouterSpec.kt new file mode 100644 index 0000000000..2564bbcd2a --- /dev/null +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/AbstractKotlinRouterSpec.kt @@ -0,0 +1,58 @@ +/* + * 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.springframework.integration.router.AbstractMessageRouter +import org.springframework.messaging.MessageChannel + +/** + * An [AbstractRouterSpec] wrapped for Kotlin DSL. + * + * @property delegate the [AbstractRouterSpec] this instance is delegating to. + * + * @author Artem Bilan + * + * @since 5.3 + */ +abstract class AbstractKotlinRouterSpec, R : AbstractMessageRouter>( + open val delegate: AbstractRouterSpec) { + + fun ignoreSendFailures(ignoreSendFailures: Boolean) { + this.delegate.ignoreSendFailures(ignoreSendFailures) + } + + fun applySequence(applySequence: Boolean) { + this.delegate.applySequence(applySequence) + } + + fun defaultOutputChannel(channelName: String) { + this.delegate.defaultOutputChannel(channelName) + } + + fun defaultOutputChannel(channel: MessageChannel) { + this.delegate.defaultOutputChannel(channel) + } + + fun defaultSubFlowMapping(subFlow: KotlinIntegrationFlowDefinition.() -> Unit) { + this.delegate.defaultSubFlowMapping { subFlow(KotlinIntegrationFlowDefinition(it)) } + } + + fun defaultOutputToParentFlow() { + this.delegate.defaultOutputToParentFlow() + } + +} diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt index afefc24f04..dcf7bbcd11 100644 --- a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt @@ -151,9 +151,9 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ */ inline fun route( crossinline function: (P) -> T, - crossinline configurer: RouterSpec.() -> Unit) { + crossinline configurer: KotlinRouterSpec.() -> Unit) { - this.delegate.route(P::class.java, { function(it) }) { configurer(it) } + this.delegate.route(P::class.java, { function(it) }) { configurer(KotlinRouterSpec(it)) } } /** @@ -711,10 +711,12 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ /** * Populate the [MethodInvokingRouter] for provided bean and its method - * with provided options from [RouterSpec]. + * with provided options from [KotlinRouterSpec]. */ - fun route(beanName: String, method: String?, routerConfigurer: RouterSpec.() -> Unit) { - this.delegate.route(beanName, method, routerConfigurer) + fun route(beanName: String, method: String?, + routerConfigurer: KotlinRouterSpec.() -> Unit) { + + this.delegate.route(beanName, method) { routerConfigurer(KotlinRouterSpec(it)) } } /** @@ -727,18 +729,22 @@ 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]. + * of the provided service and its method with provided options from [KotlinRouterSpec]. */ - fun route(service: Any, methodName: String?, routerConfigurer: RouterSpec.() -> Unit) { - this.delegate.route(service, methodName, routerConfigurer) + fun route(service: Any, methodName: String?, + routerConfigurer: KotlinRouterSpec.() -> Unit) { + + this.delegate.route(service, methodName) { routerConfigurer(KotlinRouterSpec(it)) } } /** * Populate the [ExpressionEvaluatingRouter] for provided SpEL expression - * with provided options from [RouterSpec]. + * with provided options from [KotlinRouterSpec]. */ - fun route(expression: String, routerConfigurer: RouterSpec.() -> Unit = {}) { - this.delegate.route(expression, routerConfigurer) + fun route(expression: String, + routerConfigurer: KotlinRouterSpec.() -> Unit = {}) { + + this.delegate.route(expression) { routerConfigurer(KotlinRouterSpec(it)) } } /** @@ -747,25 +753,25 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * from the provided [MessageProcessorSpec] with default options. */ fun route(messageProcessorSpec: MessageProcessorSpec<*>, - routerConfigurer: RouterSpec.() -> Unit = {}) { + routerConfigurer: KotlinRouterSpec.() -> Unit = {}) { - this.delegate.route(messageProcessorSpec, routerConfigurer) + this.delegate.route(messageProcessorSpec) { routerConfigurer(KotlinRouterSpec(it)) } } /** - * Populate the [RecipientListRouter] with options from the [RecipientListRouterSpec]. + * Populate the [RecipientListRouter] with options from the [KotlinRecipientListRouterSpec]. */ - fun routeToRecipients(routerConfigurer: RecipientListRouterSpec.() -> Unit) { - this.delegate.routeToRecipients(routerConfigurer) + fun routeToRecipients(routerConfigurer: KotlinRecipientListRouterSpec.() -> Unit) { + this.delegate.routeToRecipients { routerConfigurer(KotlinRecipientListRouterSpec(it)) } } /** - * Populate the [ErrorMessageExceptionTypeRouter] with options from the [RouterSpec]. + * Populate the [ErrorMessageExceptionTypeRouter] with options from the [KotlinRouterSpec]. */ fun routeByException( - routerConfigurer: RouterSpec, ErrorMessageExceptionTypeRouter>.() -> Unit) { + routerConfigurer: KotlinRouterSpec, ErrorMessageExceptionTypeRouter>.() -> Unit) { - this.delegate.routeByException(routerConfigurer) + this.delegate.routeByException { routerConfigurer(KotlinRouterSpec(it)) } } /** @@ -971,31 +977,33 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ /** * Populate a [ScatterGatherHandler] to the current integration flow position - * based on the provided [RecipientListRouterSpec] for scattering function + * based on the provided [KotlinRecipientListRouterSpec] for scattering function * and default [AggregatorSpec] for gathering function. */ - fun scatterGather(scatterer: RecipientListRouterSpec.() -> Unit) { - this.delegate.scatterGather(scatterer) + fun scatterGather(scatterer: KotlinRecipientListRouterSpec.() -> Unit) { + this.delegate.scatterGather(Consumer { scatterer(KotlinRecipientListRouterSpec(it)) }) } /** * Populate a [ScatterGatherHandler] to the current integration flow position - * based on the provided [RecipientListRouterSpec] for scattering function + * based on the provided [KotlinRecipientListRouterSpec] for scattering function * and [AggregatorSpec] for gathering function. */ - fun scatterGather(scatterer: RecipientListRouterSpec.() -> Unit, gatherer: AggregatorSpec.() -> Unit) { - this.delegate.scatterGather(scatterer, gatherer) + fun scatterGather(scatterer: KotlinRecipientListRouterSpec.() -> Unit, gatherer: AggregatorSpec.() -> Unit) { + this.delegate.scatterGather(Consumer { scatterer(KotlinRecipientListRouterSpec(it)) }, + Consumer { gatherer(it) }) } /** * Populate a [ScatterGatherHandler] to the current integration flow position - * based on the provided [RecipientListRouterSpec] for scattering function + * based on the provided [KotlinRecipientListRouterSpec] for scattering function * and [AggregatorSpec] for gathering function. */ - fun scatterGather(scatterer: RecipientListRouterSpec.() -> Unit, gatherer: AggregatorSpec.() -> Unit, + fun scatterGather(scatterer: KotlinRecipientListRouterSpec.() -> Unit, gatherer: AggregatorSpec.() -> Unit, scatterGather: ScatterGatherSpec.() -> Unit) { - this.delegate.scatterGather(scatterer, gatherer, scatterGather) + this.delegate.scatterGather(Consumer { scatterer(KotlinRecipientListRouterSpec(it)) }, + Consumer { gatherer(it) }, Consumer { scatterGather(it) }) } /** diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRecipientListRouterSpec.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRecipientListRouterSpec.kt new file mode 100644 index 0000000000..c9312c817f --- /dev/null +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRecipientListRouterSpec.kt @@ -0,0 +1,99 @@ +/* + * 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.springframework.expression.Expression +import org.springframework.integration.core.GenericSelector +import org.springframework.integration.core.MessageSelector +import org.springframework.integration.router.RecipientListRouter +import org.springframework.messaging.Message +import org.springframework.messaging.MessageChannel + +/** + * A [RecipientListRouterSpec] wrapped for Kotlin DSL. + * + * @property delegate the [RecipientListRouterSpec] this instance is delegating to. + * + * @author Artem Bilan + * + * @since 5.3 + */ +class KotlinRecipientListRouterSpec(override val delegate: RecipientListRouterSpec) + : AbstractKotlinRouterSpec(delegate) { + + fun recipient(channelName: String) { + this.delegate.recipient(channelName) + } + + fun recipient(channelName: String, expression: String) { + this.delegate.recipient(channelName, expression) + } + + fun recipient(channelName: String, expression: Expression) { + this.delegate.recipient(channelName, expression) + } + + inline fun recipient(channelName: String, crossinline selector: (P) -> Boolean) { + if (Message::class.java.isAssignableFrom(P::class.java)) + this.delegate.recipientMessageSelector(channelName) { selector(it as P) } + else + this.delegate.recipient

(channelName) { selector(it) } + } + + fun recipient(channel: MessageChannel) { + this.delegate.recipient(channel) + } + + fun recipient(channel: MessageChannel, expression: String) { + this.delegate.recipient(channel, expression) + } + + fun recipient(channel: MessageChannel, expression: Expression) { + this.delegate.recipient(channel, expression) + } + + inline fun recipient(channel: MessageChannel, crossinline selector: (P) -> Boolean) { + if (Message::class.java.isAssignableFrom(P::class.java)) + this.delegate.recipientMessageSelector(channel, MessageSelector { selector(it as P) }) + else + this.delegate.recipient

(channel, GenericSelector { selector(it) }) + } + + inline fun recipientFlow(crossinline selector: (P) -> Boolean, + crossinline subFlow: KotlinIntegrationFlowDefinition.() -> Unit) { + + if (Message::class.java.isAssignableFrom(P::class.java)) + this.delegate.recipientMessageSelectorFlow({ selector(it as P) }) + { subFlow(KotlinIntegrationFlowDefinition(it)) } + else + this.delegate.recipientFlow

({ selector(it) }) { subFlow(KotlinIntegrationFlowDefinition(it)) } + + } + + fun recipientFlow(subFlow: KotlinIntegrationFlowDefinition.() -> Unit) { + this.delegate.recipientFlow { subFlow(KotlinIntegrationFlowDefinition(it)) } + } + + fun recipientFlow(expression: String, subFlow: KotlinIntegrationFlowDefinition.() -> Unit) { + this.delegate.recipientFlow(expression) { subFlow(KotlinIntegrationFlowDefinition(it)) } + } + + fun recipientFlow(expression: Expression, subFlow: KotlinIntegrationFlowDefinition.() -> Unit) { + this.delegate.recipientFlow(expression) { subFlow(KotlinIntegrationFlowDefinition(it)) } + } + +} diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRouterSpec.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRouterSpec.kt new file mode 100644 index 0000000000..e7eb85dbd5 --- /dev/null +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRouterSpec.kt @@ -0,0 +1,66 @@ +/* + * 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.springframework.integration.router.AbstractMappingMessageRouter +import org.springframework.messaging.MessageChannel + +/** + * A [RouterSpec] wrapped for Kotlin DSL. + * + * @property delegate the [RouterSpec] this instance is delegating to. + * + * @author Artem Bilan + * + * @since 5.3 + */ +class KotlinRouterSpec(override val delegate: RouterSpec) + : AbstractKotlinRouterSpec, R>(delegate) { + + fun resolutionRequired(resolutionRequired: Boolean) { + this.delegate.resolutionRequired(resolutionRequired) + } + + fun dynamicChannelLimit(dynamicChannelLimit: Int) { + this.delegate.dynamicChannelLimit(dynamicChannelLimit) + } + + fun prefix(prefix: String) { + this.delegate.prefix(prefix) + } + + fun suffix(suffix: String) { + this.delegate.suffix(suffix) + } + + fun noChannelKeyFallback() { + this.delegate.noChannelKeyFallback() + } + + fun channelMapping(key: K, channelName: String) { + this.delegate.channelMapping(key, channelName) + } + + fun channelMapping(key: K, channel: MessageChannel) { + this.delegate.channelMapping(key, channel) + } + + fun subFlowMapping(key: K, subFlow: KotlinIntegrationFlowDefinition.() -> Unit) { + this.delegate.subFlowMapping(key) { subFlow(KotlinIntegrationFlowDefinition(it)) } + } + +} diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt index bab49fbc66..fd5f86a527 100644 --- a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt +++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt @@ -315,9 +315,9 @@ class KotlinDslTests { 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 } }) + recipientFlow({ true }) { handle { _, _ -> Math.random() * 10 } } + recipientFlow({ true }) { handle { _, _ -> Math.random() * 10 } } + recipientFlow({ true }) { handle { _, _ -> Math.random() * 10 } } }, { releaseStrategy { diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt index e680aa7681..9dcc9f7685 100644 --- a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt +++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt @@ -102,8 +102,8 @@ class RouterDslTests { integrationFlow { split() route({ it % 2 == 0 }) { - subFlowMapping(true) { sf -> sf.handle { p, _ -> p * 2 } } - subFlowMapping(false) { sf -> sf.handle { p, _ -> p * 3 } } + subFlowMapping(true) { handle { p, _ -> p * 2 } } + subFlowMapping(false) { handle { p, _ -> p * 3 } } } aggregate() channel { queue("routerTwoSubFlowsOutput") } @@ -114,8 +114,8 @@ class RouterDslTests { integrationFlow { split() route({ it % 2 == 0 }) { - subFlowMapping(true) { sf -> sf.gateway(oddFlow()) } - subFlowMapping(false) { sf -> sf.gateway(evenFlow()) } + subFlowMapping(true) { gateway(oddFlow().inputChannel) } + subFlowMapping(false) { gateway(evenFlow().inputChannel) } } aggregate() }