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 index 2564bbcd2a..ba95249be2 100644 --- 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 @@ -29,7 +29,8 @@ import org.springframework.messaging.MessageChannel * @since 5.3 */ abstract class AbstractKotlinRouterSpec, R : AbstractMessageRouter>( - open val delegate: AbstractRouterSpec) { + open val delegate: AbstractRouterSpec) + : ConsumerEndpointSpec(delegate.handler) { fun ignoreSendFailures(ignoreSendFailures: Boolean) { this.delegate.ignoreSendFailures(ignoreSendFailures) diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinEnricherSpec.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinEnricherSpec.kt new file mode 100644 index 0000000000..2bf51a3b4a --- /dev/null +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinEnricherSpec.kt @@ -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(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

requestPayload(function: (Message

) -> 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 property(key: String, value: V) { + this.delegate.property(key, value) + } + + fun propertyExpression(key: String, expression: String) { + this.delegate.propertyExpression(key, expression) + } + + fun

propertyFunction(key: String, function: (Message

) -> Any) { + this.delegate.propertyFunction(key, function) + } + + fun 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

headerFunction(name: String, function: (Message

) -> Any, overwrite: Boolean?) { + this.delegate.header(name, function, overwrite) + } + + fun header(headerName: String, headerValueMessageProcessor: HeaderValueMessageProcessor) { + this.delegate.header(headerName, headerValueMessageProcessor) + } + +} diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinFilterEndpointSpec.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinFilterEndpointSpec.kt new file mode 100644 index 0000000000..be16b8efab --- /dev/null +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinFilterEndpointSpec.kt @@ -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(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)) } + } + +} 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 dcf7bbcd11..b0ad7695f3 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 @@ -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 split( crossinline function: (P) -> Any, - crossinline configurer: SplitterEndpointSpec.() -> Unit) { + crossinline configurer: KotlinSplitterEndpointSpec.() -> 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 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 { 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.() -> Unit = {}) { + endpointConfigurer: KotlinSplitterEndpointSpec.() -> 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.() -> Unit) { + splitterConfigurer: KotlinSplitterEndpointSpec.() -> 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.() -> Unit) { + splitterConfigurer: KotlinSplitterEndpointSpec.() -> 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.() -> Unit = {}) { + splitterConfigurer: KotlinSplitterEndpointSpec.() -> Unit = {}) { - this.delegate.split(messageProcessorSpec, endpointConfigurer) + this.delegate.split(messageProcessorSpec) { splitterConfigurer(KotlinSplitterEndpointSpec(it)) } } /** * Populate the provided [AbstractMessageSplitter] to the current integration flow position. */ fun split(splitterMessageHandlerSpec: MessageHandlerSpec<*, S>, - endpointConfigurer: SplitterEndpointSpec.() -> Unit = {}) { + splitterConfigurer: KotlinSplitterEndpointSpec.() -> 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 split(splitter: S, - endpointConfigurer: SplitterEndpointSpec.() -> Unit = {}) { + splitterConfigurer: KotlinSplitterEndpointSpec.() -> Unit = {}) { - this.delegate.split(splitter, endpointConfigurer) + this.delegate.split(splitter) { splitterConfigurer(KotlinSplitterEndpointSpec(it)) } } /** diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinSplitterEndpointSpec.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinSplitterEndpointSpec.kt new file mode 100644 index 0000000000..9f565f29f0 --- /dev/null +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinSplitterEndpointSpec.kt @@ -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(val delegate: SplitterEndpointSpec) + : ConsumerEndpointSpec, 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)) } + } + +} 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 fd5f86a527..a58c50f5d4 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 @@ -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(LoggingHandler.Level.WARN) { "From second subscriber: ${it.payload}"} + }) } @Bean diff --git a/src/reference/asciidoc/index-single.adoc b/src/reference/asciidoc/index-single.adoc index bace77b040..e7ef170b63 100644 --- a/src/reference/asciidoc/index-single.adoc +++ b/src/reference/asciidoc/index-single.adoc @@ -23,6 +23,8 @@ include::./messaging-endpoints.adoc[] include::./dsl.adoc[] +include::./kotlin-dsl.adoc[] + include::./system-management.adoc[] include::./endpoint-summary.adoc[] diff --git a/src/reference/asciidoc/index.adoc b/src/reference/asciidoc/index.adoc index 2047d53a31..4ca5dabc23 100644 --- a/src/reference/asciidoc/index.adoc +++ b/src/reference/asciidoc/index.adoc @@ -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>> ::