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

INT-3566: JMS Recover Connection During Req/Resp

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3566

When using a named replyTo queue/topic if the connection was lost between the
request and reply, the connection was not recovered.

Retry consuming until the timeout expires.

INT-3566: Polishing; PR Comments

INT-3566: Remove Broker from Test Case
This commit is contained in:
Gary Russell
2015-07-15 17:16:16 -04:00
committed by Artem Bilan
parent 774e5868f0
commit 585fb0349c
2 changed files with 185 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -801,7 +801,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private javax.jms.Message doSendAndReceiveWithGeneratedCorrelationId(Destination requestDestination,
javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
MessageProducer messageProducer = null;
MessageConsumer messageConsumer = null;
try {
messageProducer = session.createProducer(requestDestination);
Assert.state(this.correlationKey != null, "correlationKey must not be null");
@@ -822,13 +821,11 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
messageSelector = "JMSCorrelationID = '" + jmsRequest.getJMSCorrelationID() + "'";
}
messageConsumer = session.createConsumer(replyTo, messageSelector);
this.sendRequestMessage(jmsRequest, messageProducer, priority);
return this.receiveReplyMessage(messageConsumer);
return retryableReceiveReply(session, replyTo, messageSelector);
}
finally {
JmsUtils.closeMessageProducer(messageProducer);
JmsUtils.closeMessageConsumer(messageConsumer);
}
}
@@ -864,17 +861,88 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
"will be created before the request Message is sent.");
}
MessageProducer messageProducer = null;
MessageConsumer messageConsumer = null;
try {
messageProducer = session.createProducer(requestDestination);
this.sendRequestMessage(jmsRequest, messageProducer, priority);
String messageId = jmsRequest.getJMSMessageID().replaceAll("'", "''");
String messageSelector = "JMSCorrelationID = '" + messageId + "'";
messageConsumer = session.createConsumer(replyTo, messageSelector);
return this.receiveReplyMessage(messageConsumer);
return retryableReceiveReply(session, replyTo, messageSelector);
}
finally {
JmsUtils.closeMessageProducer(messageProducer);
}
}
/*
* If the replyTo is not temporary, and the connection is lost while waiting for a reply, reconnect for
* up to receiveTimeout.
*/
private javax.jms.Message retryableReceiveReply(Session session, Destination replyTo, String messageSelector)
throws JMSException {
Connection consumerConnection = null;
Session consumerSession = session;
MessageConsumer messageConsumer = null;
JMSException exception = null;
boolean isTemporaryReplyTo = replyTo instanceof TemporaryQueue || replyTo instanceof TemporaryTopic;
long replyTimeout = isTemporaryReplyTo
? Long.MIN_VALUE
: this.receiveTimeout < 0
? Long.MAX_VALUE
: System.currentTimeMillis() + this.receiveTimeout;
try {
do {
try {
messageConsumer = consumerSession.createConsumer(replyTo, messageSelector);
javax.jms.Message reply = receiveReplyMessage(messageConsumer);
if (reply == null) {
if (replyTimeout > System.currentTimeMillis()) {
throw new JMSException("Consumer closed before timeout");
}
}
return reply;
}
catch (JMSException e) {
exception = e;
if (logger.isDebugEnabled()) {
logger.debug("Connection lost waiting for reply, retrying: " + e.getMessage());
}
do {
try {
consumerConnection = createConnection();
consumerSession = createSession(consumerConnection);
break;
}
catch (JMSException ee) {
exception = ee;
if (logger.isDebugEnabled()) {
logger.debug("Could not reconnect, retrying: " + ee.getMessage());
}
try {
Thread.sleep(1000);
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
return null;
}
}
}
while (replyTimeout > System.currentTimeMillis());
}
}
while (replyTimeout > System.currentTimeMillis());
if (isTemporaryReplyTo) {
return null;
}
else {
throw exception;
}
}
finally {
if (consumerSession != session) {
JmsUtils.closeSession(consumerSession);
JmsUtils.closeConnection(consumerConnection);
}
JmsUtils.closeMessageConsumer(messageConsumer);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 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.
@@ -16,6 +16,7 @@
package org.springframework.integration.jms;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyLong;
@@ -32,11 +33,14 @@ import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
@@ -45,10 +49,15 @@ import org.mockito.stubbing.Answer;
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.util.TestUtils;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
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.messaging.support.GenericMessage;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ErrorHandler;
import org.springframework.util.ObjectUtils;
@@ -154,4 +163,103 @@ public class JmsOutboundGatewayTests {
gateway.stop();
}
}
@Test
public void testConnectionBreakOnReplyMessageIdCorrelation() throws Exception {
CachingConnectionFactory connectionFactory1 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setConnectionFactory(connectionFactory1);
String requestQ = "requests1";
gateway.setRequestDestinationName(requestQ);
String replyQ = "replies1";
gateway.setReplyDestinationName(replyQ);
QueueChannel queueChannel = new QueueChannel();
gateway.setOutputChannel(queueChannel);
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.setReceiveTimeout(60000);
gateway.afterPropertiesSet();
gateway.start();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
gateway.handleMessage(new GenericMessage<String>("foo"));
}
});
CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
JmsTemplate template = new JmsTemplate(connectionFactory2);
template.setReceiveTimeout(5000);
template.afterPropertiesSet();
final Message request = template.receive(requestQ);
assertNotNull(request);
connectionFactory1.resetConnection();
MessageCreator reply = new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage reply = session.createTextMessage("bar");
reply.setJMSCorrelationID(request.getJMSMessageID());
return reply;
}
};
template.send(replyQ, reply);
org.springframework.messaging.Message<?> received = queueChannel.receive(10000);
assertNotNull(received);
assertEquals("bar", received.getPayload());
gateway.stop();
connectionFactory1.destroy();
connectionFactory2.destroy();
}
@Test
public void testConnectionBreakOnReplyCustomCorrelation() throws Exception {
CachingConnectionFactory connectionFactory1 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setConnectionFactory(connectionFactory1);
String requestQ = "requests2";
gateway.setRequestDestinationName(requestQ);
String replyQ = "replies2";
gateway.setReplyDestinationName(replyQ);
QueueChannel queueChannel = new QueueChannel();
gateway.setOutputChannel(queueChannel);
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.setReceiveTimeout(60000);
gateway.setCorrelationKey("JMSCorrelationID");
gateway.afterPropertiesSet();
gateway.start();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
gateway.handleMessage(new GenericMessage<String>("foo"));
}
});
CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
JmsTemplate template = new JmsTemplate(connectionFactory2);
template.setReceiveTimeout(5000);
template.afterPropertiesSet();
final Message request = template.receive(requestQ);
assertNotNull(request);
connectionFactory1.resetConnection();
MessageCreator reply = new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage reply = session.createTextMessage("bar");
reply.setJMSCorrelationID(request.getJMSCorrelationID());
return reply;
}
};
template.send(replyQ, reply);
org.springframework.messaging.Message<?> received = queueChannel.receive(10000);
assertNotNull(received);
assertEquals("bar", received.getPayload());
connectionFactory1.destroy();
connectionFactory2.destroy();
}
}