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

INT-4458: Do not expose recursive generics API (#2432)

Browse Source 
* INT-4458: Do not expose recursive generics API

JIRA: https://jira.spring.io/browse/INT-4458

According Kotlin generics system restrictions we can't expose API
based on the recursive generics, like we have with Java DSL for JMS

* Replace explicit recursive generics on factory method with the wildcard (`?`).
This way both Kotlin and Java are able to instantiate target object and perform
the proper chain API auto-completion
* Fix `JmsInboundChannelAdapterSpec` to deal with provided `S` type for the
`configureListenerContainer()` - the way we can configure a `JmsDefaultListenerContainerSpec`
* Add `kotlin-spring` Gradle plugin to avoid extra `open` modificator on `@Bean` methods
* Add `JmsDslKotlinTests.kt`

**Cherry-pick to 5.0.x excluding Kotlin support**

* Fix `Amqp` DSL factory for recursive generics

* Polishing for Kotlin tests
This commit is contained in:
Artem Bilan
2018-05-01 08:47:07 -04:00
committed by Gary Russell
parent f970bf0332
commit a62832c707
7 changed files with 185 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
build.gradle
View File

@@ -1,5 +1,5 @@
buildscript {
ext.kotlinVersion = '1.2.40'
ext.kotlinVersion = '1.2.41'
repositories {
maven { url 'https://repo.spring.io/plugins-release' }
}
@@ -7,6 +7,7 @@ buildscript {
classpath 'io.spring.gradle:docbook-reference-plugin:0.3.1'
classpath 'org.asciidoctor:asciidoctor-gradle-plugin:1.5.3'
classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlinVersion"
classpath "org.jetbrains.kotlin:kotlin-allopen:$kotlinVersion"
}
}
@@ -61,6 +62,7 @@ subprojects { subproject ->
apply plugin: 'jacoco'
apply plugin: 'checkstyle'
apply plugin: 'kotlin'
apply plugin: 'kotlin-spring'
sourceSets {
test {
@@ -77,6 +79,7 @@ subprojects { subproject ->
compileTestKotlin {
kotlinOptions {
freeCompilerArgs = ['-Xjsr305=strict']
jvmTarget = '1.8'
}
}
@@ -176,7 +179,7 @@ subprojects { subproject ->
exclude group: 'org.jetbrains.kotlin', module: 'kotlin-reflect'
}
testCompile "org.jetbrains.kotlin:kotlin-stdlib:$kotlinVersion"
testCompile "org.jetbrains.kotlin:kotlin-stdlib-jdk8:$kotlinVersion"
testRuntime "org.jetbrains.kotlin:kotlin-reflect:$kotlinVersion"
}

34
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java
View File

@@ -53,8 +53,8 @@ public final class Amqp {
* @param queueNames the queueNames.
* @return the AmqpInboundGatewaySpec.
*/
public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate,
String... queueNames) {
public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory,
AmqpTemplate amqpTemplate, String... queueNames) {
SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
listenerContainer.setQueueNames(queueNames);
@@ -81,8 +81,8 @@ public final class Amqp {
* @param queues the queues.
* @return the AmqpInboundGatewaySpec.
*/
public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate,
Queue... queues) {
public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory,
AmqpTemplate amqpTemplate, Queue... queues) {
SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
listenerContainer.setQueues(queues);
@@ -266,12 +266,9 @@ public final class Amqp {
/**
* Create an initial AmqpPollableMessageChannelSpec.
* @param connectionFactory the connectionFactory.
* @param <S> the spec type.
* @return the AmqpPollableMessageChannelSpec.
*/
public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S> pollableChannel(
ConnectionFactory connectionFactory) {
public static AmqpPollableMessageChannelSpec<?> pollableChannel(ConnectionFactory connectionFactory) {
return pollableChannel(null, connectionFactory);
}
@@ -279,24 +276,19 @@ public final class Amqp {
* Create an initial AmqpPollableMessageChannelSpec.
* @param id the id.
* @param connectionFactory the connectionFactory.
* @param <S> the spec type.
* @return the AmqpPollableMessageChannelSpec.
*/
public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S> pollableChannel(
String id, ConnectionFactory connectionFactory) {
return new AmqpPollableMessageChannelSpec<S>(connectionFactory).id(id);
public static AmqpPollableMessageChannelSpec<?> pollableChannel(String id, ConnectionFactory connectionFactory) {
return new AmqpPollableMessageChannelSpec<>(connectionFactory)
.id(id);
}
/**
* Create an initial AmqpMessageChannelSpec.
* @param connectionFactory the connectionFactory.
* @param <S> the spec type.
* @return the AmqpMessageChannelSpec.
*/
public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(
ConnectionFactory connectionFactory) {
public static AmqpMessageChannelSpec<?> channel(ConnectionFactory connectionFactory) {
return channel(null, connectionFactory);
}
@@ -304,13 +296,11 @@ public final class Amqp {
* Create an initial AmqpMessageChannelSpec.
* @param id the id.
* @param connectionFactory the connectionFactory.
* @param <S> the spec type.
* @return the AmqpMessageChannelSpec.
*/
public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(String id,
ConnectionFactory connectionFactory) {
return new AmqpMessageChannelSpec<S>(connectionFactory).id(id);
public static AmqpMessageChannelSpec<?> channel(String id, ConnectionFactory connectionFactory) {
return new AmqpMessageChannelSpec<>(connectionFactory)
.id(id);
}
/**

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

@@ -68,10 +68,10 @@ class RouterDslTests {
@Configuration
@EnableIntegration
open class Config {
class Config {
@Bean
open fun routerTwoSubFlows() =
fun routerTwoSubFlows() =
IntegrationFlow { f ->
f.split()
.route<Int, Boolean>({ p -> p % 2 == 0 },

91
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/Jms.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2017 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.
@@ -36,11 +36,9 @@ public final class Jms {
/**
* The factory to produce a {@link JmsPollableMessageChannelSpec}.
* @param connectionFactory the JMS ConnectionFactory to build on
* @param <S> the {@link JmsPollableMessageChannelSpec} inheritor type
* @return the {@link JmsPollableMessageChannelSpec} instance
*/
public static <S extends JmsPollableMessageChannelSpec<S>> JmsPollableMessageChannelSpec<S> pollableChannel(
ConnectionFactory connectionFactory) {
public static JmsPollableMessageChannelSpec<?> pollableChannel(ConnectionFactory connectionFactory) {
return pollableChannel(null, connectionFactory);
}
@@ -48,22 +46,18 @@ public final class Jms {
* The factory to produce a {@link JmsPollableMessageChannelSpec}.
* @param id the bean name for the target {@code PollableChannel} component
* @param connectionFactory the JMS ConnectionFactory to build on
* @param <S> the {@link JmsPollableMessageChannelSpec} inheritor type
* @return the {@link JmsPollableMessageChannelSpec} instance
*/
public static <S extends JmsPollableMessageChannelSpec<S>> JmsPollableMessageChannelSpec<S> pollableChannel(
String id, ConnectionFactory connectionFactory) {
return new JmsPollableMessageChannelSpec<S>(connectionFactory).id(id);
public static JmsPollableMessageChannelSpec<?> pollableChannel(String id, ConnectionFactory connectionFactory) {
return new JmsPollableMessageChannelSpec<>(connectionFactory).id(id);
}
/**
* The factory to produce a {@link JmsMessageChannelSpec}.
* @param connectionFactory the JMS ConnectionFactory to build on
* @param <S> the {@link JmsMessageChannelSpec} inheritor type
* @return the {@link JmsMessageChannelSpec} instance
*/
public static <S extends JmsMessageChannelSpec<S>> JmsMessageChannelSpec<S> channel(
ConnectionFactory connectionFactory) {
public static JmsMessageChannelSpec<?> channel(ConnectionFactory connectionFactory) {
return channel(null, connectionFactory);
}
@@ -71,12 +65,11 @@ public final class Jms {
* The factory to produce a {@link JmsMessageChannelSpec}.
* @param id the bean name for the target {@code MessageChannel} component
* @param connectionFactory the JMS ConnectionFactory to build on
* @param <S> the {@link JmsMessageChannelSpec} inheritor type
* @return the {@link JmsMessageChannelSpec} instance
*/
public static <S extends JmsMessageChannelSpec<S>> JmsMessageChannelSpec<S> channel(String id,
ConnectionFactory connectionFactory) {
return new JmsMessageChannelSpec<S>(connectionFactory).id(id);
public static JmsMessageChannelSpec<?> channel(String id, ConnectionFactory connectionFactory) {
return new JmsMessageChannelSpec<>(connectionFactory)
.id(id);
}
/**
@@ -96,18 +89,17 @@ public final class Jms {
*/
public static JmsPublishSubscribeMessageChannelSpec publishSubscribeChannel(String id,
ConnectionFactory connectionFactory) {
return new JmsPublishSubscribeMessageChannelSpec(connectionFactory).id(id);
}
/**
* The factory to produce a {@link JmsOutboundChannelAdapterSpec}.
* @param jmsTemplate the JmsTemplate to build on
* @param <S> the {@link JmsOutboundChannelAdapterSpec} inheritor type
* @return the {@link JmsOutboundChannelAdapterSpec} instance
*/
public static <S extends JmsOutboundChannelAdapterSpec<S>> JmsOutboundChannelAdapterSpec<S> outboundAdapter(
JmsTemplate jmsTemplate) {
return new JmsOutboundChannelAdapterSpec<S>(jmsTemplate);
public static JmsOutboundChannelAdapterSpec<?> outboundAdapter(JmsTemplate jmsTemplate) {
return new JmsOutboundChannelAdapterSpec<>(jmsTemplate);
}
/**
@@ -117,18 +109,17 @@ public final class Jms {
*/
public static JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware outboundAdapter(
ConnectionFactory connectionFactory) {
return new JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware(connectionFactory);
}
/**
* The factory to produce a {@link JmsInboundChannelAdapterSpec}.
* @param jmsTemplate the JmsTemplate to build on
* @param <S> the {@link JmsInboundChannelAdapterSpec} inheritor type
* @return the {@link JmsInboundChannelAdapterSpec} instance
*/
public static <S extends JmsInboundChannelAdapterSpec<S>> JmsInboundChannelAdapterSpec<S> inboundAdapter(
JmsTemplate jmsTemplate) {
return new JmsInboundChannelAdapterSpec<S>(jmsTemplate);
public static JmsInboundChannelAdapterSpec<?> inboundAdapter(JmsTemplate jmsTemplate) {
return new JmsInboundChannelAdapterSpec<>(jmsTemplate);
}
/**
@@ -138,6 +129,7 @@ public final class Jms {
*/
public static JmsInboundChannelAdapterSpec.JmsInboundChannelSpecTemplateAware inboundAdapter(
ConnectionFactory connectionFactory) {
return new JmsInboundChannelAdapterSpec.JmsInboundChannelSpecTemplateAware(connectionFactory);
}
@@ -153,12 +145,10 @@ public final class Jms {
/**
* The factory to produce a {@link JmsOutboundGatewaySpec}.
* @param listenerContainer the JMS {@link AbstractMessageListenerContainer} to build on
* @param <S> the {@link JmsInboundGatewaySpec} inheritor type
* @return the {@link JmsOutboundGatewaySpec} instance
*/
public static <S extends JmsInboundGatewaySpec<S>> JmsInboundGatewaySpec<S> inboundGateway(
AbstractMessageListenerContainer listenerContainer) {
return new JmsInboundGatewaySpec<S>(listenerContainer);
public static JmsInboundGatewaySpec<?> inboundGateway(AbstractMessageListenerContainer listenerContainer) {
return new JmsInboundGatewaySpec<>(listenerContainer);
}
/**
@@ -168,7 +158,15 @@ public final class Jms {
*/
public static JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<JmsDefaultListenerContainerSpec, DefaultMessageListenerContainer>
inboundGateway(ConnectionFactory connectionFactory) {
return inboundGateway(connectionFactory, DefaultMessageListenerContainer.class);
try {
return new JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<>(
new JmsDefaultListenerContainerSpec()
.connectionFactory(connectionFactory));
}
catch (Exception e) {
throw new IllegalStateException(e);
}
}
/**
@@ -176,18 +174,19 @@ public final class Jms {
* @param connectionFactory the JMS ConnectionFactory to build on
* @param containerClass the {@link AbstractMessageListenerContainer} implementation class
* to instantiate listener container
* @param <S> the {@link JmsListenerContainerSpec} inheritor type
* @param <C> the {@link AbstractMessageListenerContainer} inheritor type
* @return the {@link JmsOutboundGatewaySpec} instance
*/
public static <S extends JmsListenerContainerSpec<S, C>, C extends AbstractMessageListenerContainer>
JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<S, C> inboundGateway(ConnectionFactory connectionFactory,
Class<C> containerClass) {
@SuppressWarnings({ "rawtypes", "unchecked" })
public static <C extends AbstractMessageListenerContainer>
JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<?, C> inboundGateway(
ConnectionFactory connectionFactory, Class<C> containerClass) {
try {
JmsListenerContainerSpec<S, C> spec =
new JmsListenerContainerSpec<S, C>(containerClass)
JmsListenerContainerSpec<?, C> spec =
new JmsListenerContainerSpec<>(containerClass)
.connectionFactory(connectionFactory);
return new JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<S, C>(spec);
return new JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec(spec);
}
catch (Exception e) {
throw new IllegalStateException(e);
@@ -197,13 +196,10 @@ public final class Jms {
/**
* The factory to produce a {@link JmsMessageDrivenChannelAdapterSpec}.
* @param listenerContainer the {@link AbstractMessageListenerContainer} to build on
* @param <S> the {@link JmsMessageDrivenChannelAdapterSpec} inheritor type
* @return the {@link JmsMessageDrivenChannelAdapterSpec} instance
*/
public static <S extends JmsMessageDrivenChannelAdapterSpec<S>>
JmsMessageDrivenChannelAdapterSpec<S> messageDrivenChannelAdapter(
AbstractMessageListenerContainer listenerContainer) {
return new JmsMessageDrivenChannelAdapterSpec<S>(listenerContainer);
public static JmsMessageDrivenChannelAdapterSpec<?> messageDrivenChannelAdapter(AbstractMessageListenerContainer listenerContainer) {
return new JmsMessageDrivenChannelAdapterSpec<>(listenerContainer);
}
/**
@@ -215,7 +211,8 @@ public final class Jms {
messageDrivenChannelAdapter(ConnectionFactory connectionFactory) {
try {
return new JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<>(
new JmsDefaultListenerContainerSpec().connectionFactory(connectionFactory));
new JmsDefaultListenerContainerSpec()
.connectionFactory(connectionFactory));
}
catch (Exception e) {
throw new IllegalStateException(e);
@@ -227,18 +224,18 @@ public final class Jms {
* @param connectionFactory the JMS ConnectionFactory to build on
* @param containerClass the {@link AbstractMessageListenerContainer} implementation class
* to instantiate listener container
* @param <S> the {@link JmsListenerContainerSpec} inheritor type
* @param <C> the {@link AbstractMessageListenerContainer} inheritor type
* @return the {@link JmsMessageDrivenChannelAdapterSpec} instance
*/
public static <S extends JmsListenerContainerSpec<S, C>, C extends AbstractMessageListenerContainer>
JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<S, C>
@SuppressWarnings({ "rawtypes", "unchecked" })
public static <C extends AbstractMessageListenerContainer>
JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<?, C>
messageDrivenChannelAdapter(ConnectionFactory connectionFactory, Class<C> containerClass) {
try {
S spec =
new JmsListenerContainerSpec<S, C>(containerClass)
JmsListenerContainerSpec<?, C> spec =
new JmsListenerContainerSpec<>(containerClass)
.connectionFactory(connectionFactory);
return new JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<>(spec);
return new JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec(spec);
}
catch (Exception e) {
throw new IllegalStateException(e);

9
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-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.
@@ -35,6 +35,7 @@ import org.springframework.util.Assert;
* @param <S> the target {@link JmsInboundGatewaySpec} implementation type.
*
* @author Artem Bilan
*
* @since 5.0
*/
public class JmsInboundGatewaySpec<S extends JmsInboundGatewaySpec<S>>
@@ -184,9 +185,9 @@ public class JmsInboundGatewaySpec<S extends JmsInboundGatewaySpec<S>>
public static class JmsInboundGatewayListenerContainerSpec<S extends JmsListenerContainerSpec<S, C>, C extends AbstractMessageListenerContainer>
extends JmsInboundGatewaySpec<JmsInboundGatewayListenerContainerSpec<S, C>> {
private final JmsListenerContainerSpec<S, C> spec;
private final S spec;
JmsInboundGatewayListenerContainerSpec(JmsListenerContainerSpec<S, C> spec) {
JmsInboundGatewayListenerContainerSpec(S spec) {
super(spec.get());
this.spec = spec;
this.spec.get().setAutoStartup(false);
@@ -213,7 +214,7 @@ public class JmsInboundGatewaySpec<S extends JmsInboundGatewaySpec<S>>
}
public JmsInboundGatewayListenerContainerSpec<S, C> configureListenerContainer(
Consumer<JmsListenerContainerSpec<S, C>> configurer) {
Consumer<S> configurer) {
Assert.notNull(configurer, "'configurer' must not be null");
configurer.accept(this.spec);
return _this();

15
spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java
View File

@@ -65,6 +65,7 @@ import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.MessageListenerContainer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -326,9 +327,10 @@ public class JmsTests {
@Bean
public IntegrationFlow jmsMessageDrivenFlow() {
return IntegrationFlows
.from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory())
.from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory(), DefaultMessageListenerContainer.class)
.outputChannel(jmsMessageDrivenInputChannel())
.destination("jmsMessageDriven"))
.destination("jmsMessageDriven")
.configureListenerContainer(c -> c.clientId("foo")))
.<String, String>transform(String::toLowerCase)
.channel(jmsOutboundInboundReplyChannel())
.get();
@@ -377,9 +379,12 @@ public class JmsTests {
@Bean
public IntegrationFlow jmsInboundGatewayFlow() {
return IntegrationFlows.from(Jms.inboundGateway(jmsConnectionFactory())
.requestChannel(jmsInboundGatewayInputChannel())
.destination("jmsPipelineTest"))
return IntegrationFlows.from(
Jms.inboundGateway(jmsConnectionFactory())
.requestChannel(jmsInboundGatewayInputChannel())
.destination("jmsPipelineTest")
.configureListenerContainer(c ->
c.transactionManager(mock(PlatformTransactionManager.class))))
.<String, String>transform(String::toUpperCase)
.get();
}

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

@@ -0,0 +1,107 @@
/*
* 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.jms.dsl
import assertk.assertions.isEqualTo
import assertk.assertions.isNotNull
import org.apache.activemq.ActiveMQConnectionFactory
import org.junit.jupiter.api.Test
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.integration.config.EnableIntegration
import org.springframework.integration.dsl.IntegrationFlow
import org.springframework.integration.dsl.IntegrationFlows
import org.springframework.integration.dsl.MessageChannels
import org.springframework.integration.support.MessageBuilder
import org.springframework.messaging.MessageChannel
import org.springframework.messaging.PollableChannel
import org.springframework.messaging.simp.SimpMessageHeaderAccessor
import org.springframework.test.annotation.DirtiesContext
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
import java.util.concurrent.Executors
/**
* @author Artem Bilan
*
* @since 5.0.5
*/
@SpringJUnitConfig
@DirtiesContext
class JmsDslKotlinTests {
@Autowired
@Qualifier("jmsOutboundFlow.input")
private lateinit var jmsOutboundInboundChannel: MessageChannel
@Autowired
private lateinit var jmsOutboundInboundReplyChannel: PollableChannel
@Test
fun `test JMS Channel Adapters DSL`() {
this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload(" foo ")
.setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "containerSpecDestination")
.build())
val receive = this.jmsOutboundInboundReplyChannel.receive(10000)
val payload = receive?.payload
assertk.assert(payload).isNotNull {
it.isEqualTo("foo")
}
}
@Configuration
@EnableIntegration
class Config {
@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))
}
@Bean
fun jmsMessageDrivenFlowWithContainer() =
IntegrationFlows.from(
Jms.messageDrivenChannelAdapter(
Jms.container(jmsConnectionFactory(), "containerSpecDestination")
.pubSubDomain(false)
.taskExecutor(Executors.newCachedThreadPool())
.get()))
.transform({ it: String -> it.trim({ it <= ' ' }) })
.channel(jmsOutboundInboundReplyChannel())
.get()
@Bean
fun jmsOutboundInboundReplyChannel() = MessageChannels.queue().get()
}
}