From 69157dea823ad4dec7c0f5bc6ae3360cda243b9b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 16 Dec 2019 15:35:10 -0500 Subject: [PATCH] Rework KotlinIntFlowDefinition for delegation According Kotlin best practice it is better to not extend from Java classes, but instead provide a clean Kotlin API and delegate to Java one internally * Implement all the `IntegrationFlowDefinition` methods in the `KotlinIntegrationFlowDefinition` Kotlin way and delegate to the provided `IntegrationFlowDefinition` instance * Remove `@UseExperimental` and `@BuilderInference` since we now have all the API in Kotlin * Rework `KotlinDslTests` for support now Kotlin code style * Remove `-Xuse-experimental` from Gradle config since we don't use experimental API any more --- spring-integration-kotlin-dsl/build.gradle | 2 +- .../integration/dsl/IntegrationFlowDsl.kt | 26 +- .../dsl/KotlinIntegrationFlowDefinition.kt | 1007 +++++++++++++++-- .../dsl/kotlin/test/KotlinDslTests.kt | 27 +- 4 files changed, 946 insertions(+), 116 deletions(-) diff --git a/spring-integration-kotlin-dsl/build.gradle b/spring-integration-kotlin-dsl/build.gradle index 0057783..bbe9115 100644 --- a/spring-integration-kotlin-dsl/build.gradle +++ b/spring-integration-kotlin-dsl/build.gradle @@ -66,7 +66,7 @@ dependencyManagement { compileKotlin { kotlinOptions { jvmTarget = '1.8' - freeCompilerArgs = ['-Xjsr305=strict', '-Xuse-experimental=kotlin.Experimental'] + freeCompilerArgs = ['-Xjsr305=strict'] allWarningsAsErrors = true } } diff --git a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt index 453d0c7..0f7302d 100644 --- a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt +++ b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt @@ -14,8 +14,6 @@ * limitations under the License. */ -@file:UseExperimental(kotlin.experimental.ExperimentalTypeInference::class) - package org.springframework.integration.dsl import org.reactivestreams.Publisher @@ -38,7 +36,7 @@ private fun buildIntegrationFlow(flowBuilder: IntegrationFlowBuilder, * * @author Artem Bilan */ -fun integrationFlow(@BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = +fun integrationFlow(flow: KotlinIntegrationFlowDefinition.() -> Unit) = IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) } @@ -51,7 +49,7 @@ fun integrationFlow(@BuilderInference flow: KotlinIntegrationFlowDefinition.() - */ inline fun integrationFlow( crossinline gateway: (GatewayProxySpec) -> Unit = {}, - @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit): IntegrationFlow { + flow: KotlinIntegrationFlowDefinition.() -> Unit): IntegrationFlow { val flowBuilder = IntegrationFlows.from(T::class.java) { gateway(it) } flow(KotlinIntegrationFlowDefinition(flowBuilder)) @@ -65,7 +63,7 @@ inline fun integrationFlow( * @author Artem Bilan */ fun integrationFlow(channelName: String, fixedSubscriber: Boolean = false, - @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = + flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(channelName, fixedSubscriber), flow) /** @@ -74,7 +72,7 @@ fun integrationFlow(channelName: String, fixedSubscriber: Boolean = false, * * @author Artem Bilan */ -fun integrationFlow(channel: MessageChannel, @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = +fun integrationFlow(channel: MessageChannel, flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(channel), flow) /** @@ -85,7 +83,7 @@ fun integrationFlow(channel: MessageChannel, @BuilderInference flow: KotlinInteg */ fun integrationFlow(messageSource: MessageSource<*>, options: (SourcePollingChannelAdapterSpec) -> Unit = {}, - @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = + flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(messageSource, Consumer { options(it) }), flow) /** @@ -96,7 +94,7 @@ fun integrationFlow(messageSource: MessageSource<*>, */ fun integrationFlow(messageSource: MessageSourceSpec<*, out MessageSource<*>>, options: (SourcePollingChannelAdapterSpec) -> Unit = {}, - @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = + flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(messageSource, options), flow) /** @@ -107,7 +105,7 @@ fun integrationFlow(messageSource: MessageSourceSpec<*, out MessageSource<*>>, */ fun integrationFlow(source: () -> Any, options: (SourcePollingChannelAdapterSpec) -> Unit = {}, - @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = + flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(source, options), flow) /** @@ -117,7 +115,7 @@ fun integrationFlow(source: () -> Any, * @author Artem Bilan */ fun integrationFlow(publisher: Publisher>, - @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = + flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(publisher), flow) /** @@ -127,7 +125,7 @@ fun integrationFlow(publisher: Publisher>, * @author Artem Bilan */ fun integrationFlow(gateway: MessagingGatewaySupport, - @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = + flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(gateway), flow) /** @@ -137,7 +135,7 @@ fun integrationFlow(gateway: MessagingGatewaySupport, * @author Artem Bilan */ fun integrationFlow(gatewaySpec: MessagingGatewaySpec<*, *>, - @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = + flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(gatewaySpec), flow) /** @@ -147,7 +145,7 @@ fun integrationFlow(gatewaySpec: MessagingGatewaySpec<*, *>, * @author Artem Bilan */ fun integrationFlow(producer: MessageProducerSupport, - @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = + flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(producer), flow) /** @@ -157,5 +155,5 @@ fun integrationFlow(producer: MessageProducerSupport, * @author Artem Bilan */ fun integrationFlow(producerSpec: MessageProducerSpec<*, *>, - @BuilderInference flow: KotlinIntegrationFlowDefinition.() -> Unit) = + flow: KotlinIntegrationFlowDefinition.() -> Unit) = buildIntegrationFlow(IntegrationFlows.from(producerSpec), flow) diff --git a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt index 6dca00d..ae37cf1 100644 --- a/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt +++ b/spring-integration-kotlin-dsl/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt @@ -16,95 +16,56 @@ 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.FluxMessageChannel +import org.springframework.integration.channel.interceptor.WireTap +import org.springframework.integration.core.MessageSelector +import org.springframework.integration.dsl.support.MessageChannelReference +import org.springframework.integration.filter.MessageFilter +import org.springframework.integration.filter.MethodInvokingSelector +import org.springframework.integration.handler.BridgeHandler +import org.springframework.integration.handler.DelayHandler +import org.springframework.integration.handler.GenericHandler +import org.springframework.integration.handler.LoggingHandler +import org.springframework.integration.handler.MessageProcessor +import org.springframework.integration.handler.MessageTriggerAction +import org.springframework.integration.handler.ServiceActivatingHandler +import org.springframework.integration.router.AbstractMessageRouter +import org.springframework.integration.router.ErrorMessageExceptionTypeRouter +import org.springframework.integration.router.ExpressionEvaluatingRouter import org.springframework.integration.router.MethodInvokingRouter +import org.springframework.integration.router.RecipientListRouter +import org.springframework.integration.scattergather.ScatterGatherHandler +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.integration.store.MessageStore +import org.springframework.integration.support.MapBuilder +import org.springframework.integration.transformer.ClaimCheckInTransformer +import org.springframework.integration.transformer.ClaimCheckOutTransformer +import org.springframework.integration.transformer.ExpressionEvaluatingTransformer +import org.springframework.integration.transformer.HeaderFilter import org.springframework.integration.transformer.MessageTransformingHandler +import org.springframework.integration.transformer.MethodInvokingTransformer +import org.springframework.messaging.Message import org.springframework.messaging.MessageChannel +import org.springframework.messaging.MessageHandler +import reactor.core.publisher.Flux +import java.util.concurrent.Executor +import java.util.function.Consumer +import java.util.function.Function /** - * A [BaseIntegrationFlowDefinition] extension for Kotlin-specif inline functions with reified - * generic types. + * An [IntegrationFlowDefinition] wrapped for Kotlin DSL. * - * @property adaptee the [IntegrationFlowDefinition] this instance is adapted. + * @property delegate the [IntegrationFlowDefinition] this instance is delegating to. * * @author Artem Bilan */ -class KotlinIntegrationFlowDefinition(private val adaptee: IntegrationFlowDefinition<*>) : - BaseIntegrationFlowDefinition() { - - /** - * Delegate a provided component into an `adaptee` set of components. - */ - override fun addComponent(component: Any): KotlinIntegrationFlowDefinition { - return addComponent(component, null) - } - - /** - * Delegate a provided component into an `adaptee` set of components. - */ - override fun addComponent(component: Any, beanName: String?): KotlinIntegrationFlowDefinition { - this.adaptee.addComponent(component, beanName) - return _this() - } - - /** - * Delegate provided components into an `adaptee` set of components. - */ - override fun addComponents(components: Map?): KotlinIntegrationFlowDefinition { - this.adaptee.addComponents(components) - return _this() - } - - /** - * Get a [Map] of components from `adaptee`. - */ - override fun getIntegrationComponents(): Map { - return this.adaptee.getIntegrationComponents() - } - - /** - * Set a provided [MessageChannel] as a current in the `adaptee`. - */ - override fun currentMessageChannel(currentMessageChannel: MessageChannel?): KotlinIntegrationFlowDefinition { - this.adaptee.currentMessageChannel(currentMessageChannel) - return _this() - } - - /** - * Get a current [MessageChannel] from the `adaptee`. - */ - override fun getCurrentMessageChannel(): MessageChannel? { - return this.adaptee.getCurrentMessageChannel() - } - - /** - * Delegate a provided component into an `adaptee` current component. - */ - override fun currentComponent(component: Any?): KotlinIntegrationFlowDefinition { - this.adaptee.currentComponent(component) - return _this() - } - - /** - * Get a current component from `adaptee`. - */ - override fun getCurrentComponent(): Any? { - return this.adaptee.getCurrentComponent() - } - - /** - * Set a flag for an implicit channel on the `adaptee`. - */ - override fun setImplicitChannel(implicitChannel: Boolean) { - this.adaptee.setImplicitChannel(implicitChannel) - } - - /** - * Get an implicit channel flag from the `adaptee`. - */ - override fun isImplicitChannel(): Boolean { - return this.adaptee.isImplicitChannel() - } +class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: IntegrationFlowDefinition<*>) { /** * Inline function for [IntegrationFlowDefinition.convert] providing a `convert()` variant @@ -113,7 +74,7 @@ class KotlinIntegrationFlowDefinition(private val adaptee: IntegrationFlowDefini inline fun convert( crossinline configurer: (GenericEndpointSpec) -> Unit = {}) { - convert(T::class.java) { configurer(it) } + this.delegate.convert(T::class.java) { configurer(it) } } /** @@ -121,7 +82,7 @@ class KotlinIntegrationFlowDefinition(private val adaptee: IntegrationFlowDefini * with reified generic type. */ inline fun transform(crossinline function: (P) -> T) { - transform(P::class.java) { function(it) } + this.delegate.transform(P::class.java) { function(it) } } /** @@ -132,15 +93,15 @@ class KotlinIntegrationFlowDefinition(private val adaptee: IntegrationFlowDefini crossinline function: (P) -> T, crossinline configurer: (GenericEndpointSpec) -> Unit) { - transform(P::class.java, { function(it) }) { configurer(it) } + this.delegate.transform(P::class.java, { function(it) }) { configurer(it) } } /** * Inline function for [IntegrationFlowDefinition.split] providing a `split()` variant * with reified generic type. */ - inline fun split(crossinline function: (P) -> Any) { - split(P::class.java) { function(it) } + inline fun split(noinline function: (P) -> Any) { + this.delegate.split(P::class.java) { function(it) } } @@ -152,7 +113,7 @@ class KotlinIntegrationFlowDefinition(private val adaptee: IntegrationFlowDefini crossinline function: (P) -> Any, crossinline configurer: (SplitterEndpointSpec) -> Unit) { - split(P::class.java, { function(it) }) { configurer(it) } + this.delegate.split(P::class.java, { function(it) }) { configurer(it) } } /** @@ -160,7 +121,7 @@ class KotlinIntegrationFlowDefinition(private val adaptee: IntegrationFlowDefini * with reified generic type. */ inline fun filter(crossinline function: (P) -> Boolean) { - filter(P::class.java) { function(it) } + this.delegate.filter(P::class.java) { function(it) } } /** @@ -171,7 +132,7 @@ class KotlinIntegrationFlowDefinition(private val adaptee: IntegrationFlowDefini crossinline function: (P) -> Boolean, crossinline configurer: (FilterEndpointSpec) -> Unit) { - filter(P::class.java, { function(it) }) { configurer(it) } + this.delegate.filter(P::class.java, { function(it) }) { configurer(it) } } @@ -180,7 +141,7 @@ class KotlinIntegrationFlowDefinition(private val adaptee: IntegrationFlowDefini * with reified generic type. */ inline fun route(crossinline function: (P) -> T) { - route(P::class.java) { function(it) } + this.delegate.route(P::class.java, Function { function(it) }) } /** @@ -191,7 +152,875 @@ class KotlinIntegrationFlowDefinition(private val adaptee: IntegrationFlowDefini crossinline function: (P) -> T, crossinline configurer: (RouterSpec) -> Unit) { - route(P::class.java, { function(it) }) { configurer(it) } + this.delegate.route(P::class.java, { function(it) }) { configurer(it) } + } + + /** + * Populate an [org.springframework.integration.channel.FixedSubscriberChannel] instance + * at the current [IntegrationFlow] chain position. + * The provided `messageChannelName` is used for the bean registration. + */ + fun fixedSubscriberChannel(messageChannelName: String? = null) { + this.delegate.fixedSubscriberChannel(messageChannelName) + } + + /** + * Populate a [MessageChannelReference] instance + * at the current [IntegrationFlow] chain position. + * The provided `messageChannelName` is used for the bean registration + * ([org.springframework.integration.channel.DirectChannel]), if there is no such a bean + * in the application context. Otherwise the existing [MessageChannel] bean is used + * to wire integration endpoints. + */ + fun channel(messageChannelName: String) { + this.delegate.channel(messageChannelName) + } + + /** + * Populate a [MessageChannel] instance + * at the current [IntegrationFlow] chain position using the [MessageChannelSpec] + * fluent API. + */ + fun channel(messageChannelSpec: MessageChannelSpec<*, *>) { + this.delegate.channel(messageChannelSpec) + } + + /** + * Populate the provided [MessageChannel] instance + * at the current [IntegrationFlow] chain position. + * The `messageChannel` can be an existing bean, or fresh instance, in which case + * the [org.springframework.integration.dsl.context.IntegrationFlowBeanPostProcessor] + * will populate it as a bean with a generated name. + */ + fun channel(messageChannel: MessageChannel) { + this.delegate.channel(messageChannel) + } + + /** + * Populate a [MessageChannel] instance + * at the current [IntegrationFlow] chain position using the [Channels] + * factory fluent API. + */ + fun channel(channels: (Channels) -> MessageChannelSpec<*, *>) { + this.delegate.channel(channels) + } + + /** + * The [org.springframework.integration.channel.PublishSubscribeChannel] `channel()` + * method specific implementation to allow the use of the 'subflow' subscriber capability. + */ + fun publishSubscribeChannel(publishSubscribeChannelConfigurer: (PublishSubscribeSpec) -> Unit) { + this.delegate.publishSubscribeChannel(publishSubscribeChannelConfigurer) + } + + /** + * 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)) + } + + /** + * Populate the `Wire Tap` EI Pattern specific + * [org.springframework.messaging.support.ChannelInterceptor] implementation + * to the current channel. + * This method can be used after any `channel()` for explicit [MessageChannel], + * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. + */ + fun wireTap(flow: KotlinIntegrationFlowDefinition.() -> Unit) { + this.delegate.wireTap(IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) }) + } + + /** + * Populate the `Wire Tap` EI Pattern specific + * [org.springframework.messaging.support.ChannelInterceptor] implementation + * to the current channel. + * This method can be used after any `channel()` for explicit [MessageChannel], + * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. + */ + fun wireTap(flow: KotlinIntegrationFlowDefinition.() -> Unit, + wireTapConfigurer: (WireTapSpec) -> Unit) { + + this.delegate.wireTap( + IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) }, + Consumer(wireTapConfigurer)) + } + + /** + * Populate the `Wire Tap` EI Pattern specific + * [org.springframework.messaging.support.ChannelInterceptor] implementation + * to the current channel. + * This method can be used after any `channel()` for explicit [MessageChannel], + * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. + */ + fun wireTap(wireTapChannel: String, wireTapConfigurer: (WireTapSpec) -> Unit = {}) { + this.delegate.wireTap(wireTapChannel, wireTapConfigurer) + } + + /** + * Populate the `Wire Tap` EI Pattern specific + * [org.springframework.messaging.support.ChannelInterceptor] implementation + * to the current channel. + * This method can be used after any `channel()` for explicit [MessageChannel], + * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. + */ + fun wireTap(wireTapChannel: MessageChannel, wireTapConfigurer: (WireTapSpec) -> Unit = {}) { + this.delegate.wireTap(wireTapChannel, Consumer(wireTapConfigurer)) + } + + /** + * Populate the `Wire Tap` EI Pattern specific + * [org.springframework.messaging.support.ChannelInterceptor] implementation + * to the current channel. + * This method can be used after any `channel()` for explicit [MessageChannel], + * but with the caution do not impact existing [org.springframework.messaging.support.ChannelInterceptor]s. + */ + fun wireTap(wireTapSpec: WireTapSpec) { + this.delegate.wireTap(wireTapSpec) + } + + /** + * Populate the `Control Bus` EI Pattern specific [MessageHandler] implementation + * at the current [IntegrationFlow] chain position. + */ + fun controlBus(endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + this.delegate.controlBus(endpointConfigurer) + } + + /** + * Populate the `Transformer` EI Pattern specific [MessageHandler] implementation + * for the SpEL [Expression]. + */ + fun transform(expression: String, + endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + + this.delegate.transform(expression, endpointConfigurer) + } + + /** + * Populate the `MessageTransformingHandler` for the [MethodInvokingTransformer] + * to invoke the service method at runtime. + */ + fun transform(service: Any, methodName: String? = null) { + this.delegate.transform(service, methodName) + } + + /** + * Populate the `MessageTransformingHandler` for the [MethodInvokingTransformer] + * to invoke the service method at runtime. + * @param service the service to use. + * @param methodName the method to invoke. + * @param endpointConfigurer the consumer to provide integration endpoint options. + * @return the current [BaseIntegrationFlowDefinition]. + * @see ExpressionEvaluatingTransformer + */ + fun transform(service: Any, methodName: String?, + endpointConfigurer: (GenericEndpointSpec) -> Unit) { + + this.delegate.transform(service, methodName, endpointConfigurer) + } + + /** + * Populate the [MessageTransformingHandler] instance for the + * [org.springframework.integration.handler.MessageProcessor] from provided [MessageProcessorSpec]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun transform(messageProcessorSpec: MessageProcessorSpec<*>, + endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + + this.delegate.transform(messageProcessorSpec, endpointConfigurer) + } + + /** + * Populate a [MessageFilter] with [MessageSelector] for the provided SpEL expression. + * In addition accept options for the integration endpoint using [FilterEndpointSpec]: + */ + fun filter(expression: String, endpointConfigurer: (FilterEndpointSpec) -> Unit = {}) { + this.delegate.filter(expression, endpointConfigurer) + } + + /** + * Populate a [MessageFilter] with [MethodInvokingSelector] for the + * method of the provided service. + */ + fun filter(service: Any, methodName: String? = null) { + this.delegate.filter(service, methodName) + } + + /** + * 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) + } + + /** + * Populate a [MessageFilter] with [MethodInvokingSelector] + * for the [MessageProcessor] from + * the provided [MessageProcessorSpec]. + * In addition accept options for the integration endpoint using [FilterEndpointSpec]. + */ + fun filter(messageProcessorSpec: MessageProcessorSpec<*>, endpointConfigurer: (FilterEndpointSpec) -> Unit = {}) { + this.delegate.filter(messageProcessorSpec, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the selected protocol specific + * [MessageHandler] implementation from `Namespace Factory`: + */ + fun handle(messageHandlerSpec: MessageHandlerSpec<*, H>) { + this.delegate.handle(messageHandlerSpec) + } + + /** + * Populate a [ServiceActivatingHandler] for the provided + * [MessageHandler] implementation. + */ + fun handle(messageHandler: MessageHandler) { + this.delegate.handle(messageHandler) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the `method` for provided `bean` at runtime. + */ + fun handle(beanName: String, methodName: String? = null) { + this.delegate.handle(beanName, methodName) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the `method` for provided `bean` at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(beanName: String, methodName: String?, + endpointConfigurer: (GenericEndpointSpec) -> Unit) { + + this.delegate.handle(beanName, methodName, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the `method` for provided `bean` at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(service: Any, methodName: String? = null) { + this.delegate.handle(service, methodName) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the `method` for provided `bean` at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(service: Any, methodName: String?, + endpointConfigurer: (GenericEndpointSpec) -> Unit) { + + this.delegate.handle(service, methodName, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [org.springframework.integration.handler.MethodInvokingMessageProcessor] + * to invoke the provided [GenericHandler] at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + inline fun handle( + handler: GenericHandler

, + crossinline endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + + this.delegate.handle(P::class.java, handler, Consumer { endpointConfigurer(it) }) + } + + /** + * Populate a [ServiceActivatingHandler] for the + * [MessageProcessor] from the provided + * [MessageProcessorSpec]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(messageProcessorSpec: MessageProcessorSpec<*>, + endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + + this.delegate.handle(messageProcessorSpec, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the selected protocol specific + * [MessageHandler] implementation from `Namespace Factory`: + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(messageHandlerSpec: MessageHandlerSpec<*, H>, + endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + + this.delegate.handle(messageHandlerSpec, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] for the provided + * [MessageHandler] implementation. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun handle(messageHandler: H, endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + this.delegate.handle(messageHandler, endpointConfigurer) + } + + /** + * Populate a [BridgeHandler] to the current integration flow position. + */ + fun bridge(endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + this.delegate.bridge(endpointConfigurer) + } + + /** + * Populate a [DelayHandler] to the current integration flow position. + */ + fun delay(groupId: String, endpointConfigurer: (DelayerEndpointSpec) -> Unit = {}) { + this.delegate.delay(groupId, endpointConfigurer) + } + + /** + * Populate a [org.springframework.integration.transformer.ContentEnricher] + * to the current integration flow position + * with provided options. + */ + fun enrich(enricherConfigurer: (EnricherSpec) -> Unit) { + this.delegate.enrich(enricherConfigurer) + } + + /** + * Populate a [MessageTransformingHandler] for + * a [org.springframework.integration.transformer.HeaderEnricher] + * using header values from provided [MapBuilder]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun enrichHeaders(headers: MapBuilder<*, String, Any>, + endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + + this.delegate.enrichHeaders(headers, endpointConfigurer) + } + + /** + * Accept a [Map] of values to be used for the + * [Message] header enrichment. + * `values` can apply an [Expression] + * to be evaluated against a request [Message]. + */ + fun enrichHeaders(headers: Map, + endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + + this.delegate.enrichHeaders(headers, endpointConfigurer) + } + + /** + * Populate a [MessageTransformingHandler] for + * a [org.springframework.integration.transformer.HeaderEnricher] + * as the result of provided consumer. + */ + fun enrichHeaders(headerEnricherConfigurer: (HeaderEnricherSpec) -> Unit) { + this.delegate.enrichHeaders(headerEnricherConfigurer) + } + + /** + * Populate the [DefaultMessageSplitter] with provided options + * to the current integration flow position. + */ + fun split() { + this.delegate.split() + } + + /** + * Populate the [ExpressionEvaluatingSplitter] with provided + * SpEL expression. + */ + fun split(expression: String, + endpointConfigurer: (SplitterEndpointSpec) -> Unit = {}) { + + this.delegate.split(expression, endpointConfigurer) + } + + /** + * Populate the [MethodInvokingSplitter] to evaluate the provided + * `method` of the `service` at runtime. + */ + fun split(service: Any, methodName: String? = null) { + this.delegate.split(service, methodName) + } + + /** + * Populate the [MethodInvokingSplitter] to evaluate the provided + * `method` of the `bean` at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun split(service: Any, methodName: String?, + endpointConfigurer: (SplitterEndpointSpec) -> Unit) { + + this.delegate.split(service, methodName, endpointConfigurer) + } + + /** + * Populate the [MethodInvokingSplitter] to evaluate the provided + * `method` of the `bean` at runtime. + */ + fun split(beanName: String, methodName: String? = null) { + this.delegate.split(beanName, methodName) + } + + /** + * Populate the [MethodInvokingSplitter] to evaluate the provided + * `method` of the `bean` at runtime. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun split(beanName: String, methodName: String?, + endpointConfigurer: (SplitterEndpointSpec) -> Unit) { + + this.delegate.split(beanName, methodName, endpointConfigurer) + } + + /** + * Populate the [MethodInvokingSplitter] to evaluate the + * [MessageProcessor] at runtime + * from provided [MessageProcessorSpec]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun split(messageProcessorSpec: MessageProcessorSpec<*>, + endpointConfigurer: (SplitterEndpointSpec) -> Unit = {}) { + + this.delegate.split(messageProcessorSpec, endpointConfigurer) + } + + /** + * Populate the provided [AbstractMessageSplitter] to the current integration + * flow position. + */ + fun split(splitterMessageHandlerSpec: MessageHandlerSpec<*, S>, + endpointConfigurer: (SplitterEndpointSpec) -> Unit = {}) { + + this.delegate.split(splitterMessageHandlerSpec, endpointConfigurer) + } + + /** + * Populate the provided [AbstractMessageSplitter] to the current integration + * flow position. + */ + fun split(splitter: S, + endpointConfigurer: (SplitterEndpointSpec) -> Unit = {}) { + + this.delegate.split(splitter, endpointConfigurer) + } + + /** + * Provide the [HeaderFilter] to the current [StandardIntegrationFlow]. + */ + fun headerFilter(headersToRemove: String, patternMatch: Boolean = true) { + this.delegate.headerFilter(headersToRemove, patternMatch) + } + + /** + * Populate the provided [MessageTransformingHandler] for the provided + * [HeaderFilter]. + */ + fun headerFilter(headerFilter: HeaderFilter, + endpointConfigurer: (GenericEndpointSpec) -> Unit) { + + this.delegate.headerFilter(headerFilter, endpointConfigurer) + } + + /** + * Populate the [MessageTransformingHandler] for the [ClaimCheckInTransformer] + * with provided [MessageStore]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun claimCheckIn(messageStore: MessageStore, + endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + + this.delegate.claimCheckIn(messageStore, endpointConfigurer) + } + + /** + * Populate the [MessageTransformingHandler] for the [ClaimCheckOutTransformer] + * with provided [MessageStore] and `removeMessage` flag. + */ + fun claimCheckOut(messageStore: MessageStore, removeMessage: Boolean = false) { + this.delegate.claimCheckOut(messageStore, removeMessage) + } + + /** + * Populate the [MessageTransformingHandler] for the [ClaimCheckOutTransformer] + * with provided [MessageStore] and `removeMessage` flag. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun claimCheckOut(messageStore: MessageStore, removeMessage: Boolean, + endpointConfigurer: (GenericEndpointSpec) -> Unit) { + + this.delegate.claimCheckOut(messageStore, removeMessage, endpointConfigurer) + } + + /** + * Populate the + * [org.springframework.integration.aggregator.ResequencingMessageHandler] with + * provided options from [ResequencerSpec]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun resequence(resequencer: (ResequencerSpec) -> Unit = {}) { + this.delegate.resequence(resequencer) + } + + /** + * Populate the [AggregatingMessageHandler] with provided options from [AggregatorSpec]. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun aggregate(aggregator: (AggregatorSpec) -> Unit = {}) { + this.delegate.aggregate(aggregator) + } + + /** + * Populate the [MethodInvokingRouter] for provided bean and its method + * with default options. + */ + fun route(beanName: String, method: String? = null) { + this.delegate.route(beanName, method) + } + + /** + * Populate the [MethodInvokingRouter] for provided bean and its method + * with provided options from [RouterSpec]. + */ + fun route(beanName: String, method: String?, routerConfigurer: (RouterSpec) -> Unit) { + this.delegate.route(beanName, method, routerConfigurer) + } + + /** + * Populate the [MethodInvokingRouter] for the method + * of the provided service and its method with default options. + */ + fun route(service: Any, methodName: String? = null) { + this.delegate.route(service, methodName) + } + + /** + * Populate the [MethodInvokingRouter] for the method + * of the provided service and its method with provided options from [RouterSpec]. + */ + fun route(service: Any, methodName: String?, routerConfigurer: (RouterSpec) -> Unit) { + this.delegate.route(service, methodName, routerConfigurer) + } + + /** + * Populate the [ExpressionEvaluatingRouter] for provided SpEL expression + * with provided options from [RouterSpec]. + */ + fun route(expression: String, routerConfigurer: (RouterSpec) -> Unit = {}) { + this.delegate.route(expression, routerConfigurer) + } + + /** + * Populate the [MethodInvokingRouter] for the + * [MessageProcessor] + * from the provided [MessageProcessorSpec] with default options. + */ + fun route(messageProcessorSpec: MessageProcessorSpec<*>, + routerConfigurer: (RouterSpec) -> Unit = {}) { + + this.delegate.route(messageProcessorSpec, routerConfigurer) + } + + /** + * Populate the [RecipientListRouter] with options from the [RecipientListRouterSpec]. + */ + fun routeToRecipients(routerConfigurer: (RecipientListRouterSpec) -> Unit) { + this.delegate.routeToRecipients(routerConfigurer) + } + + /** + * Populate the [ErrorMessageExceptionTypeRouter] with options from the [RouterSpec]. + */ + fun routeByException( + routerConfigurer: (RouterSpec, ErrorMessageExceptionTypeRouter>) -> Unit) { + + this.delegate.routeByException(routerConfigurer) + } + + /** + * Populate the provided [AbstractMessageRouter] implementation to the + * current integration flow position. + * In addition accept options for the integration endpoint using [GenericEndpointSpec]. + */ + fun route(router: R, endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + this.delegate.route(router, endpointConfigurer) + } + + /** + * Populate the "artificial" + * [org.springframework.integration.gateway.GatewayMessageHandler] for the + * provided `requestChannel` to send a request with options from + * [GatewayEndpointSpec]. Uses + * [org.springframework.integration.gateway.RequestReplyExchanger] Proxy on the + * background. + */ + fun gateway(requestChannel: String, endpointConfigurer: (GatewayEndpointSpec) -> Unit = {}) { + this.delegate.gateway(requestChannel, endpointConfigurer) + } + + /** + * Populate the "artificial" + * [org.springframework.integration.gateway.GatewayMessageHandler] for the + * provided `requestChannel` to send a request with options from + * [GatewayEndpointSpec]. Uses + * [org.springframework.integration.gateway.RequestReplyExchanger] Proxy on the + * background. + */ + fun gateway(requestChannel: MessageChannel, endpointConfigurer: (GatewayEndpointSpec) -> Unit = {}) { + this.delegate.gateway(requestChannel, Consumer(endpointConfigurer)) + } + + /** + * Populate the "artificial" + * [org.springframework.integration.gateway.GatewayMessageHandler] for the + * provided `subflow` with options from [GatewayEndpointSpec]. + */ + fun gateway(flow: KotlinIntegrationFlowDefinition.() -> Unit) { + + this.delegate.gateway(IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) }) + } + + /** + * Populate the "artificial" + * [org.springframework.integration.gateway.GatewayMessageHandler] for the + * provided `subflow` with options from [GatewayEndpointSpec]. + */ + fun gateway(flow: KotlinIntegrationFlowDefinition.() -> Unit, + endpointConfigurer: (GatewayEndpointSpec) -> Unit) { + + this.delegate.gateway( + IntegrationFlow { flow(KotlinIntegrationFlowDefinition(it)) }, + Consumer(endpointConfigurer)) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the `INFO` + * logging level and `org.springframework.integration.handler.LoggingHandler` + * as a default logging category. + */ + fun log() { + this.delegate.log() + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for provided [LoggingHandler.Level] + * logging level and `org.springframework.integration.handler.LoggingHandler` + * as a default logging category. + */ + fun log(level: LoggingHandler.Level, category: String? = null) { + this.delegate.log(level, category) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided logging category + * and `INFO` logging level. + */ + fun log(category: String) { + this.delegate.log(category) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, logging category + * and SpEL expression for the log message. + */ + fun log(level: LoggingHandler.Level, category: String, logExpression: String) { + this.delegate.log(level, category, logExpression) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the `INFO` logging level, + * the `org.springframework.integration.handler.LoggingHandler` + * as a default logging category and function for the log message. + */ + fun

log(function: (Message

) -> Any) { + this.delegate.log(function) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the `INFO` logging level, + * the `org.springframework.integration.handler.LoggingHandler` + * as a default logging category and SpEL expression to evaluate + * logger message at runtime against the request [Message]. + */ + fun log(logExpression: Expression) { + this.delegate.log(logExpression) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, + * the `org.springframework.integration.handler.LoggingHandler` + * as a default logging category and SpEL expression to evaluate + * logger message at runtime against the request [Message]. + * When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + */ + fun log(level: LoggingHandler.Level, logExpression: Expression) { + this.delegate.log(level, logExpression) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the `INFO` + * [LoggingHandler.Level] logging level, + * the provided logging category and SpEL expression to evaluate + * logger message at runtime against the request [Message]. + */ + fun log(category: String, logExpression: Expression) { + this.delegate.log(category, logExpression) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, + * the `org.springframework.integration.handler.LoggingHandler` + * as a default logging category and function for the log message. + */ + fun

log(level: LoggingHandler.Level, function: (Message

) -> Any) { + this.delegate.log(level, function) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, + * the provided logging category and function for the log message. + */ + fun

log(category: String, function: (Message

) -> Any) { + this.delegate.log(category, function) + } + + /** + * Populate a [WireTap] for the current channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, logging category + * and function for the log message. + */ + fun

log(level: LoggingHandler.Level, category: String, function: (Message

) -> Any) { + this.delegate.log(level, category, function) + } + + /** + * Populate a [WireTap] for thecurrent channel + * with the [LoggingHandler] subscriber for the provided + * [LoggingHandler.Level] logging level, logging category + * and SpEL expression for the log message. + */ + fun log(level: LoggingHandler.Level, category: String, logExpression: Expression) { + this.delegate.log(level, category, logExpression) + } + + /** + * Populate a [ScatterGatherHandler] to the current integration flow position + * based on the provided [MessageChannel] for scattering function + * and [AggregatorSpec] for gathering function. + */ + fun scatterGather(scatterChannel: MessageChannel, gatherer: (AggregatorSpec) -> Unit = {}) { + this.delegate.scatterGather(scatterChannel, Consumer(gatherer)) + } + + /** + * Populate a [ScatterGatherHandler] to the current integration flow position + * based on the provided [MessageChannel] for scattering function + * and [AggregatorSpec] for gathering function. + */ + fun scatterGather(scatterChannel: MessageChannel, gatherer: (AggregatorSpec) -> Unit, + scatterGather: (ScatterGatherSpec) -> Unit) { + + this.delegate.scatterGather(scatterChannel, Consumer(gatherer), Consumer(scatterGather)) + } + + /** + * Populate a [ScatterGatherHandler] to the current integration flow position + * based on the provided [RecipientListRouterSpec] for scattering function + * and default [AggregatorSpec] for gathering function. + */ + fun scatterGather(scatterer: (RecipientListRouterSpec) -> Unit) { + this.delegate.scatterGather(scatterer) + } + + /** + * Populate a [ScatterGatherHandler] to the current integration flow position + * based on the provided [RecipientListRouterSpec] for scattering function + * and [AggregatorSpec] for gathering function. + */ + fun scatterGather(scatterer: (RecipientListRouterSpec) -> Unit, gatherer: (AggregatorSpec) -> Unit) { + this.delegate.scatterGather(scatterer, gatherer) + } + + /** + * Populate a [ScatterGatherHandler] to the current integration flow position + * based on the provided [RecipientListRouterSpec] for scattering function + * and [AggregatorSpec] for gathering function. + * @param scatterer the consumer for [RecipientListRouterSpec] to configure scatterer. + * @param gatherer the consumer for [AggregatorSpec] to configure gatherer. + * @param scatterGather the consumer for [ScatterGatherSpec] to configure + * [ScatterGatherHandler] and its endpoint. Can be `null`. + * @return the current [BaseIntegrationFlowDefinition]. + */ + fun scatterGather(scatterer: (RecipientListRouterSpec) -> Unit, gatherer: (AggregatorSpec) -> Unit, + scatterGather: (ScatterGatherSpec) -> Unit) { + + this.delegate.scatterGather(scatterer, gatherer, scatterGather) + } + + /** + * Populate a [org.springframework.integration.aggregator.BarrierMessageHandler] + * instance for provided timeout and options from [BarrierSpec] and endpoint + * options from [GenericEndpointSpec]. + */ + fun barrier(timeout: Long, barrierConfigurer: (BarrierSpec) -> Unit = {}) { + this.delegate.barrier(timeout, barrierConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] instance to perform [MessageTriggerAction] + * and endpoint options from [GenericEndpointSpec]. + */ + fun trigger(triggerActionId: String, + endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + + this.delegate.trigger(triggerActionId, endpointConfigurer) + } + + /** + * Populate a [ServiceActivatingHandler] instance to perform [MessageTriggerAction] + * and endpoint options from [GenericEndpointSpec]. + */ + fun trigger(triggerAction: MessageTriggerAction, + endpointConfigurer: (GenericEndpointSpec) -> Unit = {}) { + + this.delegate.trigger(triggerAction, Consumer(endpointConfigurer)) + } + + /** + * Populate a [FluxMessageChannel] to start a reactive processing for upstream data, + * wrap it to a [Flux], apply provided function via [Flux.transform] + * and emit the result to one more [FluxMessageChannel], subscribed in the downstream flow. + */ + fun fluxTransform(fluxFunction: (Flux>) -> Publisher) { + this.delegate.fluxTransform(fluxFunction) } } diff --git a/spring-integration-kotlin-dsl/src/test/kotlin/org/springframework/integration/dsl/kotlin/test/KotlinDslTests.kt b/spring-integration-kotlin-dsl/src/test/kotlin/org/springframework/integration/dsl/kotlin/test/KotlinDslTests.kt index 35fb618..980ada0 100644 --- a/spring-integration-kotlin-dsl/src/test/kotlin/org/springframework/integration/dsl/kotlin/test/KotlinDslTests.kt +++ b/spring-integration-kotlin-dsl/src/test/kotlin/org/springframework/integration/dsl/kotlin/test/KotlinDslTests.kt @@ -207,8 +207,10 @@ class KotlinDslTests { fun functionFlow() = integrationFlow>({ it.beanName("functionGateway") }) { transform { it.toUpperCase() } - split { p -> p } - split({ p -> p }) { it.id("splitterEndpoint") } + split> { it.payload } + split({ it }) { it.id("splitterEndpoint") } + resequence() + aggregate { it.id("aggregator").outputProcessor { it.one } } } @Bean @@ -223,43 +225,44 @@ class KotlinDslTests { fun messageSourceFlow() = integrationFlow(MessageProcessorMessageSource { "testSource" }, { it.poller { it.trigger(OnlyOnceTrigger()) } }) { - channel { c -> c.queue("fromSupplierQueue") } + channel { it.queue("fromSupplierQueue") } } @Bean fun messageSourceFlow2() = integrationFlow(MessageProcessorMessageSource { "testSource2" }) { - channel { c -> c.queue("fromSupplierQueue2") } + channel { it.queue("fromSupplierQueue2") } } @Bean fun fixedSubscriberFlow() = integrationFlow("fixedSubscriberInput", true) { - logAndReply(LoggingHandler.Level.WARN) + log(LoggingHandler.Level.WARN) { it.payload } + transform("payload") { it.id("spelTransformer") } } @Bean fun flowFromSupplier() = integrationFlow({ "testSupplier" }) { - channel { c -> c.queue("testSupplierResult") } + channel { it.queue("testSupplierResult") } } @Bean fun flowFromSupplier2() = integrationFlow({ "testSupplier2" }, { it.poller { it.trigger(OnlyOnceTrigger()) } }) { - filter> { m -> m.payload is String } - channel { c -> c.queue("testSupplierResult2") } + filter> { m -> m.payload is String } + channel { it.queue("testSupplierResult2") } } @Bean fun flowLambda() = integrationFlow { filter({ it === "test" }) { it.id("filterEndpoint") } - wireTap( - integrationFlow { - channel { c -> c.queue("wireTapChannel") } - }) + wireTap { + channel { it.queue("wireTapChannel") } + } + delay("delayGroup") { it.defaultDelay(100) } transform { it.toUpperCase() } }