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

Fix JmsOutboundGatewayTests to always close CF

Browse Source 
Related to https://build.spring.io/browse/INT-MASTERSPRING40-1393

The connection factory must be closed independently of test outcome.
Otherwise the ActiveMQ resource is kept in the memory and the rest
of test in JMS module fails with misleading message like:
`javax.management.InstanceAlreadyExistsException: org.apache.activemq:type=Broker,brokerName=localhost`
This commit is contained in:
Artem Bilan
2021-03-25 10:08:04 -04:00
parent 67c5cf7458
commit 6a3ef1e647
2 changed files with 54 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-jms/src/test/java/org/springframework/integration/jms/ActiveMQMultiContextTests.java
View File

@@ -49,7 +49,7 @@ public abstract class ActiveMQMultiContextTests {
@AfterAll
public static void shutDown() {
connectionFactory.resetConnection();
connectionFactory.destroy();
}
}

98
spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java
View File

@@ -82,7 +82,6 @@ public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
".replyListener");
}
@SuppressWarnings("serial")
@Test
public void testReplyContainerRecovery() throws Exception {
JmsOutboundGateway gateway = new JmsOutboundGateway();
@@ -156,8 +155,8 @@ public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
@Test
public void testConnectionBreakOnReplyMessageIdCorrelation() {
CachingConnectionFactory connectionFactory1 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
CachingConnectionFactory connectionFactory1 =
new CachingConnectionFactory(new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
connectionFactory1.setCacheConsumers(false);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setConnectionFactory(connectionFactory1);
@@ -172,35 +171,40 @@ public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
gateway.afterPropertiesSet();
gateway.start();
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> gateway.handleMessage(new GenericMessage<>("foo")));
CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
CachingConnectionFactory connectionFactory2 =
new CachingConnectionFactory(new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
connectionFactory2.setCacheConsumers(false);
JmsTemplate template = new JmsTemplate(connectionFactory2);
template.setReceiveTimeout(10000);
template.afterPropertiesSet();
final Message request = template.receive(requestQ);
assertThat(request).isNotNull();
connectionFactory1.resetConnection();
MessageCreator reply = session -> {
TextMessage reply1 = session.createTextMessage("bar");
reply1.setJMSCorrelationID(request.getJMSMessageID());
return reply1;
};
template.send(replyQ, reply);
org.springframework.messaging.Message<?> received = queueChannel.receive(20000);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("bar");
gateway.stop();
connectionFactory1.destroy();
connectionFactory2.destroy();
exec.shutdownNow();
try {
exec.execute(() -> gateway.handleMessage(new GenericMessage<>("foo")));
final Message request = template.receive(requestQ);
assertThat(request).isNotNull();
connectionFactory1.resetConnection();
MessageCreator reply =
session -> {
TextMessage reply1 = session.createTextMessage("bar");
reply1.setJMSCorrelationID(request.getJMSMessageID());
return reply1;
};
template.send(replyQ, reply);
org.springframework.messaging.Message<?> received = queueChannel.receive(20000);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("bar");
}
finally {
gateway.stop();
connectionFactory1.destroy();
connectionFactory2.destroy();
exec.shutdownNow();
}
}
@Test
public void testConnectionBreakOnReplyCustomCorrelation() {
CachingConnectionFactory connectionFactory1 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
CachingConnectionFactory connectionFactory1 =
new CachingConnectionFactory(new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
connectionFactory1.setCacheConsumers(false);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setConnectionFactory(connectionFactory1);
@@ -216,31 +220,35 @@ public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
gateway.afterPropertiesSet();
gateway.start();
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> gateway.handleMessage(new GenericMessage<>("foo")));
CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
CachingConnectionFactory connectionFactory2 =
new CachingConnectionFactory(new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
connectionFactory2.setCacheConsumers(false);
JmsTemplate template = new JmsTemplate(connectionFactory2);
template.setReceiveTimeout(10000);
template.afterPropertiesSet();
final Message request = template.receive(requestQ);
assertThat(request).isNotNull();
connectionFactory1.resetConnection();
MessageCreator reply = session -> {
TextMessage reply1 = session.createTextMessage("bar");
reply1.setJMSCorrelationID(request.getJMSCorrelationID());
return reply1;
};
logger.debug("Sending reply to: " + replyQ);
template.send(replyQ, reply);
logger.debug("Sent reply to: " + replyQ);
org.springframework.messaging.Message<?> received = queueChannel.receive(20000);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("bar");
gateway.stop();
connectionFactory1.destroy();
connectionFactory2.destroy();
exec.shutdownNow();
try {
exec.execute(() -> gateway.handleMessage(new GenericMessage<>("foo")));
Message request = template.receive(requestQ);
assertThat(request).isNotNull();
connectionFactory1.resetConnection();
MessageCreator reply = session -> {
TextMessage reply1 = session.createTextMessage("bar");
reply1.setJMSCorrelationID(request.getJMSCorrelationID());
return reply1;
};
logger.debug("Sending reply to: " + replyQ);
template.send(replyQ, reply);
logger.debug("Sent reply to: " + replyQ);
org.springframework.messaging.Message<?> received = queueChannel.receive(20000);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("bar");
}
finally {
gateway.stop();
connectionFactory1.destroy();
connectionFactory2.destroy();
exec.shutdownNow();
}
}
}