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

GH-3288: Add Kotlin DSL transform(Transformer)

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3288

* For better end-user experience with Kotlin DSL and get a gain from
existing `Transformer` implementations add a `transform(Transformer)`
EI-method into the `KotlinIntegrationFlowDefinition`
* Also add `filter(MessageSelector)` for any out-of-the-box `MessageSelector`

**Cherry-pick to 5.3.x**
This commit is contained in:
Artem Bilan
2020-05-27 13:20:44 -04:00
committed by Gary Russell
parent 8d1dd0b68d
commit e98169146b
2 changed files with 38 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -50,6 +50,7 @@ import org.springframework.integration.transformer.ClaimCheckOutTransformer
import org.springframework.integration.transformer.HeaderFilter
import org.springframework.integration.transformer.MessageTransformingHandler
import org.springframework.integration.transformer.MethodInvokingTransformer
import org.springframework.integration.transformer.Transformer
import org.springframework.messaging.Message
import org.springframework.messaging.MessageChannel
import org.springframework.messaging.MessageHandler
@@ -287,8 +288,20 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
this.delegate.controlBus(endpointConfigurer)
}
/**
* Populate the `Transformer` EI Pattern specific [MessageHandler] implementation
* Populate the [Transformer] EI Pattern specific [MessageHandler] implementation
* for the provided `Transformer` instance.
* @since 5.3.1
*/
fun transform(transformer: Transformer,
endpointConfigurer: GenericEndpointSpec<MessageTransformingHandler>.() -> Unit = {}) {
this.delegate.transform(transformer, Consumer { endpointConfigurer(it) })
}
/**
* Populate the [Transformer] EI Pattern specific [MessageHandler] implementation
* for the SpEL [Expression].
*/
fun transform(expression: String,
@@ -298,7 +311,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
}
/**
* Populate the `MessageTransformingHandler` for the [MethodInvokingTransformer]
* Populate the [MessageTransformingHandler] for the [MethodInvokingTransformer]
* to invoke the service method at runtime.
*/
fun transform(service: Any, methodName: String? = null) {
@@ -306,7 +319,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
}
/**
* Populate the `MessageTransformingHandler` for the [MethodInvokingTransformer]
* Populate the [MessageTransformingHandler] for the [MethodInvokingTransformer]
* to invoke the service method at runtime.
*/
fun transform(service: Any, methodName: String?,
@@ -358,9 +371,23 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
*/
fun filter(messageProcessorSpec: MessageProcessorSpec<*>,
filterConfigurer: KotlinFilterEndpointSpec.() -> Unit = {}) {
this.delegate.filter(messageProcessorSpec) { filterConfigurer(KotlinFilterEndpointSpec(it)) }
}
/**
* Populate a [MessageFilter] with the provided [MessageSelector].
* In addition accept options for the integration endpoint using [KotlinFilterEndpointSpec].
* @since 5.3.1
*/
fun filter(messageSelector: MessageSelector,
filterConfigurer: KotlinFilterEndpointSpec.() -> Unit = {}) {
this.delegate.filter(Message::class.java, messageSelector,
Consumer { filterConfigurer(KotlinFilterEndpointSpec(it)) })
}
/**
* Populate a [ServiceActivatingHandler] for the selected protocol specific
* [MessageHandler] implementation from `Namespace Factory`:

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

@@ -21,6 +21,7 @@ import assertk.assertions.isEqualTo
import assertk.assertions.isGreaterThanOrEqualTo
import assertk.assertions.isInstanceOf
import assertk.assertions.isNotNull
import assertk.assertions.isTrue
import assertk.assertions.size
import org.junit.jupiter.api.Test
import org.springframework.beans.factory.BeanFactory
@@ -37,6 +38,7 @@ import org.springframework.integration.dsl.context.IntegrationFlowContext
import org.springframework.integration.endpoint.MessageProcessorMessageSource
import org.springframework.integration.handler.LoggingHandler
import org.springframework.integration.scheduling.PollerMetadata
import org.springframework.integration.selector.UnexpiredMessageSelector
import org.springframework.integration.support.MessageBuilder
import org.springframework.integration.test.util.OnlyOnceTrigger
import org.springframework.messaging.Message
@@ -88,11 +90,12 @@ class KotlinDslTests {
@Autowired
@Qualifier("functionGateway")
private lateinit var upperCaseFunction: Function<String, String>
private lateinit var upperCaseFunction: Function<ByteArray, String>
@Test
fun `uppercase function`() {
assertThat(this.upperCaseFunction.apply("test")).isEqualTo("TEST")
assertThat(beanFactory.containsBean("objectToStringTransformer")).isTrue()
assertThat(this.upperCaseFunction.apply("test".toByteArray())).isEqualTo("TEST")
}
@Autowired
@@ -225,7 +228,8 @@ class KotlinDslTests {
@Bean
fun functionFlow() =
integrationFlow<Function<String, String>>({ beanName("functionGateway") }) {
integrationFlow<Function<ByteArray, String>>({ beanName("functionGateway") }) {
transform(Transformers.objectToString()) { id("objectToStringTransformer") }
transform<String> { it.toUpperCase() }
split<Message<*>> { it.payload }
split<String>({ it }) { id("splitterEndpoint") }
@@ -240,6 +244,7 @@ class KotlinDslTests {
fun functionFlow2() =
integrationFlow<Function<*, *>> {
transform<String> { it.toLowerCase() }
filter(UnexpiredMessageSelector())
route<Message<*>, Any?>({ null }) { defaultOutputToParentFlow() }
route<Message<*>> { m -> m.headers.replyChannel }
}