INT-3587: JMS OG - Lazy Reply Container Init.

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

Start the reply container on demand; stop after a timeout.

INT-3587: Polishing; PR Comments
This commit is contained in:
Gary Russell
2015-07-16 12:42:56 -04:00
committed by Artem Bilan
parent 9e0b2fb319
commit a8f47ee343
8 changed files with 241 additions and 26 deletions

View File

@@ -60,6 +60,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
@@ -144,6 +145,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private volatile boolean requiresReply;
private long lastSend;
private volatile long idleReplyContainerTimeout;
private ScheduledFuture<?> idleTask;
/**
* Set whether message delivery should be persistent or non-persistent,
* specified as a boolean value ("true" or "false"). This will set the delivery
@@ -425,6 +432,29 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
this.requiresReply = requiresReply;
}
/**
* Set the target timeout for idle containers, in seconds. Setting this greater than zero enables lazy
* starting of the reply listener container. The container will be started when a message is sent. It will be
* stopped when idle for at least this time. The actual stop time may be up to 1.5x this time.
* @param idleReplyContainerTimeout the timeout in seconds.
* @since 4.2
*/
public void setIdleReplyContainerTimeout(long idleReplyContainerTimeout) {
setIdleReplyContainerTimeout(idleReplyContainerTimeout, TimeUnit.SECONDS);
}
/**
* Set the target timeout for idle containers. Setting this greater than zero enables lazy
* starting of the reply listener container. The container will be started when a message is sent. It will be
* stopped when idle for at least this time. The actual stop time may be up to 1.5x this time.
* @param idleReplyContainerTimeout the timeout in seconds.
* @param unit the time unit.
* @since 4.2
*/
public void setIdleReplyContainerTimeout(long idleReplyContainerTimeout, TimeUnit unit) {
this.idleReplyContainerTimeout = unit.toMillis(idleReplyContainerTimeout);
}
private Destination determineRequestDestination(Message<?> message, Session session) throws JMSException {
if (this.requestDestination != null) {
return this.requestDestination;
@@ -620,9 +650,16 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
synchronized (this.lifeCycleMonitor) {
if (!this.active) {
if (this.replyContainer != null) {
this.replyContainer.start();
TaskScheduler taskScheduler = getTaskScheduler();
if (this.idleReplyContainerTimeout <= 0) {
this.replyContainer.start();
}
else {
Assert.state(taskScheduler != null, "'taskScheduler' is required.");
}
if (this.receiveTimeout >= 0) {
this.reaper = this.getTaskScheduler().schedule(new LateReplyReaper(), new Date());
Assert.state(taskScheduler != null, "'taskScheduler' is required.");
this.reaper = taskScheduler.schedule(new LateReplyReaper(), new Date());
}
}
this.active = true;
@@ -638,6 +675,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
this.deleteDestinationIfTemporary(this.replyContainer.getDestination());
this.reaper.cancel(false);
}
if (this.idleTask != null) {
this.idleTask.cancel(true);
this.idleTask = null;
}
this.active = false;
}
}
@@ -659,6 +700,19 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
jmsReply = this.sendAndReceiveWithoutContainer(requestMessage);
}
else {
if (this.idleReplyContainerTimeout > 0) {
synchronized(this.lifeCycleMonitor) {
this.lastSend = System.currentTimeMillis();
if (!this.replyContainer.isRunning()) {
if (logger.isDebugEnabled()) {
logger.debug(this.getComponentName() + ": Starting reply container.");
}
this.replyContainer.start();
this.idleTask = getTaskScheduler().scheduleAtFixedRate(new IdleContainerStopper(),
this.idleReplyContainerTimeout / 2);
}
}
}
jmsReply = this.sendAndReceiveWithContainer(requestMessage);
}
if (jmsReply == null) {
@@ -1275,7 +1329,29 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
}
private class IdleContainerStopper implements Runnable {
@Override
public void run() {
synchronized(JmsOutboundGateway.this.lifeCycleMonitor) {
if (System.currentTimeMillis() - lastSend > idleReplyContainerTimeout
&& replies.size() == 0) {
if (replyContainer.isRunning()) {
if (logger.isDebugEnabled()) {
logger.debug(getComponentName() + ": Stopping idle reply container.");
}
replyContainer.stop();
idleTask.cancel(false);
idleTask = null;
}
}
}
}
}
public static class ReplyContainerProperties {
private volatile Boolean sessionTransacted;
private volatile Integer sessionAcknowledgeMode;

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,8 @@
package org.springframework.integration.jms.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.ExpressionFactoryBean;
@@ -24,7 +26,6 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* Parser for the &lt;outbound-gateway&gt; element of the integration 'jms' namespace.
@@ -66,13 +67,15 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "priority");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "explicit-qos-enabled");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "idle-reply-listener-timeout",
"idleReplyContainerTimeout");
String deliveryPersistent = element.getAttribute("delivery-persistent");
if (StringUtils.hasText(deliveryPersistent)) {
builder.addPropertyValue("deliveryPersistent", deliveryPersistent);
}
Element container = DomUtils.getChildElementByTagName(element, "reply-listener");
if (container != null) {
this.parseReplyContainer(builder, parserContext, container);
}

View File

@@ -1108,6 +1108,15 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="idle-reply-listener-timeout" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
When using a 'reply-listener', specify whether the container should be started on-demand
and stopped when idle for this time (in seconds). When omitted (or &lt;= 0), the reply container
is started/stopped according to the gateway's lifecycle.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
</xsd:complexType>
</xsd:element>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 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.
@@ -15,6 +15,7 @@
*/
package org.springframework.integration.jms;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
@@ -34,12 +35,16 @@ import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.support.GenericMessage;
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;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -60,19 +65,15 @@ public class OutboundGatewayFunctionTests {
private static Destination requestQueue3 = new ActiveMQQueue("request3");
private static Destination replyQueue3 = new ActiveMQQueue("reply3");
private static Destination requestQueue4 = new ActiveMQQueue("request4");
private static Destination replyQueue4 = new ActiveMQQueue("reply4");
private static Destination requestQueue5 = new ActiveMQQueue("request5");
private static Destination replyQueue5 = new ActiveMQQueue("reply5");
private static Destination requestQueue6 = new ActiveMQQueue("request6");
private static Destination replyQueue6 = new ActiveMQQueue("reply6");
private static Destination requestQueue7 = new ActiveMQQueue("request7");
private static Destination replyQueue7 = new ActiveMQQueue("reply7");
@Test
public void testContainerWithDest() throws Exception {
@@ -95,6 +96,7 @@ public class OutboundGatewayFunctionTests {
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
@@ -114,6 +116,7 @@ public class OutboundGatewayFunctionTests {
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
@@ -144,6 +147,7 @@ public class OutboundGatewayFunctionTests {
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
@@ -163,6 +167,7 @@ public class OutboundGatewayFunctionTests {
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
@@ -195,6 +200,7 @@ public class OutboundGatewayFunctionTests {
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
@@ -214,6 +220,7 @@ public class OutboundGatewayFunctionTests {
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
@@ -244,6 +251,7 @@ public class OutboundGatewayFunctionTests {
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
@@ -263,6 +271,7 @@ public class OutboundGatewayFunctionTests {
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
@@ -294,6 +303,7 @@ public class OutboundGatewayFunctionTests {
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
@@ -313,6 +323,7 @@ public class OutboundGatewayFunctionTests {
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
@@ -342,6 +353,7 @@ public class OutboundGatewayFunctionTests {
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
@@ -361,6 +373,7 @@ public class OutboundGatewayFunctionTests {
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
@@ -372,6 +385,68 @@ public class OutboundGatewayFunctionTests {
gateway.stop();
}
@Test
public void testLazyContainerWithDest() 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);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getGatewayConnectionFactory());
gateway.setRequestDestination(requestQueue7);
gateway.setReplyDestination(replyQueue7);
gateway.setCorrelationKey("JMSCorrelationID");
gateway.setUseReplyContainer(true);
gateway.setIdleReplyContainerTimeout(1, TimeUnit.SECONDS);
gateway.afterPropertiesSet();
gateway.start();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getTemplateConnectionFactory());
template.setReceiveTimeout(10000);
receiveAndSend(template);
receiveAndSend(template);
}
private void receiveAndSend(JmsTemplate template) {
javax.jms.Message request = template.receive(requestQueue7);
final javax.jms.Message jmsReply = request;
try {
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
});
}
catch (JmsException e) {
}
catch (JMSException e) {
}
}
});
assertNotNull(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
DefaultMessageListenerContainer container = TestUtils.getPropertyValue(gateway, "replyContainer",
DefaultMessageListenerContainer.class);
int n = 0;
while (n++ < 100 && container.isRunning()) {
Thread.sleep(100);
}
assertFalse(container.isRunning());
assertNotNull(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
assertTrue(container.isRunning());
gateway.stop();
assertFalse(container.isRunning());
}
private ConnectionFactory getTemplateConnectionFactory() {
ConnectionFactory amqConnectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
return amqConnectionFactory;

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.
@@ -16,8 +16,17 @@
package org.springframework.integration.jms.config;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.lang.reflect.Method;
import java.util.Properties;
@@ -94,6 +103,8 @@ public class JmsOutboundGatewayParserTests {
assertEquals(3, TestUtils.getPropertyValue(container, "cacheLevel"));
assertTrue(container.isSessionTransacted());
assertSame(context.getBean("exec"), TestUtils.getPropertyValue(container, "taskExecutor"));
assertEquals(1234000L, TestUtils.getPropertyValue(gateway, "idleReplyContainerTimeout"));
context.close();
}
@Test
@@ -105,6 +116,7 @@ public class JmsOutboundGatewayParserTests {
gateway.handleMessage(new GenericMessage<String>("foo"));
assertEquals(1, adviceCalled);
assertEquals(3, TestUtils.getPropertyValue(gateway, "replyContainer.sessionAcknowledgeMode"));
context.close();
}
@Test
@@ -117,6 +129,7 @@ public class JmsOutboundGatewayParserTests {
accessor = new DirectFieldAccessor(gateway);
MessageConverter converter = (MessageConverter)accessor.getPropertyValue("messageConverter");
assertTrue("Wrong message converter", converter instanceof StubMessageConverter);
context.close();
}
@Test
@@ -129,6 +142,7 @@ public class JmsOutboundGatewayParserTests {
Object order = accessor.getPropertyValue("order");
assertEquals(99, order);
assertEquals(Boolean.TRUE, accessor.getPropertyValue("requiresReply"));
context.close();
}
@Test
@@ -140,6 +154,7 @@ public class JmsOutboundGatewayParserTests {
JmsOutboundGateway gateway = (JmsOutboundGateway) accessor.getPropertyValue("handler");
accessor = new DirectFieldAccessor(gateway);
assertSame(context.getBean("replyQueue"), accessor.getPropertyValue("replyDestination"));
context.close();
}
@Test
@@ -151,6 +166,7 @@ public class JmsOutboundGatewayParserTests {
JmsOutboundGateway gateway = (JmsOutboundGateway) accessor.getPropertyValue("handler");
accessor = new DirectFieldAccessor(gateway);
assertEquals("replyQueueName", accessor.getPropertyValue("replyDestinationName"));
context.close();
}
@Test
@@ -176,6 +192,7 @@ public class JmsOutboundGatewayParserTests {
when(session.createQueue("foo")).thenReturn(queue);
Destination replyQ = (Destination) method.invoke(gateway, message, session);
assertSame(queue, replyQ);
context.close();
}
@Test
@@ -191,13 +208,14 @@ public class JmsOutboundGatewayParserTests {
Expression.class);
assertEquals("@replyQueue", expression.getExpressionString());
assertSame(context.getBean("replyQueue"), processor.processMessage(null));
context.close();
}
@Test
public void gatewayWithDestAndDestExpression() {
try {
new ClassPathXmlApplicationContext(
"jmsOutboundGatewayReplyDestOptions-fail.xml", this.getClass());
"jmsOutboundGatewayReplyDestOptions-fail.xml", this.getClass()).close();
fail("Exception expected");
}
catch (BeanDefinitionParsingException e) {
@@ -214,15 +232,16 @@ public class JmsOutboundGatewayParserTests {
SampleGateway gateway = context.getBean("gateway", SampleGateway.class);
SubscribableChannel jmsInput = context.getBean("jmsInput", SubscribableChannel.class);
MessageHandler handler = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
MessageHistory history = MessageHistory.read(message);
assertNotNull(history);
Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "inboundGateway", 0);
assertNotNull(componentHistoryRecord);
assertEquals("jms:inbound-gateway", componentHistoryRecord.get("type"));
MessagingTemplate messagingTemplate = new MessagingTemplate();
messagingTemplate.setDefaultDestination((MessageChannel)message.getHeaders().getReplyChannel());
messagingTemplate.send(message);
MessagingTemplate messagingTemplate = new MessagingTemplate();
messagingTemplate.setDefaultDestination((MessageChannel)message.getHeaders().getReplyChannel());
messagingTemplate.send(message);
}
};
handler = spy(handler);
@@ -230,6 +249,7 @@ public class JmsOutboundGatewayParserTests {
String result = gateway.echo("hello");
verify(handler, times(1)).handleMessage(Mockito.any(Message.class));
assertEquals("hello", result);
context.close();
}
@Test
@@ -241,6 +261,7 @@ public class JmsOutboundGatewayParserTests {
new DirectFieldAccessor(endpoint).getPropertyValue("handler"));
assertFalse((Boolean) accessor.getPropertyValue("requestPubSubDomain"));
assertFalse((Boolean) accessor.getPropertyValue("replyPubSubDomain"));
context.close();
}
@Test
@@ -252,6 +273,7 @@ public class JmsOutboundGatewayParserTests {
new DirectFieldAccessor(endpoint).getPropertyValue("handler"));
assertTrue((Boolean) accessor.getPropertyValue("requestPubSubDomain"));
assertTrue((Boolean) accessor.getPropertyValue("replyPubSubDomain"));
context.close();
}

View File

@@ -30,6 +30,7 @@
request-destination-name="requestQueue"
request-channel="requestChannel"
delivery-persistent="true"
idle-reply-listener-timeout="1234"
auto-startup="false">
<jms:reply-listener
acknowledge="${jmsAcknowledgeModeTransacted}"