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

Improve functions support

Browse Source 
* Add `functions-support.adoc` chapter
* Add more tests
* Improve `InboundChannelAdapterAnnotationPostProcessor` to support
Kotlin `Function0`
* Add `FunctionsTests.kt`
* Reformat Kotlin classes to use tabs
* Upgrade to Kotlin `1.2.71`

* Add `What's New` bullet

Doc Polishing
This commit is contained in:
Artem Bilan
2018-10-04 13:18:44 -04:00
committed by Gary Russell
parent e7bc060e55
commit 7176690e4d
12 changed files with 524 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -1,5 +1,5 @@
buildscript {
ext.kotlinVersion = '1.2.61'
ext.kotlinVersion = '1.2.71'
repositories {
maven { url 'https://repo.spring.io/plugins-release' }
}

40
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2016 the original author or authors.
* Copyright 2014-2018 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.
@@ -34,6 +34,7 @@ import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.util.MessagingAnnotationUtils;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
/**
@@ -42,11 +43,27 @@ import org.springframework.util.ReflectionUtils;
* @author Artem Bilan
* @author Gary Russell
* @author Oleg Zhurakousky
*
* @since 4.0
*/
public class InboundChannelAdapterAnnotationPostProcessor extends
AbstractMethodAnnotationPostProcessor<InboundChannelAdapter> {
private static final Class<?> kotlinFunction0Class;
static {
Class<?> kotlinClass = null;
try {
kotlinClass = ClassUtils.forName("kotlin.jvm.functions.Function0", ClassUtils.getDefaultClassLoader());
}
catch (ClassNotFoundException e) {
//Ignore: assume no Kotlin in classpath
}
finally {
kotlinFunction0Class = kotlinClass;
}
}
public InboundChannelAdapterAnnotationPostProcessor(ConfigurableListableBeanFactory beanFactory) {
super(beanFactory);
}
@@ -58,7 +75,8 @@ public class InboundChannelAdapterAnnotationPostProcessor extends
@Override
public Object postProcess(Object bean, String beanName, Method method, List<Annotation> annotations) {
String channelName = MessagingAnnotationUtils.resolveAttribute(annotations, AnnotationUtils.VALUE, String.class);
String channelName = MessagingAnnotationUtils
.resolveAttribute(annotations, AnnotationUtils.VALUE, String.class);
Assert.hasText(channelName, "The channel ('value' attribute of @InboundChannelAdapter) can't be empty.");
MessageSource<?> messageSource = null;
@@ -86,15 +104,24 @@ public class InboundChannelAdapterAnnotationPostProcessor extends
MessageSource<?> messageSource = null;
if (AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) {
Object target = this.resolveTargetBeanFromMethodWithBeanAnnotation(method);
Assert.isTrue(target instanceof MessageSource || target instanceof Supplier, "The '" + this.annotationType + "' on @Bean method " +
"level is allowed only for: " + MessageSource.class.getName() + " or " + Supplier.class.getName() + " beans");
Class<?> targetClass = target.getClass();
Assert.isTrue(MessageSource.class.isAssignableFrom(targetClass) ||
Supplier.class.isAssignableFrom(targetClass) ||
(kotlinFunction0Class == null || kotlinFunction0Class.isAssignableFrom(targetClass)),
"The '" + this.annotationType + "' on @Bean method " + "level is allowed only for: "
+ MessageSource.class.getName() + " or " + Supplier.class.getName()
+ (kotlinFunction0Class != null ? " or " + kotlinFunction0Class.getName() : "") + " beans");
if (target instanceof MessageSource<?>) {
messageSource = (MessageSource<?>) target;
}
else {
else if (target instanceof Supplier<?>) {
method = ReflectionUtils.findMethod(Supplier.class, "get");
bean = target;
}
else if (kotlinFunction0Class != null) {
method = ReflectionUtils.findMethod(kotlinFunction0Class, "invoke");
bean = target;
}
}
if (messageSource == null) {
MethodInvokingMessageSource methodInvokingMessageSource = new MethodInvokingMessageSource();
@@ -102,7 +129,8 @@ public class InboundChannelAdapterAnnotationPostProcessor extends
methodInvokingMessageSource.setMethod(method);
String messageSourceBeanName = this.generateHandlerBeanName(beanName, method);
this.beanFactory.registerSingleton(messageSourceBeanName, methodInvokingMessageSource);
messageSource = (MessageSource<?>) this.beanFactory.initializeBean(methodInvokingMessageSource, messageSourceBeanName);
messageSource = (MessageSource<?>) this.beanFactory
.initializeBean(methodInvokingMessageSource, messageSourceBeanName);
}
return messageSource;
}

15
spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java
View File

@@ -34,6 +34,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
import org.aopalliance.aop.Advice;
import org.aopalliance.intercept.MethodInterceptor;
@@ -526,10 +527,20 @@ public class IntegrationFlowTests {
@EnableIntegration
public static class SupplierContextConfiguration1 {
@Bean
public Function<String, String> toUpperCaseFunction() {
return String::toUpperCase;
}
@Bean
public Supplier<String> stringSupplier() {
return () -> "foo";
}
@Bean
public IntegrationFlow supplierFlow() {
return IntegrationFlows.from(() -> "foo")
.<String, String>transform(String::toUpperCase)
return IntegrationFlows.from(stringSupplier())
.transform(toUpperCaseFunction())
.channel("suppliedChannel")
.get();
}

16
spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml
View File

@@ -1,11 +1,17 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:annotation-config/>
<beans:bean class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$FunctionConfiguration"/>
<message-history/>
@@ -76,4 +82,6 @@
<queue />
</channel>
<service-activator input-channel="processorViaFunctionChannel" ref="functionAsService"/>
</beans:beans>

31
spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -31,6 +31,7 @@ import static org.junit.Assert.fail;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeaderKey;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import org.hamcrest.Matchers;
import org.junit.Test;
@@ -38,6 +39,7 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -107,6 +109,9 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
@Autowired
private PollableChannel errorChannel;
@Autowired
private MessageChannel processorViaFunctionChannel;
@Test
public void testGateway() {
QueueChannel replyChannel = new QueueChannel();
@@ -280,6 +285,16 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
assertEquals("test", ((MessagingException) error.getPayload()).getFailedMessage().getPayload());
}
@Test
public void testFunctionFromXml() {
QueueChannel replyChannel = new QueueChannel();
Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build();
this.processorViaFunctionChannel.send(message);
Message<?> reply = replyChannel.receive(0);
assertNotNull(reply);
assertEquals("TEST", reply.getPayload());
}
public static void throwIllegalStateException(String message) {
throw new IllegalStateException(message);
}
@@ -316,6 +331,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel",
"MethodInvokerHelper", st)); // close to the metal
}
}
private static class TestMessageProcessor implements MessageProcessor<String> {
@@ -331,6 +347,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
public String processMessage(Message<?> message) {
return prefix + ":" + message.getPayload();
}
}
private static class AsyncService {
@@ -341,11 +358,21 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
@SuppressWarnings("unused")
public ListenableFuture<String> process(String payload) {
this.future = new SettableListenableFuture<String>();
this.future = new SettableListenableFuture<>();
this.payload = payload;
return this.future;
}
}
public static class FunctionConfiguration {
@Bean
public Function<String, String> functionAsService() {
return String::toUpperCase;
}
}
}

150
spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt
View File

@@ -46,100 +46,100 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
@DirtiesContext
class RouterDslTests {
@Autowired
@Qualifier("routerTwoSubFlows.input")
private lateinit var routerTwoSubFlowsInput: MessageChannel
@Autowired
@Qualifier("routerTwoSubFlows.input")
private lateinit var routerTwoSubFlowsInput: MessageChannel
@Autowired
@Qualifier("routerTwoSubFlowsOutput")
private lateinit var routerTwoSubFlowsOutput: PollableChannel
@Autowired
@Qualifier("routerTwoSubFlowsOutput")
private lateinit var routerTwoSubFlowsOutput: PollableChannel
@Test
fun `route to two subflows using lambda`() {
@Test
fun `route to two subflows using lambda`() {
this.routerTwoSubFlowsInput.send(GenericMessage<Any>(listOf(1, 2, 3, 4, 5, 6)))
val receive = this.routerTwoSubFlowsOutput.receive(10000)
this.routerTwoSubFlowsInput.send(GenericMessage<Any>(listOf(1, 2, 3, 4, 5, 6)))
val receive = this.routerTwoSubFlowsOutput.receive(10000)
val payload = receive?.payload
val payload = receive?.payload
assert(payload).isNotNull {
it.isInstanceOf(List::class.java)
}
assert(payload).isNotNull {
it.isInstanceOf(List::class.java)
}
assert(payload).isEqualTo(listOf(3, 4, 9, 8, 15, 12))
}
assert(payload).isEqualTo(listOf(3, 4, 9, 8, 15, 12))
}
@Autowired
@Qualifier("splitRouteAggregate.input")
private lateinit var splitRouteAggregateInput: MessageChannel
@Autowired
@Qualifier("splitRouteAggregate.input")
private lateinit var splitRouteAggregateInput: MessageChannel
@Test
fun `route to two subflows using them as bean references`() {
@Test
fun `route to two subflows using them as bean references`() {
val replyChannel = QueueChannel()
val message = MessageBuilder.withPayload(arrayOf(1, 2, 3))
.setReplyChannel(replyChannel)
.build()
val replyChannel = QueueChannel()
val message = MessageBuilder.withPayload(arrayOf(1, 2, 3))
.setReplyChannel(replyChannel)
.build()
this.splitRouteAggregateInput.send(message)
this.splitRouteAggregateInput.send(message)
val receive = replyChannel.receive(10000)
val receive = replyChannel.receive(10000)
val payload = receive?.payload
val payload = receive?.payload
assert(payload).isNotNull {
it.isInstanceOf(List::class.java)
it.isEqualTo(listOf("even", "odd", "even"))
}
}
assert(payload).isNotNull {
it.isInstanceOf(List::class.java)
it.isEqualTo(listOf("even", "odd", "even"))
}
}
@Configuration
@EnableIntegration
class Config {
@Configuration
@EnableIntegration
class Config {
@Bean
fun routerTwoSubFlows() =
IntegrationFlow { f ->
f.split()
.route<Int, Boolean>({ p -> p % 2 == 0 },
{ m ->
m.subFlowMapping(true, { sf -> sf.handle<Int> { p, _ -> p * 2 } })
.subFlowMapping(false, { sf -> sf.handle<Int> { p, _ -> p * 3 } })
})
.aggregate()
.channel { c -> c.queue("routerTwoSubFlowsOutput") }
}
@Bean
fun routerTwoSubFlows() =
IntegrationFlow { f ->
f.split()
.route<Int, Boolean>({ p -> p % 2 == 0 },
{ m ->
m.subFlowMapping(true, { sf -> sf.handle<Int> { p, _ -> p * 2 } })
.subFlowMapping(false, { sf -> sf.handle<Int> { p, _ -> p * 3 } })
})
.aggregate()
.channel { c -> c.queue("routerTwoSubFlowsOutput") }
}
@Bean
fun splitRouteAggregate() =
IntegrationFlow { f ->
f.split()
.route<Int, Boolean>({ o -> o % 2 == 0 },
{ m ->
m.subFlowMapping(true) { sf -> sf.gateway(oddFlow()) }
.subFlowMapping(false) { sf -> sf.gateway(evenFlow()) }
})
.aggregate()
}
@Bean
fun splitRouteAggregate() =
IntegrationFlow { f ->
f.split()
.route<Int, Boolean>({ o -> o % 2 == 0 },
{ m ->
m.subFlowMapping(true) { sf -> sf.gateway(oddFlow()) }
.subFlowMapping(false) { sf -> sf.gateway(evenFlow()) }
})
.aggregate()
}
@Bean
fun oddFlow() =
IntegrationFlow { flow ->
flow.handle<Any> { _, _ -> "odd" }
}
@Bean
fun oddFlow() =
IntegrationFlow { flow ->
flow.handle<Any> { _, _ -> "odd" }
}
@Bean
fun evenFlow() =
IntegrationFlow { flow ->
flow.handle<Any> { _, _ -> "even" }
}
@Bean
fun evenFlow() =
IntegrationFlow { flow ->
flow.handle<Any> { _, _ -> "even" }
}
@Bean
fun publishSubscribe() =
MessageChannels.publishSubscribe()
.ignoreFailures(true)
.applySequence(false)
}
@Bean
fun publishSubscribe() =
MessageChannels.publishSubscribe()
.ignoreFailures(true)
.applySequence(false)
}
}

144
spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt Normal file
View File

@@ -0,0 +1,144 @@
/*
* Copyright 2018 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
*
* http://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.function
import assertk.assert
import assertk.assertions.isEqualTo
import assertk.assertions.isNotNull
import assertk.assertions.isTrue
import assertk.assertions.size
import org.junit.jupiter.api.Test
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.integration.annotation.InboundChannelAdapter
import org.springframework.integration.annotation.Poller
import org.springframework.integration.annotation.ServiceActivator
import org.springframework.integration.annotation.Transformer
import org.springframework.integration.channel.DirectChannel
import org.springframework.integration.channel.QueueChannel
import org.springframework.integration.config.EnableIntegration
import org.springframework.integration.dsl.IntegrationFlows
import org.springframework.integration.endpoint.SourcePollingChannelAdapter
import org.springframework.messaging.Message
import org.springframework.messaging.MessageChannel
import org.springframework.messaging.SubscribableChannel
import org.springframework.messaging.support.GenericMessage
import org.springframework.messaging.support.MessageBuilder
import org.springframework.test.annotation.DirtiesContext
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
import java.util.*
import java.util.concurrent.CountDownLatch
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.function.Supplier
/**
* @author Artem Bilan
*
* @since 5.1
*/
@SpringJUnitConfig
@DirtiesContext
class FunctionsTests {
@Autowired
private lateinit var functionServiceChannel: MessageChannel
@Autowired
private lateinit var messageConsumerServiceChannel: MessageChannel
@Autowired
private lateinit var messageCollector: ArrayList<Message<Any>>
@Autowired
private lateinit var counterChannel: SubscribableChannel
@Autowired
private lateinit var kotlinSupplierInboundChannelAdapter: SourcePollingChannelAdapter
@Test
fun `invoke function via transformer`() {
val replyChannel = QueueChannel()
val message = MessageBuilder.withPayload("foo")
.setReplyChannel(replyChannel)
.build()
this.functionServiceChannel.send(message)
val receive = replyChannel.receive(10000)
val payload = receive?.payload
assertk.assert(payload).isNotNull {
it.isEqualTo("FOO")
}
}
@Test
fun `invoke consumer via service activator`() {
this.messageConsumerServiceChannel.send(GenericMessage("bar"))
assert(this.messageCollector).size().isEqualTo(1)
val message = this.messageCollector[0]
assert(message.payload).isEqualTo("bar")
}
@Test
fun `verify supplier`() {
val countDownLatch = CountDownLatch(10)
this.counterChannel.subscribe { countDownLatch.countDown() }
this.kotlinSupplierInboundChannelAdapter.start()
assert(countDownLatch.await(10, TimeUnit.SECONDS)).isTrue()
}
@Configuration
@EnableIntegration
class Config {
@Bean
@Transformer(inputChannel = "functionServiceChannel")
fun kotlinFunction(): (String) -> String {
return { it.toUpperCase() }
}
@Bean
fun messageCollector() = ArrayList<Message<Any>>()
@Bean
@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
fun kotlinConsumer(): (Message<Any>) -> Unit {
return { messageCollector().add(it) }
}
@Bean
fun counterChannel() = DirectChannel()
@Bean
@InboundChannelAdapter(value = "counterChannel", autoStartup = "false",
poller = [Poller(fixedRate = "10", maxMessagesPerPoll = "1")])
fun kotlinSupplier(): () -> String {
return { "baz" }
}
}
}

124
spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt
View File

@@ -52,83 +52,83 @@ import javax.jms.DeliveryMode
@DirtiesContext
class JmsDslKotlinTests {
@Autowired
@Qualifier("jmsOutboundFlow.input")
private lateinit var jmsOutboundInboundChannel: MessageChannel
@Autowired
@Qualifier("jmsOutboundFlow.input")
private lateinit var jmsOutboundInboundChannel: MessageChannel
@Autowired
private lateinit var jmsOutboundInboundReplyChannel: PollableChannel
@Autowired
private lateinit var jmsOutboundInboundReplyChannel: PollableChannel
@Test
fun `test JMS Channel Adapters DSL`() {
@Test
fun `test JMS Channel Adapters DSL`() {
this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload(" foo ")
.setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "containerSpecDestination")
.setPriority(9)
.build())
this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload(" foo ")
.setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "containerSpecDestination")
.setPriority(9)
.build())
val receive = this.jmsOutboundInboundReplyChannel.receive(10000)
val receive = this.jmsOutboundInboundReplyChannel.receive(10000)
val payload = receive?.payload
val payload = receive?.payload
assert(payload).isNotNull {
it.isEqualTo("foo")
}
assert(payload).isNotNull {
it.isEqualTo("foo")
}
assert(receive?.headers).isNotNull {
it.contains(IntegrationMessageHeaderAccessor.PRIORITY, 9)
it.contains(JmsHeaders.DELIVERY_MODE, 1)
}
assert(receive?.headers).isNotNull {
it.contains(IntegrationMessageHeaderAccessor.PRIORITY, 9)
it.contains(JmsHeaders.DELIVERY_MODE, 1)
}
val expiration = receive!!.headers[JmsHeaders.EXPIRATION] as Long
assert(expiration).isGreaterThan(System.currentTimeMillis())
}
val expiration = receive!!.headers[JmsHeaders.EXPIRATION] as Long
assert(expiration).isGreaterThan(System.currentTimeMillis())
}
@Configuration
@EnableIntegration
class Config {
@Configuration
@EnableIntegration
class Config {
@Bean
fun jmsConnectionFactory(): ActiveMQConnectionFactory {
val activeMQConnectionFactory = ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")
activeMQConnectionFactory.isTrustAllPackages = true
return activeMQConnectionFactory
}
@Bean
fun jmsConnectionFactory(): ActiveMQConnectionFactory {
val activeMQConnectionFactory = ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")
activeMQConnectionFactory.isTrustAllPackages = true
return activeMQConnectionFactory
}
@Bean
fun jmsOutboundFlow() =
IntegrationFlow { f ->
f.handle(Jms.outboundAdapter(jmsConnectionFactory())
.destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)
.deliveryModeFunction<Any> { _ -> DeliveryMode.NON_PERSISTENT }
.timeToLiveExpression("10000")
.configureJmsTemplate { t -> t.explicitQosEnabled(true) })
}
@Bean
fun jmsOutboundFlow() =
IntegrationFlow { f ->
f.handle(Jms.outboundAdapter(jmsConnectionFactory())
.destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)
.deliveryModeFunction<Any> { _ -> DeliveryMode.NON_PERSISTENT }
.timeToLiveExpression("10000")
.configureJmsTemplate { t -> t.explicitQosEnabled(true) })
}
@Bean
fun jmsHeaderMapper(): DefaultJmsHeaderMapper {
val jmsHeaderMapper = DefaultJmsHeaderMapper()
jmsHeaderMapper.setMapInboundDeliveryMode(true)
jmsHeaderMapper.setMapInboundExpiration(true)
return jmsHeaderMapper
}
@Bean
fun jmsHeaderMapper(): DefaultJmsHeaderMapper {
val jmsHeaderMapper = DefaultJmsHeaderMapper()
jmsHeaderMapper.setMapInboundDeliveryMode(true)
jmsHeaderMapper.setMapInboundExpiration(true)
return jmsHeaderMapper
}
@Bean
fun jmsOutboundInboundReplyChannel() = MessageChannels.queue().get()
@Bean
fun jmsOutboundInboundReplyChannel() = MessageChannels.queue().get()
@Bean
fun jmsMessageDrivenFlowWithContainer() =
IntegrationFlows.from(
Jms.messageDrivenChannelAdapter(
Jms.container(jmsConnectionFactory(), "containerSpecDestination")
.pubSubDomain(false)
.taskExecutor(Executors.newCachedThreadPool()))
.headerMapper(jmsHeaderMapper()))
.transform({ it: String -> it.trim({ it <= ' ' }) })
.channel(jmsOutboundInboundReplyChannel())
.get()
@Bean
fun jmsMessageDrivenFlowWithContainer() =
IntegrationFlows.from(
Jms.messageDrivenChannelAdapter(
Jms.container(jmsConnectionFactory(), "containerSpecDestination")
.pubSubDomain(false)
.taskExecutor(Executors.newCachedThreadPool()))
.headerMapper(jmsHeaderMapper()))
.transform({ it: String -> it.trim({ it <= ' ' }) })
.channel(jmsOutboundInboundReplyChannel())
.get()
}
}
}

141
src/reference/asciidoc/functions-support.adoc Normal file
View File

@@ -0,0 +1,141 @@
[[functions-support]]
=== `java.util.function` Interfaces Support
Starting with version 5.1, Spring Integration provides direct support for interfaces in the `java.util.function` package.
All messaging endpoints, (Service Activator, Transformer, Filter, etc.) can now refer to `Function` (or `Consumer`) beans.
The <<annotations,Messaging Annotations>> can be applied directly on these beans similar to regular `MessageHandler` definitions.
For example if you have this `Function` bean definition:
====
[source, java]
----
@Configuration
public class FunctionConfiguration {
@Bean
public Function<String, String> functionAsService() {
return String::toUpperCase;
}
}
----
====
You can use it as a simple reference in an XML configuration file:
====
[source, xml]
----
<service-activator input-channel="processorViaFunctionChannel" ref="functionAsService"/>
----
====
When we configure our flow with Messaging Annotations, the code is straightforward:
====
[source, java]
----
@Bean
@Transformer(inputChannel = "functionServiceChannel")
public Function<String, String> functionAsService() {
return String::toUpperCase;
}
----
====
When the function returns an array, `Collection` (essentially, any `Iterable`), `Stream` or Reactor `Flux`, `@Splitter` can be used on such a bean to perform iteration over the result content.
The `java.util.function.Consumer` interface can be used for an `<int:outbound-channel-adapter>` or, together with the `@ServiceActivator` annotation, to perform the final step of a flow:
====
[source, java]
----
@Bean
@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
public Consumer<Message<?>> messageConsumerAsService() {
// Has to be an anonymous class for proper type inference
return new Consumer<Message<?>>() {
@Override
public void accept(Message<?> e) {
collector().add(e);
}
};
}
----
====
Also, pay attention to the comment in the code snippet above: if you would like to deal with the whole message in your `Function`/`Consumer` you cannot use a lambda definition.
Because of Java type erasure we cannot determine the target type for the `apply()/accept()` method call.
The `java.util.function.Supplier` interface can simply be used together with the `@InboundChannelAdapter` annotation, or as a `ref` in an `<int:inbound-channel-adapter>`:
====
[source, java]
----
@Bean
@InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000"))
public Supplier<String> pojoSupplier() {
return () -> "foo";
}
----
====
With the Java DSL we just need to use a reference to the function bean in the endpoint definitions.
Meanwhile an implementation of the `Supplier` interface can be used as regular `MessageSource` definition:
====
[source, java]
----
@Bean
public Function<String, String> toUpperCaseFunction() {
return String::toUpperCase;
}
@Bean
public Supplier<String> stringSupplier() {
return () -> "foo";
}
@Bean
public IntegrationFlow supplierFlow() {
return IntegrationFlows.from(stringSupplier())
.transform(toUpperCaseFunction())
.channel("suppliedChannel")
.get();
}
----
====
This function support is useful when used together with the https://cloud.spring.io/spring-cloud-function/[Spring Cloud Function] framework, where we have a function catalog and can refer to its member functions from an integration flow definition.
[[kotlin-functions-support]]
==== Kotlin Lambdas
The Framework also has been improved to support Kotlin lambdas for functions, so now you can get a gain from combination of Kotlin language and Spring Integration flow definitions:
====
[source, java]
----
@Bean
@Transformer(inputChannel = "functionServiceChannel")
fun kotlinFunction(): (String) -> String {
return { it.toUpperCase() }
}
@Bean
@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
fun kotlinConsumer(): (Message<Any>) -> Unit {
return { print(it) }
}
@Bean
@InboundChannelAdapter(value = "counterChannel",
poller = [Poller(fixedRate = "10", maxMessagesPerPoll = "1")])
fun kotlinSupplier(): () -> String {
return { "baz" }
}
----
====

2
src/reference/asciidoc/logging-adapter.adoc
View File

@@ -11,7 +11,7 @@ With a `NullChannel`, you would see only the discarded message when logging at t
The following listing shows all the possible attributes for the `logging-channel-adapter` element:
====
[source]
[source, xml]
----
<int:logging-channel-adapter

6
src/reference/asciidoc/messaging-endpoints.adoc
View File

@@ -1,7 +1,7 @@
[[messaging-endpoints-chapter]]
== Messaging Endpoints
// BE SURE TO PRECEDE ALL include:: with a blank line - see https://github.com/asciidoctor/asciidoctor/issues/1297
// BE SURE TO PRECEDE ALL include:: with a blank line - see https://asciidoctor.org/docs/user-manual/#include-partitioning
include::./endpoint.adoc[]
include::./gateway.adoc[]
@@ -17,4 +17,6 @@ include::./groovy.adoc[]
include::./handler-advice.adoc[]
include::./logging-adapter.adoc[]
// BE SURE TO PRECEDE ALL include:: with a blank line - see https://github.com/asciidoctor/asciidoctor/issues/1297
include::./functions-support.adoc[]
// BE SURE TO PRECEDE ALL include:: with a blank line - see https://asciidoctor.org/docs/user-manual/#include-partitioning

8
src/reference/asciidoc/whats-new.adoc
View File

@@ -18,6 +18,14 @@ The following components are new in 5.1:
See <<amqp-strict-ordering>>.
[[x5.1-Functions]]
==== Improved Function Support
The `java.util.function` interfaces now have improved integration support in the Framework components.
Also Kotlin lambdas now can be used for handler and source methods.
See <<functions-support>>.
[[x5.1-general]]
=== General Changes