Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add Kotlin wrappers for router specs (#3218)

Browse Source 
* 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
This commit is contained in:
Artem Bilan
2020-03-17 12:56:45 -04:00
committed by GitHub
parent 71e6b27b41
commit 92421f7575
6 changed files with 266 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/AbstractKotlinRouterSpec.kt Normal file
View File

@@ -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<S : AbstractRouterSpec<S, R>, R : AbstractMessageRouter>(
open val delegate: AbstractRouterSpec<S, R>) {
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()
}
}

64
spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt
View File

@@ -151,9 +151,9 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
*/
inline fun <reified P, T> route(
crossinline function: (P) -> T,
crossinline configurer: RouterSpec<T, MethodInvokingRouter>.() -> Unit) {
crossinline configurer: KotlinRouterSpec<T, MethodInvokingRouter>.() -> 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<Any, MethodInvokingRouter>.() -> Unit) {
this.delegate.route(beanName, method, routerConfigurer)
fun route(beanName: String, method: String?,
routerConfigurer: KotlinRouterSpec<Any, MethodInvokingRouter>.() -> 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<Any, MethodInvokingRouter>.() -> Unit) {
this.delegate.route(service, methodName, routerConfigurer)
fun route(service: Any, methodName: String?,
routerConfigurer: KotlinRouterSpec<Any, MethodInvokingRouter>.() -> 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 <T> route(expression: String, routerConfigurer: RouterSpec<T, ExpressionEvaluatingRouter>.() -> Unit = {}) {
this.delegate.route(expression, routerConfigurer)
fun <T> route(expression: String,
routerConfigurer: KotlinRouterSpec<T, ExpressionEvaluatingRouter>.() -> Unit = {}) {
this.delegate.route<T>(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<Any, MethodInvokingRouter>.() -> Unit = {}) {
routerConfigurer: KotlinRouterSpec<Any, MethodInvokingRouter>.() -> 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<Class<out Throwable>, ErrorMessageExceptionTypeRouter>.() -> Unit) {
routerConfigurer: KotlinRouterSpec<Class<out Throwable>, 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) })
}
/**

99
spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRecipientListRouterSpec.kt Normal file
View File

@@ -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<RecipientListRouterSpec, RecipientListRouter>(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 <reified P> 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<P>(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 <reified P> 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<P>(channel, GenericSelector { selector(it) })
}
inline fun <reified P> 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<P>({ 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)) }
}
}

66
spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRouterSpec.kt Normal file
View File

@@ -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<K, R : AbstractMappingMessageRouter>(override val delegate: RouterSpec<K, R>)
: AbstractKotlinRouterSpec<RouterSpec<K, R>, 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)) }
}
}

6
spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt
View File

@@ -315,9 +315,9 @@ class KotlinDslTests {
scatterGather(
{
applySequence(true)
recipientFlow(GenericSelector<Any> { true }, integrationFlow { handle<Any> { _, _ -> Math.random() * 10 } })
recipientFlow(GenericSelector<Any> { true }, integrationFlow { handle<Any> { _, _ -> Math.random() * 10 } })
recipientFlow(GenericSelector<Any> { true }, integrationFlow { handle<Any> { _, _ -> Math.random() * 10 } })
recipientFlow<Any>({ true }) { handle<Any> { _, _ -> Math.random() * 10 } }
recipientFlow<Any>({ true }) { handle<Any> { _, _ -> Math.random() * 10 } }
recipientFlow<Any>({ true }) { handle<Any> { _, _ -> Math.random() * 10 } }
},
{
releaseStrategy {

8
spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt
View File

@@ -102,8 +102,8 @@ class RouterDslTests {
integrationFlow {
split()
route<Int, Boolean>({ it % 2 == 0 }) {
subFlowMapping(true) { sf -> sf.handle<Int> { p, _ -> p * 2 } }
subFlowMapping(false) { sf -> sf.handle<Int> { p, _ -> p * 3 } }
subFlowMapping(true) { handle<Int> { p, _ -> p * 2 } }
subFlowMapping(false) { handle<Int> { p, _ -> p * 3 } }
}
aggregate()
channel { queue("routerTwoSubFlowsOutput") }
@@ -114,8 +114,8 @@ class RouterDslTests {
integrationFlow {
split()
route<Int, Boolean>({ 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()
}