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

Add more Kotlin DSL for sub-flows (#3233)

Browse Source 
* Add more Kotlin DSL for sub-flows

* Introduce `KotlinSplitterEndpointSpec`, `KotlinEnricherSpec` &
`KotlinFilterEndpointSpec` to avoid an explicit use of `IntegrationFlow` usage
from Kotlin DSL
* Fix `AbstractKotlinRouterSpec` to extend `ConsumerEndpointSpec` for access
to more endpoint options as it is with Java DSL for similar specs
* Introduce a `KotlinIntegrationFlowDefinition.publishSubscribe()`
for a `BroadcastCapableChannel` and `vararg` of `KotlinIntegrationFlowDefinition` builders.
This allows us to not introduce a `BroadcastPublishSubscribeSpec` wrapper for Kotlin
and let us to remove `publishSubscribeChannel()` DSL methods for `PublishSubscribeSpec`
since it is covered with the `publishSubscribe(PublishSubscribeChannel())` and
respective set of sub-flows in Kotlin DSL as subscribers
* Use new Kotlin specs in the `KotlinIntegrationFlowDefinition` instead of their
Java variants

* * Mention `kotlin-dsl.adoc` in the ToC
This commit is contained in:
Artem Bilan
2020-04-01 11:06:54 -04:00
committed by GitHub
parent 6d5690fd58
commit 7433e41036
8 changed files with 276 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -29,7 +29,8 @@ import org.springframework.messaging.MessageChannel
* @since 5.3
*/
abstract class AbstractKotlinRouterSpec<S : AbstractRouterSpec<S, R>, R : AbstractMessageRouter>(
open val delegate: AbstractRouterSpec<S, R>) {
open val delegate: AbstractRouterSpec<S, R>)
: ConsumerEndpointSpec<S, R>(delegate.handler) {
fun ignoreSendFailures(ignoreSendFailures: Boolean) {
this.delegate.ignoreSendFailures(ignoreSendFailures)

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

@@ -0,0 +1,112 @@
/*
* 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.transformer.ContentEnricher
import org.springframework.integration.transformer.support.HeaderValueMessageProcessor
import org.springframework.messaging.Message
import org.springframework.messaging.MessageChannel
/**
* An [EnricherSpec] wrapped for Kotlin DSL.
*
* @property delegate the [EnricherSpec] this instance is delegating to.
*
* @author Artem Bilan
*
* @since 5.3
*/
class KotlinEnricherSpec(val delegate: EnricherSpec)
: ConsumerEndpointSpec<EnricherSpec, ContentEnricher>(delegate.handler) {
fun requestChannel(requestChannel: MessageChannel) {
this.delegate.requestChannel(requestChannel)
}
fun requestChannel(requestChannel: String) {
this.delegate.requestChannel(requestChannel)
}
fun replyChannel(replyChannel: MessageChannel) {
this.delegate.replyChannel(replyChannel)
}
fun replyChannel(replyChannel: String) {
this.delegate.replyChannel(replyChannel)
}
fun errorChannel(errorChannel: MessageChannel) {
this.delegate.errorChannel(errorChannel)
}
fun errorChannel(errorChannel: String) {
this.delegate.errorChannel(errorChannel)
}
fun requestTimeout(requestTimeout: Long) {
this.delegate.requestTimeout(requestTimeout)
}
fun replyTimeout(replyTimeout: Long) {
this.delegate.replyTimeout(replyTimeout)
}
fun requestPayloadExpression(requestPayloadExpression: String) {
this.delegate.requestPayloadExpression(requestPayloadExpression)
}
fun <P> requestPayload(function: (Message<P>) -> Any) {
this.delegate.requestPayload(function)
}
fun requestSubFlow(subFlow: KotlinIntegrationFlowDefinition.() -> Unit) {
this.delegate.requestSubFlow { subFlow(KotlinIntegrationFlowDefinition(it)) }
}
fun shouldClonePayload(shouldClonePayload: Boolean) {
this.delegate.shouldClonePayload(shouldClonePayload)
}
fun <V> property(key: String, value: V) {
this.delegate.property(key, value)
}
fun propertyExpression(key: String, expression: String) {
this.delegate.propertyExpression(key, expression)
}
fun <P> propertyFunction(key: String, function: (Message<P>) -> Any) {
this.delegate.propertyFunction(key, function)
}
fun <V> header(name: String, value: V, overwrite: Boolean?) {
this.delegate.header(name, value, overwrite)
}
fun headerExpression(name: String, expression: String, overwrite: Boolean?) {
this.delegate.header(name, expression, overwrite)
}
fun <P> headerFunction(name: String, function: (Message<P>) -> Any, overwrite: Boolean?) {
this.delegate.header(name, function, overwrite)
}
fun <V> header(headerName: String, headerValueMessageProcessor: HeaderValueMessageProcessor<V>) {
this.delegate.header(headerName, headerValueMessageProcessor)
}
}

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

@@ -0,0 +1,54 @@
/*
* 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.filter.MessageFilter
import org.springframework.messaging.MessageChannel
/**
* An [FilterEndpointSpec] wrapped for Kotlin DSL.
*
* @property delegate the [FilterEndpointSpec] this instance is delegating to.
*
* @author Artem Bilan
*
* @since 5.3
*/
class KotlinFilterEndpointSpec(val delegate: FilterEndpointSpec)
: ConsumerEndpointSpec<FilterEndpointSpec, MessageFilter>(delegate.handler) {
fun throwExceptionOnRejection(throwExceptionOnRejection: Boolean) {
this.delegate.throwExceptionOnRejection(throwExceptionOnRejection)
}
fun discardChannel(discardChannel: MessageChannel) {
this.delegate.discardChannel(discardChannel)
}
fun discardChannel(discardChannelName: String) {
this.delegate.discardChannel(discardChannelName)
}
fun discardWithinAdvice(discardWithinAdvice: Boolean) {
this.delegate.discardWithinAdvice(discardWithinAdvice)
}
fun discardFlow(subFlow: KotlinIntegrationFlowDefinition.() -> Unit) {
this.delegate.discardFlow { subFlow(KotlinIntegrationFlowDefinition(it)) }
}
}

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

@@ -19,6 +19,7 @@ package org.springframework.integration.dsl
import org.reactivestreams.Publisher
import org.springframework.expression.Expression
import org.springframework.integration.aggregator.AggregatingMessageHandler
import org.springframework.integration.channel.BroadcastCapableChannel
import org.springframework.integration.channel.FluxMessageChannel
import org.springframework.integration.channel.interceptor.WireTap
import org.springframework.integration.core.MessageSelector
@@ -54,7 +55,6 @@ 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
/**
@@ -112,9 +112,9 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
*/
inline fun <reified P> split(
crossinline function: (P) -> Any,
crossinline configurer: SplitterEndpointSpec<MethodInvokingSplitter>.() -> Unit) {
crossinline configurer: KotlinSplitterEndpointSpec<MethodInvokingSplitter>.() -> Unit) {
this.delegate.split(P::class.java, { function(it) }) { configurer(it) }
this.delegate.split(P::class.java, { function(it) }) { configurer(KotlinSplitterEndpointSpec(it)) }
}
/**
@@ -131,9 +131,9 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
*/
inline fun <reified P> filter(
crossinline function: (P) -> Boolean,
crossinline configurer: FilterEndpointSpec.() -> Unit) {
crossinline filterConfigurer: KotlinFilterEndpointSpec.() -> Unit) {
this.delegate.filter(P::class.java, { function(it) }) { configurer(it) }
this.delegate.filter(P::class.java, { function(it) }) { filterConfigurer(KotlinFilterEndpointSpec(it)) }
}
@@ -207,22 +207,19 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
}
/**
* The [org.springframework.integration.channel.PublishSubscribeChannel] `channel()`
* The [org.springframework.integration.channel.BroadcastCapableChannel] `channel()`
* method specific implementation to allow the use of the 'subflow' subscriber capability.
*/
fun publishSubscribeChannel(publishSubscribeChannelConfigurer: PublishSubscribeSpec.() -> Unit) {
this.delegate.publishSubscribeChannel(publishSubscribeChannelConfigurer)
}
fun publishSubscribe(broadcastCapableChannel: BroadcastCapableChannel,
vararg subscribeSubFlows: KotlinIntegrationFlowDefinition.() -> Unit) {
/**
* The [org.springframework.integration.channel.PublishSubscribeChannel] `channel()`
* method specific implementation to allow the use of the 'subflow' subscriber capability.
* Use the provided [Executor] for the target subscribers.
*/
fun publishSubscribeChannel(executor: Executor,
publishSubscribeChannelConfigurer: PublishSubscribeSpec.() -> Unit) {
this.delegate.publishSubscribeChannel(executor, Consumer(publishSubscribeChannelConfigurer))
val publishSubscribeChannelConfigurer =
Consumer<BroadcastPublishSubscribeSpec> { spec ->
subscribeSubFlows.forEach { subFlow ->
spec.subscribe { subFlow(KotlinIntegrationFlowDefinition(it)) }
}
}
this.delegate.publishSubscribeChannel(broadcastCapableChannel, publishSubscribeChannelConfigurer)
}
/**
@@ -331,10 +328,10 @@ 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]:
* In addition accept options for the integration endpoint using [KotlinFilterEndpointSpec]:
*/
fun filter(expression: String, endpointConfigurer: FilterEndpointSpec.() -> Unit = {}) {
this.delegate.filter(expression, endpointConfigurer)
fun filter(expression: String, filterConfigurer: KotlinFilterEndpointSpec.() -> Unit = {}) {
this.delegate.filter(expression) { filterConfigurer(KotlinFilterEndpointSpec(it)) }
}
/**
@@ -349,18 +346,19 @@ 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) {
this.delegate.filter(service, methodName, endpointConfigurer)
fun filter(service: Any, methodName: String?, filterConfigurer: KotlinFilterEndpointSpec.() -> Unit) {
this.delegate.filter(service, methodName) { filterConfigurer(KotlinFilterEndpointSpec(it)) }
}
/**
* Populate a [MessageFilter] with [MethodInvokingSelector]
* for the [MessageProcessor] from
* the provided [MessageProcessorSpec].
* In addition accept options for the integration endpoint using [FilterEndpointSpec].
* In addition accept options for the integration endpoint using [KotlinFilterEndpointSpec].
*/
fun filter(messageProcessorSpec: MessageProcessorSpec<*>, endpointConfigurer: FilterEndpointSpec.() -> Unit = {}) {
this.delegate.filter(messageProcessorSpec, endpointConfigurer)
fun filter(messageProcessorSpec: MessageProcessorSpec<*>,
filterConfigurer: KotlinFilterEndpointSpec.() -> Unit = {}) {
this.delegate.filter(messageProcessorSpec) { filterConfigurer(KotlinFilterEndpointSpec(it)) }
}
/**
@@ -512,8 +510,8 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
* to the current integration flow position
* with provided options.
*/
fun enrich(enricherConfigurer: EnricherSpec.() -> Unit) {
this.delegate.enrich(enricherConfigurer)
fun enrich(enricherConfigurer: KotlinEnricherSpec.() -> Unit) {
this.delegate.enrich { enricherConfigurer(KotlinEnricherSpec(it)) }
}
/**
@@ -562,9 +560,9 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
* SpEL expression.
*/
fun split(expression: String,
endpointConfigurer: SplitterEndpointSpec<ExpressionEvaluatingSplitter>.() -> Unit = {}) {
endpointConfigurer: KotlinSplitterEndpointSpec<ExpressionEvaluatingSplitter>.() -> Unit = {}) {
this.delegate.split(expression, endpointConfigurer)
this.delegate.split(expression) { endpointConfigurer(KotlinSplitterEndpointSpec(it)) }
}
/**
@@ -578,12 +576,12 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
/**
* Populate the [MethodInvokingSplitter] to evaluate the provided
* `method` of the `bean` at runtime.
* In addition accept options for the integration endpoint using [GenericEndpointSpec].
* In addition accept options for the integration endpoint using [KotlinSplitterEndpointSpec].
*/
fun split(service: Any, methodName: String?,
endpointConfigurer: SplitterEndpointSpec<MethodInvokingSplitter>.() -> Unit) {
splitterConfigurer: KotlinSplitterEndpointSpec<MethodInvokingSplitter>.() -> Unit) {
this.delegate.split(service, methodName, endpointConfigurer)
this.delegate.split(service, methodName) { splitterConfigurer(KotlinSplitterEndpointSpec(it)) }
}
/**
@@ -597,33 +595,33 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
/**
* Populate the [MethodInvokingSplitter] to evaluate the provided
* `method` of the `bean` at runtime.
* In addition accept options for the integration endpoint using [GenericEndpointSpec].
* In addition accept options for the integration endpoint using [KotlinSplitterEndpointSpec].
*/
fun split(beanName: String, methodName: String?,
endpointConfigurer: SplitterEndpointSpec<MethodInvokingSplitter>.() -> Unit) {
splitterConfigurer: KotlinSplitterEndpointSpec<MethodInvokingSplitter>.() -> Unit) {
this.delegate.split(beanName, methodName, endpointConfigurer)
this.delegate.split(beanName, methodName) { splitterConfigurer(KotlinSplitterEndpointSpec(it)) }
}
/**
* Populate the [MethodInvokingSplitter] to evaluate the
* [MessageProcessor] at runtime
* from provided [MessageProcessorSpec].
* In addition accept options for the integration endpoint using [GenericEndpointSpec].
* In addition accept options for the integration endpoint using [KotlinSplitterEndpointSpec].
*/
fun split(messageProcessorSpec: MessageProcessorSpec<*>,
endpointConfigurer: SplitterEndpointSpec<MethodInvokingSplitter>.() -> Unit = {}) {
splitterConfigurer: KotlinSplitterEndpointSpec<MethodInvokingSplitter>.() -> Unit = {}) {
this.delegate.split(messageProcessorSpec, endpointConfigurer)
this.delegate.split(messageProcessorSpec) { splitterConfigurer(KotlinSplitterEndpointSpec(it)) }
}
/**
* Populate the provided [AbstractMessageSplitter] to the current integration flow position.
*/
fun <S : AbstractMessageSplitter> split(splitterMessageHandlerSpec: MessageHandlerSpec<*, S>,
endpointConfigurer: SplitterEndpointSpec<S>.() -> Unit = {}) {
splitterConfigurer: KotlinSplitterEndpointSpec<S>.() -> Unit = {}) {
this.delegate.split(splitterMessageHandlerSpec, endpointConfigurer)
this.delegate.split(splitterMessageHandlerSpec) { splitterConfigurer(KotlinSplitterEndpointSpec(it)) }
}
/**
@@ -631,9 +629,9 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
* flow position.
*/
fun <S : AbstractMessageSplitter> split(splitter: S,
endpointConfigurer: SplitterEndpointSpec<S>.() -> Unit = {}) {
splitterConfigurer: KotlinSplitterEndpointSpec<S>.() -> Unit = {}) {
this.delegate.split(splitter, endpointConfigurer)
this.delegate.split(splitter) { splitterConfigurer(KotlinSplitterEndpointSpec(it)) }
}
/**

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

@@ -0,0 +1,54 @@
/*
* 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.splitter.AbstractMessageSplitter
import org.springframework.messaging.MessageChannel
/**
* An [SplitterEndpointSpec] wrapped for Kotlin DSL.
*
* @property delegate the [SplitterEndpointSpec] this instance is delegating to.
*
* @author Artem Bilan
*
* @since 5.3
*/
class KotlinSplitterEndpointSpec<H : AbstractMessageSplitter>(val delegate: SplitterEndpointSpec<H>)
: ConsumerEndpointSpec<KotlinSplitterEndpointSpec<H>, H>(delegate.handler) {
fun applySequence(applySequence: Boolean) {
this.delegate.applySequence(applySequence)
}
fun delimiters(delimiters: String) {
this.delegate.delimiters(delimiters)
}
fun discardChannel(discardChannel: MessageChannel) {
this.delegate.discardChannel(discardChannel)
}
fun discardChannel(discardChannelName: String) {
this.delegate.discardChannel(discardChannelName)
}
fun discardFlow(discardFlow: KotlinIntegrationFlowDefinition.() -> Unit) {
this.delegate.discardFlow { discardFlow(KotlinIntegrationFlowDefinition(it)) }
}
}

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

@@ -29,9 +29,9 @@ 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.PublishSubscribeChannel
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.context.IntegrationFlowContext
import org.springframework.integration.endpoint.MessageProcessorMessageSource
@@ -248,7 +248,13 @@ class KotlinDslTests {
fun messageSourceFlow() =
integrationFlow(MessageProcessorMessageSource { "testSource" },
{ poller { it.trigger(OnlyOnceTrigger()) } }) {
channel { queue("fromSupplierQueue") }
publishSubscribe(PublishSubscribeChannel(),
{
channel { queue("fromSupplierQueue") }
},
{
log<Any>(LoggingHandler.Level.WARN) { "From second subscriber: ${it.payload}"}
})
}
@Bean

2
src/reference/asciidoc/index-single.adoc
View File

@@ -23,6 +23,8 @@ include::./messaging-endpoints.adoc[]
include::./dsl.adoc[]
include::./kotlin-dsl.adoc[]
include::./system-management.adoc[]
include::./endpoint-summary.adoc[]

5
src/reference/asciidoc/index.adoc
View File

@@ -2,8 +2,8 @@
include::./index-header.adoc[]
Welcome to the Spring Integration reference documentation! This documentation is also available
as single searchable link:index-single.html[html] and link:../pdf/spring-integration-reference.pdf[pdf] documents.
Welcome to the Spring Integration reference documentation!
This documentation is also available as single searchable link:index-single.html[html] and link:../pdf/spring-integration-reference.pdf[pdf] documents.
[horizontal]
<<./preface.adoc#preface,Preface>> ::
@@ -15,6 +15,7 @@ as single searchable link:index-single.html[html] and link:../pdf/spring-integra
<<./message-transformation.adoc#messaging-transformation-chapter,Message Transformation>> ::
<<./messaging-endpoints.adoc#messaging-endpoints-chapter,Messaging Endpoints>> ::
<<./dsl.adoc#java-dsl,Java DSL>> ::
<<./kotlin-dsl.adoc#kotlin-dsl,Kotlin DSL>> ::
<<./system-management.adoc#system-management-chapter,System Management>> ::
<<./reactive-streams.adoc#reactive-streams,Reactive Streams Support>> ::