diff --git a/build.gradle b/build.gradle index f7cb669d2e..c96f9f8a42 100644 --- a/build.gradle +++ b/build.gradle @@ -1,5 +1,5 @@ buildscript { - ext.kotlinVersion = '1.2.61' + ext.kotlinVersion = '1.2.71' repositories { maven { url 'https://repo.spring.io/plugins-release' } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java index 5727c5c182..11debb52cb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java @@ -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 { + 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 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; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java index c61db01064..5d3b9b1838 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java @@ -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 toUpperCaseFunction() { + return String::toUpperCase; + } + + @Bean + public Supplier stringSupplier() { + return () -> "foo"; + } + @Bean public IntegrationFlow supplierFlow() { - return IntegrationFlows.from(() -> "foo") - .transform(String::toUpperCase) + return IntegrationFlows.from(stringSupplier()) + .transform(toUpperCaseFunction()) .channel("suppliedChannel") .get(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml index 4005c18515..456103f080 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml @@ -1,11 +1,17 @@ + http://www.springframework.org/schema/integration/spring-integration.xsd + http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> + + + + @@ -76,4 +82,6 @@ + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java index 3269509dd0..981caca168 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java @@ -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 { @@ -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 process(String payload) { - this.future = new SettableListenableFuture(); + this.future = new SettableListenableFuture<>(); this.payload = payload; return this.future; } } + public static class FunctionConfiguration { + + @Bean + public Function functionAsService() { + return String::toUpperCase; + } + + } + + } diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt index f8d9c63f7e..9211c17841 100644 --- a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt +++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt @@ -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(listOf(1, 2, 3, 4, 5, 6))) - val receive = this.routerTwoSubFlowsOutput.receive(10000) + this.routerTwoSubFlowsInput.send(GenericMessage(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({ p -> p % 2 == 0 }, - { m -> - m.subFlowMapping(true, { sf -> sf.handle { p, _ -> p * 2 } }) - .subFlowMapping(false, { sf -> sf.handle { p, _ -> p * 3 } }) - }) - .aggregate() - .channel { c -> c.queue("routerTwoSubFlowsOutput") } - } + @Bean + fun routerTwoSubFlows() = + IntegrationFlow { f -> + f.split() + .route({ p -> p % 2 == 0 }, + { m -> + m.subFlowMapping(true, { sf -> sf.handle { p, _ -> p * 2 } }) + .subFlowMapping(false, { sf -> sf.handle { p, _ -> p * 3 } }) + }) + .aggregate() + .channel { c -> c.queue("routerTwoSubFlowsOutput") } + } - @Bean - fun splitRouteAggregate() = - IntegrationFlow { f -> - f.split() - .route({ 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({ 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 { _, _ -> "odd" } - } + @Bean + fun oddFlow() = + IntegrationFlow { flow -> + flow.handle { _, _ -> "odd" } + } - @Bean - fun evenFlow() = - IntegrationFlow { flow -> - flow.handle { _, _ -> "even" } - } + @Bean + fun evenFlow() = + IntegrationFlow { flow -> + flow.handle { _, _ -> "even" } + } - @Bean - fun publishSubscribe() = - MessageChannels.publishSubscribe() - .ignoreFailures(true) - .applySequence(false) - } + @Bean + fun publishSubscribe() = + MessageChannels.publishSubscribe() + .ignoreFailures(true) + .applySequence(false) + } } diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt new file mode 100644 index 0000000000..695c41f009 --- /dev/null +++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt @@ -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> + + @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>() + + @Bean + @ServiceActivator(inputChannel = "messageConsumerServiceChannel") + fun kotlinConsumer(): (Message) -> 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" } + } + + } + +} diff --git a/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt b/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt index 80e0f3b85a..a92e7990f6 100644 --- a/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt +++ b/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt @@ -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 { _ -> 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 { _ -> 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() - } + } } diff --git a/src/reference/asciidoc/functions-support.adoc b/src/reference/asciidoc/functions-support.adoc new file mode 100644 index 0000000000..b210700113 --- /dev/null +++ b/src/reference/asciidoc/functions-support.adoc @@ -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 <> 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 functionAsService() { + return String::toUpperCase; + } + +} +---- +==== + +You can use it as a simple reference in an XML configuration file: + +==== +[source, xml] +---- + +---- +==== + +When we configure our flow with Messaging Annotations, the code is straightforward: + +==== +[source, java] +---- +@Bean +@Transformer(inputChannel = "functionServiceChannel") +public Function 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 `` or, together with the `@ServiceActivator` annotation, to perform the final step of a flow: + +==== +[source, java] +---- +@Bean +@ServiceActivator(inputChannel = "messageConsumerServiceChannel") +public Consumer> messageConsumerAsService() { + // Has to be an anonymous class for proper type inference + return new Consumer>() { + + @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 ``: + +==== +[source, java] +---- +@Bean +@InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000")) +public Supplier 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 toUpperCaseFunction() { + return String::toUpperCase; +} + +@Bean +public Supplier 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) -> Unit { + return { print(it) } +} + +@Bean +@InboundChannelAdapter(value = "counterChannel", + poller = [Poller(fixedRate = "10", maxMessagesPerPoll = "1")]) +fun kotlinSupplier(): () -> String { + return { "baz" } +} +---- +==== diff --git a/src/reference/asciidoc/logging-adapter.adoc b/src/reference/asciidoc/logging-adapter.adoc index a44db3242c..203ed4847d 100644 --- a/src/reference/asciidoc/logging-adapter.adoc +++ b/src/reference/asciidoc/logging-adapter.adoc @@ -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] ---- >. +[[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 <>. + [[x5.1-general]] === General Changes