diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinConsumerEndpointSpec.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinConsumerEndpointSpec.kt index c023f18452..535d761ec6 100644 --- a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinConsumerEndpointSpec.kt +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinConsumerEndpointSpec.kt @@ -1,5 +1,5 @@ /* - * Copyright 2023 the original author or authors. + * Copyright 2023-2024 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. @@ -37,75 +37,74 @@ import reactor.core.publisher.Mono * * @since 5.5.19 */ -abstract class KotlinConsumerEndpointSpec, H : MessageHandler>(open val delegate: S) - : ConsumerEndpointSpec(delegate.handler) { +abstract class KotlinConsumerEndpointSpec, H : MessageHandler>(open val delegate: S) { - override fun phase(phase: Int): S { - return this.delegate.phase(phase) + fun phase(phase: Int) { + this.delegate.phase(phase) } - override fun autoStartup(autoStartup: Boolean): S { - return this.delegate.autoStartup(autoStartup) + fun autoStartup(autoStartup: Boolean) { + this.delegate.autoStartup(autoStartup) } - override fun poller(pollerMetadata: PollerMetadata): S { - return this.delegate.poller(pollerMetadata) + fun poller(pollerMetadata: PollerMetadata) { + this.delegate.poller(pollerMetadata) } - override fun reactive(): S { - return this.delegate.reactive() + fun reactive() { + this.delegate.reactive() } fun reactive(reactiveCustomizer: (Flux>) -> Publisher>) { this.delegate.reactive(reactiveCustomizer) } - override fun role(role: String): S { - return this.delegate.role(role) + fun role(role: String) { + this.delegate.role(role) } - override fun taskScheduler(taskScheduler: TaskScheduler): S { - return this.delegate.taskScheduler(taskScheduler) + fun taskScheduler(taskScheduler: TaskScheduler) { + this.delegate.taskScheduler(taskScheduler) } - override fun handleMessageAdvice(vararg interceptors: MethodInterceptor?): S { - return this.delegate.handleMessageAdvice(*interceptors) + fun handleMessageAdvice(vararg interceptors: MethodInterceptor?) { + this.delegate.handleMessageAdvice(*interceptors) } - override fun advice(vararg advice: Advice?): S { - return this.delegate.advice(*advice) + fun advice(vararg advice: Advice?) { + this.delegate.advice(*advice) } - override fun transactional(transactionManager: TransactionManager): S { - return this.delegate.transactional(transactionManager) + fun transactional(transactionManager: TransactionManager) { + this.delegate.transactional(transactionManager) } - override fun transactional(transactionManager: TransactionManager, handleMessageAdvice: Boolean): S { - return this.delegate.transactional(transactionManager, handleMessageAdvice) + fun transactional(transactionManager: TransactionManager, handleMessageAdvice: Boolean) { + this.delegate.transactional(transactionManager, handleMessageAdvice) } - override fun transactional(transactionInterceptor: TransactionInterceptor): S { - return this.delegate.transactional(transactionInterceptor) + fun transactional(transactionInterceptor: TransactionInterceptor) { + this.delegate.transactional(transactionInterceptor) } - override fun transactional(): S { - return this.delegate.transactional() + fun transactional() { + this.delegate.transactional() } - override fun transactional(handleMessageAdvice: Boolean): S { - return this.delegate.transactional(handleMessageAdvice) + fun transactional(handleMessageAdvice: Boolean) { + this.delegate.transactional(handleMessageAdvice) } fun customizeMonoReply(replyCustomizer: (Message<*>, Mono) -> Publisher) { this.delegate.customizeMonoReply(replyCustomizer) } - override fun id(id: String?): S { - return this.delegate.id(id) + fun id(id: String?) { + this.delegate.id(id) } - override fun poller(pollerMetadataSpec: PollerSpec): S { - return this.delegate.poller(pollerMetadataSpec) + fun poller(pollerMetadataSpec: PollerSpec) { + this.delegate.poller(pollerMetadataSpec) } fun poller(pollers: (PollerFactory) -> PollerSpec) { 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 6c5781802a..c6d96a1d00 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 @@ -93,8 +93,8 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate a transformer endpoint. * @since 6.2 */ - fun transformWith(configurer: KotlinTransformerEndpointSpec.() -> Unit) { - this.delegate.register(KotlinTransformerEndpointSpec(), configurer) + fun transformWith(transformerConfigurer: KotlinTransformerEndpointSpec.() -> Unit) { + this.delegate.transformWith { transformerConfigurer(KotlinTransformerEndpointSpec(it)) } } /** @@ -109,8 +109,8 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ * Populate a splitter endpoint. * @since 6.2 */ - fun splitWith(configurer: KotlinSplitterSpec.() -> Unit) { - this.delegate.register(KotlinSplitterSpec(), configurer) + fun splitWith(splitterConfigurer: KotlinSplitterSpec.() -> Unit) { + this.delegate.splitWith { splitterConfigurer(KotlinSplitterSpec(it)) } } /** diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinSplitterSpec.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinSplitterSpec.kt index e1edfc9c64..4e75ac2731 100644 --- a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinSplitterSpec.kt +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinSplitterSpec.kt @@ -1,5 +1,5 @@ /* - * Copyright 2023 the original author or authors. + * Copyright 2023-2024 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. @@ -16,6 +16,14 @@ package org.springframework.integration.dsl +import org.springframework.expression.Expression +import org.springframework.integration.handler.BeanNameMessageProcessor +import org.springframework.integration.splitter.AbstractMessageSplitter +import org.springframework.integration.splitter.DefaultMessageSplitter +import org.springframework.integration.splitter.ExpressionEvaluatingSplitter +import org.springframework.integration.splitter.MethodInvokingSplitter +import org.springframework.messaging.MessageChannel + /** * A [SplitterSpec] wrapped for Kotlin DSL. * @@ -23,7 +31,8 @@ package org.springframework.integration.dsl * * @since 6.2 */ -class KotlinSplitterSpec : SplitterSpec() { +class KotlinSplitterSpec(override val delegate: SplitterSpec) + : KotlinConsumerEndpointSpec(delegate) { /** * Provide a Kotlin function as a direct delegate for @@ -32,12 +41,102 @@ class KotlinSplitterSpec : SplitterSpec() { * @param

the input type. */ inline fun function(crossinline function: (P) -> Any) { - expectedType(P::class.java) - function

{ function(it) } + this.delegate.expectedType(P::class.java) + this.delegate.function

{ function(it) } } + /** + * Set delimiters to tokenize String values. The default is + * `null` indicating that no tokenizing should occur. + * If delimiters are provided, they will be applied to any String payload. + * Only applied if provided `splitter` is instance of [DefaultMessageSplitter]. + * @param delimiters The delimiters. + */ + fun delimiters(delimiters: String) { + this.delegate.delimiters(delimiters) + } + + /** + * Provide an expression to use an [ExpressionEvaluatingSplitter] for the target handler. + * @param expression the SpEL expression to use. + */ + fun expression(expression: String){ + this.delegate.expression(expression) + } + + /** + * Provide an expression to use an [ExpressionEvaluatingSplitter] for the target handler. + * @param expression the SpEL expression to use. + */ + fun expression(expression: Expression) { + this.delegate.expression(expression) + } + + /** + * Provide a service to use a [MethodInvokingSplitter] for the target handler. + * This option can be set to an [AbstractMessageSplitter] implementation, + * a [MessageHandlerSpec] providing an [AbstractMessageSplitter], + * or [MessageProcessorSpec]. + * @param ref the service to call as a splitter POJO. + */ + fun ref(ref: Any) { + this.delegate.ref(ref) + } + + /** + * Provide a bean name to use a [MethodInvokingSplitter] + * (based on [BeanNameMessageProcessor]) for the target handler. + * @param refName the bean name for service to call as a splitter POJO. + */ + fun refName(refName: String) { + this.delegate.refName(refName) + } + + /** + * Provide a service method name to call. Optional. + * Use only together with [.ref] or [.refName]. + * @param method the service method name to call. + */ + fun method(method: String?) { + this.delegate.method(method) + } + + /** + * Set the applySequence flag to the specified value. Defaults to `true`. + * @param applySequence the applySequence. + */ + fun applySequence(applySequence: Boolean) { + this.delegate.applySequence(applySequence) + } + + /** + * Specify a channel where rejected Messages should be sent. If the discard + * channel is null (the default), rejected Messages will be dropped. + * A "Rejected Message" means that split function has returned an empty result (but not null): + * no items to iterate for sending. + * @param discardChannel The discard channel. + */ + fun discardChannel(discardChannel: MessageChannel) { + this.delegate.discardChannel(discardChannel) + } + + /** + * Specify a channel bean name where rejected Messages should be sent. If the discard + * channel is null (the default), rejected Messages will be dropped. + * A "Rejected Message" means that split function has returned an empty result (but not null): + * no items to iterate for sending. + * @param discardChannelName The discard channel bean name. + */ + fun discardChannel(discardChannelName: String) { + this.delegate.discardChannel(discardChannelName) + } + + /** + * Configure a subflow to run for discarded messages instead of a [discardChannel]. + * @param discardFlow the discard flow. + */ fun discardFlow(discardFlow: KotlinIntegrationFlowDefinition.() -> Unit) { - discardFlow {definition -> discardFlow(KotlinIntegrationFlowDefinition(definition)) } + this.delegate.discardFlow {definition -> discardFlow(KotlinIntegrationFlowDefinition(definition)) } } } diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinTransformerEndpointSpec.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinTransformerEndpointSpec.kt index c997a59020..be5f59e150 100644 --- a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinTransformerEndpointSpec.kt +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinTransformerEndpointSpec.kt @@ -1,5 +1,5 @@ /* - * Copyright 2023 the original author or authors. + * Copyright 2023-2024 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. @@ -16,7 +16,12 @@ package org.springframework.integration.dsl +import org.springframework.expression.Expression +import org.springframework.integration.core.GenericTransformer +import org.springframework.integration.handler.BeanNameMessageProcessor +import org.springframework.integration.transformer.ExpressionEvaluatingTransformer import org.springframework.integration.transformer.MessageTransformingHandler +import org.springframework.integration.transformer.MethodInvokingTransformer /** * A [TransformerEndpointSpec] wrapped for Kotlin DSL. @@ -27,7 +32,8 @@ import org.springframework.integration.transformer.MessageTransformingHandler * * @since 6.2 */ -class KotlinTransformerEndpointSpec : TransformerEndpointSpec() { +class KotlinTransformerEndpointSpec(override val delegate: TransformerEndpointSpec) + : KotlinConsumerEndpointSpec(delegate) { /** * Provide a Kotlin function as a direct delegate for [MessageTransformingHandler]. @@ -35,8 +41,68 @@ class KotlinTransformerEndpointSpec : TransformerEndpointSpec() { * @param

the input type. */ inline fun transformer(crossinline function: (P) -> Any) { - expectedType(P::class.java) - transformer { function(it) } + this.delegate.expectedType(P::class.java) + this.delegate.transformer { function(it) } + } + + /** + * Provide a [GenericTransformer] as a direct delegate for [MessageTransformingHandler]. + * @param transformer the [GenericTransformer] instance to use. + * @param

the input type. + * @param the output type. + */ + fun transformer(transformer: GenericTransformer) { + this.delegate.transformer(transformer) + } + + /** + * Provide an expression to use an [ExpressionEvaluatingTransformer] for the target handler. + * @param expression the SpEL expression to use. + */ + fun expression(expression: String) { + this.delegate.expression(expression) + } + + /** + * Provide an expression to use an [ExpressionEvaluatingTransformer] for the target handler. + * @param expression the SpEL expression to use. + */ + fun expression(expression: Expression) { + this.delegate.expression(expression) + } + + /** + * Provide a service to use a [MethodInvokingTransformer] for the target handler. + * @param ref the service to call as a transformer POJO. + */ + fun ref(ref: Any) { + this.delegate.ref(ref) + } + + /** + * Provide a bean name to use a [MethodInvokingTransformer] + * (based on [BeanNameMessageProcessor]) for the target handler. + * @param refName the bean name for service to call as a transformer POJO. + */ + fun refName(refName: String) { + this.delegate.refName(refName) + } + + /** + * Provide a service method name to call. Optional. + * Use only together with [.ref] or [.refName]. + * @param method the service method name to call. + */ + fun method(method: String?) { + this.delegate.method(method) + } + + /** + * Provide a [MessageProcessorSpec] as a factory for [MethodInvokingTransformer] delegate. + * @param processor the [MessageProcessorSpec] to use. + */ + fun processor(processor: MessageProcessorSpec<*>) { + this.delegate.processor(processor) } } 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 8e9094c7b7..796847a0f7 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 @@ -1,5 +1,5 @@ /* - * Copyright 2020-2023 the original author or authors. + * Copyright 2020-2024 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. @@ -214,7 +214,6 @@ class KotlinDslTests { .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) }