diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index e249bb8195..3fa43c4902 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -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); } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java index 225003bc58..7ff1063d70 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java @@ -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("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("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(); + } + }