Rework most of JMS tests to JUnit 5

* Optimize some resources usage relying on a shared object
* Remove some redundant `ActiveMqTestUtils`
This commit is contained in:
Artem Bilan
2021-03-05 16:51:49 -05:00
parent 9761c7f0b7
commit f154088935
44 changed files with 739 additions and 1019 deletions

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 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.
@@ -18,8 +18,8 @@ package org.springframework.integration.jms;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.transport.vm.VMTransport;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.springframework.jms.connection.CachingConnectionFactory;
@@ -29,26 +29,25 @@ import org.springframework.jms.connection.CachingConnectionFactory;
* connection is closed).
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 3.0
*
*/
public abstract class ActiveMQMultiContextTests {
public static final ActiveMQConnectionFactory amqFactory =
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
public static final CachingConnectionFactory connectionFactory =
new CachingConnectionFactory(amqFactory);
public static final CachingConnectionFactory connectionFactory = new CachingConnectionFactory(amqFactory);
@BeforeClass
@BeforeAll
public static void startUp() throws Exception {
amqFactory.setTrustAllPackages(true);
connectionFactory.setCacheConsumers(false);
connectionFactory.createConnection().close();
}
@AfterClass
@AfterAll
public static void shutDown() {
connectionFactory.resetConnection();
}

View File

@@ -13,12 +13,15 @@
<context:property-placeholder properties-ref="props"/>
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.amqFactory"/>
<util:properties id="props">
<prop key="jmsAcknowledgeModeTransacted">transacted</prop>
</util:properties>
<int-jms:inbound-channel-adapter channel="out" session-transacted="true"
connection-factory="jmsConnectionFactory" destination-name="incatQ"
destination-name="incatQ"
receive-timeout="500" acknowledge="${jmsAcknowledgeModeTransacted}">
<int:poller fixed-delay="500"/>
</int-jms:inbound-channel-adapter>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-2021 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.
@@ -18,29 +18,22 @@ package org.springframework.integration.jms;
import static org.assertj.core.api.Assertions.assertThat;
import javax.jms.ConnectionFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
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.context.annotation.ImportResource;
import org.springframework.integration.jms.JmsInboundChannelAdapterTests.CFConfig;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.0
*
*/
@ContextConfiguration(classes = CFConfig.class)
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class JmsInboundChannelAdapterTests extends ActiveMQMultiContextTests {
@@ -58,14 +51,4 @@ public class JmsInboundChannelAdapterTests extends ActiveMQMultiContextTests {
*/
}
@Configuration
@ImportResource("org/springframework/integration/jms/JmsInboundChannelAdapterTests-context.xml")
public static class CFConfig {
@Bean
public ConnectionFactory jmsConnectionFactory() {
return amqFactory;
}
}
}

View File

@@ -1,26 +1,31 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd">
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.amqFactory"/>
<int-jms:message-driven-channel-adapter channel="toOut"
connection-factory="jmsConnectionFactory"
acknowledge="transacted"
destination-name="outcatQ1"/>
connection-factory="jmsConnectionFactory"
acknowledge="transacted"
destination-name="outcatQ1"/>
<int:publish-subscribe-channel id="toOut" />
<int:publish-subscribe-channel id="toOut"/>
<int-jms:outbound-channel-adapter channel="toOut" order="1"
connection-factory="jmsConnectionFactory"
destination-name="outcatQ2"
session-transacted="true" />
connection-factory="jmsConnectionFactory"
destination-name="outcatQ2"
session-transacted="true"/>
<int:service-activator input-channel="toOut" order="2" ref="aborter"/>
<bean id="aborter" class="org.springframework.integration.jms.JmsOutboundChannelAdapterTests$Aborter" />
<bean id="aborter" class="org.springframework.integration.jms.JmsOutboundChannelAdapterTests$Aborter"/>
</beans>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-2021 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.
@@ -18,28 +18,21 @@ package org.springframework.integration.jms;
import static org.assertj.core.api.Assertions.assertThat;
import javax.jms.ConnectionFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
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.context.annotation.ImportResource;
import org.springframework.integration.jms.JmsOutboundChannelAdapterTests.CFConfig;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.0
*
*/
@ContextConfiguration(classes = CFConfig.class)
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class JmsOutboundChannelAdapterTests extends ActiveMQMultiContextTests {
@@ -63,16 +56,6 @@ public class JmsOutboundChannelAdapterTests extends ActiveMQMultiContextTests {
endpoint.stop();
}
@Configuration
@ImportResource("org/springframework/integration/jms/JmsOutboundChannelAdapterTests-context.xml")
public static class CFConfig {
@Bean
public ConnectionFactory jmsConnectionFactory() {
return connectionFactory;
}
}
public static class Aborter {
private volatile boolean abort;
@@ -82,6 +65,7 @@ public class JmsOutboundChannelAdapterTests extends ActiveMQMultiContextTests {
throw new RuntimeException("intentional");
}
}
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -42,14 +42,12 @@ import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.jms.JmsOutboundGateway.ReplyContainerProperties;
import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.jms.JmsException;
@@ -70,11 +68,6 @@ public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
private final Log logger = LogFactory.getLog(this.getClass());
@Rule
public Log4j2LevelAdjuster adjuster =
Log4j2LevelAdjuster.trace()
.categories(true, "org.springframework.jms", "org.apache");
@Test
public void testContainerBeanNameWhenNoGatewayBeanName() {
JmsOutboundGateway gateway = new JmsOutboundGateway();
@@ -98,7 +91,7 @@ public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
gateway.setRequestDestinationName("foo");
gateway.setUseReplyContainer(true);
ReplyContainerProperties replyContainerProperties = new ReplyContainerProperties();
final List<Throwable> errors = new ArrayList<Throwable>();
final List<Throwable> errors = new ArrayList<>();
ExecutorService exec = Executors.newFixedThreadPool(10);
ErrorHandlingTaskExecutor errorHandlingTaskExecutor =
new ErrorHandlingTaskExecutor(exec, t -> {
@@ -162,7 +155,7 @@ public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
}
@Test
public void testConnectionBreakOnReplyMessageIdCorrelation() throws Exception {
public void testConnectionBreakOnReplyMessageIdCorrelation() {
CachingConnectionFactory connectionFactory1 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
connectionFactory1.setCacheConsumers(false);
@@ -179,7 +172,7 @@ public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
gateway.afterPropertiesSet();
gateway.start();
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> gateway.handleMessage(new GenericMessage<String>("foo")));
exec.execute(() -> gateway.handleMessage(new GenericMessage<>("foo")));
CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
connectionFactory2.setCacheConsumers(false);
@@ -205,7 +198,7 @@ public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
}
@Test
public void testConnectionBreakOnReplyCustomCorrelation() throws Exception {
public void testConnectionBreakOnReplyCustomCorrelation() {
CachingConnectionFactory connectionFactory1 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
connectionFactory1.setCacheConsumers(false);
@@ -223,7 +216,7 @@ public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
gateway.afterPropertiesSet();
gateway.start();
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> gateway.handleMessage(new GenericMessage<String>("foo")));
exec.execute(() -> gateway.handleMessage(new GenericMessage<>("foo")));
CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
connectionFactory2.setCacheConsumers(false);

View File

@@ -1,13 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:jms="http://www.springframework.org/schema/integration/jms"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:jms="http://www.springframework.org/schema/integration/jms"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd">
<int:channel id="receiveChannel">
<int:queue/>
@@ -44,8 +46,7 @@
<constructor-arg value="test.queue2"/>
</bean>
<bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
</bean>
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.amqFactory"/>
</beans>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -18,8 +18,7 @@ package org.springframework.integration.jms;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.support.MessageBuilder;
@@ -27,17 +26,13 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* //INT-2275
*
* @author Artem Bilan
* @author Gary Russell
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class JmsOutboundInsideChainTests extends ActiveMQMultiContextTests {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -32,8 +32,8 @@ import javax.jms.Destination;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.context.IntegrationContextUtils;
@@ -55,14 +55,15 @@ public class OutboundGatewayConnectionTests {
private final Destination replyQueue1 = new ActiveMQQueue("reply1");
@Test @Ignore // need a more reliable stop/start for AMQ
@Test
@Disabled("need a more reliable stop/start for AMQ")
public void testContainerWithDestBrokenConnection() throws Exception {
BeanFactory beanFactory = mock(BeanFactory.class);
when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();
when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
.thenReturn(scheduler);
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
BrokerService broker = new BrokerService();

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -26,23 +26,17 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.JmsException;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.TaskScheduler;
@@ -77,9 +71,6 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
private static Destination replyQueue7 = new ActiveMQQueue("reply7");
@Rule
public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.trace();
@Test
public void testContainerWithDest() throws Exception {
BeanFactory beanFactory = mock(BeanFactory.class);
@@ -90,7 +81,6 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
ConnectionFactory connectionFactory = getConnectionFactory();
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestination(requestQueue1);
gateway.setReplyDestination(replyQueue1);
@@ -98,14 +88,14 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
gateway.start();
final AtomicReference<Object> reply = new AtomicReference<Object>();
final AtomicReference<Object> reply = new AtomicReference<>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo")));
}
finally {
latch2.countDown();
@@ -118,7 +108,7 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
javax.jms.Message request = template.receive(requestQueue1);
assertThat(request).isNotNull();
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), (MessageCreator) session -> jmsReply);
template.send(request.getJMSReplyTo(), session -> jmsReply);
assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(reply.get()).isNotNull();
@@ -137,20 +127,20 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getConnectionFactory());
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestination(requestQueue2);
gateway.setReplyDestination(replyQueue2);
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
gateway.start();
final AtomicReference<Object> reply = new AtomicReference<Object>();
final AtomicReference<Object> reply = new AtomicReference<>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo")));
}
finally {
latch2.countDown();
@@ -158,12 +148,12 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
});
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setConnectionFactory(connectionFactory);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue2);
assertThat(request).isNotNull();
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), (MessageCreator) session -> {
template.send(request.getJMSReplyTo(), session -> {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
});
@@ -185,21 +175,21 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getConnectionFactory());
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestination(requestQueue3);
gateway.setReplyDestinationName("reply3");
gateway.setCorrelationKey("JMSCorrelationID");
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
gateway.start();
final AtomicReference<Object> reply = new AtomicReference<Object>();
final AtomicReference<Object> reply = new AtomicReference<>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo")));
}
finally {
latch2.countDown();
@@ -207,7 +197,7 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
});
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setConnectionFactory(connectionFactory);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue3);
assertThat(request).isNotNull();
@@ -231,7 +221,7 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getConnectionFactory());
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestination(requestQueue4);
gateway.setReplyDestinationName("reply4");
gateway.setUseReplyContainer(true);
@@ -252,7 +242,7 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
});
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setConnectionFactory(connectionFactory);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue4);
assertThat(request).isNotNull();
@@ -279,7 +269,7 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getConnectionFactory());
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestination(requestQueue5);
gateway.setCorrelationKey("JMSCorrelationID");
gateway.setUseReplyContainer(true);
@@ -301,7 +291,7 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
});
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setConnectionFactory(connectionFactory);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue5);
assertThat(request).isNotNull();
@@ -325,7 +315,7 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getConnectionFactory());
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestination(requestQueue6);
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
@@ -345,7 +335,7 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
});
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setConnectionFactory(connectionFactory);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue6);
assertThat(request).isNotNull();
@@ -372,7 +362,7 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getConnectionFactory());
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestination(requestQueue7);
gateway.setReplyDestination(replyQueue7);
gateway.setCorrelationKey("JMSCorrelationID");
@@ -385,7 +375,7 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setConnectionFactory(connectionFactory);
template.setReceiveTimeout(20000);
receiveAndSend(template);
receiveAndSend(template);
@@ -418,12 +408,4 @@ public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
}
}
private ConnectionFactory getConnectionFactory() {
ActiveMQConnectionFactory activeMQConnectionFactory =
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(activeMQConnectionFactory);
cachingConnectionFactory.setCacheConsumers(false);
return cachingConnectionFactory;
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -33,14 +33,11 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.jms.Destination;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.jms.config.JmsChannelFactoryBean;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jms.connection.CachingConnectionFactory;
@@ -58,33 +55,23 @@ import org.springframework.messaging.support.GenericMessage;
*/
public class PollableJmsChannelTests extends ActiveMQMultiContextTests {
private ActiveMQConnectionFactory connectionFactory;
private Destination queue;
@Before
public void setup() {
ActiveMqTestUtils.prepare();
this.connectionFactory = new ActiveMQConnectionFactory();
this.connectionFactory.setBrokerURL("vm://localhost?broker.persistent=false");
this.connectionFactory.setTrustAllPackages(true);
}
@Test
public void queueReference() throws Exception {
this.queue = new ActiveMQQueue("pollableJmsChannelTestQueue");
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
CachingConnectionFactory ccf = new CachingConnectionFactory(this.connectionFactory);
CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
ccf.setCacheConsumers(false);
factoryBean.setConnectionFactory(ccf);
factoryBean.setDestination(this.queue);
factoryBean.setBeanFactory(mock(BeanFactory.class));
factoryBean.afterPropertiesSet();
PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject();
boolean sent1 = channel.send(new GenericMessage<String>("foo"));
boolean sent1 = channel.send(new GenericMessage<>("foo"));
assertThat(sent1).isTrue();
boolean sent2 = channel.send(new GenericMessage<String>("bar"));
boolean sent2 = channel.send(new GenericMessage<>("bar"));
assertThat(sent2).isTrue();
Message<?> result1 = channel.receive(10000);
assertThat(result1).isNotNull();
@@ -97,7 +84,7 @@ public class PollableJmsChannelTests extends ActiveMQMultiContextTests {
@Test
public void queueName() throws Exception {
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
CachingConnectionFactory ccf = new CachingConnectionFactory(this.connectionFactory);
CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
ccf.setCacheConsumers(false);
factoryBean.setConnectionFactory(ccf);
factoryBean.setDestinationName("someDynamicQueue");
@@ -105,9 +92,9 @@ public class PollableJmsChannelTests extends ActiveMQMultiContextTests {
factoryBean.setBeanFactory(mock(BeanFactory.class));
factoryBean.afterPropertiesSet();
PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject();
boolean sent1 = channel.send(new GenericMessage<String>("foo"));
boolean sent1 = channel.send(new GenericMessage<>("foo"));
assertThat(sent1).isTrue();
boolean sent2 = channel.send(new GenericMessage<String>("bar"));
boolean sent2 = channel.send(new GenericMessage<>("bar"));
assertThat(sent2).isTrue();
Message<?> result1 = channel.receive(10000);
assertThat(result1).isNotNull();
@@ -120,19 +107,19 @@ public class PollableJmsChannelTests extends ActiveMQMultiContextTests {
@Test
public void queueNameWithFalsePreReceiveInterceptors() throws Exception {
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
CachingConnectionFactory ccf = new CachingConnectionFactory(this.connectionFactory);
CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
ccf.setCacheConsumers(false);
factoryBean.setConnectionFactory(ccf);
factoryBean.setDestinationName("someDynamicQueue");
factoryBean.setPubSubDomain(false);
List<ChannelInterceptor> interceptorList = new ArrayList<ChannelInterceptor>();
List<ChannelInterceptor> interceptorList = new ArrayList<>();
ChannelInterceptor interceptor = spy(new SampleInterceptor(false));
interceptorList.add(interceptor);
factoryBean.setInterceptors(interceptorList);
factoryBean.setBeanFactory(mock(BeanFactory.class));
factoryBean.afterPropertiesSet();
PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject();
boolean sent1 = channel.send(new GenericMessage<String>("foo"));
boolean sent1 = channel.send(new GenericMessage<>("foo"));
assertThat(sent1).isTrue();
Message<?> result1 = channel.receive(10000);
assertThat(result1).isNull();
@@ -143,19 +130,19 @@ public class PollableJmsChannelTests extends ActiveMQMultiContextTests {
@Test
public void queueNameWithTruePreReceiveInterceptors() throws Exception {
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
CachingConnectionFactory ccf = new CachingConnectionFactory(this.connectionFactory);
CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
ccf.setCacheConsumers(false);
factoryBean.setConnectionFactory(ccf);
factoryBean.setDestinationName("someDynamicQueue");
factoryBean.setPubSubDomain(false);
List<ChannelInterceptor> interceptorList = new ArrayList<ChannelInterceptor>();
List<ChannelInterceptor> interceptorList = new ArrayList<>();
ChannelInterceptor interceptor = spy(new SampleInterceptor(true));
interceptorList.add(interceptor);
factoryBean.setInterceptors(interceptorList);
factoryBean.setBeanFactory(mock(BeanFactory.class));
factoryBean.afterPropertiesSet();
PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject();
boolean sent1 = channel.send(new GenericMessage<String>("foo"));
boolean sent1 = channel.send(new GenericMessage<>("foo"));
assertThat(sent1).isTrue();
Message<?> result1 = channel.receive(10000);
assertThat(result1).isNotNull();
@@ -180,10 +167,10 @@ public class PollableJmsChannelTests extends ActiveMQMultiContextTests {
factoryBean.setBeanFactory(mock(BeanFactory.class));
factoryBean.afterPropertiesSet();
PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject();
final JmsTemplate receiver = new JmsTemplate(this.connectionFactory);
boolean sent1 = channel.send(new GenericMessage<String>("foo"));
final JmsTemplate receiver = new JmsTemplate(connectionFactory);
boolean sent1 = channel.send(new GenericMessage<>("foo"));
assertThat(sent1).isTrue();
final AtomicReference<javax.jms.Message> message = new AtomicReference<javax.jms.Message>();
final AtomicReference<javax.jms.Message> message = new AtomicReference<>();
final CountDownLatch latch1 = new CountDownLatch(1);
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
@@ -216,7 +203,7 @@ public class PollableJmsChannelTests extends ActiveMQMultiContextTests {
this.queue = new ActiveMQQueue("pollableJmsChannelSelectorTestQueue");
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
CachingConnectionFactory ccf = new CachingConnectionFactory(this.connectionFactory);
CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
ccf.setCacheConsumers(false);
factoryBean.setConnectionFactory(ccf);
factoryBean.setDestination(this.queue);
@@ -226,12 +213,12 @@ public class PollableJmsChannelTests extends ActiveMQMultiContextTests {
factoryBean.setBeanFactory(mock(BeanFactory.class));
factoryBean.afterPropertiesSet();
PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject();
boolean sent1 = channel.send(new GenericMessage<String>("foo"));
boolean sent1 = channel.send(new GenericMessage<>("foo"));
assertThat(sent1).isTrue();
Message<?> result1 = channel.receive(100);
assertThat(result1).isNull();
JmsTemplate jmsTemplate = new JmsTemplate(this.connectionFactory);
JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
jmsTemplate.setDefaultDestinationName("pollableJmsChannelSelectorTestQueue");
jmsTemplate.send(session -> {
TextMessage message = session.createTextMessage("bar");

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2019 the original author or authors.
* Copyright 2015-2021 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.
@@ -22,8 +22,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
@@ -46,15 +45,14 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Artem Bilan
*
* @since 4.3
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class SplitterAggregatorTests extends ActiveMQMultiContextTests {
@@ -68,7 +66,7 @@ public class SplitterAggregatorTests extends ActiveMQMultiContextTests {
@Test
public void testSplitterAggregatorOverJms() {
List<Integer> payload = Arrays.asList(1, 2, 3, 4, 5, 6);
this.splitChannel.send(new GenericMessage<List<Integer>>(payload));
this.splitChannel.send(new GenericMessage<>(payload));
Message<?> message = this.resultChannel.receive(10000);
assertThat(message).isNotNull();
Collections.sort(((List<Integer>) message.getPayload()));

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -33,22 +33,18 @@ import java.util.concurrent.TimeUnit;
import javax.jms.Destination;
import javax.jms.MessageListener;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.commons.logging.Log;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.jms.config.JmsChannelFactoryBean;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.messaging.Message;
@@ -67,43 +63,31 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
private static final int TIMEOUT = 30000;
private CachingConnectionFactory connectionFactory;
private Destination topic;
private Destination queue;
@Before
@BeforeEach
public void setup() {
ActiveMqTestUtils.prepare();
ActiveMQConnectionFactory targetConnectionFactory = new ActiveMQConnectionFactory();
targetConnectionFactory.setBrokerURL("vm://localhost?broker.persistent=false");
targetConnectionFactory.setTrustAllPackages(true);
this.connectionFactory = new CachingConnectionFactory(targetConnectionFactory);
this.topic = new ActiveMQTopic("testTopic");
this.queue = new ActiveMQQueue("testQueue");
}
@After
public void tearDown() throws Exception {
this.connectionFactory.resetConnection();
}
@Test
public void queueReference() throws Exception {
final CountDownLatch latch = new CountDownLatch(2);
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<Message<?>>());
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<>());
MessageHandler handler1 = message -> {
receivedList1.add(message);
latch.countDown();
};
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<Message<?>>());
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<>());
MessageHandler handler2 = message -> {
receivedList2.add(message);
latch.countDown();
};
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
factoryBean.setConnectionFactory(this.connectionFactory);
factoryBean.setConnectionFactory(connectionFactory);
factoryBean.setDestination(this.queue);
factoryBean.setBeanFactory(mock(BeanFactory.class));
factoryBean.afterPropertiesSet();
@@ -112,8 +96,8 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
channel.start();
channel.subscribe(handler1);
channel.subscribe(handler2);
channel.send(new GenericMessage<String>("foo"));
channel.send(new GenericMessage<String>("bar"));
channel.send(new GenericMessage<>("foo"));
channel.send(new GenericMessage<>("bar"));
latch.await(TIMEOUT, TimeUnit.MILLISECONDS);
assertThat(receivedList1.size()).isEqualTo(1);
assertThat(receivedList1.get(0)).isNotNull();
@@ -127,18 +111,18 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
@Test
public void topicReference() throws Exception {
final CountDownLatch latch = new CountDownLatch(4);
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<Message<?>>());
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<>());
MessageHandler handler1 = message -> {
receivedList1.add(message);
latch.countDown();
};
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<Message<?>>());
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<>());
MessageHandler handler2 = message -> {
receivedList2.add(message);
latch.countDown();
};
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
factoryBean.setConnectionFactory(this.connectionFactory);
factoryBean.setConnectionFactory(connectionFactory);
factoryBean.setDestination(this.topic);
factoryBean.setBeanFactory(mock(BeanFactory.class));
factoryBean.afterPropertiesSet();
@@ -150,8 +134,8 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
if (!waitUntilRegisteredWithDestination(channel, 10000)) {
fail("Listener failed to subscribe to topic");
}
channel.send(new GenericMessage<String>("foo"));
channel.send(new GenericMessage<String>("bar"));
channel.send(new GenericMessage<>("foo"));
channel.send(new GenericMessage<>("bar"));
latch.await(TIMEOUT, TimeUnit.MILLISECONDS);
assertThat(receivedList1.size()).isEqualTo(2);
assertThat(receivedList1.get(0).getPayload()).isEqualTo("foo");
@@ -165,18 +149,18 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
@Test
public void queueName() throws Exception {
final CountDownLatch latch = new CountDownLatch(2);
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<Message<?>>());
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<>());
MessageHandler handler1 = message -> {
receivedList1.add(message);
latch.countDown();
};
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<Message<?>>());
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<>());
MessageHandler handler2 = message -> {
receivedList2.add(message);
latch.countDown();
};
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
factoryBean.setConnectionFactory(this.connectionFactory);
factoryBean.setConnectionFactory(connectionFactory);
factoryBean.setDestinationName("dynamicQueue");
factoryBean.setPubSubDomain(false);
factoryBean.setBeanFactory(mock(BeanFactory.class));
@@ -187,8 +171,8 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
channel.start();
channel.subscribe(handler1);
channel.subscribe(handler2);
channel.send(new GenericMessage<String>("foo"));
channel.send(new GenericMessage<String>("bar"));
channel.send(new GenericMessage<>("foo"));
channel.send(new GenericMessage<>("bar"));
assertThat(latch.await(TIMEOUT, TimeUnit.MILLISECONDS))
.as("Countdown latch should have counted down to 0 but was "
@@ -206,19 +190,19 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
@Test
public void topicName() throws Exception {
final CountDownLatch latch = new CountDownLatch(4);
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<Message<?>>());
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<>());
MessageHandler handler1 = message -> {
receivedList1.add(message);
latch.countDown();
};
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<Message<?>>());
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<>());
MessageHandler handler2 = message -> {
receivedList2.add(message);
latch.countDown();
};
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
factoryBean.setConnectionFactory(this.connectionFactory);
factoryBean.setConnectionFactory(connectionFactory);
factoryBean.setDestinationName("dynamicTopic");
factoryBean.setPubSubDomain(true);
factoryBean.setBeanFactory(mock(BeanFactory.class));
@@ -231,8 +215,8 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
}
channel.subscribe(handler1);
channel.subscribe(handler2);
channel.send(new GenericMessage<String>("foo"));
channel.send(new GenericMessage<String>("bar"));
channel.send(new GenericMessage<>("foo"));
channel.send(new GenericMessage<>("bar"));
latch.await(TIMEOUT, TimeUnit.MILLISECONDS);
assertThat(receivedList1.size()).isEqualTo(2);
assertThat(receivedList1.get(0).getPayload()).isEqualTo("foo");
@@ -247,7 +231,7 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
public void contextManagesLifecycle() {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsChannelFactoryBean.class);
builder.addConstructorArgValue(true);
builder.addPropertyValue("connectionFactory", this.connectionFactory);
builder.addPropertyValue("connectionFactory", connectionFactory);
builder.addPropertyValue("destinationName", "dynamicQueue");
builder.addPropertyValue("pubSubDomain", false);
StaticApplicationContext context = new StaticApplicationContext();
@@ -264,7 +248,7 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
@Test
public void dispatcherHasNoSubscribersQueue() throws Exception {
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
factoryBean.setConnectionFactory(this.connectionFactory);
factoryBean.setConnectionFactory(connectionFactory);
factoryBean.setDestinationName("noSubscribersQueue");
factoryBean.setBeanName("noSubscribersChannel");
factoryBean.setBeanFactory(mock(BeanFactory.class));
@@ -289,7 +273,7 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
@Test
public void dispatcherHasNoSubscribersTopic() throws Exception {
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
factoryBean.setConnectionFactory(this.connectionFactory);
factoryBean.setConnectionFactory(connectionFactory);
factoryBean.setDestinationName("noSubscribersTopic");
factoryBean.setBeanName("noSubscribersChannel");
factoryBean.setPubSubDomain(true);
@@ -302,7 +286,7 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
.getPropertyValue(channel, "container",
AbstractMessageListenerContainer.class);
MessageListener listener = (MessageListener) container.getMessageListener();
List<String> logList = insertMockLoggerInListener(channel);
List<String> logList = insertMockLoggerInListener(channel);
listener.onMessage(new StubTextMessage("Hello, world!"));
verifyLogReceived(logList);
}
@@ -312,7 +296,7 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
AbstractMessageListenerContainer container = TestUtils.getPropertyValue(
channel, "container", AbstractMessageListenerContainer.class);
Log logger = mock(Log.class);
final ArrayList<String> logList = new ArrayList<String>();
final ArrayList<String> logList = new ArrayList<>();
doAnswer(invocation -> {
String message = invocation.getArgument(0);
if (message.startsWith("Dispatcher has no subscribers")) {
@@ -357,7 +341,7 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
(AbstractMessageListenerContainer) new DirectFieldAccessor(channel).getPropertyValue("container");
if (container instanceof DefaultMessageListenerContainer) {
DefaultMessageListenerContainer listenerContainer =
(DefaultMessageListenerContainer) container;
(DefaultMessageListenerContainer) container;
if (listenerContainer.getCacheLevel() != DefaultMessageListenerContainer.CACHE_CONSUMER) {
return true;
}

View File

@@ -1,58 +0,0 @@
/*
* Copyright 2002-2019 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
*
* https://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.config;
import java.io.File;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
*
*/
public class ActiveMqTestUtils {
private static final Log logger = LogFactory.getLog(ActiveMqTestUtils.class);
private ActiveMqTestUtils() {
super();
}
@Before
public static void prepare() {
logger.info("####### Refreshing ActiveMq ########");
File activeMqTempDir = new File("activemq-data");
deleteDir(activeMqTempDir);
}
private static void deleteDir(File directory) {
if (directory.exists()) {
String[] children = directory.list();
if (children != null) {
for (int i = 0; i < children.length; i++) {
deleteDir(new File(directory, children[i]));
}
}
}
directory.delete();
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -18,8 +18,7 @@ package org.springframework.integration.jms.config;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
@@ -31,17 +30,16 @@ import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.ErrorHandler;
/**
* @author Mark Fisher
* @author Artem Bilan
*
* @since 1.0.3
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class DefaultConfigurationTests {

View File

@@ -1,15 +1,17 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:task="http://www.springframework.org/schema/task"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/jms https://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/task https://www.springframework.org/schema/task/spring-task.xsd"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:task="http://www.springframework.org/schema/task">
http://www.springframework.org/schema/task https://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd">
<int:gateway id="sampleGateway"
service-interface="org.springframework.integration.jms.config.ExceptionHandlingSiConsumerTests$SampleGateway"
@@ -71,14 +73,7 @@
<constructor-arg value="request.queue.c"/>
</bean>
<bean id="jmsConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="false"/>
</bean>
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.connectionFactory"/>
</beans>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -18,33 +18,37 @@ package org.springframework.integration.jms.config;
import static org.assertj.core.api.Assertions.assertThat;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.TextMessage;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Oleg Zhurakousky
* @author Artem Bilan
*/
public class ExceptionHandlingSiConsumerTests {
@SpringJUnitConfig
@DirtiesContext
public class ExceptionHandlingSiConsumerTests extends ActiveMQMultiContextTests {
@Autowired
ApplicationContext applicationContext;
@SuppressWarnings("resource")
@Test
public void nonSiProducer_siConsumer_sync_withReturn() throws Exception {
ActiveMqTestUtils.prepare();
ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("Exception-nonSiProducer-siConsumer.xml", ExceptionHandlingSiConsumerTests.class);
JmsTemplate jmsTemplate = new JmsTemplate(applicationContext.getBean("jmsConnectionFactory", ConnectionFactory.class));
public void nonSiProducer_siConsumer_sync_withReturn() {
JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
Destination request = applicationContext.getBean("requestQueueA", Destination.class);
final Destination reply = applicationContext.getBean("replyQueueA", Destination.class);
jmsTemplate.send(request, (MessageCreator) session -> {
jmsTemplate.send(request, session -> {
TextMessage message = session.createTextMessage();
message.setText("echoChannel");
message.setJMSReplyTo(reply);
@@ -52,18 +56,14 @@ public class ExceptionHandlingSiConsumerTests {
});
Message message = jmsTemplate.receive(reply);
assertThat(message).isNotNull();
applicationContext.close();
}
@SuppressWarnings("resource")
@Test
public void nonSiProducer_siConsumer_sync_withReturnNoException() throws Exception {
ActiveMqTestUtils.prepare();
ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("Exception-nonSiProducer-siConsumer.xml", ExceptionHandlingSiConsumerTests.class);
JmsTemplate jmsTemplate = new JmsTemplate(applicationContext.getBean("jmsConnectionFactory", ConnectionFactory.class));
JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
Destination request = applicationContext.getBean("requestQueueB", Destination.class);
final Destination reply = applicationContext.getBean("replyQueueB", Destination.class);
jmsTemplate.send(request, (MessageCreator) session -> {
jmsTemplate.send(request, session -> {
TextMessage message = session.createTextMessage();
message.setText("echoWithExceptionChannel");
message.setJMSReplyTo(reply);
@@ -72,17 +72,13 @@ public class ExceptionHandlingSiConsumerTests {
Message message = jmsTemplate.receive(reply);
assertThat(message).isNotNull();
assertThat(((TextMessage) message).getText()).isEqualTo("echoWithException");
applicationContext.close();
}
@Test
public void nonSiProducer_siConsumer_sync_withOutboundGateway() throws Exception {
ActiveMqTestUtils.prepare();
final ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("Exception-nonSiProducer-siConsumer.xml", ExceptionHandlingSiConsumerTests.class);
public void nonSiProducer_siConsumer_sync_withOutboundGateway() {
SampleGateway gateway = applicationContext.getBean("sampleGateway", SampleGateway.class);
String reply = gateway.echo("echoWithExceptionChannel");
assertThat(reply).isEqualTo("echoWithException");
applicationContext.close();
}
@@ -95,26 +91,33 @@ public class ExceptionHandlingSiConsumerTests {
public String echo(String value) {
return value;
}
}
@SuppressWarnings("serial")
public static class SampleException extends RuntimeException {
public SampleException(String message) {
super(message);
}
}
public interface SampleGateway {
String echo(String value);
}
public static class SampleErrorTransformer {
public org.springframework.messaging.Message<?> transform(Throwable t) throws Exception {
public org.springframework.messaging.Message<?> transform(Throwable t) {
return MessageBuilder.withPayload(t.getCause().getMessage()).build();
}
}
}

View File

@@ -25,15 +25,8 @@
<int-jms:channel id="jmsChannel" cache="${CACHELEVEL}" queue="jmsQueue" container-type="${listenerContainer}"/>
<bean id="jmsConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="false"/>
</bean>
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.connectionFactory"/>
<bean id="jmsQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="jmsQueue.queue"/>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -20,13 +20,14 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.List;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Oleg Zhurakousky
@@ -35,20 +36,19 @@ import org.springframework.messaging.support.InterceptableChannel;
*
* @since 2.0.1
*/
public class GlobalChannelInterceptorTests {
@SpringJUnitConfig
@DirtiesContext
public class GlobalChannelInterceptorTests extends ActiveMQMultiContextTests {
@Autowired
InterceptableChannel jmsChannel;
@Test
public void testJmsChannel() {
ActiveMqTestUtils.prepare();
try (ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"GlobalChannelInterceptorTests-context.xml", GlobalChannelInterceptorTests.class)) {
InterceptableChannel jmsChannel = context.getBean("jmsChannel", AbstractMessageChannel.class);
List<ChannelInterceptor> interceptors = jmsChannel.getInterceptors();
assertThat(interceptors).isNotNull();
assertThat(interceptors.size()).isEqualTo(1);
assertThat(interceptors.get(0) instanceof SampleInterceptor).isTrue();
}
List<ChannelInterceptor> interceptors = this.jmsChannel.getInterceptors();
assertThat(interceptors).isNotNull();
assertThat(interceptors.size()).isEqualTo(1);
assertThat(interceptors.get(0) instanceof SampleInterceptor).isTrue();
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -22,7 +22,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.context.ConfigurableApplicationContext;
@@ -40,21 +40,20 @@ import org.springframework.messaging.support.GenericMessage;
* @author Oleg Zhurakousky
* @author Gunnar Hillert
* @author Gary Russell
*
* @author Artem Bilan
*/
public class JmsChannelHistoryTests {
@Test
public void testMessageHistory() throws Exception {
public void testMessageHistory() {
AbstractMessageListenerContainer mlContainer = mock(AbstractMessageListenerContainer.class);
JmsTemplate template = mock(JmsTemplate.class);
SubscribableJmsChannel channel = new SubscribableJmsChannel(mlContainer, template);
channel.setShouldTrack(true);
channel.setBeanName("jmsChannel");
Message<String> message = new GenericMessage<String>("hello");
Message<String> message = new GenericMessage<>("hello");
doAnswer(invocation -> {
@SuppressWarnings("unchecked")
Message<String> msg = invocation.getArgument(0);
MessageHistory history = MessageHistory.read(msg);
assertThat(history.get(0).contains("jmsChannel")).isTrue();
@@ -66,17 +65,17 @@ public class JmsChannelHistoryTests {
}
@Test
public void testFullConfig() throws Exception {
ActiveMqTestUtils.prepare();
ConfigurableApplicationContext ac =
new ClassPathXmlApplicationContext("JmsChannelHistoryTests-context.xml", this.getClass());
SubscribableChannel channel = ac.getBean("jmsChannel", SubscribableChannel.class);
PollableChannel resultChannel = ac.getBean("resultChannel", PollableChannel.class);
channel.send(new GenericMessage<String>("hello"));
Message<?> resultMessage = resultChannel.receive(10000);
MessageHistory history = MessageHistory.read(resultMessage);
assertThat(history.get(0).contains("jmsChannel")).isTrue();
ac.close();
public void testFullConfig() {
try (ConfigurableApplicationContext ac =
new ClassPathXmlApplicationContext("JmsChannelHistoryTests-context.xml", this.getClass())) {
SubscribableChannel channel = ac.getBean("jmsChannel", SubscribableChannel.class);
PollableChannel resultChannel = ac.getBean("resultChannel", PollableChannel.class);
channel.send(new GenericMessage<>("hello"));
Message<?> resultMessage = resultChannel.receive(10000);
MessageHistory history = MessageHistory.read(resultMessage);
assertThat(history.get(0).contains("jmsChannel")).isTrue();
}
}
}

View File

@@ -1,13 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/jms https://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd">
<int-jms:outbound-channel-adapter id="channelAdapterChannel" destination-expression="'queue.test.dynamic.adapter.' + headers.destinationNumber"/>
@@ -27,15 +29,8 @@
<int:queue capacity="1"/>
</int:channel>
<bean id="jmsConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="false"/>
</bean>
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.connectionFactory"/>
<jms:listener-container connection-factory="jmsConnectionFactory">
<jms:listener destination="queue.test.dynamic.gateway.1" ref="responder" method="one"/>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -20,9 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import javax.jms.TextMessage;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.core.MessagingTemplate;
@@ -31,15 +29,15 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Mark Fisher
* @author Artem Bilan
*
* @since 2.1
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class JmsDynamicDestinationTests {
@@ -55,12 +53,6 @@ public class JmsDynamicDestinationTests {
@Autowired
private PollableChannel channelAdapterResults2;
@Before
public void prepareActiveMq() {
ActiveMqTestUtils.prepare();
}
@Test
public void channelAdapter() throws Exception {
Message<?> message1 = MessageBuilder.withPayload("test-1").setHeader("destinationNumber", 1).build();

View File

@@ -0,0 +1,39 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd">
<int:gateway id="sampleGateway"
service-interface="org.springframework.integration.jms.config.JmsMessageHistoryTests$SampleGateway"
default-request-channel="outbound-channel">
</int:gateway>
<int:channel id="outbound-channel"/>
<int-jms:outbound-channel-adapter id="jmsOutbound" channel="outbound-channel" destination-name="request.queue_c"/>
<!-- <int-jms:outbound-gateway id="jmsOutbound" request-channel="outbound-channel" request-destination-name="request.queue_c"/>-->
<int-jms:inbound-channel-adapter id="sampleJmsInboundAdapter" channel="jmsInputChannel"
destination-name="request.queue_c"/>
<!-- <int-jms:inbound-gateway id="sampleJmsInboundAdapter" request-channel="jmsInputChannel" request-destination-name="request.queue_c" reply-timeout="60000" request-timeout="60000"/>-->
<int:channel id="jmsInputChannel">
<int:queue capacity="2"/>
</int:channel>
<int:poller id="poller" default="true" fixed-delay="10"/>
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.connectionFactory"/>
<bean class="org.springframework.integration.history.MessageHistoryConfigurer"/>
</beans>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -25,11 +25,11 @@ import java.util.Map;
import java.util.Properties;
import java.util.StringTokenizer;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.jms.DefaultJmsHeaderMapper;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.messaging.Message;
@@ -37,23 +37,30 @@ import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Oleg Zhurakousky
* @author Mark Fisher
* @author Artem Bilan
*
* @since 2.0
*/
public class JmsMessageHistoryTests {
@SpringJUnitConfig
@DirtiesContext
public class JmsMessageHistoryTests extends ActiveMQMultiContextTests {
@Autowired
SampleGateway sampleGateway;
@Autowired
PollableChannel jmsInputChannel;
@Test
public void testInboundAdapter() throws Exception {
ActiveMqTestUtils.prepare();
ConfigurableApplicationContext applicationContext =
new ClassPathXmlApplicationContext("MessageHistoryTests-context.xml", JmsMessageHistoryTests.class);
SampleGateway gateway = applicationContext.getBean("sampleGateway", SampleGateway.class);
PollableChannel jmsInputChannel = applicationContext.getBean("jmsInputChannel", PollableChannel.class);
gateway.send("hello");
Message<?> message = jmsInputChannel.receive(5000);
public void testInboundAdapter() {
sampleGateway.send("hello");
Message<?> message = this.jmsInputChannel.receive(5000);
Iterator<Properties> historyIterator = message.getHeaders()
.get(MessageHistory.HEADER_NAME, MessageHistory.class)
.iterator();
@@ -63,7 +70,6 @@ public class JmsMessageHistoryTests {
Properties event2 = historyIterator.next();
assertThat(event2.getProperty(MessageHistory.TYPE_PROPERTY)).isEqualTo("channel");
assertThat(event2.getProperty(MessageHistory.NAME_PROPERTY)).isEqualTo("jmsInputChannel");
applicationContext.close();
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -17,7 +17,7 @@
package org.springframework.integration.jms.config;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -29,7 +29,7 @@ import javax.jms.Destination;
import javax.jms.Queue;
import javax.jms.Session;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.NotReadablePropertyException;
@@ -106,7 +106,7 @@ public class JmsOutboundGatewayParserTests {
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("advised");
JmsOutboundGateway gateway = TestUtils.getPropertyValue(endpoint, "handler", JmsOutboundGateway.class);
assertThat(TestUtils.getPropertyValue(gateway, "async", Boolean.class)).isFalse();
gateway.handleMessage(new GenericMessage<String>("foo"));
gateway.handleMessage(new GenericMessage<>("foo"));
assertThat(adviceCalled).isEqualTo(1);
assertThat(TestUtils.getPropertyValue(gateway, "replyContainer.sessionAcknowledgeMode")).isEqualTo(3);
context.close();
@@ -169,15 +169,17 @@ public class JmsOutboundGatewayParserTests {
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("jmsGatewayDestExpression");
DirectFieldAccessor accessor = new DirectFieldAccessor(endpoint);
JmsOutboundGateway gateway = (JmsOutboundGateway) accessor.getPropertyValue("handler");
ExpressionEvaluatingMessageProcessor<?> processor = TestUtils.getPropertyValue(gateway, "replyDestinationExpressionProcessor",
ExpressionEvaluatingMessageProcessor.class);
ExpressionEvaluatingMessageProcessor<?> processor =
TestUtils.getPropertyValue(gateway, "replyDestinationExpressionProcessor",
ExpressionEvaluatingMessageProcessor.class);
Expression expression = TestUtils.getPropertyValue(gateway, "replyDestinationExpressionProcessor.expression",
Expression.class);
assertThat(expression.getExpressionString()).isEqualTo("payload");
Message<?> message = MessageBuilder.withPayload("foo").build();
assertThat(processor.processMessage(message)).isEqualTo("foo");
Method method = JmsOutboundGateway.class.getDeclaredMethod("determineReplyDestination", Message.class, Session.class);
Method method =
JmsOutboundGateway.class.getDeclaredMethod("determineReplyDestination", Message.class, Session.class);
method.setAccessible(true);
Session session = mock(Session.class);
@@ -195,8 +197,9 @@ public class JmsOutboundGatewayParserTests {
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("jmsGatewayDestExpressionBeanRef");
DirectFieldAccessor accessor = new DirectFieldAccessor(endpoint);
JmsOutboundGateway gateway = (JmsOutboundGateway) accessor.getPropertyValue("handler");
ExpressionEvaluatingMessageProcessor<?> processor = TestUtils.getPropertyValue(gateway, "replyDestinationExpressionProcessor",
ExpressionEvaluatingMessageProcessor.class);
ExpressionEvaluatingMessageProcessor<?> processor =
TestUtils.getPropertyValue(gateway, "replyDestinationExpressionProcessor",
ExpressionEvaluatingMessageProcessor.class);
Expression expression = TestUtils.getPropertyValue(gateway, "replyDestinationExpressionProcessor.expression",
Expression.class);
assertThat(expression.getExpressionString()).isEqualTo("@replyQueue");
@@ -206,22 +209,16 @@ public class JmsOutboundGatewayParserTests {
@Test
public void gatewayWithDestAndDestExpression() {
try {
new ClassPathXmlApplicationContext(
"jmsOutboundGatewayReplyDestOptions-fail.xml", this.getClass()).close();
fail("Exception expected");
}
catch (BeanDefinitionParsingException e) {
assertThat(e.getMessage().startsWith("Configuration problem: Only one of the " +
"'replyQueue', 'reply-destination-name', or 'reply-destination-expression' attributes is allowed" +
"."))
.isTrue();
}
assertThatExceptionOfType(BeanDefinitionParsingException.class)
.isThrownBy(() ->
new ClassPathXmlApplicationContext("jmsOutboundGatewayReplyDestOptions-fail.xml", getClass()))
.withMessageStartingWith("Configuration problem: Only one of the " +
"'replyQueue', 'reply-destination-name', or 'reply-destination-expression' " +
"attributes is allowed.");
}
@Test
public void gatewayMaintainsReplyChannelAndInboundHistory() {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"gatewayMaintainsReplyChannel.xml", this.getClass());
SampleGateway gateway = context.getBean("gateway", SampleGateway.class);

View File

@@ -1,35 +1,34 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/jms https://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd">
<int-jms:outbound-channel-adapter id="channelAdapterChannel" destination-name="queue.test.priority.channelAdapter" priority="3" explicit-qos-enabled="true"/>
<int-jms:outbound-channel-adapter id="channelAdapterChannel" destination-name="queue.test.priority.channelAdapter"
priority="3" explicit-qos-enabled="true"/>
<int-jms:message-driven-channel-adapter channel="channelAdapterResults" destination-name="queue.test.priority.channelAdapter" extract-payload="false"/>
<int-jms:message-driven-channel-adapter channel="channelAdapterResults"
destination-name="queue.test.priority.channelAdapter"
extract-payload="false"/>
<int:channel id="gatewayChannel"/>
<int-jms:outbound-gateway request-channel="gatewayChannel" request-destination-name="queue.test.priority.gateway" priority="2" explicit-qos-enabled="true"/>
<int-jms:outbound-gateway request-channel="gatewayChannel" request-destination-name="queue.test.priority.gateway"
priority="2" explicit-qos-enabled="true"/>
<int:channel id="channelAdapterResults">
<int:queue capacity="2"/>
</int:channel>
<bean id="jmsConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="false"/>
</bean>
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.connectionFactory"/>
<jms:listener-container connection-factory="jmsConnectionFactory">
<jms:listener destination="queue.test.priority.gateway" ref="priorityReader"/>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -23,29 +23,28 @@ import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Mark Fisher
* @author Artem Bilan
*
* @since 2.0.2
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class JmsPriorityTests {
public class JmsPriorityTests extends ActiveMQMultiContextTests {
@Autowired
private MessageChannel channelAdapterChannel;
@@ -56,12 +55,6 @@ public class JmsPriorityTests {
@Autowired
private MessageChannel gatewayChannel;
@Before
public void prepareActiveMq() {
ActiveMqTestUtils.prepare();
}
@Test
public void verifyPrioritySettingOnChannelAdapterUsedAsJmsPriorityIfNoHeader() throws Exception {
Message<?> message = MessageBuilder.withPayload("test").build();
@@ -85,7 +78,7 @@ public class JmsPriorityTests {
}
@Test
public void verifyPrioritySettingOnGatewayUsedAsJmsPriorityIfNoHeader() throws Exception {
public void verifyPrioritySettingOnGatewayUsedAsJmsPriorityIfNoHeader() {
QueueChannel replyChannel = new QueueChannel();
Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build();
gatewayChannel.send(message);
@@ -95,7 +88,7 @@ public class JmsPriorityTests {
}
@Test
public void verifyPriorityHeaderUsedAsJmsPriorityWithGateway() throws Exception {
public void verifyPriorityHeaderUsedAsJmsPriorityWithGateway() {
QueueChannel replyChannel = new QueueChannel();
Message<?> message = MessageBuilder.withPayload("test").setPriority(8).setReplyChannel(replyChannel).build();
gatewayChannel.send(message);
@@ -114,6 +107,7 @@ public class JmsPriorityTests {
reply.setJMSCorrelationID(request.getJMSMessageID());
producer.send(reply);
}
}
}

View File

@@ -1,45 +1,48 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd">
<int:channel id="outbound-gateway-channel"/>
<int:channel id="output">
<int:queue/>
</int:channel>
<int-jms:outbound-gateway request-channel="outbound-gateway-channel"
<int-jms:outbound-gateway request-channel="outbound-gateway-channel"
reply-destination="replyQueue"
reply-channel="output"
request-destination="requestQueue"
message-converter="marshallingMessageConverter"/>
<int-jms:inbound-gateway request-destination="requestQueue"
request-channel="inbound-gateway-channel"
message-converter="marshallingMessageConverter"/>
<int:channel id="inbound-gateway-channel"/>
<int:service-activator input-channel="inbound-gateway-channel" method="echo">
<bean class="org.springframework.integration.jms.config.JmsWithMarshallingMessageConverterTests$SampleService"/>
</int:service-activator>
<bean id="marshallingMessageConverter" class="org.springframework.jms.support.converter.MarshallingMessageConverter">
<constructor-arg>
<bean class="org.springframework.integration.jms.config.JmsWithMarshallingMessageConverterTests$SampleMarshaller"/>
</constructor-arg>
<constructor-arg>
<bean class="org.springframework.integration.jms.config.JmsWithMarshallingMessageConverterTests$SampleUnmarshaller"/>
</constructor-arg>
</bean>
<bean id="requestQueue" class="org.apache.activemq.command.ActiveMQQueue">
request-destination="requestQueue"
message-converter="marshallingMessageConverter"/>
<int-jms:inbound-gateway request-destination="requestQueue"
request-channel="inbound-gateway-channel"
message-converter="marshallingMessageConverter"/>
<int:channel id="inbound-gateway-channel"/>
<int:service-activator input-channel="inbound-gateway-channel" method="echo">
<bean class="org.springframework.integration.jms.config.JmsWithMarshallingMessageConverterTests$SampleService"/>
</int:service-activator>
<bean id="marshallingMessageConverter"
class="org.springframework.jms.support.converter.MarshallingMessageConverter">
<constructor-arg>
<bean class="org.springframework.integration.jms.config.JmsWithMarshallingMessageConverterTests$SampleMarshaller"/>
</constructor-arg>
<constructor-arg>
<bean class="org.springframework.integration.jms.config.JmsWithMarshallingMessageConverterTests$SampleUnmarshaller"/>
</constructor-arg>
</bean>
<bean id="requestQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="request.queue"/>
</bean>
@@ -47,14 +50,7 @@
<constructor-arg value="reply.queue"/>
</bean>
<bean id="jmsConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="false"/>
</bean>
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.amqFactory"/>
</beans>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -26,10 +26,11 @@ import javax.xml.transform.Source;
import javax.xml.transform.stream.StreamResult;
import javax.xml.transform.stream.StreamSource;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
@@ -38,28 +39,33 @@ import org.springframework.messaging.support.GenericMessage;
import org.springframework.oxm.Marshaller;
import org.springframework.oxm.Unmarshaller;
import org.springframework.oxm.XmlMappingException;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Oleg Zhurakousky
* @author Gunnar Hillert
* @author Artem Bilan
*/
public class JmsWithMarshallingMessageConverterTests {
@SpringJUnitConfig
@DirtiesContext
public class JmsWithMarshallingMessageConverterTests extends ActiveMQMultiContextTests {
@Autowired
@Qualifier("outbound-gateway-channel")
MessageChannel input;
@Autowired
PollableChannel output;
@Test
@SuppressWarnings("unchecked")
public void demoWithMarshallingConverter() {
ActiveMqTestUtils.prepare();
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext(
"JmsWithMarshallingMessageConverterTests-context.xml", JmsWithMarshallingMessageConverterTests.class);
MessageChannel input = ac.getBean("outbound-gateway-channel", MessageChannel.class);
PollableChannel output = ac.getBean("output", PollableChannel.class);
input.send(new GenericMessage<String>("hello"));
Message<String> replyMessage = (Message<String>) output.receive();
this.input.send(new GenericMessage<>("hello"));
Message<?> replyMessage = this.output.receive();
MessageHeaders headers = replyMessage.getHeaders();
// check for couple of JMS headers, make sure they are present
assertThat(headers.get("jms_redelivered")).isNotNull();
assertThat(replyMessage.getPayload()).isEqualTo("HELLO");
ac.close();
}
@@ -68,13 +74,14 @@ public class JmsWithMarshallingMessageConverterTests {
public String echo(String value) {
return value.toUpperCase();
}
}
public static class SampleMarshaller implements Marshaller {
public void marshal(Object graph, Result result) throws IOException, XmlMappingException {
String payload = null;
String payload;
if (graph instanceof Message<?>) {
payload = (String) ((Message<?>) graph).getPayload();
}
@@ -101,7 +108,7 @@ public class JmsWithMarshallingMessageConverterTests {
InputStream io = ((StreamSource) source).getInputStream();
byte[] bytes = new byte[io.available()];
io.read(bytes);
return new GenericMessage<String>(new String(bytes));
return new GenericMessage<>(new String(bytes));
}
}

View File

@@ -1,47 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/jms https://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/task https://www.springframework.org/schema/task/spring-task.xsd"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:task="http://www.springframework.org/schema/task">
<int:gateway id="sampleGateway"
service-interface="org.springframework.integration.jms.config.JmsMessageHistoryTests$SampleGateway"
default-request-channel="outbound-channel">
</int:gateway>
<int:channel id="outbound-channel"/>
<int-jms:outbound-channel-adapter id="jmsOutbound" channel="outbound-channel" destination-name="request.queue_c"/>
<!-- <int-jms:outbound-gateway id="jmsOutbound" request-channel="outbound-channel" request-destination-name="request.queue_c"/>-->
<int-jms:inbound-channel-adapter id="sampleJmsInboundAdapter" channel="jmsInputChannel" destination-name="request.queue_c"/>
<!-- <int-jms:inbound-gateway id="sampleJmsInboundAdapter" request-channel="jmsInputChannel" request-destination-name="request.queue_c" reply-timeout="60000" request-timeout="60000"/>-->
<int:channel id="jmsInputChannel">
<int:queue capacity="2"/>
</int:channel>
<int:poller id="poller" default="true" fixed-delay="10"/>
<bean id="jmsConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="false"/>
</bean>
<bean class="org.springframework.integration.history.MessageHistoryConfigurer"/>
</beans>

View File

@@ -25,16 +25,12 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.ConnectionFactory;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.ListableBeanFactory;
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.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.annotation.InboundChannelAdapter;
@@ -60,7 +56,6 @@ import org.springframework.integration.jms.JmsDestinationPollingSource;
import org.springframework.integration.scheduling.PollerMetadata;
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;
@@ -72,7 +67,6 @@ import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.stereotype.Component;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.transaction.PlatformTransactionManager;
@@ -117,9 +111,6 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Autowired
private TestChannelInterceptor testChannelInterceptor;
@Autowired
private ConnectionFactory cachingConnectionFactory;
@Autowired
private PollableChannel jmsPubSubBridgeChannel;
@@ -246,7 +237,7 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Test
public void testPubSubFlow() {
JmsTemplate template = new JmsTemplate(this.cachingConnectionFactory);
JmsTemplate template = new JmsTemplate(connectionFactory);
template.setPubSubDomain(true);
template.setDefaultDestinationName("pubsub");
template.convertAndSend("foo");
@@ -284,25 +275,11 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Configuration
@EnableIntegration
@IntegrationComponentScan
@ComponentScan
public static class ContextConfiguration {
@Bean
public ActiveMQConnectionFactory jmsConnectionFactory() {
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
"vm://localhost?broker.persistent=false");
activeMQConnectionFactory.setTrustAllPackages(true);
return activeMQConnectionFactory;
}
@Bean
public ConnectionFactory cachingConnectionFactory() {
return new CachingConnectionFactory(jmsConnectionFactory());
}
@Bean
public JmsTemplate jmsTemplate() {
return new JmsTemplate(cachingConnectionFactory());
return new JmsTemplate(connectionFactory);
}
@Bean(name = PollerMetadata.DEFAULT_POLLER)
@@ -329,14 +306,14 @@ public class JmsTests extends ActiveMQMultiContextTests {
return f -> f
.fixedSubscriberChannel("integerChannel")
.transform("payload.toString()")
.channel(Jms.pollableChannel("flow1QueueChannel", jmsConnectionFactory())
.channel(Jms.pollableChannel("flow1QueueChannel", amqFactory)
.destination("flow1QueueChannel"));
}
@Bean
public IntegrationFlow jmsOutboundFlow() {
return f -> f
.handle(Jms.outboundAdapter(cachingConnectionFactory())
.handle(Jms.outboundAdapter(connectionFactory)
.destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)
.configureJmsTemplate(t -> t.id("jmsOutboundFlowTemplate")));
}
@@ -349,12 +326,19 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Bean
public IntegrationFlow jmsInboundFlow() {
return IntegrationFlows
.from(Jms.inboundAdapter(jmsConnectionFactory()).destination("jmsInbound"))
.from(Jms.inboundAdapter(amqFactory).destination("jmsInbound"))
.<String, String>transform(String::toUpperCase)
.channel(this.jmsOutboundInboundReplyChannel())
.get();
}
@Bean
public BroadcastCapableChannel jmsPublishSubscribeChannel() {
return Jms.publishSubscribeChannel(amqFactory)
.destination("pubsub")
.get();
}
@Bean
public IntegrationFlow pubSubFlow() {
return f -> f
@@ -366,17 +350,10 @@ public class JmsTests extends ActiveMQMultiContextTests {
.channel(c -> c.queue("jmsPubSubBridgeChannel2"))));
}
@Bean
public BroadcastCapableChannel jmsPublishSubscribeChannel() {
return Jms.publishSubscribeChannel(jmsConnectionFactory())
.destination("pubsub")
.get();
}
@Bean
public IntegrationFlow jmsMessageDrivenFlow() {
return IntegrationFlows
.from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory(),
.from(Jms.messageDrivenChannelAdapter(amqFactory,
DefaultMessageListenerContainer.class)
.outputChannel(jmsMessageDrivenInputChannel())
.destination("jmsMessageDriven")
@@ -410,7 +387,7 @@ public class JmsTests extends ActiveMQMultiContextTests {
public IntegrationFlow jmsMessageDrivenFlowWithContainer() {
return IntegrationFlows
.from(Jms.messageDrivenChannelAdapter(
Jms.container(jmsConnectionFactory(), "containerSpecDestination")
Jms.container(amqFactory, "containerSpecDestination")
.pubSubDomain(false)
.taskExecutor(Executors.newCachedThreadPool())
.get()))
@@ -421,7 +398,7 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Bean
public IntegrationFlow jmsOutboundGatewayFlow() {
return f -> f.handle(Jms.outboundGateway(cachingConnectionFactory())
return f -> f.handle(Jms.outboundGateway(connectionFactory)
.replyContainer(c -> c.idleReplyContainerTimeout(10))
.requestDestination("jmsPipelineTest"),
e -> e.id("jmsOutboundGateway"));
@@ -430,7 +407,7 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Bean
public IntegrationFlow jmsInboundGatewayFlow() {
return IntegrationFlows.from(
Jms.inboundGateway(jmsConnectionFactory())
Jms.inboundGateway(amqFactory)
.requestChannel(jmsInboundGatewayInputChannel())
.replyTimeout(1)
.errorOnTimeout(true)
@@ -475,7 +452,7 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Bean
public IntegrationFlow jmsMessageDrivenRedeliveryFlow() {
return IntegrationFlows
.from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory())
.from(Jms.messageDrivenChannelAdapter(amqFactory)
.errorChannel("errorChannelForRedelivery")
.destination("jmsMessageDrivenRedelivery")
.configureListenerContainer(c -> c
@@ -505,10 +482,14 @@ public class JmsTests extends ActiveMQMultiContextTests {
.get();
}
@Bean
@GlobalChannelInterceptor(patterns = "flow1QueueChannel")
ChannelInterceptor testChannelInterceptor() {
return new TestChannelInterceptor();
}
}
@Component
@GlobalChannelInterceptor(patterns = "flow1QueueChannel")
public static class TestChannelInterceptor implements ChannelInterceptor {
private final AtomicInteger invoked = new AtomicInteger();

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2021 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.
@@ -21,9 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import javax.jms.Message;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
@@ -33,16 +31,13 @@ import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.integration.jms.JmsTimeoutException;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.support.JmsHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
@@ -51,14 +46,10 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @since 4.3
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class AsyncGatewayTests extends ActiveMQMultiContextTests {
@Autowired
private CachingConnectionFactory ccf;
@Autowired
private JmsOutboundGateway gateway1;
@@ -73,11 +64,11 @@ public class AsyncGatewayTests extends ActiveMQMultiContextTests {
this.gateway1.handleMessage(MessageBuilder.withPayload("foo")
.setHeader(JmsHeaders.CORRELATION_ID, "baz")// make sure it's restored in case we're from an upstream gw
.build());
JmsTemplate template = new JmsTemplate(this.ccf);
JmsTemplate template = new JmsTemplate(connectionFactory);
template.setReceiveTimeout(10000);
final Message received = template.receive("asyncTest1");
assertThat(received).isNotNull();
template.send(received.getJMSReplyTo(), (MessageCreator) session -> {
template.send(received.getJMSReplyTo(), session -> {
TextMessage textMessage = session.createTextMessage("bar");
textMessage.setJMSCorrelationID(received.getJMSCorrelationID());
return textMessage;
@@ -90,12 +81,12 @@ public class AsyncGatewayTests extends ActiveMQMultiContextTests {
}
@Test
public void testWithTimeout() throws Exception {
public void testWithTimeout() {
QueueChannel errors = new QueueChannel();
this.gateway2.setOutputChannel(errors);
this.gateway2.start();
this.gateway2.handleMessage(MessageBuilder.withPayload("foo").setErrorChannel(errors).build());
JmsTemplate template = new JmsTemplate(this.ccf);
JmsTemplate template = new JmsTemplate(connectionFactory);
template.setReceiveTimeout(10000);
final Message received = template.receive("asyncTest3");
assertThat(received).isNotNull();
@@ -110,13 +101,13 @@ public class AsyncGatewayTests extends ActiveMQMultiContextTests {
@Test
@DirtiesContext
public void testWithTimeoutNoReplyRequired() throws Exception {
public void testWithTimeoutNoReplyRequired() {
QueueChannel errors = new QueueChannel();
this.gateway2.setOutputChannel(errors);
this.gateway2.setRequiresReply(false);
this.gateway2.start();
this.gateway2.handleMessage(MessageBuilder.withPayload("foo").setErrorChannel(errors).build());
JmsTemplate template = new JmsTemplate(this.ccf);
JmsTemplate template = new JmsTemplate(connectionFactory);
template.setReceiveTimeout(10000);
final Message received = template.receive("asyncTest3");
assertThat(received).isNotNull();
@@ -129,19 +120,11 @@ public class AsyncGatewayTests extends ActiveMQMultiContextTests {
@EnableIntegration
public static class Config {
@Bean
public CachingConnectionFactory ccf() {
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
cachingConnectionFactory.setCacheConsumers(false);
return cachingConnectionFactory;
}
@Bean
public JmsOutboundGateway gateway1() {
JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setUseReplyContainer(true);
gateway.setConnectionFactory(ccf());
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestinationName("asyncTest1");
gateway.setReplyDestinationName("asyncTest2");
gateway.setRequiresReply(true);
@@ -155,7 +138,7 @@ public class AsyncGatewayTests extends ActiveMQMultiContextTests {
public JmsOutboundGateway gateway2() {
JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setUseReplyContainer(true);
gateway.setConnectionFactory(ccf());
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestinationName("asyncTest3");
gateway.setReplyDestinationName("asyncTest4");
gateway.setRequiresReply(true);

View File

@@ -1,15 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd">
<bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
</bean>
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.amqFactory"/>
<int:channel id="input" />

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2019 the original author or authors.
* Copyright 2013-2021 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.
@@ -18,27 +18,27 @@ package org.springframework.integration.jms.request_reply;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 3.0
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class GatewaySerializedReplyChannelTests {
public class GatewaySerializedReplyChannelTests extends ActiveMQMultiContextTests {
@Autowired
MessageChannel input;
@@ -48,7 +48,7 @@ public class GatewaySerializedReplyChannelTests {
@Test
public void test() {
input.send(new GenericMessage<String>("foo"));
input.send(new GenericMessage<>("foo"));
Message<?> reply = output.receive(0);
assertThat(reply).isNotNull();
assertThat(reply.getPayload()).isEqualTo("echo:foo");

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -21,24 +21,21 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.condition.LongRunningTest;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.StopWatch;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
*/
@LongRunningTest
public class MiscellaneousTests {
@Rule
public LongRunningIntegrationTest longRunning = new LongRunningIntegrationTest();
/**
* Asserts that receive-timeout is honored even if
* requests (once in process), takes less then receive-timeout value
@@ -46,7 +43,6 @@ public class MiscellaneousTests {
*/
@Test
public void testTimeoutHonoringWhenRequestsQueuedUp() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("honor-timeout.xml", this.getClass());
final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
final CountDownLatch latch = new CountDownLatch(3);
@@ -64,10 +60,10 @@ public class MiscellaneousTests {
}
private void exchange(final CountDownLatch latch, final RequestReplyExchanger gateway, final AtomicInteger replies) {
private void exchange(CountDownLatch latch, RequestReplyExchanger gateway, AtomicInteger replies) {
new Thread(() -> {
try {
gateway.exchange(new GenericMessage<String>(""));
gateway.exchange(new GenericMessage<>(""));
replies.incrementAndGet();
}
catch (Exception e) {
@@ -76,4 +72,5 @@ public class MiscellaneousTests {
latch.countDown();
}).start();
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 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.
@@ -25,32 +25,30 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.condition.LongRunningTest;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Ali Moghadam
* @author Artem Bilan
*/
@LongRunningTest
public class PipelineJmsTests extends ActiveMQMultiContextTests {
private final ExecutorService executor = Executors.newFixedThreadPool(30);
private static final Log logger = LogFactory.getLog(PipelineJmsTests.class);
@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@After
@AfterEach
public void tearDown() {
this.executor.shutdownNow();
}
@@ -148,7 +146,6 @@ public class PipelineJmsTests extends ActiveMQMultiContextTests {
}
public void test(String contextConfig) throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(contextConfig, this.getClass());
final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
final CountDownLatch latch = new CountDownLatch(requests);
@@ -160,7 +157,7 @@ public class PipelineJmsTests extends ActiveMQMultiContextTests {
final int y = i;
executor.execute(() -> {
try {
assertThat(gateway.exchange(new GenericMessage<Integer>(y)).getPayload()).isEqualTo(y);
assertThat(gateway.exchange(new GenericMessage<>(y)).getPayload()).isEqualTo(y);
successCounter.incrementAndGet();
}
catch (MessageTimeoutException e) {
@@ -189,4 +186,5 @@ public class PipelineJmsTests extends ActiveMQMultiContextTests {
context.close();
}
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 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.
@@ -28,34 +28,31 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.condition.LongRunningTest;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
* @author Ali Moghadam
*/
@LongRunningTest
public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests {
private final ExecutorService executor = Executors.newFixedThreadPool(30);
private static final Log logger = LogFactory.getLog(PipelineJmsTests.class);
@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@After
@AfterEach
public void tearDown() {
this.executor.shutdownNow();
}
@@ -160,7 +157,6 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
}
public int test(String contextConfig, final int offset) throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(contextConfig, this.getClass());
final AtomicInteger successCounter = new AtomicInteger();
final AtomicInteger timeoutCounter = new AtomicInteger();
@@ -173,7 +169,7 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
final int y = i;
executor.execute(() -> {
try {
assertThat(gateway.exchange(new GenericMessage<Integer>(y)).getPayload()).isEqualTo(y + offset);
assertThat(gateway.exchange(new GenericMessage<>(y)).getPayload()).isEqualTo(y + offset);
successCounter.incrementAndGet();
}
catch (MessageTimeoutException e) {
@@ -210,7 +206,7 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
public static class Capture {
private final BlockingQueue<String> messages = new LinkedBlockingQueue<String>();
private final BlockingQueue<String> messages = new LinkedBlockingQueue<>();
public Message<?> capture(Message<?> message) {
messages.add("\n[" + Thread.currentThread().getName() + "] " + message);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -17,6 +17,7 @@
package org.springframework.integration.jms.request_reply;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import java.util.concurrent.CountDownLatch;
@@ -27,20 +28,17 @@ import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.TextMessage;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.condition.LongRunningTest;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.jms.support.converter.SimpleMessageConverter;
@@ -50,65 +48,53 @@ import org.springframework.messaging.support.GenericMessage;
* @author Oleg Zhurakousky
* @author Gary Russell
*/
@LongRunningTest
public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMultiContextTests {
private final SimpleMessageConverter converter = new SimpleMessageConverter();
@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@Test
public void messageCorrelationBasedOnRequestMessageIdOptimized() {
try (ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-cached-consumers.xml", getClass())) {
@Test(expected = MessageTimeoutException.class)
public void messageCorrelationBasedOnRequestMessageIdOptimized() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml",
this
.getClass());
try {
RequestReplyExchanger gateway = context
.getBean("standardMessageIdCopyingConsumerWithOptimization", RequestReplyExchanger.class);
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueueOptimizedA", Destination.class);
final Destination replyDestination = context.getBean("siInQueueOptimizedA", Destination.class);
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
jmsTemplate.send(replyDestination, session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
});
}).start();
gateway.exchange(new GenericMessage<String>("foo"));
}
finally {
context.close();
}
assertThatExceptionOfType(MessageTimeoutException.class)
.isThrownBy(() -> gateway.exchange(new GenericMessage<>("foo")));
}
}
@Test
public void messageCorrelationBasedOnRequestMessageIdNonOptimized() throws Exception {
ActiveMqTestUtils.prepare();
public void messageCorrelationBasedOnRequestMessageIdNonOptimized() {
try (ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-cached-consumers.xml", getClass())) {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml",
this
.getClass());
try {
RequestReplyExchanger gateway = context
.getBean("standardMessageIdCopyingConsumerWithoutOptimization", RequestReplyExchanger.class);
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueueNonOptimizedB", Destination.class);
final Destination replyDestination = context.getBean("siInQueueNonOptimizedB", Destination.class);
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
jmsTemplate.send(replyDestination, session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
@@ -116,21 +102,16 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway
.exchange(new GenericMessage<String>("foo"));
.exchange(new GenericMessage<>("foo"));
assertThat(siReplyMessage.getPayload()).isEqualTo("bar");
}
finally {
context.close();
}
}
@Test
public void messageCorrelationBasedOnRequestCorrelationIdOptimized() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml",
this
.getClass());
try {
public void messageCorrelationBasedOnRequestCorrelationIdOptimized() {
try (ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-cached-consumers.xml", getClass())) {
RequestReplyExchanger gateway = context
.getBean("correlationPropagatingConsumerWithOptimization", RequestReplyExchanger.class);
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
@@ -141,7 +122,7 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti
final Destination replyDestination = context.getBean("siInQueueOptimizedC", Destination.class);
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
jmsTemplate.send(replyDestination, session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
@@ -149,21 +130,16 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway
.exchange(new GenericMessage<String>("foo"));
.exchange(new GenericMessage<>("foo"));
assertThat(siReplyMessage.getPayload()).isEqualTo("bar");
}
finally {
context.close();
}
}
@Test(expected = MessageTimeoutException.class)
public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml",
this
.getClass());
try {
@Test
public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() {
try (ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-cached-consumers.xml", getClass())) {
RequestReplyExchanger gateway = context
.getBean("correlationPropagatingConsumerWithoutOptimization", RequestReplyExchanger.class);
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
@@ -174,26 +150,23 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti
final Destination replyDestination = context.getBean("siInQueueNonOptimizedD", Destination.class);
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
jmsTemplate.send(replyDestination, session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
});
}).start();
gateway.exchange(new GenericMessage<String>("foo"));
}
finally {
context.close();
assertThatExceptionOfType(MessageTimeoutException.class)
.isThrownBy(() -> gateway.exchange(new GenericMessage<>("foo")));
}
}
@Test
public void messageCorrelationBasedOnRequestCorrelationIdTimedOutFirstReplyOptimized() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
try {
try (ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-cached-consumers.xml", getClass())) {
RequestReplyExchanger gateway =
context.getBean("correlationPropagatingConsumerWithOptimizationDelayFirstReply",
RequestReplyExchanger.class);
@@ -244,9 +217,6 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti
Thread.sleep(1000);
assertThat(gateway.exchange(new GenericMessage<>("bar")).getPayload()).isEqualTo("bar");
}
finally {
context.close();
}
}
private Object extractPayload(Message jmsMessage) throws JMSException {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -20,34 +20,30 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.UUID;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.condition.LongRunningTest;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.support.JmsHeaders;
import org.springframework.messaging.Message;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*/
@LongRunningTest
public class RequestReplyScenariosWithCorrelationKeyProvidedTests extends ActiveMQMultiContextTests {
@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@Test
public void messageCorrelationBasedCustomCorrelationKey() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("explicitCorrelationKeyGateway", RequestReplyExchanger.class);
gateway.exchange(MessageBuilder.withPayload("foo").build());
@@ -56,9 +52,8 @@ public class RequestReplyScenariosWithCorrelationKeyProvidedTests extends Active
@Test
public void messageCorrelationBasedCustomCorrelationKeyAsJMSCorrelationID() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("explicitCorrelationKeyGatewayB", RequestReplyExchanger.class);
gateway.exchange(MessageBuilder.withPayload("foo").build());
@@ -67,9 +62,8 @@ public class RequestReplyScenariosWithCorrelationKeyProvidedTests extends Active
@Test
public void messageCorrelationBasedOnProvidedJMSCorrelationID() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("existingCorrelationKeyGatewayB", RequestReplyExchanger.class);
String correlationId = UUID.randomUUID().toString().replaceAll("'", "''");
@@ -82,9 +76,8 @@ public class RequestReplyScenariosWithCorrelationKeyProvidedTests extends Active
@Test
public void messageCorrelationBasedCustomCorrelationKeyDelayedReplies() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("explicitCorrelationKeyGatewayC", RequestReplyExchanger.class);
@@ -97,7 +90,8 @@ public class RequestReplyScenariosWithCorrelationKeyProvidedTests extends Active
}
}
JmsOutboundGateway outGateway = TestUtils.getPropertyValue(context.getBean("outGateway"), "handler", JmsOutboundGateway.class);
JmsOutboundGateway outGateway =
TestUtils.getPropertyValue(context.getBean("outGateway"), "handler", JmsOutboundGateway.class);
outGateway.setReceiveTimeout(5000);
assertThat(gateway.exchange(MessageBuilder.withPayload("foo").build()).getPayload()).isEqualTo("foo");
context.close();
@@ -105,9 +99,12 @@ public class RequestReplyScenariosWithCorrelationKeyProvidedTests extends Active
public static class DelayedService {
public String echo(String s) throws Exception {
Thread.sleep(200);
return s;
}
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -17,39 +17,37 @@
package org.springframework.integration.jms.request_reply;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.TextMessage;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.condition.LongRunningTest;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Arte Bilan
*/
@LongRunningTest
public class RequestReplyScenariosWithNonCachedConsumersTests extends ActiveMQMultiContextTests {
@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@Test
public void messageCorrelationBasedOnRequestMessageIdOptimized() {
try (AbstractApplicationContext context =
new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass())) {
@Test(expected = MessageTimeoutException.class)
public void messageCorrelationBasedOnRequestMessageIdOptimized() throws Exception {
ActiveMqTestUtils.prepare();
AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
try {
RequestReplyExchanger gateway = context.getBean("optimizedMessageId", RequestReplyExchanger.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
@@ -58,26 +56,24 @@ public class RequestReplyScenariosWithNonCachedConsumersTests extends ActiveMQMu
final Destination replyDestination = context.getBean("siInQueueC", Destination.class);
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
jmsTemplate.send(replyDestination, session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertThat(siReplyMessage.getPayload()).isEqualTo("bar");
}
finally {
context.close();
assertThatExceptionOfType(MessageTimeoutException.class)
.isThrownBy(() -> gateway.exchange(new GenericMessage<>("foo")));
}
}
@Test
public void messageCorrelationBasedOnRequestMessageIdNonOptimized() throws Exception {
ActiveMqTestUtils.prepare();
AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
try {
public void messageCorrelationBasedOnRequestMessageIdNonOptimized() {
try (AbstractApplicationContext context =
new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass())) {
RequestReplyExchanger gateway = context.getBean("nonoptimizedMessageId", RequestReplyExchanger.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
@@ -86,26 +82,23 @@ public class RequestReplyScenariosWithNonCachedConsumersTests extends ActiveMQMu
final Destination replyDestination = context.getBean("siInQueueD", Destination.class);
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
jmsTemplate.send(replyDestination, session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<>("foo"));
assertThat(siReplyMessage.getPayload()).isEqualTo("bar");
}
finally {
context.close();
}
}
@Test
public void messageCorrelationBasedOnRequestCorrelationIdOptimized() throws Exception {
ActiveMqTestUtils.prepare();
AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
try {
public void messageCorrelationBasedOnRequestCorrelationIdOptimized() {
try (AbstractApplicationContext context =
new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass())) {
RequestReplyExchanger gateway = context.getBean("optimized", RequestReplyExchanger.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
@@ -114,26 +107,23 @@ public class RequestReplyScenariosWithNonCachedConsumersTests extends ActiveMQMu
final Destination replyDestination = context.getBean("siInQueueA", Destination.class);
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
jmsTemplate.send(replyDestination, session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<>("foo"));
assertThat(siReplyMessage.getPayload()).isEqualTo("bar");
}
finally {
context.close();
}
}
@Test(expected = MessageTimeoutException.class)
public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() throws Exception {
ActiveMqTestUtils.prepare();
AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
try {
@Test
public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() {
try (AbstractApplicationContext context =
new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass())) {
RequestReplyExchanger gateway = context.getBean("nonoptimized", RequestReplyExchanger.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
@@ -142,17 +132,17 @@ public class RequestReplyScenariosWithNonCachedConsumersTests extends ActiveMQMu
final Destination replyDestination = context.getBean("siInQueueB", Destination.class);
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
jmsTemplate.send(replyDestination, session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
});
}).start();
gateway.exchange(new GenericMessage<String>("foo"));
}
finally {
context.close();
assertThatExceptionOfType(MessageTimeoutException.class)
.isThrownBy(() -> gateway.exchange(new GenericMessage<>("foo")));
}
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -37,14 +37,12 @@ import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.condition.LongRunningTest;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
@@ -58,109 +56,105 @@ import org.springframework.messaging.support.GenericMessage;
* @author Oleg Zhurakousky
* @author Gary Russell
*/
@LongRunningTest
public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMultiContextTests {
private final Log logger = LogFactory.getLog(getClass());
private final SimpleMessageConverter converter = new SimpleMessageConverter();
@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@SuppressWarnings("resource")
@Test
public void messageCorrelationBasedOnRequestMessageId() {
ActiveMqTestUtils.prepare();
try (ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass())) {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-temp-reply-consumers" +
".xml", this
.getClass());
RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueue", Destination.class);
final Destination requestDestination = context.getBean("siOutQueue", Destination.class);
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
Destination replyTo = null;
try {
replyTo = requestMessage.getJMSReplyTo();
}
catch (Exception ex) {
fail("Test failed", ex);
}
jmsTemplate.send(replyTo,
session -> {
try {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
catch (Exception e) {
// ignore
}
return null;
});
}).start();
gateway.exchange(new GenericMessage<>("foo"));
context.close();
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
Destination replyTo = null;
try {
replyTo = requestMessage.getJMSReplyTo();
}
catch (Exception ex) {
fail("Test failed", ex);
}
jmsTemplate.send(replyTo,
session -> {
try {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
catch (Exception e) {
// ignore
}
return null;
});
}).start();
gateway.exchange(new GenericMessage<>("foo"));
}
}
@Test
public void messageCorrelationBasedOnRequestCorrelationIdTimedOutFirstReply() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
try (ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass())) {
final Destination requestDestination = context.getBean("siOutQueue", Destination.class);
RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
dmlc.setConnectionFactory(connectionFactory);
dmlc.setDestination(requestDestination);
dmlc.setMessageListener((SessionAwareMessageListener<Message>) (message, session) -> {
Destination replyTo = null;
try {
replyTo = message.getJMSReplyTo();
}
catch (Exception e1) {
fail("Test failed", e1);
}
String requestPayload = (String) extractPayload(message);
if (requestPayload.equals("foo")) {
final Destination requestDestination = context.getBean("siOutQueue", Destination.class);
DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
dmlc.setConnectionFactory(connectionFactory);
dmlc.setDestination(requestDestination);
dmlc.setMessageListener((SessionAwareMessageListener<Message>) (message, session) -> {
Destination replyTo = null;
try {
Thread.sleep(6000);
replyTo = message.getJMSReplyTo();
}
catch (Exception e2) {
/*ignore*/
catch (Exception e1) {
fail("Test failed", e1);
}
}
try {
TextMessage replyMessage = session.createTextMessage();
replyMessage.setText(requestPayload);
replyMessage.setJMSCorrelationID(message.getJMSMessageID());
MessageProducer producer = session.createProducer(replyTo);
producer.send(replyMessage);
}
catch (Exception e3) {
// ignore. the test will fail
}
});
dmlc.afterPropertiesSet();
dmlc.start();
String requestPayload = (String) extractPayload(message);
if (requestPayload.equals("foo")) {
try {
Thread.sleep(6000);
}
catch (Exception e2) {
/*ignore*/
}
}
try {
TextMessage replyMessage = session.createTextMessage();
replyMessage.setText(requestPayload);
replyMessage.setJMSCorrelationID(message.getJMSMessageID());
MessageProducer producer = session.createProducer(replyTo);
producer.send(replyMessage);
}
catch (Exception e3) {
// ignore. the test will fail
}
});
dmlc.afterPropertiesSet();
dmlc.start();
try {
gateway.exchange(new GenericMessage<>("foo"));
try {
gateway.exchange(new GenericMessage<>("foo"));
}
catch (Exception e) {
// ignore
}
Thread.sleep(1000);
assertThat(gateway.exchange(new GenericMessage<>("bar")).getPayload()).isEqualTo("bar");
}
catch (Exception e) {
// ignore
}
Thread.sleep(1000);
assertThat(gateway.exchange(new GenericMessage<>("bar")).getPayload()).isEqualTo("bar");
context.close();
}
/**
@@ -169,11 +163,10 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
*/
@Test
public void brokenBrokerTest() throws Exception {
BrokerService broker = new BrokerService();
broker.setPersistent(false);
broker.setUseJmx(false);
broker.setTransportConnectorURIs(new String[] { "tcp://localhost:61623" });
broker.setTransportConnectorURIs(new String[]{ "tcp://localhost:61623" });
broker.setDeleteAllMessagesOnStartup(true);
broker.start();
@@ -203,55 +196,58 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
}
assertThat(replyCounter + timeoutCounter).isEqualTo(50);
context.close();
broker.stop();
}
@Test
public void testConcurrently() throws Exception {
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("mult-producer-and-consumers-temp-reply.xml", this.getClass());
final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
ExecutorService executor = Executors.newFixedThreadPool(10);
final int testNumbers = 30;
final CountDownLatch latch = new CountDownLatch(testNumbers);
final AtomicInteger failures = new AtomicInteger();
final AtomicInteger timeouts = new AtomicInteger();
final AtomicInteger missmatches = new AtomicInteger();
for (int i = 0; i < testNumbers; i++) {
final int y = i;
executor.execute(() -> {
try {
try (ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("mult-producer-and-consumers-temp-reply.xml", this.getClass())) {
String reply = (String) gateway.exchange(new GenericMessage<>(String.valueOf(y)))
.getPayload();
if (!String.valueOf(y).equals(reply)) {
missmatches.incrementAndGet();
final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
ExecutorService executor = Executors.newFixedThreadPool(10);
final int testNumbers = 30;
final CountDownLatch latch = new CountDownLatch(testNumbers);
final AtomicInteger failures = new AtomicInteger();
final AtomicInteger timeouts = new AtomicInteger();
final AtomicInteger mismatches = new AtomicInteger();
for (int i = 0; i < testNumbers; i++) {
final int y = i;
executor.execute(() -> {
try {
String reply = (String) gateway.exchange(new GenericMessage<>(String.valueOf(y)))
.getPayload();
if (!String.valueOf(y).equals(reply)) {
mismatches.incrementAndGet();
}
}
}
catch (Exception e) {
if (e instanceof MessageDeliveryException) {
timeouts.incrementAndGet();
catch (Exception e) {
if (e instanceof MessageDeliveryException) {
timeouts.incrementAndGet();
}
else {
failures.incrementAndGet();
}
}
else {
failures.incrementAndGet();
}
}
latch.countDown();
});
latch.countDown();
});
}
assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
print(failures, timeouts, mismatches, testNumbers);
assertThat(mismatches.get()).isEqualTo(0);
assertThat(failures.get()).isEqualTo(0);
assertThat(timeouts.get()).isEqualTo(0);
executor.shutdownNow();
}
assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
print(failures, timeouts, missmatches, testNumbers);
assertThat(missmatches.get()).isEqualTo(0);
assertThat(failures.get()).isEqualTo(0);
assertThat(timeouts.get()).isEqualTo(0);
context.close();
executor.shutdownNow();
}
private void print(AtomicInteger failures, AtomicInteger timeouts, AtomicInteger mismatches,
long echangesProcessed) {
long exchangesProcessed) {
logger.info("============================");
logger.info(echangesProcessed + " exchanges processed");
logger.info(exchangesProcessed + " exchanges processed");
logger.info("Failures: " + failures.get());
logger.info("Timeouts: " + timeouts.get());
logger.info("Missmatches: " + mismatches.get());
@@ -262,7 +258,7 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
Random random = new Random();
public String secho(String value) throws Exception {
public String echo(String value) throws Exception {
int i = random.nextInt(2000);
Thread.sleep(i);
return value;

View File

@@ -1,11 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/jms https://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd">
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd">
<int:channel id="test" />
@@ -26,16 +28,7 @@
<int:transformer input-channel="upcase" expression="payload.toUpperCase()" />
<bean id="jmsConnectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost?broker.persistent=false" />
</bean>
</property>
<property name="sessionCacheSize" value="10" />
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="false" />
</bean>
<util:constant id="jmsConnectionFactory"
static-field="org.springframework.integration.jms.ActiveMQMultiContextTests.amqFactory"/>
</beans>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2019 the original author or authors.
* Copyright 2015-2021 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.
@@ -18,27 +18,26 @@ package org.springframework.integration.jms.request_reply;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.Lifecycle;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @author Artem Bilan
* @since 4.2
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class StopStartTests {
public class StopStartTests extends ActiveMQMultiContextTests {
@Autowired
@Qualifier("out")

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-2021 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.
@@ -22,7 +22,6 @@ import assertk.assertions.contains
import assertk.assertions.isEqualTo
import assertk.assertions.isGreaterThan
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
@@ -32,6 +31,7 @@ import org.springframework.integration.IntegrationMessageHeaderAccessor
import org.springframework.integration.config.EnableIntegration
import org.springframework.integration.dsl.MessageChannels
import org.springframework.integration.dsl.integrationFlow
import org.springframework.integration.jms.ActiveMQMultiContextTests
import org.springframework.integration.jms.DefaultJmsHeaderMapper
import org.springframework.integration.support.MessageBuilder
import org.springframework.jms.support.JmsHeaders
@@ -50,7 +50,7 @@ import javax.jms.DeliveryMode
*/
@SpringJUnitConfig
@DirtiesContext
class JmsDslKotlinTests {
class JmsDslKotlinTests : ActiveMQMultiContextTests() {
@Autowired
@Qualifier("jmsOutboundFlow.input")
@@ -62,10 +62,12 @@ class JmsDslKotlinTests {
@Test
fun `test JMS Channel Adapters DSL`() {
this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload(" foo ")
this.jmsOutboundInboundChannel.send(
MessageBuilder.withPayload(" foo ")
.setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "containerSpecDestination")
.setPriority(9)
.build())
.build()
)
val receive = this.jmsOutboundInboundReplyChannel.receive(10000)
@@ -74,11 +76,11 @@ class JmsDslKotlinTests {
assertThat(payload).isNotNull().isEqualTo("foo")
assertThat(receive?.headers)
.isNotNull()
.all {
contains(IntegrationMessageHeaderAccessor.PRIORITY, 9)
contains(JmsHeaders.DELIVERY_MODE, 1)
}
.isNotNull()
.all {
contains(IntegrationMessageHeaderAccessor.PRIORITY, 9)
contains(JmsHeaders.DELIVERY_MODE, 1)
}
val expiration = receive!!.headers[JmsHeaders.EXPIRATION] as Long
assertThat(expiration).isGreaterThan(System.currentTimeMillis())
@@ -88,25 +90,18 @@ class JmsDslKotlinTests {
@EnableIntegration
class Config {
@Bean
fun jmsConnectionFactory(): ActiveMQConnectionFactory {
val activeMQConnectionFactory = ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")
activeMQConnectionFactory.isTrustAllPackages = true
return activeMQConnectionFactory
}
@Bean
fun jmsOutboundFlow() =
integrationFlow {
handle(Jms.outboundAdapter(jmsConnectionFactory())
.apply {
destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)
deliveryModeFunction<Any> { DeliveryMode.NON_PERSISTENT }
timeToLiveExpression("10000")
configureJmsTemplate { it.explicitQosEnabled(true) }
}
)
}
integrationFlow {
handle(Jms.outboundAdapter(connectionFactory)
.apply {
destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)
deliveryModeFunction<Any> { DeliveryMode.NON_PERSISTENT }
timeToLiveExpression("10000")
configureJmsTemplate { it.explicitQosEnabled(true) }
}
)
}
@Bean
fun jmsHeaderMapper(): DefaultJmsHeaderMapper {
@@ -121,17 +116,18 @@ class JmsDslKotlinTests {
@Bean
fun jmsMessageDrivenFlowWithContainer() =
integrationFlow(
Jms.messageDrivenChannelAdapter(
Jms.container(jmsConnectionFactory(), "containerSpecDestination")
.pubSubDomain(false)
.taskExecutor(Executors.newCachedThreadPool()))
.headerMapper(jmsHeaderMapper())) {
transform { it: String -> it.trim { it <= ' ' } }
channel(jmsOutboundInboundReplyChannel())
}
integrationFlow(
Jms.messageDrivenChannelAdapter(
Jms.container(amqFactory, "containerSpecDestination")
.pubSubDomain(false)
.taskExecutor(Executors.newCachedThreadPool())
)
.headerMapper(jmsHeaderMapper())
) {
transform { it: String -> it.trim { it <= ' ' } }
channel(jmsOutboundInboundReplyChannel())
}
}
}